diff --git a/scripts/contract_identity.py b/scripts/contract_identity.py
new file mode 100644
index 0000000..e93f039
--- /dev/null
+++ b/scripts/contract_identity.py
@@ -0,0 +1,394 @@
+"""Pure canonical contract identity schema and digest helpers."""
+
+from __future__ import annotations
+
+import hashlib
+import hmac
+import json
+import re
+import unicodedata
+from collections import defaultdict
+from dataclasses import dataclass
+from typing import Any
+
+SCHEMA = "contract_identity.v2"
+CANONICALIZER_VERSION = "operator_tokens.v1"
+ALLOWED_CATEGORIES = frozenset(
+    {"bug", "contract", "logic", "performance", "reliability", "security"}
+)
+ALLOWED_SEVERITIES = frozenset({"critical", "high", "medium", "low"})
+ALLOWED_ANCHOR_KINDS = frozenset(
+    {"config", "endpoint", "field", "identifier", "schema", "symbol", "type"}
+)
+
+MAX_REPO_BYTES = 200
+MAX_OWNER_LENGTH = 39
+MAX_REPOSITORY_NAME_LENGTH = 100
+MAX_FILE_BYTES = 512
+MAX_ANCHOR_BYTES = 256
+MAX_TOKEN_BYTES = 256
+MAX_EVIDENCE_BYTES = 256
+MAX_ITEMS = 32
+MAX_TOKENS_PER_CLAUSE = 128
+MAX_CLAUSE_BYTES = MAX_TOKEN_BYTES * MAX_TOKENS_PER_CLAUSE
+MAX_CANONICAL_BYTES = 16_384
+
+_TOP_REQUIRED = frozenset(
+    {
+        "schema",
+        "canonicalizer_version",
+        "scope",
+        "anchors",
+        "predicates",
+        "required_behavior",
+        "forbidden_behavior",
+        "ordering_constraints",
+        "evidence",
+    }
+)
+_DIGEST_FIELDS = frozenset({"contract_key", "behavior_digest", "fingerprint_v2"})
+_OPERATORS = ("===", "!==", ">=", "<=", "==", "!=", "->", "=>", "::", ">", "<")
+_TOKEN_RE = re.compile(
+    r"<SECRET:[A-Z_]+:\d+>|===|!==|>=|<=|==|!=|->|=>|::|>|<|[\w.$]+|[^\s]",
+    re.UNICODE,
+)
+_PLACEHOLDER_RE = re.compile(r"<SECRET:[A-Z_]+:\d+>")
+_SAFE_CREDENTIAL_STATES = frozenset(
+    {"absent", "disabled", "enabled", "forbidden", "invalid", "missing", "none", "optional", "present", "redacted", "required", "valid"}
+)
+_SECRET_PATTERNS = (
+    ("PRIVATE_KEY", re.compile(r"-----BEGIN [^-]+-----.*?-----END [^-]+-----", re.I | re.S)),
+    ("AWS", re.compile(r"\b(?:AKIA|ASIA|AIDA|AROA)[A-Z0-9]{16}\b")),
+    ("SLACK", re.compile(r"\bxox[baprs]-[A-Za-z0-9-]{10,}\b")),
+    ("GITLAB", re.compile(r"\bglpat-[A-Za-z0-9_-]{10,}\b")),
+    ("JWT", re.compile(r"\beyJ[A-Za-z0-9_-]+\.[A-Za-z0-9_-]+\.[A-Za-z0-9_-]+\b")),
+    ("DSN", re.compile(r"\b[a-z][a-z0-9+.-]*://[^\s/@:]+:[^\s/@]+@[^\s]+", re.I)),
+    ("BEARER", re.compile(r"(?i)\bbearer\s+\S+")),
+    ("CREDENTIAL", re.compile(
+        r"(?i)(?<!<)\b(?P<key>token|secret|password|api[ _-]?key|authorization|cookie|aws[_ -]?secret[_ -]?access[_ -]?key)\b\s*[:=]\s*(?P<value>\S+)"
+    )),
+    ("API_KEY", re.compile(r"\b(?:gh[pousr]_|github_pat_|sk-)[A-Za-z0-9_-]{20,}\b")),
+)
+
+
+class IdentityValidationError(ValueError):
+    """Raised when canonical identity input cannot be safely validated."""
+
+
+@dataclass(frozen=True)
+class Scope:
+    repo: str
+    file: str
+    category: str
+
+    def as_dict(self) -> dict[str, str]:
+        return {"repo": self.repo, "file": self.file, "category": self.category}
+
+
+@dataclass(frozen=True)
+class Anchor:
+    kind: str
+    value: str
+
+    def as_dict(self) -> dict[str, str]:
+        return {"kind": self.kind, "value": self.value}
+
+
+@dataclass(frozen=True)
+class Evidence:
+    head_sha: str
+    diff_digest: str
+    file: str
+    location_or_hunk_digest: str
+
+    def as_dict(self) -> dict[str, str]:
+        return {
+            "head_sha": self.head_sha,
+            "diff_digest": self.diff_digest,
+            "file": self.file,
+            "location_or_hunk_digest": self.location_or_hunk_digest,
+        }
+
+
+@dataclass(frozen=True)
+class ContractIdentity:
+    scope: Scope
+    anchors: tuple[Anchor, ...]
+    predicates: tuple[tuple[str, ...], ...]
+    required_behavior: tuple[tuple[str, ...], ...]
+    forbidden_behavior: tuple[tuple[str, ...], ...]
+    ordering_constraints: tuple[tuple[str, ...], ...]
+    evidence: Evidence
+    severity: str | None
+    contract_key: str
+    behavior_digest: str
+    fingerprint_v2: str
+
+    def as_record(self) -> dict[str, Any]:
+        record: dict[str, Any] = {
+            "schema": SCHEMA,
+            "canonicalizer_version": CANONICALIZER_VERSION,
+            "scope": self.scope.as_dict(),
+            "anchors": [anchor.as_dict() for anchor in self.anchors],
+            "predicates": [list(clause) for clause in self.predicates],
+            "required_behavior": [list(clause) for clause in self.required_behavior],
+            "forbidden_behavior": [list(clause) for clause in self.forbidden_behavior],
+            "ordering_constraints": [list(clause) for clause in self.ordering_constraints],
+            "evidence": self.evidence.as_dict(),
+            "contract_key": self.contract_key,
+            "behavior_digest": self.behavior_digest,
+            "fingerprint_v2": self.fingerprint_v2,
+        }
+        if self.severity is not None:
+            record["severity"] = self.severity
+        return record
+
+
+class _SecretRedactor:
+    def __init__(self, *, allow_placeholders: bool) -> None:
+        self.allow_placeholders = allow_placeholders
+        self.counts: dict[str, int] = defaultdict(int)
+
+    @staticmethod
+    def _safe_credential(match: re.Match[str]) -> bool:
+        return match.groupdict().get("value", "").lower() in _SAFE_CREDENTIAL_STATES
+
+    def _has_unredacted_secret(self, value: str) -> bool:
+        for secret_type, pattern in _SECRET_PATTERNS:
+            for match in pattern.finditer(value):
+                if secret_type != "CREDENTIAL" or not self._safe_credential(match):
+                    return True
+        return False
+
+    def redact(self, value: str, field: str) -> str:
+        if "<SECRET:" in value and not self.allow_placeholders:
+            raise IdentityValidationError(f"{field} contains a reserved secret placeholder")
+        if self.allow_placeholders and self._has_unredacted_secret(value):
+            raise IdentityValidationError(f"{field} persists an unredacted secret")
+        text = value
+        for secret_type, pattern in _SECRET_PATTERNS:
+            def replacement(_match: re.Match[str], kind: str = secret_type) -> str:
+                if kind == "CREDENTIAL" and self._safe_credential(_match):
+                    return _match.group(0)
+                self.counts[kind] += 1
+                return f"<SECRET:{kind}:{self.counts[kind]}>"
+
+            text = pattern.sub(replacement, text)
+        remainder = _PLACEHOLDER_RE.sub("", text)
+        if "<SECRET:" in remainder:
+            raise IdentityValidationError(f"{field} contains an invalid secret placeholder")
+        return text
+
+
+def _exact_fields(value: Any, required: frozenset[str], optional: frozenset[str], field: str) -> dict[str, Any]:
+    if not isinstance(value, dict) or len(value) > len(required) + len(optional):
+        raise IdentityValidationError(f"{field} has missing or extra fields")
+    keys = set(value)
+    if not required.issubset(keys) or not keys.difference(required).issubset(optional):
+        raise IdentityValidationError(f"{field} has missing or extra fields")
+    return value
+
+
+def _text(value: Any, field: str, max_bytes: int, redactor: _SecretRedactor) -> str:
+    if not isinstance(value, str):
+        raise IdentityValidationError(f"{field} must be a string")
+    if len(value) > max_bytes:
+        raise IdentityValidationError(f"{field} exceeds {max_bytes} bytes")
+    raw_bytes = value.encode("utf-8")
+    if len(raw_bytes) > max_bytes:
+        raise IdentityValidationError(f"{field} exceeds {max_bytes} bytes")
+    normalized = unicodedata.normalize("NFC", value)
+    if not normalized or any(unicodedata.category(char).startswith("C") for char in normalized):
+        raise IdentityValidationError(f"{field} is empty or contains control characters")
+    if len(normalized.encode("utf-8")) > max_bytes:
+        raise IdentityValidationError(f"{field} exceeds {max_bytes} bytes")
+    normalized = redactor.redact(normalized, field)
+    if len(normalized.encode("utf-8")) > max_bytes:
+        raise IdentityValidationError(f"{field} exceeds {max_bytes} bytes")
+    return normalized
+
+
+def _repo(value: Any, field: str, redactor: _SecretRedactor) -> str:
+    repo = _text(value, field, MAX_REPO_BYTES, redactor)
+    parts = repo.split("/")
+    if len(parts) != 2:
+        raise IdentityValidationError(f"{field} must be owner/name")
+    owner, name = parts
+    owner_valid = re.fullmatch(r"(?!.*--)[A-Za-z0-9](?:[A-Za-z0-9-]{0,37}[A-Za-z0-9])?", owner)
+    name_valid = re.fullmatch(r"[A-Za-z0-9._-]+", name)
+    if (
+        not owner_valid
+        or not name_valid
+        or len(owner) > MAX_OWNER_LENGTH
+        or len(name) > MAX_REPOSITORY_NAME_LENGTH
+        or name in {".", ".."}
+    ):
+        raise IdentityValidationError(f"{field} must be owner/name")
+    return repo
+
+
+def _relative_path(value: Any, field: str, redactor: _SecretRedactor) -> str:
+    path = _text(value, field, MAX_FILE_BYTES, redactor)
+    parts = path.split("/")
+    if (
+        "<SECRET:" in path
+        or path.startswith("/")
+        or "\\" in path
+        or any(not part or part in {".", ".."} for part in parts)
+    ):
+        raise IdentityValidationError(f"{field} must be a repository-relative POSIX path")
+    return path
+
+
+def _reference(value: Any, field: str, redactor: _SecretRedactor) -> str:
+    reference = _text(value, field, MAX_EVIDENCE_BYTES, redactor)
+    if "<SECRET:" in reference:
+        raise IdentityValidationError(f"{field} must not contain a secret placeholder")
+    return reference
+
+
+def _enum(value: Any, field: str, allowed: frozenset[str]) -> str:
+    if not isinstance(value, str) or len(value) > 64:
+        raise IdentityValidationError(f"{field} is not a bounded string")
+    normalized = value.lower()
+    if normalized not in allowed:
+        raise IdentityValidationError(f"{field} is not controlled")
+    return normalized
+
+
+def _clauses(
+    value: Any, field: str, redactor: _SecretRedactor, *, allow_empty: bool
+) -> tuple[tuple[str, ...], ...]:
+    if not isinstance(value, list) or len(value) > MAX_ITEMS or (not value and not allow_empty):
+        raise IdentityValidationError(f"{field} must be a bounded list")
+    clauses: list[tuple[str, ...]] = []
+    for index, clause in enumerate(value):
+        if not isinstance(clause, list) or not clause or len(clause) > MAX_TOKENS_PER_CLAUSE:
+            raise IdentityValidationError(f"{field}[{index}] must be a non-empty token list")
+        if any(not isinstance(raw_token, str) or len(raw_token) > MAX_TOKEN_BYTES for raw_token in clause):
+            raise IdentityValidationError(f"{field}[{index}] contains an oversized token")
+        if sum(len(raw_token) for raw_token in clause) > MAX_CLAUSE_BYTES:
+            raise IdentityValidationError(f"{field}[{index}] exceeds aggregate size")
+        joined = _text(" ".join(clause), f"{field}[{index}]", MAX_CLAUSE_BYTES, redactor)
+        tokens = _TOKEN_RE.findall(joined)
+        if any(len(token.encode("utf-8")) > MAX_TOKEN_BYTES for token in tokens):
+            raise IdentityValidationError(f"{field}[{index}] contains an oversized token")
+        if not tokens or len(tokens) > MAX_TOKENS_PER_CLAUSE:
+            raise IdentityValidationError(f"{field}[{index}] has invalid token count")
+        clauses.append(tuple(tokens))
+    return tuple(clauses)
+
+
+def _stable_json(value: Any) -> bytes:
+    return json.dumps(value, ensure_ascii=False, sort_keys=True, separators=(",", ":")).encode("utf-8")
+
+
+def _digest(value: Any) -> str:
+    return hashlib.sha256(_stable_json(value)).hexdigest()
+
+
+def build_contract_identity(payload: dict[str, Any], *, _allow_placeholders: bool = False) -> ContractIdentity:
+    top = _exact_fields(payload, _TOP_REQUIRED, frozenset({"severity"}), "identity")
+    if top["schema"] != SCHEMA or top["canonicalizer_version"] != CANONICALIZER_VERSION:
+        raise IdentityValidationError("unsupported identity schema or canonicalizer")
+    redactor = _SecretRedactor(allow_placeholders=_allow_placeholders)
+
+    raw_scope = _exact_fields(top["scope"], frozenset({"repo", "file", "category"}), frozenset(), "scope")
+    category = _enum(raw_scope["category"], "scope.category", ALLOWED_CATEGORIES)
+    scope = Scope(
+        repo=_repo(raw_scope["repo"], "scope.repo", redactor),
+        file=_relative_path(raw_scope["file"], "scope.file", redactor),
+        category=category,
+    )
+
+    raw_anchors = top["anchors"]
+    if not isinstance(raw_anchors, list) or not raw_anchors or len(raw_anchors) > MAX_ITEMS:
+        raise IdentityValidationError("anchors must be a non-empty bounded list")
+    anchors: list[Anchor] = []
+    for index, raw_anchor in enumerate(raw_anchors):
+        anchor = _exact_fields(raw_anchor, frozenset({"kind", "value"}), frozenset(), f"anchors[{index}]")
+        kind = _enum(anchor["kind"], f"anchors[{index}].kind", ALLOWED_ANCHOR_KINDS)
+        anchors.append(Anchor(kind, _text(anchor["value"], f"anchors[{index}].value", MAX_ANCHOR_BYTES, redactor)))
+
+    predicates = _clauses(top["predicates"], "predicates", redactor, allow_empty=False)
+    required = _clauses(
+        top["required_behavior"], "required_behavior", redactor, allow_empty=False
+    )
+    forbidden = _clauses(
+        top["forbidden_behavior"], "forbidden_behavior", redactor, allow_empty=True
+    )
+    ordering = _clauses(
+        top["ordering_constraints"], "ordering_constraints", redactor, allow_empty=True
+    )
+
+    raw_evidence = _exact_fields(
+        top["evidence"],
+        frozenset({"head_sha", "diff_digest", "file", "location_or_hunk_digest"}),
+        frozenset(),
+        "evidence",
+    )
+    head_sha = _text(raw_evidence["head_sha"], "evidence.head_sha", 64, redactor).lower()
+    diff_digest = _text(raw_evidence["diff_digest"], "evidence.diff_digest", 64, redactor).lower()
+    if not re.fullmatch(r"[0-9a-f]{7,64}", head_sha) or not re.fullmatch(r"[0-9a-f]{64}", diff_digest):
+        raise IdentityValidationError("evidence digests are malformed")
+    evidence = Evidence(
+        head_sha=head_sha,
+        diff_digest=diff_digest,
+        file=_relative_path(raw_evidence["file"], "evidence.file", redactor),
+        location_or_hunk_digest=_reference(
+            raw_evidence["location_or_hunk_digest"],
+            "evidence.location_or_hunk_digest",
+            redactor,
+        ),
+    )
+    if evidence.file != scope.file:
+        raise IdentityValidationError("evidence.file must match scope.file")
+    severity = top.get("severity")
+    if severity is not None:
+        severity = _enum(severity, "severity", ALLOWED_SEVERITIES)
+
+    contract_payload = {
+        "schema": SCHEMA,
+        "canonicalizer_version": CANONICALIZER_VERSION,
+        "scope": scope.as_dict(),
+        "anchors": [anchor.as_dict() for anchor in anchors],
+        "predicates": [list(clause) for clause in predicates],
+    }
+    contract_key = _digest(contract_payload)
+    behavior_payload = {
+        "contract_key": contract_key,
+        "required_behavior": [list(clause) for clause in required],
+        "forbidden_behavior": [list(clause) for clause in forbidden],
+        "ordering_constraints": [list(clause) for clause in ordering],
+    }
+    behavior_digest = _digest(behavior_payload)
+    fingerprint_v2 = _digest(
+        {"contract_key": contract_key, "behavior_digest": behavior_digest}
+    )
+    identity = ContractIdentity(
+        scope, tuple(anchors), predicates, required, forbidden, ordering,
+        evidence, severity, contract_key, behavior_digest, fingerprint_v2,
+    )
+    if len(_stable_json(identity.as_record())) > MAX_CANONICAL_BYTES:
+        raise IdentityValidationError("canonical identity exceeds total byte limit")
+    return identity
+
+
+def verify_persisted_identity(record: dict[str, Any]) -> ContractIdentity:
+    top = _exact_fields(record, _TOP_REQUIRED | _DIGEST_FIELDS, frozenset({"severity"}), "record")
+    expected = {field: top[field] for field in _DIGEST_FIELDS}
+    if any(not isinstance(value, str) or not re.fullmatch(r"[0-9a-f]{64}", value) for value in expected.values()):
+        raise IdentityValidationError("persisted identity digest is malformed")
+    payload = {key: value for key, value in top.items() if key not in _DIGEST_FIELDS}
+    identity = build_contract_identity(payload, _allow_placeholders=True)
+    if not all(hmac.compare_digest(expected[field], getattr(identity, field)) for field in _DIGEST_FIELDS):
+        raise IdentityValidationError("persisted identity digest mismatch")
+    return identity
+
+
+def canonical_json(identity: ContractIdentity) -> str:
+    return _stable_json(identity.as_record()).decode("utf-8")
+
+
+def operators() -> tuple[str, ...]:
+    return _OPERATORS
diff --git a/tests/test_contract_identity.py b/tests/test_contract_identity.py
new file mode 100644
index 0000000..b37700c
--- /dev/null
+++ b/tests/test_contract_identity.py
@@ -0,0 +1,294 @@
+from __future__ import annotations
+
+import copy
+import json
+import unittest
+from unittest.mock import patch
+
+from scripts.contract_identity import (
+    MAX_ANCHOR_BYTES,
+    MAX_CLAUSE_BYTES,
+    MAX_ITEMS,
+    MAX_TOKEN_BYTES,
+    MAX_TOKENS_PER_CLAUSE,
+    IdentityValidationError,
+    _SecretRedactor,
+    _text,
+    build_contract_identity,
+    canonical_json,
+    operators,
+    verify_persisted_identity,
+)
+
+
+class ContractIdentityTests(unittest.TestCase):
+    def payload(self) -> dict[str, object]:
+        return {
+            "schema": "contract_identity.v2",
+            "canonicalizer_version": "operator_tokens.v1",
+            "scope": {
+                "repo": "org/audit-bridge",
+                "file": "service/review.py",
+                "category": "contract",
+            },
+            "anchors": [{"kind": "symbol", "value": "Review.validate()"}],
+            "predicates": [["score>=threshold"]],
+            "required_behavior": [["return", "blocked"]],
+            "forbidden_behavior": [],
+            "ordering_constraints": [["validate", "before", "dispatch"]],
+            "evidence": {
+                "head_sha": "deadbeef",
+                "diff_digest": "a" * 64,
+                "file": "service/review.py",
+                "location_or_hunk_digest": "hunk:review:1",
+            },
+            "severity": "high",
+        }
+
+    def test_operator_matrix_is_atomic_and_collision_resistant(self) -> None:
+        fingerprints = set()
+        for operator in operators():
+            payload = self.payload()
+            payload["predicates"] = [[f"left{operator}right"]]
+            identity = build_contract_identity(payload)
+            self.assertIn(operator, identity.predicates[0])
+            fingerprints.add(identity.fingerprint_v2)
+        self.assertEqual(len(fingerprints), len(operators()))
+
+    def test_anchor_and_predicate_matrix_changes_contract_key(self) -> None:
+        schema = self.payload()
+        schema["anchors"] = [{"kind": "schema", "value": "schema_v2"}]
+        fingerprint = copy.deepcopy(schema)
+        fingerprint["anchors"] = [{"kind": "schema", "value": "fingerprint_v2"}]
+        self.assertNotEqual(
+            build_contract_identity(schema).contract_key,
+            build_contract_identity(fingerprint).contract_key,
+        )
+
+        auth = self.payload()
+        auth["predicates"] = [["validate()", "checks", "auth_header"]]
+        database = copy.deepcopy(auth)
+        database["predicates"] = [["validate()", "prevents", "database_leak"]]
+        self.assertNotEqual(
+            build_contract_identity(auth).contract_key,
+            build_contract_identity(database).contract_key,
+        )
+
+        unrelated = self.payload()
+        unrelated["anchors"] = [{"kind": "symbol", "value": "Audit.redact()"}]
+        self.assertNotEqual(
+            build_contract_identity(auth).contract_key,
+            build_contract_identity(unrelated).contract_key,
+        )
+
+    def test_severity_is_excluded_and_opposite_behavior_is_distinct(self) -> None:
+        high = build_contract_identity(self.payload())
+        critical_payload = self.payload()
+        critical_payload["severity"] = "critical"
+        critical = build_contract_identity(critical_payload)
+        self.assertEqual(high.contract_key, critical.contract_key)
+        self.assertEqual(high.behavior_digest, critical.behavior_digest)
+        self.assertEqual(high.fingerprint_v2, critical.fingerprint_v2)
+
+        opposite_payload = self.payload()
+        opposite_payload["required_behavior"] = [["return", "success"]]
+        opposite = build_contract_identity(opposite_payload)
+        self.assertEqual(high.contract_key, opposite.contract_key)
+        self.assertNotEqual(high.behavior_digest, opposite.behavior_digest)
+
+    def test_limits_reject_instead_of_truncating_and_keep_long_tails(self) -> None:
+        exact = self.payload()
+        exact["anchors"] = [{"kind": "identifier", "value": "a" * MAX_ANCHOR_BYTES}]
+        self.assertEqual(
+            build_contract_identity(exact).anchors[0].value,
+            "a" * MAX_ANCHOR_BYTES,
+        )
+        oversize = copy.deepcopy(exact)
+        oversize["anchors"][0]["value"] += "a"
+        with self.assertRaises(IdentityValidationError):
+            build_contract_identity(oversize)
+
+        first = self.payload()
+        first["predicates"] = [["p" * 240, "tail_alpha"]]
+        second = copy.deepcopy(first)
+        second["predicates"] = [["p" * 240, "tail_beta"]]
+        self.assertNotEqual(
+            build_contract_identity(first).contract_key,
+            build_contract_identity(second).contract_key,
+        )
+        first["predicates"] = [["p" * (MAX_TOKEN_BYTES + 1)]]
+        with self.assertRaises(IdentityValidationError):
+            build_contract_identity(first)
+
+    def test_secret_literals_become_typed_placeholders_before_hashing(self) -> None:
+        payload = self.payload()
+        payload["required_behavior"] = [["token=secret-value"]]
+        identity = build_contract_identity(payload)
+        record = identity.as_record()
+        serialized = canonical_json(identity)
+        self.assertIn("<SECRET:CREDENTIAL:1>", serialized)
+        self.assertNotIn("secret-value", serialized)
+        self.assertEqual(verify_persisted_identity(record), identity)
+
+        payload["required_behavior"] = [["<SECRET:CREDENTIAL:1>"]]
+        with self.assertRaises(IdentityValidationError):
+            build_contract_identity(payload)
+        record["required_behavior"] = [["token=secret-value"]]
+        with self.assertRaises(IdentityValidationError):
+            verify_persisted_identity(record)
+        for malformed in ("<SECRET:CREDENTIAL:1> <SECRET:BAD>", "<SECRET:CREDENTIAL:1> <SECRET:"):
+            record = identity.as_record()
+            record["required_behavior"] = [[malformed]]
+            with self.assertRaises(IdentityValidationError):
+                verify_persisted_identity(record)
+
+        oversized_secret = self.payload()
+        oversized_secret["required_behavior"] = [["x" * MAX_TOKEN_BYTES + "AKIAIOSFODNN7EXAMPLE"]]
+        with self.assertRaises(IdentityValidationError):
+            build_contract_identity(oversized_secret)
+
+    def test_credential_states_are_not_collapsed_and_common_literals_are_redacted(self) -> None:
+        required = self.payload()
+        required["required_behavior"] = [["authorization=required"]]
+        forbidden = copy.deepcopy(required)
+        forbidden["required_behavior"] = [["authorization=forbidden"]]
+        self.assertEqual(
+            verify_persisted_identity(build_contract_identity(required).as_record()),
+            build_contract_identity(required),
+        )
+        self.assertEqual(
+            verify_persisted_identity(build_contract_identity(forbidden).as_record()),
+            build_contract_identity(forbidden),
+        )
+        self.assertNotEqual(
+            build_contract_identity(required).behavior_digest,
+            build_contract_identity(forbidden).behavior_digest,
+        )
+        secrets = self.payload()
+        secrets["required_behavior"] = [["github_pat_" + "x" * 30], ["aws_secret_access_key=secret-value"]]
+        identity = build_contract_identity(secrets)
+        serialized = canonical_json(identity)
+        self.assertNotIn("github_pat_", serialized)
+        self.assertNotIn("secret-value", serialized)
+
+        for state in ("required", "forbidden"):
+            split = self.payload()
+            split["required_behavior"] = [["authorization", "=", state]]
+            split_identity = build_contract_identity(split)
+            self.assertEqual(verify_persisted_identity(split_identity.as_record()), split_identity)
+        for clauses in (
+            [["token", "=", "supersecret"]],
+            [["github_pat_" + "x" * 30], ["aws_secret_access_key", "=", "secret-value"]],
+        ):
+            split = self.payload()
+            split["required_behavior"] = clauses
+            serialized = canonical_json(build_contract_identity(split))
+            self.assertNotIn("supersecret", serialized)
+            self.assertNotIn("github_pat_", serialized)
+            self.assertNotIn("secret-value", serialized)
+
+    def test_schema_fields_category_unicode_and_controls_are_strict(self) -> None:
+        for mutate in (
+            lambda value: value.pop("schema"),
+            lambda value: value.update({"description": "raw prose"}),
+            lambda value: value["scope"].update({"category": "unknown"}),
+        ):
+            payload = self.payload()
+            mutate(payload)
+            with self.assertRaises(IdentityValidationError):
+                build_contract_identity(payload)
+
+        payload = self.payload()
+        payload["anchors"] = [{"kind": "identifier", "value": "Cafe\u0301"}]
+        self.assertEqual(build_contract_identity(payload).anchors[0].value, "Café")
+        payload["anchors"] = [{"kind": "identifier", "value": "bad\nanchor"}]
+        with self.assertRaises(IdentityValidationError):
+            build_contract_identity(payload)
+
+    def test_repo_and_relative_path_boundaries(self) -> None:
+        for repo in ("QuantStrategyLab/AIAuditBridge", "owner/repo.name"):
+            payload = self.payload()
+            payload["scope"]["repo"] = repo
+            self.assertEqual(build_contract_identity(payload).scope.repo, repo)
+        for repo in (
+            "owner", "/owner/repo", "owner/", "owner//repo", "owner/repo/extra",
+            "owner_name/repo", "-owner/repo", "owner-/repo", "owner--name/repo", "./repo", "owner/.", "owner/..",
+        ):
+            payload = self.payload()
+            payload["scope"]["repo"] = repo
+            with self.assertRaises(IdentityValidationError):
+                build_contract_identity(payload)
+
+        payload = self.payload()
+        payload["scope"]["file"] = "src/nested/review.py"
+        payload["evidence"]["file"] = "src/nested/review.py"
+        self.assertEqual(build_contract_identity(payload).scope.file, "src/nested/review.py")
+        for path in ("/abs/review.py", "../escape.py", "a//b.py", "a\\b.py", "./review.py"):
+            invalid = self.payload()
+            invalid["scope"]["file"] = path
+            invalid["evidence"]["file"] = path
+            with self.assertRaises(IdentityValidationError):
+                build_contract_identity(invalid)
+
+        mismatch = self.payload()
+        mismatch["evidence"]["file"] = "service/other.py"
+        with self.assertRaises(IdentityValidationError):
+            build_contract_identity(mismatch)
+        secret_scope = self.payload()
+        secret_scope["scope"]["repo"] = "owner/token=secret-value"
+        with self.assertRaises(IdentityValidationError):
+            build_contract_identity(secret_scope)
+        secret_ref = self.payload()
+        secret_ref["evidence"]["location_or_hunk_digest"] = "token=secret-value"
+        with self.assertRaises(IdentityValidationError):
+            build_contract_identity(secret_ref)
+
+    def test_untrusted_shape_and_normalize_bounds_fail_closed(self) -> None:
+        anchors = self.payload()
+        anchors["anchors"] = anchors["anchors"] * (MAX_ITEMS + 1)
+        with self.assertRaises(IdentityValidationError):
+            build_contract_identity(anchors)
+        clauses = self.payload()
+        clauses["predicates"] = [["x"] * (MAX_TOKENS_PER_CLAUSE + 1)]
+        with self.assertRaises(IdentityValidationError):
+            build_contract_identity(clauses)
+        aggregate = self.payload()
+        aggregate["predicates"] = [["x" * MAX_TOKEN_BYTES] * MAX_TOKENS_PER_CLAUSE]
+        self.assertGreater(sum(map(len, aggregate["predicates"][0])), MAX_CLAUSE_BYTES - 1)
+        with self.assertRaises(IdentityValidationError):
+            build_contract_identity(aggregate)
+        oversized_dict = self.payload()
+        oversized_dict.update({f"extra_{index}": index for index in range(1000)})
+        with self.assertRaises(IdentityValidationError):
+            build_contract_identity(oversized_dict)
+        with patch("scripts.contract_identity.unicodedata.normalize") as normalize:
+            with self.assertRaises(IdentityValidationError):
+                _text("x" * 2_000_000, "oversized", MAX_TOKEN_BYTES, _SecretRedactor(allow_placeholders=False))
+            normalize.assert_not_called()
+
+    def test_digest_tamper_order_and_evidence_binding_are_explicit(self) -> None:
+        identity = build_contract_identity(self.payload())
+        tampered = identity.as_record()
+        tampered["behavior_digest"] = "0" * 64
+        with self.assertRaises(IdentityValidationError):
+            verify_persisted_identity(tampered)
+
+        reversed_payload = self.payload()
+        reversed_payload["ordering_constraints"] = [["dispatch", "before", "validate"]]
+        reversed_identity = build_contract_identity(reversed_payload)
+        self.assertEqual(identity.contract_key, reversed_identity.contract_key)
+        self.assertNotEqual(identity.behavior_digest, reversed_identity.behavior_digest)
+
+        evidence_change = self.payload()
+        evidence_change["evidence"]["head_sha"] = "feedface"
+        evidence_change["evidence"]["diff_digest"] = "b" * 64
+        rebound = build_contract_identity(evidence_change)
+        self.assertEqual(identity.fingerprint_v2, rebound.fingerprint_v2)
+        self.assertNotEqual(
+            json.dumps(identity.evidence.as_dict(), sort_keys=True),
+            json.dumps(rebound.evidence.as_dict(), sort_keys=True),
+        )
+
+
+if __name__ == "__main__":
+    unittest.main()