diff --git a/scripts/run_codex_pr_review.py b/scripts/run_codex_pr_review.py
index 4b04f62..cf3cd49 100644
--- a/scripts/run_codex_pr_review.py
+++ b/scripts/run_codex_pr_review.py
@@ -77,6 +77,7 @@
 FINDING_HISTORY_MAX_ENCODED_BYTES = ((FINDING_HISTORY_MAX_BYTES + 2) // 3) * 4
 FINDING_HISTORY_TEXT_LIMIT = 500
 ARBITRATION_REPEAT_THRESHOLD = 2
+CONTRACT_HISTORY_VERSION = 2
 
 
 class ReviewError(RuntimeError):
@@ -804,22 +805,96 @@ def parse_arbitration_output(
     return result
 
 
+def _normalize_contract_text(value: Any) -> str:
+    text = _sanitize_history_text(value, 240).lower()
+    words = {
+        token for token in re.findall(r"[a-z0-9]+", text)
+        if token not in {"the", "and", "for", "from", "into", "that", "this", "with", "must", "should", "will", "return"}
+    }
+    return " ".join(sorted(words))
+
+
+def _contract_subject(finding: dict[str, Any]) -> str:
+    text = " ".join(
+        str(finding.get(field) or "") for field in ("description", "suggestion")
+    )
+    anchors = re.findall(r"`([^`]{1,120})`", text)
+    anchors.extend(re.findall(r"\b[A-Za-z_]\w*(?:\.[A-Za-z_]\w*)+\b", text))
+    normalized = {_normalize_contract_text(anchor) for anchor in anchors}
+    return "|".join(sorted(anchor for anchor in normalized if anchor)) or _normalize_contract_text(
+        finding.get("suggestion") or text
+    )
+
+
+def _contract_key(finding: dict[str, Any]) -> str:
+    payload = json.dumps(
+        {
+            "file": _sanitize_history_path(finding.get("file")),
+            "category": _sanitize_history_text(finding.get("category"), 80).lower(),
+            "subject": _contract_subject(finding),
+        },
+        sort_keys=True,
+        separators=(",", ":"),
+    )
+    return hashlib.sha256(payload.encode("utf-8")).hexdigest()[:20]
+
+
+def _behavior_digest(finding: dict[str, Any]) -> str:
+    payload = json.dumps(
+        {
+            "contract_key": _contract_key(finding),
+            "behavior": _normalize_contract_text(finding.get("suggestion") or finding.get("description")),
+        },
+        sort_keys=True,
+        separators=(",", ":"),
+    )
+    return hashlib.sha256(payload.encode("utf-8")).hexdigest()[:20]
+
+
+def _contract_finding(finding: dict[str, Any]) -> dict[str, str]:
+    normalized = {
+        "severity": _sanitize_history_text(finding.get("severity"), 20).lower(),
+        "category": _sanitize_history_text(finding.get("category"), 80).lower(),
+        "file": _sanitize_history_path(finding.get("file")),
+        "description": _sanitize_history_text(finding.get("description")),
+        "suggestion": _sanitize_history_text(finding.get("suggestion")),
+    }
+    normalized["contract_key"] = _contract_key(normalized)
+    normalized["behavior_digest"] = _behavior_digest(normalized)
+    normalized["fingerprint_v2"] = hashlib.sha256(
+        json.dumps(
+            {
+                "contract_key": normalized["contract_key"],
+                "behavior_digest": normalized["behavior_digest"],
+            },
+            sort_keys=True,
+            separators=(",", ":"),
+        ).encode("utf-8")
+    ).hexdigest()[:20]
+    return normalized
+
+
+def _fingerprint_v2(finding: dict[str, Any]) -> str:
+    return _contract_finding(finding)["fingerprint_v2"]
+
+
+def _contract_identity(finding: dict[str, Any]) -> tuple[str, str]:
+    return (
+        str(finding.get("contract_key") or _contract_key(finding)),
+        str(finding.get("behavior_digest") or _behavior_digest(finding)),
+    )
+
+
 def blocking_finding_fingerprint(findings: list[dict[str, Any]]) -> str:
     """Return a stable arbitration-candidate identifier despite wording drift."""
-    normalized: list[dict[str, str]] = []
-    for finding in findings:
-        if not isinstance(finding, dict):
-            continue
-        normalized.append(
-            {
-                "category": str(finding.get("category") or "").strip().lower(),
-                "file": str(finding.get("file") or "").strip(),
-                "severity": str(finding.get("severity") or "").strip().lower(),
-            }
-        )
-    if not normalized:
+    fingerprints = sorted(
+        _fingerprint_v2(finding)
+        for finding in findings
+        if isinstance(finding, dict) and _fingerprint_v2(finding)
+    )
+    if not fingerprints:
         return ""
-    payload = json.dumps(sorted(normalized, key=lambda item: json.dumps(item, sort_keys=True)), sort_keys=True)
+    payload = json.dumps(fingerprints, sort_keys=True)
     return hashlib.sha256(payload.encode("utf-8")).hexdigest()[:20]
 
 
@@ -831,7 +906,7 @@ def blocking_finding_fingerprints(findings: list[dict[str, Any]]) -> tuple[str,
                 fingerprint
                 for finding in findings
                 if isinstance(finding, dict)
-                for fingerprint in [blocking_finding_fingerprint([finding])]
+                for fingerprint in [_fingerprint_v2(finding)]
                 if fingerprint
             }
         )
@@ -1062,13 +1137,7 @@ def _sanitize_history_text(value: Any, limit: int = FINDING_HISTORY_TEXT_LIMIT)
 
 
 def _history_finding(finding: dict[str, Any]) -> dict[str, str]:
-    return {
-        "severity": _sanitize_history_text(finding.get("severity"), 20).lower(),
-        "category": _sanitize_history_text(finding.get("category"), 80).lower(),
-        "file": _sanitize_history_path(finding.get("file")),
-        "description": _sanitize_history_text(finding.get("description")),
-        "suggestion": _sanitize_history_text(finding.get("suggestion")),
-    }
+    return _contract_finding(finding)
 
 
 def _sanitize_history_path(value: Any) -> str:
@@ -1099,7 +1168,7 @@ def build_finding_history_marker(
                 "status": status or "blocking",
             }
         )
-    payload = {"version": 1, "rounds": rounds[-FINDING_HISTORY_MAX_ROUNDS:]}
+    payload = {"version": CONTRACT_HISTORY_VERSION, "rounds": rounds[-FINDING_HISTORY_MAX_ROUNDS:]}
     raw = json.dumps(payload, ensure_ascii=True, separators=(",", ":")).encode("utf-8")
     while len(raw) > FINDING_HISTORY_MAX_BYTES and len(rounds) > 1:
         rounds.pop(0)
@@ -1110,7 +1179,7 @@ def build_finding_history_marker(
         if not re.fullmatch(r"[0-9a-f]{7,64}", overflow_head_sha):
             return build_invalid_finding_history_marker(overflow_head_sha)
         overflow_payload = {
-            "version": 1,
+            "version": CONTRACT_HISTORY_VERSION,
             "rounds": [
                 {
                     "head_sha": overflow_head_sha,
@@ -1165,7 +1234,7 @@ def parse_finding_history(body: str) -> tuple[list[dict[str, Any]], bool]:
         payload = json.loads(raw)
     except (ValueError, json.JSONDecodeError):
         return [], False
-    if not isinstance(payload, dict) or payload.get("version") != 1:
+    if not isinstance(payload, dict) or payload.get("version") not in {1, CONTRACT_HISTORY_VERSION}:
         return [], False
     rounds = payload.get("rounds")
     if not isinstance(rounds, list) or len(rounds) > FINDING_HISTORY_MAX_ROUNDS:
@@ -1188,20 +1257,37 @@ def parse_finding_history(body: str) -> tuple[list[dict[str, Any]], bool]:
             return [], False
         if not isinstance(findings, list):
             return [], False
-        if status not in {"blocking", "clear", "cleared", "overflow", "invalid_history"}:
+        if status not in {"blocking", "clear", "cleared", "overflow", "invalid_history", "initial_review", "remediation", "arbitration_required", "blocked_human_contract"}:
             return [], False
         if set(round_state) not in ({"head_sha", "findings"}, {"head_sha", "findings", "status"}):
             return [], False
         checked_findings: list[dict[str, str]] = []
         for finding in findings:
-            if not isinstance(finding, dict) or set(finding) != set(field_limits):
-                return [], False
-            if any(
-                not isinstance(finding[field], str) or len(finding[field]) > limit
-                for field, limit in field_limits.items()
-            ):
+            if not isinstance(finding, dict):
                 return [], False
-            checked_findings.append(dict(finding))
+            if payload.get("version") == 1:
+                if set(finding) != set(field_limits) or any(
+                    not isinstance(finding[field], str) or len(finding[field]) > limit
+                    for field, limit in field_limits.items()
+                ):
+                    return [], False
+                checked_findings.append(_history_finding(finding))
+            else:
+                required = {**field_limits, "contract_key": 20, "behavior_digest": 20, "fingerprint_v2": 20}
+                if set(finding) != set(required) or any(
+                    not isinstance(finding[field], str) or len(finding[field]) > limit
+                    for field, limit in required.items()
+                ):
+                    return [], False
+                if not all(re.fullmatch(r"[0-9a-f]{20}", finding[field]) for field in ("contract_key", "behavior_digest", "fingerprint_v2")):
+                    return [], False
+                recomputed = _contract_finding(finding)
+                if any(
+                    finding[field] != recomputed[field]
+                    for field in ("contract_key", "behavior_digest", "fingerprint_v2")
+                ):
+                    return [], False
+                checked_findings.append(dict(finding))
         validated.append(
             {"head_sha": head_sha, "findings": checked_findings, "status": status}
         )
@@ -1212,9 +1298,9 @@ def previous_matching_findings(
     history: list[dict[str, Any]], current_findings: list[dict[str, Any]]
 ) -> list[dict[str, Any]]:
     """Return the latest unresolved historical finding for every current key."""
-    current_keys = set(blocking_finding_fingerprints(current_findings))
-    matched: dict[str, dict[str, Any]] = {}
-    resolved: set[str] = set()
+    current_keys = {_contract_identity(finding) for finding in current_findings if isinstance(finding, dict)}
+    matched: dict[tuple[str, str], dict[str, Any]] = {}
+    resolved: set[tuple[str, str]] = set()
     for round_state in reversed(history):
         prior = round_state.get("findings")
         if not isinstance(prior, list):
@@ -1225,7 +1311,7 @@ def previous_matching_findings(
         for finding in prior:
             if not isinstance(finding, dict):
                 continue
-            key = blocking_finding_fingerprint([finding])
+            key = _contract_identity(finding)
             if key not in current_keys or key in resolved or key in matched:
                 continue
             if status in {"clear", "cleared"}:
@@ -1262,6 +1348,46 @@ def unresolved_history_findings(history: list[dict[str, Any]]) -> list[dict[str,
     return previous_matching_findings(history, all_findings)
 
 
+def conflicting_contract_findings(
+    history: list[dict[str, Any]], current_findings: list[dict[str, Any]]
+) -> list[dict[str, Any]]:
+    """Find same-contract findings whose required behavior changed."""
+    current: dict[str, set[str]] = {}
+    for finding in current_findings:
+        if not isinstance(finding, dict):
+            continue
+        contract_key, behavior_digest = _contract_identity(finding)
+        current.setdefault(contract_key, set()).add(behavior_digest)
+    prior_by_contract: dict[str, dict[str, dict[str, Any]]] = {}
+    resolved: set[tuple[str, str]] = set()
+    for round_state in reversed(history):
+        status = round_state.get("status")
+        findings = round_state.get("findings") or []
+        if status in {"clear", "cleared"} and not findings:
+            break
+        for finding in findings:
+            if not isinstance(finding, dict):
+                continue
+            identity = _contract_identity(finding)
+            if status in {"clear", "cleared"}:
+                resolved.add(identity)
+                continue
+            prior_by_contract.setdefault(identity[0], {}).setdefault(identity[1], finding)
+    conflicts: list[dict[str, Any]] = []
+    for contract_key, prior_behaviors in prior_by_contract.items():
+        current_behaviors = current.get(contract_key)
+        if not current_behaviors:
+            continue
+        for behavior_digest, prior in prior_behaviors.items():
+            if (contract_key, behavior_digest) in resolved or behavior_digest in current_behaviors:
+                continue
+            conflicts.append({
+                **prior,
+                "current_behavior_digest": sorted(current_behaviors)[0],
+            })
+    return conflicts
+
+
 def has_active_blocking_history(history: list[dict[str, Any]]) -> bool:
     if not history:
         return False
@@ -1943,6 +2069,11 @@ def main() -> int:
     findings = review.get("findings", [])
     if not isinstance(findings, list):
         findings = []
+    findings = [
+        {**finding, **_contract_finding(finding)}
+        for finding in findings
+        if isinstance(finding, dict)
+    ]
     print(f"Found {len(findings)} issue(s)")
 
     # Evaluate findings
@@ -2004,6 +2135,18 @@ def main() -> int:
         previous_findings = previous_matching_findings(
             finding_history, decision["blocking_findings"]
         )
+    contract_conflicts = conflicting_contract_findings(
+        finding_history, decision["blocking_findings"]
+    )
+    if contract_conflicts:
+        decision.update(
+            {
+                "blocked": True,
+                "contract_conflict": True,
+                "auto_fix_allowed": False,
+                "next_action": "contract_arbitration",
+            }
+        )
     matched_history_heads = sorted(
         {
             str(finding.get("history_head_sha") or "")
diff --git a/tests/fixtures/codex_pr_review_replays.json b/tests/fixtures/codex_pr_review_replays.json
new file mode 100644
index 0000000..c2e1bea
--- /dev/null
+++ b/tests/fixtures/codex_pr_review_replays.json
@@ -0,0 +1,40 @@
+{
+  "dispatch_pr_75": {
+    "repo": "org/audit-bridge",
+    "pr_number": 75,
+    "finding": {
+      "severity": "high",
+      "category": "logic",
+      "file": "service/adapters/dispatch.py",
+      "description": "`DispatchState.apply()` must distinguish pre-launch failure from provider dispatch.",
+      "suggestion": "Persist `DispatchState.not_started` before the provider boundary."
+    },
+    "reworded": {
+      "severity": "high",
+      "category": "logic",
+      "file": "service/adapters/dispatch.py",
+      "description": "A local error in `DispatchState.apply()` is not proof of provider launch.",
+      "suggestion": "Persist `DispatchState.not_started` before the provider boundary."
+    }
+  },
+  "crypto_pr_125": {
+    "repo": "org/crypto-pipelines",
+    "pr_number": 125,
+    "findings": [
+      {
+        "severity": "high",
+        "category": "logic",
+        "file": "src/strategy_lifecycle/backtest_wrapper.py",
+        "description": "`load_preflight_panel()` must validate the requested completed day.",
+        "suggestion": "Require an exact scored-date match."
+      },
+      {
+        "severity": "high",
+        "category": "logic",
+        "file": "src/strategy_lifecycle/backtest_wrapper.py",
+        "description": "`CryptoBacktestRunner.run()` must revalidate each requested window.",
+        "suggestion": "Do not reuse a cached runner without window validation."
+      }
+    ]
+  }
+}
diff --git a/tests/test_run_codex_pr_review.py b/tests/test_run_codex_pr_review.py
index cec69a0..4f759f8 100644
--- a/tests/test_run_codex_pr_review.py
+++ b/tests/test_run_codex_pr_review.py
@@ -3,6 +3,7 @@
 import base64
 import json
 import os
+import re
 import subprocess
 import sys
 import tempfile
@@ -15,6 +16,10 @@
 
 
 class RunCodexPrReviewTests(unittest.TestCase):
+    def _replay_fixtures(self) -> dict[str, object]:
+        path = Path(__file__).with_name("fixtures") / "codex_pr_review_replays.json"
+        return json.loads(path.read_text(encoding="utf-8"))
+
     def _write_event(self, tmpdir: str, files: list[str]) -> str:
         event = {
             "pull_request": {"number": 7, "head": {"sha": "abc1234"}},
@@ -73,7 +78,7 @@ def test_repeated_findings_are_fingerprinted_before_arbitration(self) -> None:
         self.assertEqual(fingerprint, run_codex_pr_review.blocking_finding_fingerprint(reordered))
         self.assertEqual(fingerprint, run_codex_pr_review.blocking_finding_fingerprint(reworded))
         self.assertNotEqual(fingerprint, run_codex_pr_review.blocking_finding_fingerprint(other))
-        self.assertNotEqual(fingerprint, run_codex_pr_review.blocking_finding_fingerprint(reclassified))
+        self.assertEqual(fingerprint, run_codex_pr_review.blocking_finding_fingerprint(reclassified))
         self.assertEqual(fingerprints, run_codex_pr_review.blocking_finding_fingerprints(reworded))
         self.assertEqual(
             run_codex_pr_review.next_blocking_streak(
@@ -111,6 +116,118 @@ def test_repeated_findings_are_fingerprinted_before_arbitration(self) -> None:
         self.assertTrue(run_codex_pr_review.should_arbitrate(blocked=True, streak=2, repeated=True, new_head=True))
         self.assertFalse(run_codex_pr_review.should_arbitrate(blocked=True, streak=2, repeated=True, new_head=False))
 
+    def test_replay_fixtures_keep_contract_identity_stable(self) -> None:
+        fixtures = self._replay_fixtures()
+        dispatch = fixtures["dispatch_pr_75"]
+        crypto = fixtures["crypto_pr_125"]
+        self.assertEqual(
+            run_codex_pr_review._contract_finding(dispatch["finding"])["fingerprint_v2"],
+            run_codex_pr_review._contract_finding(dispatch["reworded"])["fingerprint_v2"],
+        )
+        self.assertEqual(len(crypto["findings"]), 2)
+        self.assertNotEqual(
+            run_codex_pr_review._contract_key(crypto["findings"][0]),
+            run_codex_pr_review._contract_key(crypto["findings"][1]),
+        )
+
+    def test_severity_drift_is_not_contract_conflict(self) -> None:
+        finding = {
+            "severity": "high", "category": "logic", "file": "service/review.py",
+            "description": "`Review.run()` must reject invalid state.",
+            "suggestion": "Reject invalid state in `Review.run()`."}
+        marker = run_codex_pr_review.build_finding_history_marker([], [finding], "deadbeef")
+        history, valid = run_codex_pr_review.parse_finding_history(marker)
+        self.assertTrue(valid)
+        self.assertEqual(
+            run_codex_pr_review.conflicting_contract_findings(
+                history, [{**finding, "severity": "critical"}]
+            ),
+            [],
+        )
+
+    def test_opposite_behavior_conflicts_but_unrelated_same_file_does_not(self) -> None:
+        prior = {
+            "severity": "high", "category": "logic", "file": "service/review.py",
+            "description": "`ReviewContract` must reject invalid state.",
+            "suggestion": "Reject invalid state.",
+        }
+        history, valid = run_codex_pr_review.parse_finding_history(
+            run_codex_pr_review.build_finding_history_marker([], [prior], "deadbeef")
+        )
+        self.assertTrue(valid)
+        opposite = {**prior, "suggestion": "Accept invalid state."}
+        unrelated = {**prior, "description": "`AuditContract` must redact credentials.", "suggestion": "Redact credentials."}
+        self.assertEqual(len(run_codex_pr_review.conflicting_contract_findings(history, [opposite])), 1)
+        self.assertEqual(run_codex_pr_review.conflicting_contract_findings(history, [unrelated]), [])
+
+    def test_contract_conflicts_aggregate_behaviors_and_honor_clear_rounds(self) -> None:
+        prior = {
+            "severity": "high", "category": "logic", "file": "service/review.py",
+            "description": "`ReviewContract` must reject invalid state.",
+            "suggestion": "Reject invalid state.",
+        }
+        history, valid = run_codex_pr_review.parse_finding_history(
+            run_codex_pr_review.build_finding_history_marker([], [prior], "deadbeef")
+        )
+        self.assertTrue(valid)
+        matching = dict(prior)
+        opposite = dict(prior, suggestion="Accept invalid state.")
+        unrelated_behavior = dict(prior, suggestion="Ignore invalid state.")
+
+        self.assertEqual(
+            run_codex_pr_review.conflicting_contract_findings(history, [matching, opposite]),
+            [],
+        )
+        self.assertEqual(
+            len(run_codex_pr_review.conflicting_contract_findings(
+                history, [opposite, unrelated_behavior]
+            )),
+            1,
+        )
+
+        cleared_history, valid = run_codex_pr_review.parse_finding_history(
+            run_codex_pr_review.build_finding_history_marker(
+                history, [prior], "feedface", status="cleared"
+            )
+        )
+        self.assertTrue(valid)
+        self.assertEqual(
+            run_codex_pr_review.previous_matching_findings(cleared_history, [prior]),
+            [],
+        )
+        self.assertEqual(
+            run_codex_pr_review.conflicting_contract_findings(cleared_history, [opposite]),
+            [],
+        )
+
+        boundary_history, valid = run_codex_pr_review.parse_finding_history(
+            run_codex_pr_review.build_finding_history_marker(
+                cleared_history, [], "cafebabe", status="clear"
+            )
+        )
+        self.assertTrue(valid)
+        self.assertEqual(
+            run_codex_pr_review.previous_matching_findings(boundary_history, [prior]),
+            [],
+        )
+        self.assertEqual(
+            run_codex_pr_review.conflicting_contract_findings(boundary_history, [opposite]),
+            [],
+        )
+
+    def test_v2_history_rejects_tampered_identity_digest(self) -> None:
+        finding = {
+            "severity": "high", "category": "logic", "file": "service/review.py",
+            "description": "`Review.run()` must reject invalid state.",
+            "suggestion": "Reject invalid state in `Review.run()`."}
+        marker = run_codex_pr_review.build_finding_history_marker([], [finding], "deadbeef")
+        encoded = re.search(r"history:v1:([A-Za-z0-9_-]+) -->", marker).group(1)
+        payload = json.loads(base64.urlsafe_b64decode(encoded + "=" * (-len(encoded) % 4)))
+        payload["rounds"][0]["findings"][0]["behavior_digest"] = "0" * 20
+        tampered = base64.urlsafe_b64encode(json.dumps(payload, separators=(",", ":")).encode()).decode().rstrip("=")
+        body = f"<!-- codex-pr-review-history:v1:{tampered} -->"
+        self.assertEqual(run_codex_pr_review.parse_finding_history(body), ([], False))
+
     def test_parse_arbitration_output_requires_supported_verdict(self) -> None:
         self.assertEqual(
             run_codex_pr_review.parse_arbitration_output('{"verdict":"clear","reason":"covered by the regression test"}'),
@@ -1135,12 +1252,12 @@ def test_main_contract_conflict_is_consistent_across_comment_artifact_and_output
                 "severity": "high",
                 "category": "contract",
                 "file": "service/review.py",
-                "description": "Missing panel must return a structured blocked result.",
+                "description": "`PanelContract` must return a structured blocked result.",
                 "suggestion": "Return ReviewResult(blocked=True).",
             }
             current = dict(
                 prior,
-                description="Missing panel must fail fast.",
+                description="`PanelContract` must fail fast.",
                 suggestion="Raise ReviewError instead of returning a result.",
             )
             fingerprint = run_codex_pr_review.blocking_finding_fingerprint([prior])