review_proofs.py

@@ -3363,6 +3363,172 @@ def assess_reviewer_baseline_validation_proof(
     }
 
 
+# ---------------------------------------------------------------------------
+# Already-landed review gate (#292)
+# ---------------------------------------------------------------------------
+
+ALREADY_LANDED_STATE = "ALREADY_LANDED_RECONCILE_REQUIRED"
+
+ALREADY_LANDED_HANDOFF_FIELDS = (
+    "PR number/title",
+    "candidate head SHA",
+    "target branch",
+    "target branch SHA",
+    "ancestor proof",
+    "linked issue status",
+    "recommended reconciliation action",
+    "capability proof for close/comment mutations",
+)
+
+_FORBIDDEN_LANDED_STATES_RE = re.compile(
+    r"review decision\s*:\s*approved?\b|"
+    r"merge result\s*:\s*merged\b|"
+    r"\bready[_ ]to[_ ]merge\b|"
+    r"\bmerge[d]?\s*:\s*success\b",
+    re.IGNORECASE,
+)
+
+
+def assess_already_landed_review_gate(
+    *,
+    pr_number: int | None,
+    candidate_head_sha: str | None,
+    target_branch: str | None,
+    target_branch_sha: str | None,
+    head_is_ancestor_of_target: bool | None,
+    live_head_sha: str | None = None,
+    real_blocker: bool = False,
+    mergeable: bool | None = None,
+) -> dict:
+    """#292: hard gate before any review mutation on an already-landed PR.
+
+    Runs after PR selection and head pinning, before approval, request-
+    changes, or the merge API. *head_is_ancestor_of_target* is the result
+    of an ancestry check of *candidate_head_sha* against the freshly
+    fetched target branch at *target_branch_sha* (e.g. ``git merge-base
+    --is-ancestor``). ``None`` means the check was not run — the gate then
+    fails closed.
+
+    *live_head_sha*, when provided, must equal *candidate_head_sha*; a
+    changed head invalidates the pinned ancestry result until re-checked.
+
+    *mergeable* is accepted for caller convenience but never consulted:
+    conflict/mergeability handling is a separate gate.
+    """
+    del mergeable  # ancestry gate only; mergeability is a separate gate
+    reasons: list[str] = []
+
+    if not isinstance(pr_number, int) or pr_number <= 0:
+        reasons.append("PR number missing or invalid (#292)")
+    if not _FULL_SHA.match((candidate_head_sha or "").strip()):
+        reasons.append(
+            "candidate head SHA is not a full 40-hex commit SHA (#292)"
+        )
+    if not (target_branch or "").strip():
+        reasons.append("target branch missing (#292)")
+    if not _FULL_SHA.match((target_branch_sha or "").strip()):
+        reasons.append(
+            "target branch SHA is not a full 40-hex commit SHA; fetch the "
+            "target branch and record its SHA before the ancestry check "
+            "(#292)"
+        )
+    if live_head_sha is not None and candidate_head_sha and (
+        live_head_sha.strip() != candidate_head_sha.strip()
+    ):
+        reasons.append(
+            "PR head changed since the ancestry check was pinned; re-fetch "
+            "and re-run the already-landed gate (#292)"
+        )
+    if head_is_ancestor_of_target is None:
+        reasons.append(
+            "ancestry of the PR head against the target branch was not "
+            "checked; the already-landed gate must run before any review "
+            "mutation (#292)"
+        )
+
+    if reasons:
+        return {
+            "state": "GATE_NOT_PROVEN",
+            "approve_allowed": False,
+            "request_changes_allowed": False,
+            "merge_api_allowed": False,
+            "reconciliation_handoff_required": False,
+            "required_handoff_fields": (),
+            "reasons": reasons,
+            "safe_next_action": (
+                "fetch the target branch, pin the live PR head, run the "
+                "ancestry check, then re-run this gate"
+            ),
+        }
+
+    if head_is_ancestor_of_target:
+        return {
+            "state": ALREADY_LANDED_STATE,
+            "approve_allowed": False,
+            "request_changes_allowed": bool(real_blocker),
+            "merge_api_allowed": False,
+            "reconciliation_handoff_required": True,
+            "required_handoff_fields": ALREADY_LANDED_HANDOFF_FIELDS,
+            "reasons": [
+                f"PR #{pr_number} head {candidate_head_sha} is already an "
+                f"ancestor of {target_branch} @ {target_branch_sha}; the PR "
+                "is reconciliation-only (#292)"
+            ],
+            "safe_next_action": (
+                "stop before review mutation and emit an "
+                f"{ALREADY_LANDED_STATE} reconciliation handoff; any PR/issue "
+                "close or comment requires exact capability proof"
+            ),
+        }
+
+    return {
+        "state": "NORMAL_REVIEW_CANDIDATE",
+        "approve_allowed": True,
+        "request_changes_allowed": True,
+        "merge_api_allowed": True,
+        "reconciliation_handoff_required": False,
+        "required_handoff_fields": (),
+        "reasons": [],
+        "safe_next_action": "proceed with the normal review workflow gates",
+    }
+
+
+def assess_already_landed_report_state(
+    report_text: str | None,
+    *,
+    gate_fired: bool,
+) -> dict:
+    """#292: reject APPROVED/MERGED/READY_TO_MERGE after the gate fired.
+
+    *gate_fired* comes from the session's own gate result, so a report
+    that omits the already-landed markers entirely still fails closed —
+    unlike the text-only #327 wording rules this does not depend on the
+    report admitting the PR was already landed.
+    """
+    if not gate_fired:
+        return {"complete": True, "block": False, "reasons": []}
+
+    text = report_text or ""
+    reasons: list[str] = []
+
+    if _FORBIDDEN_LANDED_STATES_RE.search(text):
+        reasons.append(
+            "report claims APPROVED/MERGED/READY_TO_MERGE although the "
+            "already-landed gate fired (#292)"
+        )
+    if ALREADY_LANDED_STATE.lower() not in text.lower():
+        reasons.append(
+            f"report must state {ALREADY_LANDED_STATE} when the "
+            "already-landed gate fired (#292)"
+        )
+
+    return {
+        "complete": not reasons,
+        "block": bool(reasons),
+        "reasons": reasons,
+    }
+
+
 # ---------------------------------------------------------------------------
 # Identity disclosure (#305)
 # ---------------------------------------------------------------------------

tests/test_review_proofs.py

@@ -23,6 +23,8 @@ sys.path.insert(0, str(__import__("pathlib").Path(__file__).resolve().parent.par
 from review_proofs import (  # noqa: E402
     ISSUE_SELECTION_CONTINUATION_EXPLICIT,
     ISSUE_SELECTION_REPRESENTED_BY_OPEN_PR,
+    assess_already_landed_report_state,
+    assess_already_landed_review_gate,
     assess_author_pr_report,
     assess_reviewer_baseline_validation_proof,
     assess_capability_evidence,
@@ -2281,5 +2283,139 @@ class TestReviewerBaselineValidationProof(unittest.TestCase):
         self.assertFalse(report["baseline_validation_proven"])
 
 
+class TestAlreadyLandedReviewGate(unittest.TestCase):
+    """Issue #292: PR head already on target blocks approval and merge."""
+
+    def _gate(self, **overrides):
+        kwargs = {
+            "pr_number": 278,
+            "candidate_head_sha": PINNED,
+            "target_branch": "master",
+            "target_branch_sha": OTHER,
+            "head_is_ancestor_of_target": True,
+            "live_head_sha": PINNED,
+        }
+        kwargs.update(overrides)
+        return assess_already_landed_review_gate(**kwargs)
+
+    def test_already_landed_blocks_approval_and_merge(self):
+        result = self._gate()
+        self.assertEqual(result["state"], "ALREADY_LANDED_RECONCILE_REQUIRED")
+        self.assertFalse(result["approve_allowed"])
+        self.assertFalse(result["merge_api_allowed"])
+        self.assertFalse(result["request_changes_allowed"])
+        self.assertTrue(result["reconciliation_handoff_required"])
+
+    def test_already_landed_with_real_blocker_allows_request_changes(self):
+        result = self._gate(real_blocker=True)
+        self.assertEqual(result["state"], "ALREADY_LANDED_RECONCILE_REQUIRED")
+        self.assertFalse(result["approve_allowed"])
+        self.assertFalse(result["merge_api_allowed"])
+        self.assertTrue(result["request_changes_allowed"])
+
+    def test_normal_candidate_passes_gate(self):
+        result = self._gate(head_is_ancestor_of_target=False)
+        self.assertEqual(result["state"], "NORMAL_REVIEW_CANDIDATE")
+        self.assertTrue(result["approve_allowed"])
+        self.assertTrue(result["merge_api_allowed"])
+        self.assertFalse(result["reconciliation_handoff_required"])
+
+    def test_non_mergeable_normal_pr_still_passes_this_gate(self):
+        # Mergeability/conflicts are separate gates; ancestry gate only
+        # decides already-landed vs normal candidate.
+        result = self._gate(
+            head_is_ancestor_of_target=False, mergeable=False)
+        self.assertEqual(result["state"], "NORMAL_REVIEW_CANDIDATE")
+        self.assertTrue(result["request_changes_allowed"])
+
+    def test_unchecked_ancestry_blocks_review_mutations(self):
+        result = self._gate(head_is_ancestor_of_target=None)
+        self.assertEqual(result["state"], "GATE_NOT_PROVEN")
+        self.assertFalse(result["approve_allowed"])
+        self.assertFalse(result["merge_api_allowed"])
+
+    def test_changed_head_since_pin_blocks_until_recheck(self):
+        result = self._gate(
+            head_is_ancestor_of_target=False, live_head_sha=OTHER)
+        self.assertEqual(result["state"], "GATE_NOT_PROVEN")
+        self.assertFalse(result["approve_allowed"])
+        self.assertTrue(any(
+            "head" in reason.lower() for reason in result["reasons"]
+        ))
+
+    def test_invalid_shas_block_gate(self):
+        result = self._gate(candidate_head_sha="deadbeef")
+        self.assertEqual(result["state"], "GATE_NOT_PROVEN")
+        self.assertFalse(result["approve_allowed"])
+
+        result = self._gate(target_branch_sha=None)
+        self.assertEqual(result["state"], "GATE_NOT_PROVEN")
+
+    def test_already_landed_handoff_lists_required_fields(self):
+        result = self._gate()
+        for field in (
+            "PR number/title",
+            "candidate head SHA",
+            "target branch",
+            "target branch SHA",
+            "ancestor proof",
+            "linked issue status",
+            "recommended reconciliation action",
+            "capability proof for close/comment mutations",
+        ):
+            self.assertIn(field, result["required_handoff_fields"])
+
+
+class TestAlreadyLandedReportState(unittest.TestCase):
+    """Issue #292: reports cannot say APPROVED after the gate fired."""
+
+    GOOD_REPORT = (
+        "Eligibility class: ALREADY_LANDED_RECONCILE_REQUIRED\n"
+        "Candidate head SHA: " + PINNED + "\n"
+        "Review decision: none\n"
+        "Merge result: none\n"
+    )
+
+    def test_gate_fired_with_reconcile_state_passes(self):
+        result = assess_already_landed_report_state(
+            self.GOOD_REPORT, gate_fired=True)
+        self.assertTrue(result["complete"])
+        self.assertFalse(result["block"])
+
+    def test_gate_fired_with_approved_state_blocks(self):
+        report = self.GOOD_REPORT.replace(
+            "Review decision: none", "Review decision: APPROVED")
+        result = assess_already_landed_report_state(report, gate_fired=True)
+        self.assertTrue(result["block"])
+
+    def test_gate_fired_with_merged_state_blocks(self):
+        report = self.GOOD_REPORT.replace(
+            "Merge result: none", "Merge result: MERGED")
+        result = assess_already_landed_report_state(report, gate_fired=True)
+        self.assertTrue(result["block"])
+
+    def test_gate_fired_with_ready_to_merge_blocks(self):
+        result = assess_already_landed_report_state(
+            self.GOOD_REPORT + "Current status: READY_TO_MERGE\n",
+            gate_fired=True)
+        self.assertTrue(result["block"])
+
+    def test_gate_fired_without_reconcile_state_blocks(self):
+        result = assess_already_landed_report_state(
+            "Current status: review complete.\n", gate_fired=True)
+        self.assertTrue(result["block"])
+        self.assertTrue(any(
+            "already_landed_reconcile_required" in reason.lower()
+            for reason in result["reasons"]
+        ))
+
+    def test_gate_not_fired_reports_pass_untouched(self):
+        result = assess_already_landed_report_state(
+            "Review decision: APPROVED\nMerge result: MERGED\n",
+            gate_fired=False)
+        self.assertTrue(result["complete"])
+        self.assertFalse(result["block"])
+
+
 if __name__ == "__main__":
     unittest.main()