From cf1cedf5e33679bdf1ab94f9af81eb19d1ae7e8d Mon Sep 17 00:00:00 2001 From: Jason Walker <913443@dadeschools.net> Date: Tue, 7 Jul 2026 05:08:53 -0400 Subject: [PATCH] feat: require conflict proof for skipped non-mergeable PRs (Closes #322) Add assess_non_mergeable_skip_proof to validate reviewer reports document head SHA, mergeability result, conflict proof command, conflicting files, and head-change status for each skipped non-mergeable PR, or classify MERGEABILITY_UNVERIFIED when proof cannot be retrieved. Co-Authored-By: Claude Opus 4.8 (1M context) --- review_proofs.py | 7 + reviewer_mergeability_skip.py | 159 +++++++++++++++++++++++ tests/test_llm_workflow_split.py | 8 +- tests/test_reviewer_mergeability_skip.py | 150 +++++++++++++++++++++ 4 files changed, 323 insertions(+), 1 deletion(-) create mode 100644 reviewer_mergeability_skip.py create mode 100644 tests/test_reviewer_mergeability_skip.py diff --git a/review_proofs.py b/review_proofs.py index 9cd7e8c..97adafd 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_non_mergeable_skip_proof(report_text, **kwargs): + """#322: require conflict proof when skipping non-mergeable PRs.""" + from reviewer_mergeability_skip import assess_non_mergeable_skip_proof as _assess + + return _assess(report_text, **kwargs) diff --git a/reviewer_mergeability_skip.py b/reviewer_mergeability_skip.py new file mode 100644 index 0000000..565de40 --- /dev/null +++ b/reviewer_mergeability_skip.py @@ -0,0 +1,159 @@ +"""Non-mergeable PR skip conflict-proof verifier for reviewer reports (#322).""" + +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) +_MERGEABILITY_FALSE_RE = re.compile( + r"(?:mergeable\s*:\s*false|not mergeable|non[- ]mergeable|mergeability\s*:\s*false|" + r"mergeability result\s*:\s*false)", + re.IGNORECASE, +) +_CONFLICT_PROOF_RE = re.compile( + r"(?:merge-tree|git merge-tree|gitea_view_pr|gitea_check_pr_eligibility|" + r"conflict proof|mergeability tool|merge simulation|conflicting files?)", + re.IGNORECASE, +) +_CONFLICTING_FILES_RE = re.compile( + r"(?:conflicting files?|conflict files?)\s*:\s*(.+)$", + re.IGNORECASE | re.MULTILINE, +) +_HEAD_CHANGED_RE = re.compile( + r"(?:head (?:changed|unchanged)|head sha (?:changed|unchanged)|" + r"head changed since|head unchanged since)", + re.IGNORECASE, +) +_MERGEABILITY_UNVERIFIED_RE = re.compile( + r"\bMERGEABILITY_UNVERIFIED\b", +) + + +def _pr_section(text: str, pr_number: int) -> str: + """Return report lines likely describing one skipped PR.""" + 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_non_mergeable_skip_proof( + report_text: str, + *, + skipped_prs: list[dict] | None = None, +) -> dict[str, Any]: + """Validate skipped non-mergeable PR documentation in reviewer reports (#322).""" + 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) + mergeable = entry.get("mergeable") + verified = entry.get("mergeability_verified") + classification = (entry.get("classification") or "").strip().upper() + head_sha = (entry.get("head_sha") or "").strip() or None + + item_reasons: list[str] = [] + + 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 mergeable is False or verified is False: + if not _MERGEABILITY_UNVERIFIED_RE.search(section) and verified is False: + if "MERGEABILITY_UNVERIFIED" not in classification: + item_reasons.append( + f"PR #{pr_number} conflict proof unavailable; report must classify " + "MERGEABILITY_UNVERIFIED" + ) + elif verified is not False: + if not _MERGEABILITY_FALSE_RE.search(section): + item_reasons.append( + f"PR #{pr_number} skip missing mergeability:false proof" + ) + if not _has_head_sha(section, head_sha): + item_reasons.append( + f"PR #{pr_number} skip missing current head SHA" + ) + if not _CONFLICT_PROOF_RE.search(section): + item_reasons.append( + f"PR #{pr_number} skip missing conflict proof command/tool" + ) + if entry.get("head_changed_since_prior_blocker") is not None: + if not _HEAD_CHANGED_RE.search(section): + item_reasons.append( + f"PR #{pr_number} skip missing head-changed-since-blocker proof" + ) + + if ( + entry.get("conflicting_files") + and not _CONFLICTING_FILES_RE.search(section) + and not any( + path.lower() in section.lower() + for path in (entry.get("conflicting_files") or []) + ) + ): + item_reasons.append( + f"PR #{pr_number} has conflicting files in session proof but " + "report omits file-level conflict proof" + ) + + proven = not item_reasons + assessments.append({ + "pr_number": pr_number, + "proven": proven, + "classification": classification or ( + "MERGEABILITY_UNVERIFIED" if verified is False else "NON_MERGEABLE_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": ( + "document each skipped non-mergeable PR with head SHA, mergeability result, " + "conflict proof command, conflicting files, and head-change status" + 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..7e0fe23 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_non_mergeable_skip_verifier_exported(): + from review_proofs import assess_non_mergeable_skip_proof + + assert callable(assess_non_mergeable_skip_proof) \ No newline at end of file diff --git a/tests/test_reviewer_mergeability_skip.py b/tests/test_reviewer_mergeability_skip.py new file mode 100644 index 0000000..98fcc2d --- /dev/null +++ b/tests/test_reviewer_mergeability_skip.py @@ -0,0 +1,150 @@ +"""Tests for non-mergeable skip conflict-proof verifier (#322).""" +import sys +import unittest +from pathlib import Path + +sys.path.insert(0, str(Path(__file__).resolve().parent.parent)) + +from reviewer_mergeability_skip import assess_non_mergeable_skip_proof # noqa: E402 + +HEAD_A = "0fdc8f582026b72a229d59a172c0a63ac4aaeaf9" +HEAD_B = "1111111111111111111111111111111111111111" + + +def _good_skip_report(pr_number: int, head: str, files: str = "review_proofs.py") -> str: + return "\n".join([ + f"Skipped PR #{pr_number} (non-mergeable).", + f"- PR number: #{pr_number}", + f"- Current head SHA: {head}", + "- Mergeability result: false", + "- Conflict proof command: git merge-tree $(git merge-base prgs/master prgs/feat) prgs/master prgs/feat", + f"- Conflicting files: {files}", + "- Head unchanged since prior blocker", + "- Blocking category: merge conflict", + ]) + + +class TestNonMergeableSkipProof(unittest.TestCase): + def test_no_skips_passes(self): + report = "Selected PR #203 for review. No earlier PRs skipped." + result = assess_non_mergeable_skip_proof(report, skipped_prs=[]) + self.assertTrue(result["proven"]) + self.assertFalse(result["block"]) + + def test_documented_non_mergeable_skip_passes(self): + report = _good_skip_report(282, HEAD_A) + result = assess_non_mergeable_skip_proof( + report, + skipped_prs=[{ + "pr_number": 282, + "mergeable": False, + "head_sha": HEAD_A, + "mergeability_verified": True, + "conflicting_files": ["review_proofs.py"], + "head_changed_since_prior_blocker": False, + }], + ) + self.assertTrue(result["proven"]) + + def test_skip_without_conflict_proof_blocks(self): + report = "\n".join([ + "Skipped PR #284 because it was not mergeable.", + f"Head SHA: {HEAD_B}", + ]) + result = assess_non_mergeable_skip_proof( + report, + skipped_prs=[{ + "pr_number": 284, + "mergeable": False, + "head_sha": HEAD_B, + "mergeability_verified": True, + }], + ) + self.assertFalse(result["proven"]) + self.assertTrue(result["block"]) + self.assertTrue(any("conflict proof" in r for r in result["reasons"])) + + def test_missing_head_sha_blocks(self): + report = _good_skip_report(282, HEAD_A).replace(HEAD_A, "") + result = assess_non_mergeable_skip_proof( + report, + skipped_prs=[{ + "pr_number": 282, + "mergeable": False, + "head_sha": HEAD_A, + "mergeability_verified": True, + }], + ) + self.assertFalse(result["proven"]) + + def test_mergeability_unverified_classification_passes(self): + report = "\n".join([ + "Skipped PR #282.", + "Classification: MERGEABILITY_UNVERIFIED", + "Conflict proof could not be retrieved in this session.", + ]) + result = assess_non_mergeable_skip_proof( + report, + skipped_prs=[{ + "pr_number": 282, + "mergeable": False, + "mergeability_verified": False, + }], + ) + self.assertTrue(result["proven"]) + + def test_unverified_without_classification_blocks(self): + report = "Skipped PR #282 due to conflicts." + result = assess_non_mergeable_skip_proof( + report, + skipped_prs=[{ + "pr_number": 282, + "mergeable": False, + "mergeability_verified": False, + }], + ) + self.assertFalse(result["proven"]) + self.assertTrue(any("MERGEABILITY_UNVERIFIED" in r for r in result["reasons"])) + + def test_non_mergeable_without_file_details_still_passes_with_command(self): + report = _good_skip_report(284, HEAD_B, files="not available") + result = assess_non_mergeable_skip_proof( + report, + skipped_prs=[{ + "pr_number": 284, + "mergeable": False, + "head_sha": HEAD_B, + "mergeability_verified": True, + "conflicting_files": [], + }], + ) + self.assertTrue(result["proven"]) + + def test_stale_head_requires_head_change_field(self): + report = _good_skip_report(282, HEAD_A).replace( + "- Head unchanged since prior blocker", + "", + ) + result = assess_non_mergeable_skip_proof( + report, + skipped_prs=[{ + "pr_number": 282, + "mergeable": False, + "head_sha": HEAD_A, + "mergeability_verified": True, + "head_changed_since_prior_blocker": True, + }], + ) + self.assertFalse(result["proven"]) + self.assertTrue(any("head-changed" in r for r in result["reasons"])) + + +class TestExport(unittest.TestCase): + def test_review_proofs_reexport(self): + from review_proofs import assess_non_mergeable_skip_proof as exported + + self.assertTrue(callable(exported)) + + +if __name__ == "__main__": + unittest.main() \ No newline at end of file -- 2.43.7