feat: reject stale already-landed controller handoff fields (#299)

Add assess_already_landed_handoff_report verifier to block Pinned reviewed
head, Scratch worktree used, legacy Mutations/Workspace mutations None when
git fetch or other mutations occurred, and narrative/handoff drift after the
already-landed gate fires. Wire into build_final_report and review_proofs.

Co-Authored-By: Claude Opus 4.8 (1M context) <[email protected]>

tests/test_llm_workflow_split.py

@@ -140,3 +140,9 @@ def test_non_mergeable_skip_verifier_exported():
     from review_proofs import assess_non_mergeable_skip_proof
 
     assert callable(assess_non_mergeable_skip_proof)
+
+
+def test_already_landed_handoff_verifier_exported():
+    from review_proofs import assess_already_landed_handoff_report
+
+    assert callable(assess_already_landed_handoff_report)

tests/test_reviewer_already_landed_handoff.py

@@ -0,0 +1,131 @@
+"""Tests for already-landed handoff verifier (#299)."""
+import sys
+import unittest
+from pathlib import Path
+
+sys.path.insert(0, str(Path(__file__).resolve().parent.parent))
+
+from reviewer_already_landed_handoff import (  # noqa: E402
+    ALREADY_LANDED_STATE,
+    assess_already_landed_handoff_report,
+)
+
+
+def _good_report() -> str:
+    return "\n".join([
+        "## PR reconciliation summary",
+        f"Eligibility class: {ALREADY_LANDED_STATE}",
+        "Candidate head SHA: 2a544f582026b72a229d59a172c0a63ac4aaeaf9",
+        "Reviewed head SHA: none",
+        "Review worktree used: false",
+        "Ancestor proof: PR head is ancestor of prgs/master",
+        "",
+        "## Controller Handoff",
+        "- Selected PR: #278",
+        f"- Eligibility class: {ALREADY_LANDED_STATE}",
+        "- Candidate head SHA: 2a544f582026b72a229d59a172c0a63ac4aaeaf9",
+        "- Reviewed head SHA: none",
+        "- Target branch: master",
+        "- Target branch SHA: abc123def4567890abcdef1234567890abcdef12",
+        "- Review worktree used: false",
+        "- Git ref mutations: git fetch prgs master",
+        "- Review decision: none",
+        "- Merge result: none",
+        "- Linked issue status: open",
+        "- Safe next action: reconcile open PR and linked issue",
+    ])
+
+
+class TestAlreadyLandedHandoff(unittest.TestCase):
+    def test_clean_handoff_passes(self):
+        result = assess_already_landed_handoff_report(_good_report())
+        self.assertTrue(result["proven"], result["reasons"])
+        self.assertTrue(result["gate_active"])
+
+    def test_inactive_gate_skips_checks(self):
+        report = "## Controller Handoff\n- Selected PR: #12\n- Review decision: approve"
+        result = assess_already_landed_handoff_report(report)
+        self.assertTrue(result["proven"])
+        self.assertFalse(result["gate_active"])
+
+    def test_rejects_stale_pinned_reviewed_head(self):
+        report = _good_report().replace(
+            "- Candidate head SHA:",
+            "- Pinned reviewed head: 2a544f582026b72a229d59a172c0a63ac4aaeaf9\n"
+            "- Candidate head SHA:",
+        )
+        result = assess_already_landed_handoff_report(report)
+        self.assertFalse(result["proven"])
+        self.assertTrue(any("pinned reviewed head" in r.lower() for r in result["reasons"]))
+
+    def test_rejects_scratch_worktree_used(self):
+        report = _good_report() + "\n- Scratch worktree used: false"
+        result = assess_already_landed_handoff_report(report)
+        self.assertFalse(result["proven"])
+        self.assertTrue(any("scratch worktree" in r.lower() for r in result["reasons"]))
+
+    def test_rejects_mutations_none_with_git_fetch(self):
+        report = _good_report().replace(
+            "- Git ref mutations: git fetch prgs master",
+            "- Git ref mutations: none\n- Mutations: None",
+        )
+        result = assess_already_landed_handoff_report(
+            report,
+            command_log=[{"command": "git fetch prgs master", "performed": True}],
+        )
+        self.assertFalse(result["proven"])
+        joined = " ".join(result["reasons"]).lower()
+        self.assertIn("mutations", joined)
+        self.assertIn("git ref", joined)
+
+    def test_rejects_workspace_mutations_none(self):
+        report = _good_report() + "\n- Workspace mutations: None"
+        result = assess_already_landed_handoff_report(report)
+        self.assertFalse(result["proven"])
+        self.assertTrue(any("workspace mutations" in r.lower() for r in result["reasons"]))
+
+    def test_rejects_reviewed_head_sha_when_gate_fired(self):
+        report = _good_report().replace("- Reviewed head SHA: none", "- Reviewed head SHA: abc123")
+        result = assess_already_landed_handoff_report(report)
+        self.assertFalse(result["proven"])
+        self.assertTrue(any("reviewed head sha" in r.lower() for r in result["reasons"]))
+
+    def test_rejects_narrative_handoff_eligibility_mismatch(self):
+        report = _good_report().replace(
+            f"Eligibility class: {ALREADY_LANDED_STATE}",
+            "Eligibility class: NORMAL_REVIEW_CANDIDATE",
+            1,
+        )
+        result = assess_already_landed_handoff_report(report)
+        self.assertFalse(result["proven"])
+        self.assertTrue(any("eligibility class" in r.lower() for r in result["reasons"]))
+
+    def test_session_gate_fired_without_text_marker(self):
+        report = "## Controller Handoff\n- Selected PR: #1\n- Pinned reviewed head: abc"
+        result = assess_already_landed_handoff_report(
+            report,
+            handoff_session={"gate_fired": True},
+        )
+        self.assertFalse(result["proven"])
+        self.assertTrue(result["gate_active"])
+
+    def test_infra_stop_style_normal_handoff_not_blocked(self):
+        report = "\n".join([
+            "## Controller Handoff",
+            "- Selected PR: #12",
+            "- Review decision: request_changes",
+            "- Pinned reviewed head: abc123def4567890abcdef1234567890abcdef12",
+        ])
+        result = assess_already_landed_handoff_report(report)
+        self.assertTrue(result["proven"])
+
+
+class TestExport(unittest.TestCase):
+    def test_review_proofs_reexport(self):
+        from review_proofs import assess_already_landed_handoff_report as exported
+
+        self.assertTrue(callable(exported))
+
+
+if __name__ == "__main__":
+    unittest.main()