From c1d85b621a45438e3a95575a377077bd701fffcd Mon Sep 17 00:00:00 2001 From: Jason Walker <913443@dadeschools.net> Date: Tue, 7 Jul 2026 10:24:54 -0400 Subject: [PATCH] feat: require proof-backed claims in reviewer handoff reports (Closes #395) --- review_proofs.py | 23 ++ reviewer_proof_backed_handoff.py | 217 ++++++++++++++++++ .../schemas/review-merge-final-report.md | 21 +- tests/test_llm_workflow_split.py | 6 + tests/test_reviewer_proof_backed_handoff.py | 117 ++++++++++ 5 files changed, 383 insertions(+), 1 deletion(-) create mode 100644 reviewer_proof_backed_handoff.py create mode 100644 tests/test_reviewer_proof_backed_handoff.py diff --git a/review_proofs.py b/review_proofs.py index 332f866..cdb3cf0 100644 --- a/review_proofs.py +++ b/review_proofs.py @@ -1725,6 +1725,11 @@ def build_final_report(checkout_proof, inventory, validation, contamination, if report_text else {"proven": True, "block": False, "reasons": []} ) + proof_backed_handoff = ( + assess_proof_backed_handoff_report(report_text) + if report_text and _PROOF_BACKED_HANDOFF_HINT.search(report_text) + else {"proven": True, "block": False, "reasons": []} + ) if baseline_validation is None and report_text: baseline_validation = assess_reviewer_baseline_validation_proof( report_text=report_text, @@ -1911,6 +1916,11 @@ def build_final_report(checkout_proof, inventory, validation, contamination, "reviewer baseline validation proof missing or failed (#325)" ) downgrade_reasons.extend(baseline_validation.get("reasons", [])) + if not proof_backed_handoff.get("proven"): + downgrade_reasons.append( + "proof-sensitive handoff claims lack command/tool evidence (#395)" + ) + downgrade_reasons.extend(proof_backed_handoff.get("reasons", [])) merge_allowed = ( identity_eligible @@ -4912,6 +4922,12 @@ _PR_QUEUE_CLEANUP_REPORT_HINT = re.compile( re.I, ) +_PROOF_BACKED_HANDOFF_HINT = re.compile( + r"task:\s*review-merge-pr|task mode:\s*review-merge-pr|" + r"review-merge-pr|controller handoff", + re.I, +) + def assess_pr_queue_cleanup_report(report_text: str | None) -> dict: """#390: validate PR-only cleanup final reports (one PR per run).""" @@ -5374,3 +5390,10 @@ def assess_worktree_ownership_report(report_text, **kwargs): from reviewer_worktree_ownership import assess_worktree_ownership_report as _assess return _assess(report_text, **kwargs) + + +def assess_proof_backed_handoff_report(report_text, **kwargs): + """#395: proof-sensitive review handoff claims require explicit evidence.""" + from reviewer_proof_backed_handoff import assess_proof_backed_handoff_report as _assess + + return _assess(report_text, **kwargs) diff --git a/reviewer_proof_backed_handoff.py b/reviewer_proof_backed_handoff.py new file mode 100644 index 0000000..0cf0a79 --- /dev/null +++ b/reviewer_proof_backed_handoff.py @@ -0,0 +1,217 @@ +"""Proof-backed reviewer handoff claim verifier (#395).""" + +from __future__ import annotations + +import re +from typing import Any + +_PAGINATION_FINALITY = re.compile( + r"has_more\s*:\s*false|is_final_page\s*:\s*true|" + r"inventory_complete\s*:\s*true|pages_fetched|total_count\s*:", + re.I, +) +_PAGE_SIZE_ONLY = re.compile( + r"(?:less|fewer) than (?:the )?(?:default )?page[- ]?size|" + r"under (?:the )?50[- ]?(?:item )?limit|" + r"returned \d+ (?:open )?prs?.*less than 50", + re.I, +) +_INVENTORY_COMPLETE_CLAIM = re.compile( + r"\b(?:inventory (?:is )?complete|inventory exhaustive|" + r"complete (?:pr )?inventory)\b", + re.I, +) +_SKIP_CLAIM = re.compile( + r"\b(?:earlier prs? skipped|skipped (?:earlier )?pr|" + r"skip(?:ped)? pr #?\d+|non-mergeable|prior request.changes)\b", + re.I, +) +_CONFLICT_PROOF = re.compile( + r"merge simulation|git merge --no-commit|conflicting files|" + r"conflict proof|merge_exit|non-mergeable", + re.I, +) +_BASELINE_CLAIM = re.compile( + r"\b(?:baseline (?:validation|worktree|comparison)|" + r"same as master|pre-existing (?:on )?master|" + r"failure signatures match)\b", + re.I, +) +_BASELINE_PATH = re.compile(r"baseline worktree path\s*:\s*(\S+)", re.I) +_BASELINE_SHA = re.compile(r"baseline (?:target )?sha\s*:\s*([0-9a-f]{7,40})", re.I) +_BASELINE_DIRTY_BEFORE = re.compile( + r"baseline.*dirty before|dirty before.*baseline", re.I +) +_BASELINE_DIRTY_AFTER = re.compile( + r"baseline.*dirty after|dirty after.*baseline", re.I +) +_BASELINE_COMMAND = re.compile( + r"baseline.*(?:validation )?command|pytest.*baseline", re.I +) +_BASELINE_RESULT = re.compile( + r"baseline.*(?:validation )?result|baseline_exit|baseline failures", re.I +) +_MASTER_INTEGRATION = re.compile( + r"\b(?:merged master into|merge(?:d)? (?:remote[- ]tracking )?branch.*master|" + r"master integration|integrated master|rebase.*master)\b", + re.I, +) +_CLEANUP_CLAIM = re.compile( + r"\b(?:worktree(?:s)? (?:were )?cleaned|cleanup (?:result|mutations)|" + r"removed (?:session[- ]owned )?worktree|git worktree remove)\b", + re.I, +) +_WORKTREE_LIST = re.compile(r"git worktree list|worktree list proof", re.I) +_PROOF_SOURCE = re.compile( + r"proof source\s*:\s*(command|mcp metadata|prior blocker|not checked)", + re.I, +) +_HANDOFF_SECTION = re.compile(r"^##\s*Controller Handoff\s*$", re.I | re.M) + + +def _handoff_fields(report_text: str) -> dict[str, str]: + text = report_text or "" + match = _HANDOFF_SECTION.search(text) + if not match: + return {} + fields: dict[str, str] = {} + for line in text[match.end() :].splitlines(): + stripped = line.strip().lstrip("-*").strip() + if ":" not in stripped: + continue + key, value = stripped.split(":", 1) + fields[key.strip().lower()] = value.strip() + return fields + + +def _command_text(entry: Any) -> str: + if isinstance(entry, dict): + return str(entry.get("command") or entry.get("tool") or "").strip() + return str(entry or "").strip() + + +def _action_log_has(action_log: list | None, pattern: re.Pattern[str]) -> bool: + for entry in action_log or []: + blob = " ".join( + filter( + None, + [ + _command_text(entry), + str(entry.get("result") or "") if isinstance(entry, dict) else "", + str(entry.get("reason") or "") if isinstance(entry, dict) else "", + ], + ) + ) + if pattern.search(blob): + return True + return False + + +def assess_proof_backed_handoff_report( + report_text: str, + *, + action_log: list | None = None, + inventory_session: dict | None = None, + baseline_session: dict | None = None, + skip_session: list[dict] | None = None, +) -> dict[str, Any]: + """Require explicit command/tool evidence for proof-sensitive review claims (#395).""" + text = report_text or "" + fields = _handoff_fields(text) + reasons: list[str] = [] + inventory = dict(inventory_session or {}) + baseline = dict(baseline_session or {}) + skips = list(skip_session or []) + + pagination_field = fields.get("inventory pagination proof", "") + if _INVENTORY_COMPLETE_CLAIM.search(text) or _PAGE_SIZE_ONLY.search(text): + has_meta = bool( + inventory.get("inventory_complete") + or inventory.get("pagination_complete") + or _PAGINATION_FINALITY.search(pagination_field) + or _PAGINATION_FINALITY.search(text) + or _action_log_has(action_log, re.compile(r"gitea_list_prs", re.I)) + ) + if _PAGE_SIZE_ONLY.search(text) and not has_meta: + reasons.append( + "inventory completeness claimed from page-size assumption only; " + "require has_more=false, is_final_page=true, or inventory_complete=true" + ) + elif _INVENTORY_COMPLETE_CLAIM.search(text) and not has_meta: + reasons.append( + "inventory complete claim lacks explicit pagination metadata proof" + ) + + if _SKIP_CLAIM.search(text) or fields.get("earlier prs skipped", "").lower() not in { + "", + "none", + "n/a", + }: + skip_proof = bool( + skips + or _CONFLICT_PROOF.search(text) + or _action_log_has( + action_log, re.compile(r"git merge --no-commit|gitea_get_pr_review_feedback", re.I) + ) + ) + if not skip_proof: + reasons.append( + "earlier PR skip claim lacks command/tool conflict or blocker proof" + ) + + if _BASELINE_CLAIM.search(text) or fields.get("baseline worktree used", "").lower() == "true": + baseline_ok = bool( + baseline.get("complete") + or ( + _BASELINE_PATH.search(text) + and _BASELINE_SHA.search(text) + and (_BASELINE_DIRTY_BEFORE.search(text) or baseline.get("clean_before")) + and (_BASELINE_COMMAND.search(text) or baseline.get("command")) + and (_BASELINE_RESULT.search(text) or baseline.get("result")) + and (_BASELINE_DIRTY_AFTER.search(text) or baseline.get("clean_after")) + ) + ) + if not baseline_ok: + reasons.append( + "baseline validation claim missing worktree path, target SHA, " + "dirty-before/after status, command, or result proof" + ) + + if _MASTER_INTEGRATION.search(text): + if not _action_log_has( + action_log, + re.compile(r"git merge.*master|git rebase.*master", re.I), + ) and not re.search(r"merge_exit\s*=\s*\d+|merge simulation", text, re.I): + reasons.append( + "master integration claim lacks exact merge/rebase command and result" + ) + + if _CLEANUP_CLAIM.search(text) or fields.get("cleanup mutations", "").lower() not in { + "", + "none", + "n/a", + }: + if not (_WORKTREE_LIST.search(text) or _action_log_has(action_log, _WORKTREE_LIST)): + reasons.append( + "cleanup claim lacks final git worktree list or equivalent proof" + ) + + if re.search(r"\blive proof\b", text, re.I) and not _PROOF_SOURCE.search(text): + if not action_log: + reasons.append( + "live proof wording requires proof source classification " + "(command, MCP metadata, prior blocker, or not checked)" + ) + + proven = not reasons + return { + "proven": proven, + "block": not proven, + "reasons": reasons, + "safe_next_action": ( + "cite explicit command/tool evidence or structured MCP pagination metadata " + "for each proof-sensitive claim" + if not proven + else "proceed" + ), + } \ No newline at end of file diff --git a/skills/llm-project-workflow/schemas/review-merge-final-report.md b/skills/llm-project-workflow/schemas/review-merge-final-report.md index 08c094b..9eb3592 100644 --- a/skills/llm-project-workflow/schemas/review-merge-final-report.md +++ b/skills/llm-project-workflow/schemas/review-merge-final-report.md @@ -91,4 +91,23 @@ Verifier: `review_proofs.assess_queue_status_report()`. Narrative final report and controller handoff must agree on eligibility class, candidate/reviewed head SHA, mutation state, worktree usage, review decision, -terminal review mutation, merge result, and linked issue status. \ No newline at end of file +terminal review mutation, merge result, and linked issue status. + +### Proof-backed claims (#395) + +Proof-sensitive claims must cite explicit command/tool evidence in the report +or structured MCP metadata — not narrative alone: + +- **Inventory complete:** `has_more=false`, `is_final_page=true`, + `inventory_complete=true`, and/or `total_count` from `gitea_list_prs`. +- **Skipped earlier PRs:** merge-simulation command output, conflict file list, + or live `gitea_get_pr_review_feedback` / mergeability fields. +- **Baseline validation:** baseline worktree path, baseline target SHA, + dirty-before/after, exact command, exact result. +- **Master integration:** exact `git merge` / `git rebase` command and exit status. +- **Cleanup:** final `git worktree list` (or remove commands) proving session + worktrees were removed. + +When a claim relies on prior-session blocker state or MCP metadata only, label +the proof source explicitly (`command`, `MCP metadata`, `prior blocker`, +`not checked`). Do not use `live proof` without that classification. \ No newline at end of file diff --git a/tests/test_llm_workflow_split.py b/tests/test_llm_workflow_split.py index 255b8f0..69099d3 100644 --- a/tests/test_llm_workflow_split.py +++ b/tests/test_llm_workflow_split.py @@ -219,6 +219,12 @@ def test_inventory_worktree_verifier_exported(): assert callable(assess_inventory_worktree_report) +def test_proof_backed_handoff_verifier_exported(): + from review_proofs import assess_proof_backed_handoff_report + + assert callable(assess_proof_backed_handoff_report) + + def test_infra_stop_handoff_verifier_exported(): from review_proofs import assess_infra_stop_handoff_report diff --git a/tests/test_reviewer_proof_backed_handoff.py b/tests/test_reviewer_proof_backed_handoff.py new file mode 100644 index 0000000..d5f0740 --- /dev/null +++ b/tests/test_reviewer_proof_backed_handoff.py @@ -0,0 +1,117 @@ +"""Tests for proof-backed reviewer handoff verifier (#395).""" +import sys +import unittest +from pathlib import Path + +sys.path.insert(0, str(Path(__file__).resolve().parent.parent)) + +from reviewer_proof_backed_handoff import ( # noqa: E402 + assess_proof_backed_handoff_report, +) + + +def _good_report() -> str: + return "\n".join([ + "Inventory pagination proof: is_final_page: true, has_more: false, total_count: 11", + "Earlier PRs skipped: #372 non-mergeable — merge simulation conflicts in review_proofs.py", + "Baseline worktree path: branches/baseline-master-pr383", + "Baseline target SHA: 4243b60ca32d79f1ee5e6b0f10d7570e9dea2937", + "Baseline dirty before validation: false", + "Baseline validation command: venv/bin/python -m pytest tests/ -q", + "Baseline validation result: 1 failed (test_clean_reviewer_report_passes)", + "Baseline dirty after validation: false", + "Cleanup mutations: git worktree list showed session worktrees removed", + "## Controller Handoff", + "- Inventory pagination proof: inventory_complete: true, total_count: 11", + "- Earlier PRs skipped: #372 conflict proof via merge simulation", + "- Baseline worktree used: true", + "- Baseline worktree path: branches/baseline-master-pr383", + "- Cleanup mutations: git worktree remove branches/review-pr383", + ]) + + +class TestProofBackedHandoff(unittest.TestCase): + def test_fully_proof_backed_report_passes(self): + result = assess_proof_backed_handoff_report( + _good_report(), + action_log=[{"command": "gitea_list_prs", "result": "inventory_complete=true"}], + baseline_session={ + "complete": True, + "clean_before": True, + "clean_after": True, + "command": "pytest", + "result": "passed", + }, + ) + self.assertTrue(result["proven"], result["reasons"]) + + def test_inventory_page_size_assumption_blocks(self): + report = ( + "Inventory is complete because 13 results are less than page size 50." + ) + result = assess_proof_backed_handoff_report(report) + self.assertFalse(result["proven"]) + self.assertTrue(any("page-size" in r.lower() for r in result["reasons"])) + + def test_skip_without_conflict_proof_blocks(self): + report = "\n".join([ + "Earlier PRs skipped: #372, #374", + "## Controller Handoff", + "- Earlier PRs skipped: #372, #374", + ]) + result = assess_proof_backed_handoff_report(report) + self.assertFalse(result["proven"]) + self.assertTrue(any("skip" in r.lower() for r in result["reasons"])) + + def test_baseline_without_command_blocks(self): + report = "Baseline validation passed; same as master failures." + result = assess_proof_backed_handoff_report(report) + self.assertFalse(result["proven"]) + self.assertTrue(any("baseline" in r.lower() for r in result["reasons"])) + + def test_cleanup_without_worktree_list_blocks(self): + report = "Worktrees were cleaned after merge." + result = assess_proof_backed_handoff_report(report) + self.assertFalse(result["proven"]) + self.assertTrue(any("cleanup" in r.lower() for r in result["reasons"])) + + def test_cleanup_with_worktree_list_passes(self): + report = "Cleanup: git worktree list confirmed removal." + result = assess_proof_backed_handoff_report( + report, + action_log=[{"command": "git worktree list"}], + ) + self.assertTrue(result["proven"], result["reasons"]) + + def test_master_integration_requires_command(self): + report = "Merged master into the review worktree before validation." + result = assess_proof_backed_handoff_report(report) + self.assertFalse(result["proven"]) + self.assertTrue(any("master integration" in r.lower() for r in result["reasons"])) + + def test_master_integration_with_action_log_passes(self): + report = "Merged master into review worktree; merge_exit=0" + result = assess_proof_backed_handoff_report( + report, + action_log=[{"command": "git merge --no-commit prgs/master"}], + ) + self.assertTrue(result["proven"], result["reasons"]) + + def test_live_proof_requires_source_classification(self): + report = "Live proof shows inventory complete." + result = assess_proof_backed_handoff_report(report) + self.assertFalse(result["proven"]) + self.assertTrue(any("proof source" in r.lower() for r in result["reasons"])) + + +class TestExport(unittest.TestCase): + def test_review_proofs_reexport(self): + from review_proofs import assess_proof_backed_handoff_report as exported + + self.assertTrue(callable(exported)) + result = exported(_good_report(), inventory_session={"inventory_complete": True}) + self.assertTrue(result["proven"], result["reasons"]) + + +if __name__ == "__main__": + unittest.main() \ No newline at end of file -- 2.43.7