From 56661cd0a725abc66d34e8cbcb7afc8f46f2f5de Mon Sep 17 00:00:00 2001 From: Jason Walker <913443@dadeschools.net> Date: Tue, 7 Jul 2026 05:12:01 -0400 Subject: [PATCH] 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) --- review_proofs.py | 7 ++ reviewer_blocker_skip.py | 164 ++++++++++++++++++++++++++++ tests/test_llm_workflow_split.py | 8 +- tests/test_reviewer_blocker_skip.py | 139 +++++++++++++++++++++++ 4 files changed, 317 insertions(+), 1 deletion(-) create mode 100644 reviewer_blocker_skip.py create mode 100644 tests/test_reviewer_blocker_skip.py diff --git a/review_proofs.py b/review_proofs.py index 9cd7e8c..4a6bb36 100644 --- a/review_proofs.py +++ b/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) diff --git a/reviewer_blocker_skip.py b/reviewer_blocker_skip.py new file mode 100644 index 0000000..a95d368 --- /dev/null +++ b/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" + ), + } \ No newline at end of file diff --git a/tests/test_llm_workflow_split.py b/tests/test_llm_workflow_split.py index f8d3b04..38602c7 100644 --- a/tests/test_llm_workflow_split.py +++ b/tests/test_llm_workflow_split.py @@ -121,4 +121,10 @@ def test_start_issue_template_references_work_issue_workflow(): def test_create_issue_final_report_verifier_exported(): from review_proofs import assess_create_issue_final_report - assert callable(assess_create_issue_final_report) \ No newline at end of file + 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) \ No newline at end of file diff --git a/tests/test_reviewer_blocker_skip.py b/tests/test_reviewer_blocker_skip.py new file mode 100644 index 0000000..b037ede --- /dev/null +++ b/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() \ No newline at end of file