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) <[email protected]>

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.