diff --git a/review_proofs.py b/review_proofs.py index e41b639..7918fa2 100644 --- a/review_proofs.py +++ b/review_proofs.py @@ -3964,6 +3964,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, + } + + # --------------------------------------------------------------------------- # Full-suite failure approval gate (#323) # --------------------------------------------------------------------------- diff --git a/tests/test_review_proofs.py b/tests/test_review_proofs.py index 6b7f0ca..0d42359 100644 --- a/tests/test_review_proofs.py +++ b/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_validation_environment_proof, assess_full_suite_failure_approval_gate, @@ -2639,6 +2641,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"]) + + class TestFullSuiteFailureApprovalGate(unittest.TestCase): """Issue #323: full-suite failure blocks approval without baseline proof."""