From efb0a2d076e144bd5515eb56487d25efb3ac5edc Mon Sep 17 00:00:00 2001 From: Jason Walker <913443@dadeschools.net> Date: Tue, 7 Jul 2026 02:53:25 -0400 Subject: [PATCH] Add REQUEST_CHANGES override proof helper before approval (#326) Introduce assess_request_changes_approval_proof in review_proofs.py so review workflows can fail closed when approving an unchanged head after a prior REQUEST_CHANGES unless explicit override proof is provided. Co-Authored-By: Claude Opus 4.8 (1M context) --- review_proofs.py | 139 +++++++++++++++++++ tests/test_request_changes_approval_proof.py | 104 ++++++++++++++ 2 files changed, 243 insertions(+) create mode 100644 tests/test_request_changes_approval_proof.py diff --git a/review_proofs.py b/review_proofs.py index c539a0b..72be5fb 100644 --- a/review_proofs.py +++ b/review_proofs.py @@ -704,6 +704,145 @@ def assess_live_state_recheck(recheck): return {"proven": proven, "block": not proven, "reasons": reasons} +_REQUEST_CHANGES_OVERRIDE_REASONS = frozenset({ + "incorrect_blocker", + "wrong_validation_environment", + "resolved_externally", +}) + + +def _blocking_request_changes_reviews(feedback: dict) -> list[dict]: + """Return undismissed REQUEST_CHANGES reviews (latest verdict per reviewer).""" + latest_by_reviewer: dict[str, dict] = {} + for entry in feedback.get("reviews") or []: + verdict = (entry.get("verdict") or "").upper() + reviewer = (entry.get("reviewer") or "").strip() + if verdict not in ("APPROVED", "REQUEST_CHANGES") or not reviewer: + continue + latest_by_reviewer[reviewer] = entry + return [ + entry for entry in latest_by_reviewer.values() + if entry.get("verdict") == "REQUEST_CHANGES" and not entry.get("dismissed") + ] + + +def _primary_blocking_review(blockers: list[dict]) -> dict | None: + if not blockers: + return None + return sorted( + blockers, + key=lambda entry: (entry.get("submitted_at") or "", entry.get("reviewer") or ""), + )[-1] + + +def assess_request_changes_approval_proof( + feedback: dict | None, + *, + override_reason: str | None = None, + override_explanation: str | None = None, + report_text: str = "", +) -> dict: + """#326: prove approval is safe when prior REQUEST_CHANGES exists on same head. + + Before approving, workflows must fetch ``gitea_get_pr_review_feedback`` and + pass the result here. When a prior undismissed REQUEST_CHANGES targets the + current head, approval requires explicit override proof; otherwise fail closed. + """ + if not feedback or feedback.get("success") is not True: + return { + "approve_allowed": False, + "block": True, + "reasons": ["PR review feedback missing or unreadable; fail closed"], + "blocking_review": None, + "head_changed_since_blocker": None, + "override_required": None, + "override_proof": None, + } + + blockers = _blocking_request_changes_reviews(feedback) + primary = _primary_blocking_review(blockers) + current_head = (feedback.get("current_head_sha") or "").strip().lower() + + if not blockers: + return { + "approve_allowed": True, + "block": False, + "reasons": [], + "blocking_review": None, + "head_changed_since_blocker": False, + "override_required": False, + "override_proof": None, + } + + blocking_head = (primary.get("reviewed_head_sha") or "").strip().lower() + head_changed = bool( + feedback.get("author_pushed_after_request_changes") + or (blocking_head and current_head and blocking_head != current_head) + ) + + blocker_report = { + "blocking_reviewer": primary.get("reviewer"), + "blocking_review_timestamp": primary.get("submitted_at"), + "blocking_head_sha": primary.get("reviewed_head_sha"), + "current_head_sha": feedback.get("current_head_sha"), + "blocker_text": (primary.get("body") or "").strip(), + "head_changed_since_blocker": head_changed, + } + + if head_changed: + return { + "approve_allowed": True, + "block": False, + "reasons": [], + "blocking_review": blocker_report, + "head_changed_since_blocker": True, + "override_required": False, + "override_proof": None, + } + + reason = (override_reason or "").strip().lower() + explanation = (override_explanation or "").strip() + report_lower = (report_text or "").lower() + blocker_text = blocker_report["blocker_text"] + reasons: list[str] = [] + + if reason not in _REQUEST_CHANGES_OVERRIDE_REASONS: + reasons.append( + "unchanged head after REQUEST_CHANGES requires override_reason " + f"in {sorted(_REQUEST_CHANGES_OVERRIDE_REASONS)}" + ) + if not explanation: + reasons.append( + "unchanged head after REQUEST_CHANGES requires override_explanation" + ) + if blocker_text and blocker_text.lower() not in report_lower: + reasons.append( + "report must include the blocking REQUEST_CHANGES body text" + ) + if reason and reason.replace("_", " ") not in report_lower and reason not in report_lower: + reasons.append( + "report must state the override reason for unchanged-head approval" + ) + + override_proof = { + "override_reason": reason or None, + "override_explanation": explanation or None, + "blocker_text_in_report": bool( + blocker_text and blocker_text.lower() in report_lower + ), + } + approve_allowed = not reasons + return { + "approve_allowed": approve_allowed, + "block": not approve_allowed, + "reasons": reasons, + "blocking_review": blocker_report, + "head_changed_since_blocker": False, + "override_required": True, + "override_proof": override_proof, + } + + 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_request_changes_approval_proof.py b/tests/test_request_changes_approval_proof.py new file mode 100644 index 0000000..95ed42e --- /dev/null +++ b/tests/test_request_changes_approval_proof.py @@ -0,0 +1,104 @@ +"""Tests for REQUEST_CHANGES override proof before approval (#326).""" +import sys +import unittest +from pathlib import Path + +sys.path.insert(0, str(Path(__file__).resolve().parent.parent)) + +from review_proofs import assess_request_changes_approval_proof # noqa: E402 + +HEAD_A = "aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa" +HEAD_B = "bbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbb" + + +def _feedback(**overrides): + base = { + "success": True, + "current_head_sha": HEAD_A, + "has_blocking_change_requests": False, + "author_pushed_after_request_changes": False, + "reviews": [], + } + base.update(overrides) + return base + + +def _blocking_review(**overrides): + entry = { + "reviewer": "reviewer-a", + "verdict": "REQUEST_CHANGES", + "body": "Tests fail on pinned head; fix test_commit_files_gate.", + "submitted_at": "2026-07-07T01:00:00-05:00", + "reviewed_head_sha": HEAD_A, + "dismissed": False, + "stale": False, + } + entry.update(overrides) + return entry + + +class TestRequestChangesApprovalProof(unittest.TestCase): + def test_missing_feedback_blocks_approval(self): + result = assess_request_changes_approval_proof(None) + self.assertFalse(result["approve_allowed"]) + self.assertTrue(result["block"]) + + def test_no_prior_request_changes_allows_approval(self): + feedback = _feedback( + reviews=[{ + "reviewer": "sysadmin", + "verdict": "APPROVED", + "body": "LGTM", + "submitted_at": "2026-07-07T02:00:00-05:00", + "reviewed_head_sha": HEAD_A, + "dismissed": False, + }] + ) + result = assess_request_changes_approval_proof(feedback) + self.assertTrue(result["approve_allowed"]) + self.assertIsNone(result["blocking_review"]) + + def test_changed_head_since_blocker_allows_approval(self): + feedback = _feedback( + current_head_sha=HEAD_B, + author_pushed_after_request_changes=True, + has_blocking_change_requests=True, + reviews=[_blocking_review(reviewed_head_sha=HEAD_A)], + ) + result = assess_request_changes_approval_proof(feedback) + self.assertTrue(result["approve_allowed"]) + self.assertTrue(result["head_changed_since_blocker"]) + + def test_unchanged_head_without_override_blocks_approval(self): + feedback = _feedback( + has_blocking_change_requests=True, + reviews=[_blocking_review()], + ) + result = assess_request_changes_approval_proof(feedback) + self.assertFalse(result["approve_allowed"]) + self.assertIn("override_reason", " ".join(result["reasons"])) + + def test_unchanged_head_with_valid_override_allows_approval(self): + blocker = _blocking_review() + feedback = _feedback( + has_blocking_change_requests=True, + reviews=[blocker], + ) + report = ( + "Blocker text: Tests fail on pinned head; fix test_commit_files_gate.\n" + "Override: wrong_validation_environment — CI used stale worktree." + ) + result = assess_request_changes_approval_proof( + feedback, + override_reason="wrong_validation_environment", + override_explanation="CI used stale worktree; local rerun passed.", + report_text=report, + ) + self.assertTrue(result["approve_allowed"]) + self.assertEqual( + result["blocking_review"]["blocking_reviewer"], "reviewer-a" + ) + + +if __name__ == "__main__": + unittest.main() \ No newline at end of file