feat: require live blocker proof before skipping earlier PRs (Closes #318)

Add assess_prior_blocker_skip_proof to validate reviewer reports document
live REQUEST_CHANGES blocker proof for skipped earlier open PRs, reject
stale-memory skips when head changed after blocker, and require
BLOCKER_STATUS_UNVERIFIED when review feedback cannot be fetched.

Co-Authored-By: Claude Opus 4.8 (1M context) <[email protected]>

review_proofs.py

@@ -3215,3 +3215,10 @@ def assess_git_ref_mutation_report(report_text, *, command_log=None):
         "ref_mutating_commands": ref_commands,
         "fetch_targets": fetch_targets,
     }
+
+
+def assess_prior_blocker_skip_proof(report_text, **kwargs):
+    """#318: require live blocker proof before skipping earlier open PRs."""
+    from reviewer_blocker_skip import assess_prior_blocker_skip_proof as _assess
+
+    return _assess(report_text, **kwargs)

reviewer_blocker_skip.py

@@ -0,0 +1,164 @@
+"""Prior-blocker skip proof verifier for reviewer queue reports (#318)."""
+
+from __future__ import annotations
+
+import re
+from typing import Any
+
+_FULL_SHA = re.compile(r"\b[0-9a-f]{40}\b", re.IGNORECASE)
+_SHORT_SHA = re.compile(r"\b[0-9a-f]{7,39}\b", re.IGNORECASE)
+_PR_NUMBER_RE = re.compile(r"(?:\bPR\s*#?|#)(\d+)\b", re.IGNORECASE)
+_BLOCKING_DECISION_RE = re.compile(
+    r"(?:blocking review decision|review decision|blocking decision)\s*:\s*request[_ ]changes|"
+    r"request[_ ]changes",
+    re.IGNORECASE,
+)
+_BLOCKING_HEAD_RE = re.compile(
+    r"(?:blocking review head(?:\s+sha)?|blocker head(?:\s+sha)?|review head at blocker)",
+    re.IGNORECASE,
+)
+_HEAD_CHANGED_RE = re.compile(
+    r"(?:head (?:changed|unchanged)|head sha (?:changed|unchanged)|"
+    r"head changed (?:after|since) (?:the )?blocker|head unchanged since blocker)",
+    re.IGNORECASE,
+)
+_BLOCKER_REASON_RE = re.compile(
+    r"(?:reason (?:it )?remains blocked|remains blocked because|blocking category)",
+    re.IGNORECASE,
+)
+_LIVE_BLOCKER_PROOF_RE = re.compile(
+    r"(?:blocker revalidated live|live proof|gitea_get_pr_review_feedback|"
+    r"gitea_view_pr|review feedback fetched|current review state)",
+    re.IGNORECASE,
+)
+_BLOCKER_UNVERIFIED_RE = re.compile(r"\bBLOCKER_STATUS_UNVERIFIED\b")
+
+
+def _pr_section(text: str, pr_number: int) -> str:
+    lines = (text or "").splitlines()
+    chunks: list[str] = []
+    capture = False
+    token = f"#{pr_number}"
+    for line in lines:
+        lower = line.lower()
+        if token in lower or f"pr {pr_number}" in lower or f"pr#{pr_number}" in lower.replace(" ", ""):
+            capture = True
+            chunks.append(line)
+            continue
+        if capture:
+            if _PR_NUMBER_RE.search(line) and token not in line.lower():
+                break
+            if line.strip() == "" and len(chunks) > 3:
+                break
+            chunks.append(line)
+    if chunks:
+        return "\n".join(chunks)
+    return text or ""
+
+
+def _has_head_sha(text: str, head_sha: str | None) -> bool:
+    if not head_sha:
+        return bool(_FULL_SHA.search(text) or _SHORT_SHA.search(text))
+    head = head_sha.strip().lower()
+    if head in text.lower():
+        return True
+    if len(head) >= 7 and head[:7] in text.lower():
+        return True
+    return bool(_FULL_SHA.search(text))
+
+
+def assess_prior_blocker_skip_proof(
+    report_text: str,
+    *,
+    skipped_prs: list[dict] | None = None,
+) -> dict[str, Any]:
+    """Validate live blocker proof for skipped earlier open PRs (#318)."""
+    text = report_text or ""
+    reasons: list[str] = []
+    assessments: list[dict[str, Any]] = []
+
+    for entry in skipped_prs or []:
+        pr_number = entry.get("pr_number")
+        if pr_number is None:
+            reasons.append("skipped PR entry missing pr_number")
+            continue
+        pr_number = int(pr_number)
+        section = _pr_section(text, pr_number)
+        verified = entry.get("blocker_verified")
+        head_changed = entry.get("head_changed_since_blocker")
+        blocking_head = (entry.get("blocking_review_head_sha") or "").strip() or None
+        current_head = (entry.get("head_sha") or "").strip() or None
+        skip_reason = (entry.get("skip_reason") or entry.get("blocking_decision") or "").lower()
+
+        item_reasons: list[str] = []
+
+        if head_changed is True:
+            item_reasons.append(
+                f"PR #{pr_number} head changed after blocker; it cannot be skipped "
+                "based on stale REQUEST_CHANGES"
+            )
+
+        if f"#{pr_number}" not in text.lower() and f"pr {pr_number}" not in text.lower():
+            item_reasons.append(f"skipped PR #{pr_number} not documented in final report")
+
+        if "request_changes" in skip_reason or entry.get("blocking_decision") == "request_changes":
+            if verified is False:
+                if not _BLOCKER_UNVERIFIED_RE.search(section):
+                    item_reasons.append(
+                        f"PR #{pr_number} blocker proof unavailable; report must classify "
+                        "BLOCKER_STATUS_UNVERIFIED"
+                    )
+            else:
+                if not _BLOCKING_DECISION_RE.search(section):
+                    item_reasons.append(
+                        f"PR #{pr_number} skip missing blocking review decision proof"
+                    )
+                if not _has_head_sha(section, current_head):
+                    item_reasons.append(
+                        f"PR #{pr_number} skip missing current head SHA"
+                    )
+                if blocking_head and not _BLOCKING_HEAD_RE.search(section):
+                    if blocking_head.lower() not in section.lower():
+                        item_reasons.append(
+                            f"PR #{pr_number} skip missing blocking review head SHA"
+                        )
+                if head_changed is False and not _HEAD_CHANGED_RE.search(section):
+                    item_reasons.append(
+                        f"PR #{pr_number} skip missing head-changed-since-blocker proof"
+                    )
+                if not _BLOCKER_REASON_RE.search(section):
+                    item_reasons.append(
+                        f"PR #{pr_number} skip missing reason-it-remains-blocked"
+                    )
+                if not _LIVE_BLOCKER_PROOF_RE.search(section):
+                    item_reasons.append(
+                        f"PR #{pr_number} skip missing live blocker proof for this session"
+                    )
+
+        proven = not item_reasons
+        assessments.append({
+            "pr_number": pr_number,
+            "proven": proven,
+            "classification": (
+                "BLOCKER_STATUS_UNVERIFIED"
+                if verified is False
+                else "BLOCKED_SKIPPED"
+            ),
+            "reasons": item_reasons,
+        })
+        reasons.extend(item_reasons)
+
+    proven = not reasons
+    return {
+        "proven": proven,
+        "block": not proven,
+        "downgraded": False,
+        "assessments": assessments,
+        "reasons": reasons,
+        "safe_next_action": (
+            "fetch live review feedback and document blocker proof for each skipped PR, "
+            "or classify BLOCKER_STATUS_UNVERIFIED"
+            if reasons
+            else "proceed"
+        ),
+    }

tests/test_llm_workflow_split.py

@@ -122,3 +122,9 @@ def test_create_issue_final_report_verifier_exported():
     from review_proofs import assess_create_issue_final_report
 
     assert callable(assess_create_issue_final_report)
+
+
+def test_prior_blocker_skip_verifier_exported():
+    from review_proofs import assess_prior_blocker_skip_proof
+
+    assert callable(assess_prior_blocker_skip_proof)

tests/test_reviewer_blocker_skip.py

@@ -0,0 +1,139 @@
+"""Tests for prior-blocker skip proof verifier (#318)."""
+import sys
+import unittest
+from pathlib import Path
+
+sys.path.insert(0, str(Path(__file__).resolve().parent.parent))
+
+from reviewer_blocker_skip import assess_prior_blocker_skip_proof  # noqa: E402
+
+HEAD_CURRENT = "0fdc8f582026b72a229d59a172c0a63ac4aaeaf9"
+HEAD_BLOCKER = "1111111111111111111111111111111111111111"
+
+
+def _good_blocker_skip_report(pr_number: int) -> str:
+    return "\n".join([
+        f"Skipped PR #{pr_number} due to prior REQUEST_CHANGES.",
+        f"- PR number: #{pr_number}",
+        f"- Current head SHA: {HEAD_CURRENT}",
+        "- Blocking review decision: request_changes",
+        f"- Blocking review head SHA: {HEAD_CURRENT}",
+        "- Head unchanged since blocker",
+        "- Reason remains blocked: unresolved review feedback at same head",
+        "- Blocker revalidated live in this session via gitea_get_pr_review_feedback",
+    ])
+
+
+class TestPriorBlockerSkipProof(unittest.TestCase):
+    def test_no_skips_passes(self):
+        result = assess_prior_blocker_skip_proof(
+            "Selected PR #281 for review.",
+            skipped_prs=[],
+        )
+        self.assertTrue(result["proven"])
+
+    def test_blocked_unchanged_with_live_proof_passes(self):
+        report = _good_blocker_skip_report(280)
+        result = assess_prior_blocker_skip_proof(
+            report,
+            skipped_prs=[{
+                "pr_number": 280,
+                "head_sha": HEAD_CURRENT,
+                "blocking_decision": "request_changes",
+                "blocking_review_head_sha": HEAD_CURRENT,
+                "head_changed_since_blocker": False,
+                "blocker_verified": True,
+            }],
+        )
+        self.assertTrue(result["proven"])
+
+    def test_head_changed_after_blocker_blocks_skip(self):
+        result = assess_prior_blocker_skip_proof(
+            _good_blocker_skip_report(280),
+            skipped_prs=[{
+                "pr_number": 280,
+                "head_sha": HEAD_CURRENT,
+                "blocking_decision": "request_changes",
+                "blocking_review_head_sha": HEAD_BLOCKER,
+                "head_changed_since_blocker": True,
+                "blocker_verified": True,
+            }],
+        )
+        self.assertFalse(result["proven"])
+        self.assertTrue(any("head changed" in r for r in result["reasons"]))
+
+    def test_missing_live_proof_blocks(self):
+        report = _good_blocker_skip_report(280).replace(
+            "- Blocker revalidated live in this session via gitea_get_pr_review_feedback",
+            "",
+        )
+        result = assess_prior_blocker_skip_proof(
+            report,
+            skipped_prs=[{
+                "pr_number": 280,
+                "head_sha": HEAD_CURRENT,
+                "blocking_decision": "request_changes",
+                "blocking_review_head_sha": HEAD_CURRENT,
+                "head_changed_since_blocker": False,
+                "blocker_verified": True,
+            }],
+        )
+        self.assertFalse(result["proven"])
+        self.assertTrue(any("live blocker proof" in r for r in result["reasons"]))
+
+    def test_blocker_unverified_classification_passes(self):
+        report = "\n".join([
+            "Skipped PR #280.",
+            "Classification: BLOCKER_STATUS_UNVERIFIED",
+            "Review feedback could not be fetched live in this session.",
+        ])
+        result = assess_prior_blocker_skip_proof(
+            report,
+            skipped_prs=[{
+                "pr_number": 280,
+                "blocking_decision": "request_changes",
+                "blocker_verified": False,
+            }],
+        )
+        self.assertTrue(result["proven"])
+
+    def test_unverified_without_classification_blocks(self):
+        result = assess_prior_blocker_skip_proof(
+            "Skipped PR #280 based on memory from last session.",
+            skipped_prs=[{
+                "pr_number": 280,
+                "blocking_decision": "request_changes",
+                "blocker_verified": False,
+            }],
+        )
+        self.assertFalse(result["proven"])
+        self.assertTrue(any("BLOCKER_STATUS_UNVERIFIED" in r for r in result["reasons"]))
+
+    def test_missing_blocking_head_sha_blocks(self):
+        report = _good_blocker_skip_report(280).replace(
+            f"- Blocking review head SHA: {HEAD_CURRENT}",
+            "",
+        )
+        result = assess_prior_blocker_skip_proof(
+            report,
+            skipped_prs=[{
+                "pr_number": 280,
+                "head_sha": HEAD_CURRENT,
+                "blocking_decision": "request_changes",
+                "blocking_review_head_sha": HEAD_CURRENT,
+                "head_changed_since_blocker": False,
+                "blocker_verified": True,
+            }],
+        )
+        self.assertFalse(result["proven"])
+
+
+class TestExport(unittest.TestCase):
+    def test_review_proofs_reexport(self):
+        from review_proofs import assess_prior_blocker_skip_proof as exported
+
+        self.assertTrue(callable(exported))
+
+
+if __name__ == "__main__":
+    unittest.main()