From b842a7f9234c7b1d30899545b00bdfe310906717 Mon Sep 17 00:00:00 2001 From: Jason Walker <913443@dadeschools.net> Date: Tue, 7 Jul 2026 05:28:19 -0400 Subject: [PATCH 1/3] Add validation worktree no-edit verifier for reviewer reports (#315) Block file edits in PR validation worktrees, require separate diagnostic scratch worktrees with explicit labeling, and reject contradictory File edits by reviewer: none claims. Co-Authored-By: Claude Opus 4.8 (1M context) --- review_proofs.py | 9 + reviewer_validation_worktree_edits.py | 184 ++++++++++++++++++ tests/test_llm_workflow_split.py | 8 +- ...test_reviewer_validation_worktree_edits.py | 134 +++++++++++++ 4 files changed, 334 insertions(+), 1 deletion(-) create mode 100644 reviewer_validation_worktree_edits.py create mode 100644 tests/test_reviewer_validation_worktree_edits.py diff --git a/review_proofs.py b/review_proofs.py index 02fa543..043ec35 100644 --- a/review_proofs.py +++ b/review_proofs.py @@ -3832,3 +3832,12 @@ def assess_non_mergeable_skip_proof(report_text, **kwargs): from reviewer_mergeability_skip import assess_non_mergeable_skip_proof as _assess return _assess(report_text, **kwargs) + + +def assess_validation_worktree_edit_report(report_text, **kwargs): + """#315: block edits in PR validation worktrees; require diagnostic separation.""" + from reviewer_validation_worktree_edits import ( + assess_validation_worktree_edit_report as _assess, + ) + + return _assess(report_text, **kwargs) diff --git a/reviewer_validation_worktree_edits.py b/reviewer_validation_worktree_edits.py new file mode 100644 index 0000000..c4e904f --- /dev/null +++ b/reviewer_validation_worktree_edits.py @@ -0,0 +1,184 @@ +"""PR validation worktree read-only edit verifier (#315).""" + +from __future__ import annotations + +import re +from typing import Any + +_VALIDATION_WORKTREE_RE = re.compile( + r"branches/(?:review-pr\d+[\w/-]*|review-[\w-]+)", + re.IGNORECASE, +) +_DIAGNOSTIC_WORKTREE_RE = re.compile( + r"branches/(?:diagnostic[\w/-]*|scratch[\w/-]*)", + re.IGNORECASE, +) +_FILE_EDITS_NONE_RE = re.compile( + r"file edits by reviewer\s*:\s*none\b", + re.IGNORECASE, +) +_FILE_EDITS_FIELD_RE = re.compile( + r"file edits by reviewer\s*:\s*(.+)", + re.IGNORECASE, +) +_VALIDATION_WORKTREE_PATH_RE = re.compile( + r"(?:validation worktree path|review worktree path)\s*:\s*(\S+)", + re.IGNORECASE, +) +_DIAGNOSTIC_WORKTREE_PATH_RE = re.compile( + r"(?:diagnostic (?:scratch )?worktree path|diagnostic worktree)\s*:\s*(\S+)", + re.IGNORECASE, +) +_DIAGNOSTIC_LABEL_RE = re.compile( + r"diagnostic local experiment|not pr-head validation|diagnostic-only", + re.IGNORECASE, +) +_OFFICIAL_VALIDATION_RE = re.compile( + r"(?:official (?:pr-head )?validation|pr-head validation result)", + re.IGNORECASE, +) +_POST_EDIT_OFFICIAL_RE = re.compile( + r"(?:official validation(?:\s+result)?\s*:\s*pass|validation passed after (?:local )?edit)", + re.IGNORECASE, +) + +_PERFORMED_EDIT_KINDS = frozenset({"file_edit", "edit", "write", "edited", "created", "wrote"}) + + +def _normalize_path(path: str) -> str: + return (path or "").replace("\\", "/").strip() + + +def _is_validation_worktree(path: str) -> bool: + return bool(_VALIDATION_WORKTREE_RE.search(_normalize_path(path))) + + +def _is_diagnostic_worktree(path: str) -> bool: + normalized = _normalize_path(path) + return bool( + _DIAGNOSTIC_WORKTREE_RE.search(normalized) + or "diagnostic" in normalized.lower() + ) + + +def _performed_edits(action_log: list[dict] | None) -> list[dict[str, str]]: + edits: list[dict[str, str]] = [] + for entry in action_log or []: + if entry.get("gated_rejected") or entry.get("performed") is False: + continue + kind = (entry.get("kind") or entry.get("action") or "file_edit").strip().lower() + if kind not in _PERFORMED_EDIT_KINDS: + continue + path = (entry.get("path") or "").strip() + if not path: + continue + worktree = _normalize_path(str(entry.get("worktree_path") or entry.get("cwd") or "")) + edits.append({"path": path, "worktree_path": worktree}) + return edits + + +def _extract_validation_path(text: str, session: dict) -> str: + path = _normalize_path(str(session.get("validation_worktree_path") or "")) + if path: + return path + match = _VALIDATION_WORKTREE_PATH_RE.search(text or "") + return _normalize_path(match.group(1)) if match else "" + + +def _extract_diagnostic_path(text: str, session: dict) -> str: + path = _normalize_path(str(session.get("diagnostic_worktree_path") or "")) + if path: + return path + match = _DIAGNOSTIC_WORKTREE_PATH_RE.search(text or "") + return _normalize_path(match.group(1)) if match else "" + + +def assess_validation_worktree_edit_report( + report_text: str, + *, + validation_session: dict | None = None, + action_log: list[dict] | None = None, +) -> dict[str, Any]: + """Block edits in PR validation worktrees; require diagnostic separation (#315).""" + text = report_text or "" + session = dict(validation_session or {}) + reasons: list[str] = [] + + validation_path = _extract_validation_path(text, session) + diagnostic_path = _extract_diagnostic_path(text, session) + validation_edits = list(session.get("validation_worktree_edits") or []) + diagnostic_edits = list(session.get("diagnostic_edits") or []) + edits = _performed_edits(action_log) + + if not validation_edits: + for entry in edits: + wt = entry.get("worktree_path") or validation_path + if wt and _is_validation_worktree(wt) and not _is_diagnostic_worktree(wt): + validation_edits.append(entry["path"]) + + if not diagnostic_edits: + for entry in edits: + wt = entry.get("worktree_path") or "" + if wt and _is_diagnostic_worktree(wt): + diagnostic_edits.append(entry["path"]) + elif entry["path"] and diagnostic_path and not validation_edits: + if _is_diagnostic_worktree(diagnostic_path): + diagnostic_edits.append(entry["path"]) + + claimed_none = bool(_FILE_EDITS_NONE_RE.search(text)) + any_edits = bool(validation_edits or diagnostic_edits or edits) + + if validation_edits: + reasons.append( + "reviewer must not edit files in the PR validation worktree; " + f"observed edits: {', '.join(validation_edits)}" + ) + + if diagnostic_edits or session.get("diagnostic_experiment"): + if not diagnostic_path and not _DIAGNOSTIC_WORKTREE_PATH_RE.search(text): + reasons.append( + "diagnostic experiments require a separate diagnostic scratch worktree path" + ) + if not _DIAGNOSTIC_LABEL_RE.search(text): + reasons.append( + "diagnostic experiments must be labeled " + "'Diagnostic local experiment — not PR-head validation'" + ) + if not _FILE_EDITS_FIELD_RE.search(text) or claimed_none: + reasons.append( + "diagnostic edits must be reported under 'File edits by reviewer'" + ) + + if any_edits and claimed_none: + reasons.append( + "report claims 'File edits by reviewer: none' but reviewer file edits occurred" + ) + + if session.get("official_validation_after_edit") or _POST_EDIT_OFFICIAL_RE.search(text): + if validation_edits or (validation_path and diagnostic_edits): + reasons.append( + "post-edit test results cannot be reported as official PR-head validation" + ) + + if validation_edits and _OFFICIAL_VALIDATION_RE.search(text): + if not _DIAGNOSTIC_LABEL_RE.search(text): + reasons.append( + "validation worktree was edited; official PR-head validation is no longer valid" + ) + + proven = not reasons + return { + "proven": proven, + "block": not proven, + "reasons": reasons, + "validation_worktree_path": validation_path or None, + "diagnostic_worktree_path": diagnostic_path or None, + "validation_worktree_edits": validation_edits, + "diagnostic_edits": diagnostic_edits, + "safe_next_action": ( + "keep PR validation worktrees read-only; use a separate diagnostic " + "scratch worktree and report all reviewer file edits" + if reasons + else "proceed" + ), + } \ No newline at end of file diff --git a/tests/test_llm_workflow_split.py b/tests/test_llm_workflow_split.py index 7e0fe23..ad4524b 100644 --- a/tests/test_llm_workflow_split.py +++ b/tests/test_llm_workflow_split.py @@ -127,4 +127,10 @@ def test_create_issue_final_report_verifier_exported(): def test_non_mergeable_skip_verifier_exported(): from review_proofs import assess_non_mergeable_skip_proof - assert callable(assess_non_mergeable_skip_proof) \ No newline at end of file + assert callable(assess_non_mergeable_skip_proof) + + +def test_validation_worktree_edit_verifier_exported(): + from review_proofs import assess_validation_worktree_edit_report + + assert callable(assess_validation_worktree_edit_report) \ No newline at end of file diff --git a/tests/test_reviewer_validation_worktree_edits.py b/tests/test_reviewer_validation_worktree_edits.py new file mode 100644 index 0000000..adedaca --- /dev/null +++ b/tests/test_reviewer_validation_worktree_edits.py @@ -0,0 +1,134 @@ +"""Tests for PR validation worktree no-edit verifier (#315).""" +import sys +import unittest +from pathlib import Path + +sys.path.insert(0, str(Path(__file__).resolve().parent.parent)) + +from reviewer_validation_worktree_edits import ( # noqa: E402 + assess_validation_worktree_edit_report, +) + + +def _read_only_report() -> str: + return "\n".join([ + "Validation worktree path: branches/review-pr280-feat-issue-275-claim", + "Official PR-head validation on unmodified worktree.", + "Official validation result: failed (1 failed)", + "File edits by reviewer: none", + "Review decision: request_changes", + ]) + + +def _diagnostic_report() -> str: + return "\n".join([ + "Validation worktree path: branches/review-pr280-feat-issue-275-claim", + "Official PR-head validation on unmodified worktree.", + "Official validation result: failed (1 failed)", + "Review decision: request_changes", + "Diagnostic local experiment — not PR-head validation", + "Diagnostic scratch worktree path: branches/diagnostic-pr280-fix-test", + "File edits by reviewer: tests/test_agent_temp_artifacts.py (diagnostic only)", + "Diagnostic result: passed after local fix", + ]) + + +class TestValidationWorktreeEdits(unittest.TestCase): + def test_read_only_review_passes(self): + result = assess_validation_worktree_edit_report( + _read_only_report(), + validation_session={ + "validation_worktree_path": ( + "branches/review-pr280-feat-issue-275-claim" + ), + }, + ) + self.assertTrue(result["proven"], result["reasons"]) + + def test_validation_worktree_edit_blocks(self): + result = assess_validation_worktree_edit_report( + _read_only_report(), + validation_session={ + "validation_worktree_path": ( + "branches/review-pr280-feat-issue-275-claim" + ), + "validation_worktree_edits": ["tests/test_agent_temp_artifacts.py"], + }, + ) + self.assertFalse(result["proven"]) + self.assertTrue(result["block"]) + self.assertTrue(any("validation worktree" in r.lower() for r in result["reasons"])) + + def test_file_edits_none_contradiction_blocks(self): + result = assess_validation_worktree_edit_report( + _read_only_report(), + validation_session={ + "validation_worktree_path": ( + "branches/review-pr280-feat-issue-275-claim" + ), + }, + action_log=[ + { + "path": "tests/test_agent_temp_artifacts.py", + "kind": "file_edit", + "performed": True, + "worktree_path": "branches/review-pr280-feat-issue-275-claim", + } + ], + ) + self.assertFalse(result["proven"]) + self.assertTrue(any("file edits" in r.lower() for r in result["reasons"])) + + def test_diagnostic_scratch_with_reporting_passes(self): + result = assess_validation_worktree_edit_report( + _diagnostic_report(), + validation_session={ + "validation_worktree_path": ( + "branches/review-pr280-feat-issue-275-claim" + ), + "diagnostic_worktree_path": "branches/diagnostic-pr280-fix-test", + "diagnostic_edits": ["tests/test_agent_temp_artifacts.py"], + "diagnostic_experiment": True, + }, + ) + self.assertTrue(result["proven"], result["reasons"]) + + def test_diagnostic_without_scratch_worktree_blocks(self): + result = assess_validation_worktree_edit_report( + "File edits by reviewer: tests/test_example.py", + validation_session={ + "diagnostic_edits": ["tests/test_example.py"], + "diagnostic_experiment": True, + }, + ) + self.assertFalse(result["proven"]) + self.assertTrue(any("diagnostic" in r.lower() for r in result["reasons"])) + + def test_post_edit_official_validation_blocks(self): + report = "\n".join([ + "Validation worktree path: branches/review-pr280-feat-issue-275-claim", + "Official validation result: passed after local edit", + "File edits by reviewer: tests/test_agent_temp_artifacts.py", + ]) + result = assess_validation_worktree_edit_report( + report, + validation_session={ + "validation_worktree_path": ( + "branches/review-pr280-feat-issue-275-claim" + ), + "validation_worktree_edits": ["tests/test_agent_temp_artifacts.py"], + "official_validation_after_edit": True, + }, + ) + self.assertFalse(result["proven"]) + + +class TestExport(unittest.TestCase): + def test_review_proofs_reexport(self): + from review_proofs import assess_validation_worktree_edit_report as exported + + self.assertTrue(callable(exported)) + + +if __name__ == "__main__": + unittest.main() \ No newline at end of file From 876f1ee3a3914a364f89efa3569b48c1017d68db Mon Sep 17 00:00:00 2001 From: Jason Walker <913443@dadeschools.net> Date: Tue, 7 Jul 2026 05:36:10 -0400 Subject: [PATCH 2/3] feat: hard-stop review mutations on already-landed PR heads (Closes #292) A reviewer workflow approved and attempted to merge PR #278 although its head commit was already an ancestor of master, ending in a Gitea 405 and manual reconciliation. Add the missing pre-mutation gate: - assess_already_landed_review_gate classifies the pinned candidate head against the fetched target branch: NORMAL_REVIEW_CANDIDATE, ALREADY_LANDED_RECONCILE_REQUIRED, or GATE_NOT_PROVEN (fail closed). Already-landed PRs block approval and the merge API, allow request-changes only for a real blocker, and require a reconciliation handoff (PR number/title, candidate head SHA, target branch + SHA, ancestor proof, linked issue status, recommended action, capability proof for close/comment mutations). Unchecked ancestry, invalid or missing SHAs, and a head changed since pinning all fail closed. Mergeability stays a separate gate. - assess_already_landed_report_state rejects APPROVED / MERGED / READY_TO_MERGE wording and missing ALREADY_LANDED_RECONCILE_REQUIRED state when the session's gate fired, complementing the text-marker rules from #327 which need the report to admit the landed state. Tests cover already-landed, normal, non-mergeable, changed-head, unchecked-ancestry, and invalid-SHA gate paths plus all report-state verdicts. Co-Authored-By: Claude Opus 4.8 (1M context) --- review_proofs.py | 166 ++++++++++++++++++++++++++++++++++++ tests/test_review_proofs.py | 136 +++++++++++++++++++++++++++++ 2 files changed, 302 insertions(+) diff --git a/review_proofs.py b/review_proofs.py index e7c94e2..3e28113 100644 --- a/review_proofs.py +++ b/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) # --------------------------------------------------------------------------- diff --git a/tests/test_review_proofs.py b/tests/test_review_proofs.py index ec411d2..4eefd47 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_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() From be1462dc03ddd65c9e8f85ec8e0b6fd246ec9905 Mon Sep 17 00:00:00 2001 From: Jason Walker <913443@dadeschools.net> Date: Tue, 7 Jul 2026 05:38:32 -0400 Subject: [PATCH 3/3] feat: reject eligible/reviewed wording for already-landed PRs (Closes #298) Add assess_already_landed_report_wording to review_proofs: when the already-landed gate fires, final reports and controller handoffs must use the reconciliation wording (Oldest open PR requiring action, Eligibility class: ALREADY_LANDED_RECONCILE_REQUIRED, Candidate head SHA, Reviewed head SHA: none, Review worktree used: false) and must not use the legacy fields (oldest/next eligible PR, Pinned reviewed head, Scratch worktree used) or claim a reviewed head SHA. In non-already-landed reports, a populated Pinned reviewed head or Reviewed head SHA fails unless validation and diff review actually passed this session. Fails closed. Tests cover the correct reconciliation wording, each forbidden legacy field, missing required fields, populated reviewed SHA, normal reviewed reports, blocked-infra reports, and validation-failed reports. Co-Authored-By: Claude Opus 4.8 (1M context) --- review_proofs.py | 96 +++++++++++++ tests/test_already_landed_report_wording.py | 151 ++++++++++++++++++++ 2 files changed, 247 insertions(+) create mode 100644 tests/test_already_landed_report_wording.py diff --git a/review_proofs.py b/review_proofs.py index 78f815d..9fb5dd7 100644 --- a/review_proofs.py +++ b/review_proofs.py @@ -1171,6 +1171,102 @@ def assess_workspace_mutation_consistency(report_text, observed_commands=None): } +# ── Already-landed report wording (Issue #298) ─────────────────────────────── +# +# An already-landed PR is reconciliation-only: it must never be called +# eligible, its head SHA must never be called reviewed, and the legacy +# eligible/reviewed/scratch-worktree handoff fields must not appear. +# Reviewed-head claims in any report require validation + diff review to +# have actually passed. + +ALREADY_LANDED_FORBIDDEN_PHRASES = ( + "oldest eligible pr", + "next eligible pr", + "pinned reviewed head", + "scratch worktree used", +) + +ALREADY_LANDED_ELIGIBILITY_CLASS = "ALREADY_LANDED_RECONCILE_REQUIRED" + + +def assess_already_landed_report_wording(report_text, *, already_landed, + review_validated=False): + """#298: reject eligible/reviewed wording for already-landed PRs. + + *already_landed* states whether the already-landed gate fired for the + selected PR. *review_validated* states whether validation and diff + review actually passed this session; only then may a reviewed head + SHA be populated in a non-already-landed report. + + Returns {'complete', 'downgraded', 'reasons'}; fails closed. + """ + text_lower = (report_text or "").lower() + fields = _report_labeled_fields(report_text) + reasons = [] + + reviewed_head = fields.get("reviewed head sha") + reviewed_head_populated = bool( + reviewed_head and not reviewed_head.strip().lower().startswith("none") + ) + pinned_head_present = "pinned reviewed head" in fields + + if already_landed: + for phrase in ALREADY_LANDED_FORBIDDEN_PHRASES: + if phrase in text_lower: + reasons.append( + f"already-landed report must not use '{phrase}'; the PR " + "is reconciliation-only, not review/merge eligible" + ) + + eligibility = fields.get("eligibility class", "") + if ALREADY_LANDED_ELIGIBILITY_CLASS not in eligibility.upper(): + reasons.append( + "already-landed report missing 'Eligibility class: " + f"{ALREADY_LANDED_ELIGIBILITY_CLASS}'" + ) + + for required in ("oldest open pr requiring action", + "candidate head sha"): + if required not in fields: + reasons.append( + f"already-landed report missing '{required}' field" + ) + + if "reviewed head sha" not in fields: + reasons.append( + "already-landed report must state 'Reviewed head SHA: none'" + ) + elif reviewed_head_populated: + reasons.append( + "already-landed report must not claim a reviewed head SHA; " + "no validation or diff review passed for this PR" + ) + + worktree_used = fields.get("review worktree used", "") + if worktree_used.strip().lower() not in ("false", "no"): + reasons.append( + "already-landed report must state 'Review worktree used: " + "false'" + ) + elif not review_validated: + if pinned_head_present: + reasons.append( + "'Pinned reviewed head' populated before validation and " + "diff review passed" + ) + if reviewed_head_populated: + reasons.append( + "reviewed head SHA populated before validation and diff " + "review passed" + ) + + return { + "complete": not reasons, + "downgraded": bool(reasons), + "reasons": reasons, + } + + def assess_role_boundary(proof=None, *, task_role=None, namespaces_used=None, justification=None): """Assess reviewer/author role separation for blind queue workflows. diff --git a/tests/test_already_landed_report_wording.py b/tests/test_already_landed_report_wording.py new file mode 100644 index 0000000..ac0379f --- /dev/null +++ b/tests/test_already_landed_report_wording.py @@ -0,0 +1,151 @@ +"""Tests for already-landed final-report wording validator (#298). + +An already-landed PR is reconciliation-only, never review/merge +eligible, and a head SHA may not be called reviewed unless validation +and diff review actually passed. Final reports and controller handoffs +must use the reconciliation wording, not the legacy eligible/reviewed +fields. +""" +import sys +import unittest +from pathlib import Path + +sys.path.insert(0, str(Path(__file__).resolve().parent.parent)) + +from review_proofs import assess_already_landed_report_wording # noqa: E402 + + +GOOD_ALREADY_LANDED = ( + "Controller Handoff\n" + "- Oldest open PR requiring action: PR #278\n" + "- Eligibility class: ALREADY_LANDED_RECONCILE_REQUIRED\n" + "- Candidate head SHA: 2a544b7\n" + "- Reviewed head SHA: none\n" + "- Review worktree used: false\n" +) + + +class TestAlreadyLandedWording(unittest.TestCase): + def test_correct_reconciliation_wording_passes(self): + result = assess_already_landed_report_wording( + GOOD_ALREADY_LANDED, already_landed=True + ) + self.assertTrue(result["complete"]) + self.assertFalse(result["downgraded"]) + self.assertEqual(result["reasons"], []) + + def test_oldest_eligible_wording_fails(self): + report = GOOD_ALREADY_LANDED + "- oldest eligible PR: PR #278\n" + result = assess_already_landed_report_wording( + report, already_landed=True + ) + self.assertFalse(result["complete"]) + self.assertTrue( + any("oldest eligible pr" in r.lower() for r in result["reasons"]) + ) + + def test_next_eligible_wording_fails(self): + report = GOOD_ALREADY_LANDED + "- next eligible PR: PR #278\n" + result = assess_already_landed_report_wording( + report, already_landed=True + ) + self.assertFalse(result["complete"]) + + def test_pinned_reviewed_head_fails(self): + report = GOOD_ALREADY_LANDED + "- Pinned reviewed head: 2a544b7\n" + result = assess_already_landed_report_wording( + report, already_landed=True + ) + self.assertFalse(result["complete"]) + self.assertTrue( + any("pinned reviewed head" in r.lower() for r in result["reasons"]) + ) + + def test_scratch_worktree_field_fails(self): + report = GOOD_ALREADY_LANDED + "- Scratch worktree used: False\n" + result = assess_already_landed_report_wording( + report, already_landed=True + ) + self.assertFalse(result["complete"]) + + def test_missing_eligibility_class_fails(self): + report = ( + "Controller Handoff\n" + "- Oldest open PR requiring action: PR #278\n" + "- Candidate head SHA: 2a544b7\n" + "- Reviewed head SHA: none\n" + "- Review worktree used: false\n" + ) + result = assess_already_landed_report_wording( + report, already_landed=True + ) + self.assertFalse(result["complete"]) + self.assertTrue( + any("eligibility class" in r.lower() for r in result["reasons"]) + ) + + def test_populated_reviewed_head_sha_fails(self): + report = GOOD_ALREADY_LANDED.replace( + "- Reviewed head SHA: none\n", + "- Reviewed head SHA: 2a544b7\n", + ) + result = assess_already_landed_report_wording( + report, already_landed=True + ) + self.assertFalse(result["complete"]) + self.assertTrue( + any("reviewed head" in r.lower() for r in result["reasons"]) + ) + + +class TestNonAlreadyLandedReports(unittest.TestCase): + def test_normal_reviewed_report_passes(self): + report = ( + "- Selected PR: 281\n" + "- Pinned reviewed head: abc1234\n" + "- Review decision: approve\n" + ) + result = assess_already_landed_report_wording( + report, already_landed=False, review_validated=True + ) + self.assertTrue(result["complete"]) + + def test_blocked_infra_report_with_pinned_head_fails(self): + report = ( + "- Selected PR: 281\n" + "- Pinned reviewed head: abc1234\n" + "- Current status: blocked on infra_stop\n" + ) + result = assess_already_landed_report_wording( + report, already_landed=False, review_validated=False + ) + self.assertFalse(result["complete"]) + self.assertTrue( + any("before validation" in r.lower() for r in result["reasons"]) + ) + + def test_validation_failed_report_with_reviewed_sha_fails(self): + report = ( + "- Selected PR: 281\n" + "- Reviewed head SHA: abc1234\n" + "- Validation: full suite failed\n" + ) + result = assess_already_landed_report_wording( + report, already_landed=False, review_validated=False + ) + self.assertFalse(result["complete"]) + + def test_validation_failed_report_with_reviewed_none_passes(self): + report = ( + "- Selected PR: 281\n" + "- Reviewed head SHA: none\n" + "- Validation: full suite failed\n" + ) + result = assess_already_landed_report_wording( + report, already_landed=False, review_validated=False + ) + self.assertTrue(result["complete"]) + + +if __name__ == "__main__": + unittest.main()