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.

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()