Merge pull request 'Add final-report mutation ledger verifier (#331)' (#338) from feat/issue-331-mutation-ledger-verifier into master

review_proofs.py

@@ -843,6 +843,131 @@ def assess_request_changes_approval_proof(
     }
 
 
+_PERFORMED_FILE_ACTIONS = frozenset({"edited", "created", "wrote", "generated"})
+_FILE_EDITS_FIELD_RE = re.compile(
+    r"^\s*file edits by reviewer\s*:\s*(.+?)\s*$",
+    re.I | re.M,
+)
+_WALKTHROUGH_ARTIFACT_RE = re.compile(r"walkthrough\.md", re.I)
+
+
+def _performed_file_mutations(action_log: list[dict] | None) -> list[dict]:
+    """Return performed local file mutations, excluding gated rejections."""
+    performed: list[dict] = []
+    for entry in action_log or []:
+        if entry.get("gated_rejected") or entry.get("performed") is False:
+            continue
+        action = (entry.get("action") or "").strip().lower()
+        if action not in _PERFORMED_FILE_ACTIONS:
+            continue
+        path = (entry.get("path") or "").strip()
+        if not path:
+            continue
+        performed.append({**entry, "action": action, "path": path})
+    return performed
+
+
+def assess_mutation_ledger_report(
+    report_text: str,
+    *,
+    action_log: list[dict] | None = None,
+    final_git_status_reported: bool | None = None,
+    walkthrough_explicitly_requested: bool = False,
+) -> dict:
+    """#331: verify reviewer final reports match observed file mutations.
+
+    Compares an action log of local file writes/edits against the report's
+    mutation ledger and ``File edits by reviewer`` field. Gated rejections
+    (``performed: false`` or ``gated_rejected: true``) are excluded from the
+    performed mutation set but should still appear in a separate rejected-calls
+    section when reported.
+    """
+    text = report_text or ""
+    lower = text.lower()
+    reasons: list[str] = []
+    performed = _performed_file_mutations(action_log)
+
+    field_match = _FILE_EDITS_FIELD_RE.search(text)
+    file_edits_value = (field_match.group(1).strip() if field_match else None)
+    claimed_none = bool(
+        file_edits_value and file_edits_value.lower() == "none"
+    )
+
+    if performed and claimed_none:
+        reasons.append(
+            "report claims 'File edits by reviewer: none' but action log "
+            "records performed file mutations"
+        )
+
+    unreported: list[str] = []
+    for entry in performed:
+        path = entry["path"]
+        path_lower = path.lower()
+        if path_lower not in lower:
+            unreported.append(path)
+            reasons.append(
+                f"performed mutation path '{path}' missing from report "
+                "mutation ledger"
+            )
+            continue
+
+        if entry.get("outside_repo"):
+            if "outside repo" not in lower or path_lower not in lower:
+                reasons.append(
+                    f"outside-repo mutation '{path}' must be reported with "
+                    "'outside repo' label"
+                )
+        elif entry.get("in_repo", True):
+            tracked = entry.get("tracked")
+            if tracked is True and "tracked" not in lower:
+                reasons.append(
+                    f"in-repo tracked mutation '{path}' must state tracked/"
+                    "untracked status in the ledger"
+                )
+            elif tracked is False and "untracked" not in lower:
+                reasons.append(
+                    f"in-repo untracked mutation '{path}' must state tracked/"
+                    "untracked status in the ledger"
+                )
+
+        if (
+            _WALKTHROUGH_ARTIFACT_RE.search(path)
+            and not walkthrough_explicitly_requested
+        ):
+            reasons.append(
+                "walkthrough artifact created without explicit workflow or "
+                "operator request"
+            )
+
+        if entry.get("after_git_status") and final_git_status_reported is not True:
+            reasons.append(
+                f"mutation '{path}' occurred after an earlier git status; "
+                "report must include a final git status"
+            )
+
+    rejected = [
+        e for e in (action_log or [])
+        if e.get("gated_rejected") or e.get("performed") is False
+    ]
+    rejected_reported = "rejected" in lower or "gated" in lower or "no-op" in lower
+    if rejected and performed and not rejected_reported:
+        reasons.append(
+            "rejected/no-op gated tool calls must be reported separately from "
+            "performed mutations"
+        )
+
+    proven = not reasons
+    return {
+        "proven": proven,
+        "block": not proven,
+        "reasons": reasons,
+        "performed_mutations": performed,
+        "unreported_paths": unreported,
+        "file_edits_claimed_none": claimed_none,
+        "rejected_calls": rejected,
+    }
+
+
 def assess_role_boundary(proof=None, *, task_role=None, namespaces_used=None,
                          justification=None):
     """Assess reviewer/author role separation for blind queue workflows.

tests/test_mutation_ledger_report.py

@@ -0,0 +1,102 @@
+"""Tests for final-report mutation ledger verifier (#331)."""
+import sys
+import unittest
+from pathlib import Path
+
+sys.path.insert(0, str(Path(__file__).resolve().parent.parent))
+
+from review_proofs import assess_mutation_ledger_report  # noqa: E402
+
+
+class TestMutationLedgerReport(unittest.TestCase):
+    def test_none_claim_with_performed_edit_blocks(self):
+        report = (
+            "Review decision: request_changes.\n"
+            "File edits by reviewer: none\n"
+        )
+        action_log = [
+            {"action": "Edited", "path": "walkthrough.md", "tracked": False},
+        ]
+        result = assess_mutation_ledger_report(report, action_log=action_log)
+        self.assertFalse(result["proven"])
+        self.assertTrue(result["file_edits_claimed_none"])
+
+    def test_reported_edit_passes(self):
+        report = (
+            "File edits by reviewer: Edited walkthrough.md (untracked)\n"
+            "Mutation ledger: Edited walkthrough.md untracked in review worktree\n"
+        )
+        action_log = [
+            {
+                "action": "Edited",
+                "path": "walkthrough.md",
+                "tracked": False,
+                "in_repo": True,
+            },
+        ]
+        result = assess_mutation_ledger_report(
+            report,
+            action_log=action_log,
+            walkthrough_explicitly_requested=True,
+        )
+        self.assertTrue(result["proven"])
+
+    def test_unreported_mutation_blocks(self):
+        report = "File edits by reviewer: none\n"
+        action_log = [{"action": "Wrote", "path": "/tmp/review-notes.txt"}]
+        result = assess_mutation_ledger_report(report, action_log=action_log)
+        self.assertFalse(result["proven"])
+        self.assertIn("/tmp/review-notes.txt", result["unreported_paths"])
+
+    def test_outside_repo_requires_label(self):
+        report = (
+            "File edits by reviewer: Wrote /tmp/review-notes.txt\n"
+        )
+        action_log = [
+            {
+                "action": "Wrote",
+                "path": "/tmp/review-notes.txt",
+                "outside_repo": True,
+            },
+        ]
+        result = assess_mutation_ledger_report(report, action_log=action_log)
+        self.assertFalse(result["proven"])
+        self.assertIn("outside repo", " ".join(result["reasons"]).lower())
+
+    def test_gated_rejection_excluded_from_performed(self):
+        report = "File edits by reviewer: none\nRejected gated calls: submit_pr_review blocked\n"
+        action_log = [
+            {
+                "action": "Edited",
+                "path": "walkthrough.md",
+                "performed": False,
+                "gated_rejected": True,
+            },
+        ]
+        result = assess_mutation_ledger_report(report, action_log=action_log)
+        self.assertTrue(result["proven"])
+        self.assertEqual(result["performed_mutations"], [])
+
+    def test_post_status_artifact_requires_final_git_status(self):
+        report = (
+            "File edits by reviewer: Created scratch/notes.md (untracked)\n"
+        )
+        action_log = [
+            {
+                "action": "Created",
+                "path": "scratch/notes.md",
+                "tracked": False,
+                "after_git_status": True,
+            },
+        ]
+        result = assess_mutation_ledger_report(
+            report,
+            action_log=action_log,
+            final_git_status_reported=False,
+        )
+        self.assertFalse(result["proven"])
+        self.assertIn("final git status", " ".join(result["reasons"]).lower())
+
+
+if __name__ == "__main__":
+    unittest.main()