feat: require proof-backed claims in reviewer handoff reports (Closes #395)

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)

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"
+        ),
+    }

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.
+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.

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
 

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()