fix: scope MCP cleanup proof to mutation ledgers (#517)

Raw branch/comment delete detection now scans only Cleanup mutations,
Git ref mutations, and Worktree mutations ledger fields so narrative
§28B citations and validation notes no longer false-positive.

Authorized cleanup tool proof also applies when cleanup is claimed under
Git ref or Worktree mutation ledgers while Cleanup mutations is none.

Co-Authored-By: Claude Opus 4.8 (1M context) <[email protected]>

mcp_native_cleanup_proof.py

@@ -52,10 +52,27 @@ _CLEANUP_MUTATIONS_RE = re.compile(
     r"^\s*[-*]?\s*cleanup mutations\s*:\s*(.+)$",
     re.IGNORECASE | re.MULTILINE,
 )
+_GIT_REF_MUTATIONS_RE = re.compile(
+    r"^\s*[-*]?\s*git ref mutations\s*:\s*(.+)$",
+    re.IGNORECASE | re.MULTILINE,
+)
+_WORKTREE_MUTATIONS_RE = re.compile(
+    r"^\s*[-*]?\s*worktree mutations\s*:\s*(.+)$",
+    re.IGNORECASE | re.MULTILINE,
+)
 _MERGE_MUTATIONS_RE = re.compile(
     r"^\s*[-*]?\s*merge mutations\s*:\s*(.+)$",
     re.IGNORECASE | re.MULTILINE,
 )
+_NONE_LEDGER_VALUES = frozenset({"", "none", "n/a"})
+_MUTATION_LEDGER_PATTERNS = (
+    _CLEANUP_MUTATIONS_RE,
+    _GIT_REF_MUTATIONS_RE,
+    _WORKTREE_MUTATIONS_RE,
+)
+_RAW_WORKTREE_REMOVE_PATTERNS = (
+    re.compile(r"\bgit(?:\s+-C\s+\S+)?\s+worktree\s+remove\b", re.I),
+)
 _AUTHORIZED_TOOL_RE = re.compile(
     r"\b(" + "|".join(re.escape(t) for t in sorted(AUTHORIZED_CLEANUP_TOOLS)) + r")\b",
     re.IGNORECASE,
@@ -75,6 +92,63 @@ _HANDOFF_TO_RECONCILER_RE = re.compile(
 )
 
 
+def _ledger_field_body(text: str, pattern: re.Pattern[str]) -> str | None:
+    match = pattern.search(text)
+    if not match:
+        return None
+    return (match.group(1) or "").strip()
+
+
+def _is_none_ledger_value(value: str) -> bool:
+    return value.strip().lower() in _NONE_LEDGER_VALUES
+
+
+def scoped_cleanup_mutation_ledger_text(text: str | None) -> str:
+    """Return mutation-ledger bodies scoped to cleanup enforcement (#517)."""
+    text = text or ""
+    parts: list[str] = []
+    for pattern in _MUTATION_LEDGER_PATTERNS:
+        body = _ledger_field_body(text, pattern)
+        if body is not None and not _is_none_ledger_value(body):
+            parts.append(body)
+    return "\n".join(parts)
+
+
+def _git_ref_cleanup_claimed(body: str) -> bool:
+    return bool(raw_branch_delete_commands(body))
+
+
+def _worktree_cleanup_claimed(body: str) -> bool:
+    for pattern in _RAW_WORKTREE_REMOVE_PATTERNS:
+        if pattern.search(body):
+            return True
+    return bool(
+        re.search(
+            r"\b(?:removed|deleted)\b[^\n]{0,40}\bworktree\b",
+            body,
+            re.IGNORECASE,
+        )
+    )
+
+
+def _cleanup_claiming_ledger_fields(text: str) -> list[tuple[str, str]]:
+    claiming: list[tuple[str, str]] = []
+    cleanup_body = _ledger_field_body(text, _CLEANUP_MUTATIONS_RE)
+    if cleanup_body is not None and not _is_none_ledger_value(cleanup_body):
+        claiming.append(("Cleanup mutations", cleanup_body))
+
+    git_ref_body = _ledger_field_body(text, _GIT_REF_MUTATIONS_RE)
+    if git_ref_body is not None and not _is_none_ledger_value(git_ref_body):
+        if _git_ref_cleanup_claimed(git_ref_body):
+            claiming.append(("Git ref mutations", git_ref_body))
+
+    worktree_body = _ledger_field_body(text, _WORKTREE_MUTATIONS_RE)
+    if worktree_body is not None and not _is_none_ledger_value(worktree_body):
+        if _worktree_cleanup_claimed(worktree_body):
+            claiming.append(("Worktree mutations", worktree_body))
+    return claiming
+
+
 def raw_branch_delete_commands(text: str | None) -> list[str]:
     """Return raw git branch-delete commands cited in *text*."""
     if not text:
@@ -96,8 +170,8 @@ def raw_comment_delete_commands(text: str | None) -> list[str]:
 
 
 def assess_raw_branch_delete_report(text: str | None) -> dict[str, Any]:
-    """Fail closed when a report uses raw git branch deletion as cleanup proof."""
+    """Fail closed when mutation ledgers cite raw git branch deletion."""
-    commands = raw_branch_delete_commands(text)
+    commands = raw_branch_delete_commands(scoped_cleanup_mutation_ledger_text(text))
     reasons = [
         (
             "raw git branch deletion bypasses MCP branch.delete cleanup gates: "
@@ -120,8 +194,8 @@ def assess_raw_branch_delete_report(text: str | None) -> dict[str, Any]:
 
 
 def assess_raw_comment_delete_report(text: str | None) -> dict[str, Any]:
-    """Fail closed when a report uses raw comment deletion as cleanup proof."""
+    """Fail closed when mutation ledgers cite raw comment deletion."""
-    commands = raw_comment_delete_commands(text)
+    commands = raw_comment_delete_commands(scoped_cleanup_mutation_ledger_text(text))
     reasons = [
         (
             "raw comment deletion bypasses MCP lease cleanup gates: "
@@ -187,17 +261,8 @@ def assess_merge_cleanup_mutation_separation(text: str | None) -> dict[str, Any]
 def assess_authorized_reconciler_cleanup_path(text: str | None) -> dict[str, Any]:
     """Validate cleanup claims cite authorized MCP tools and reconciler capability."""
     text = text or ""
-    cleanup_match = _CLEANUP_MUTATIONS_RE.search(text)
+    claiming = _cleanup_claiming_ledger_fields(text)
-    if not cleanup_match:
+    if not claiming:
-        return {
-            "proven": True,
-            "block": False,
-            "reasons": [],
-            "safe_next_action": "proceed",
-        }
-
-    cleanup_body = cleanup_match.group(1) or ""
-    if cleanup_body.strip().lower() in {"", "none", "n/a"}:
         return {
             "proven": True,
             "block": False,
@@ -206,11 +271,12 @@ def assess_authorized_reconciler_cleanup_path(text: str | None) -> dict[str, Any
         }
 
     reasons: list[str] = []
-    if not _AUTHORIZED_TOOL_RE.search(cleanup_body):
+    for field_name, body in claiming:
-        reasons.append(
+        if not _AUTHORIZED_TOOL_RE.search(body):
-            "cleanup mutations must name an authorized MCP cleanup tool "
+            reasons.append(
-            f"({', '.join(sorted(AUTHORIZED_CLEANUP_TOOLS))})"
+                f"{field_name} cleanup must name an authorized MCP cleanup tool "
-        )
+                f"({', '.join(sorted(AUTHORIZED_CLEANUP_TOOLS))})"
+            )
     if not _RECONCILER_CAPABILITY_RE.search(text):
         reasons.append(
             "post-merge cleanup requires reconciler capability proof "
@@ -284,6 +350,10 @@ def assess_mcp_native_cleanup_proof(report_text: str | None) -> dict[str, Any]:
         "block": block,
         "reasons": reasons,
         "safe_next_action": safe_next,
-        "raw_branch_commands": raw_branch_delete_commands(text),
+        "raw_branch_commands": raw_branch_delete_commands(
-        "raw_comment_commands": raw_comment_delete_commands(text),
+            scoped_cleanup_mutation_ledger_text(text)
+        ),
+        "raw_comment_commands": raw_comment_delete_commands(
+            scoped_cleanup_mutation_ledger_text(text)
+        ),
     }

tests/test_mcp_native_cleanup_proof.py

@@ -38,11 +38,22 @@ class TestRawBranchDeleteBlocked(unittest.TestCase):
         self.assertTrue(result["block"])
         self.assertIn("git branch -d", result["commands"][0])
 
-    def test_git_push_delete_blocked(self):
+    def test_git_push_delete_in_ledger_blocked(self):
-        report = "Ran git push origin --delete feat/x for cleanup"
+        report = "- Git ref mutations: git push origin --delete feat/x"
         result = assess_raw_branch_delete_report(report)
         self.assertTrue(result["block"])
 
+    def test_narrative_git_push_delete_not_blocked(self):
+        report = "\n".join([
+            "## Validation",
+            "Workflow §28B forbids raw git push --delete for cleanup.",
+            "- Cleanup mutations: none",
+            "- Git ref mutations: none",
+            "- Worktree mutations: none",
+        ])
+        result = assess_raw_branch_delete_report(report)
+        self.assertFalse(result["block"])
+
     def test_authorized_mcp_path_passes(self):
         report = _authorized_cleanup_report()
         result = assess_raw_branch_delete_report(report)
@@ -50,16 +61,27 @@ class TestRawBranchDeleteBlocked(unittest.TestCase):
 
 
 class TestRawCommentDeleteBlocked(unittest.TestCase):
-    def test_curl_delete_comment_blocked(self):
+    def test_curl_delete_comment_in_ledger_blocked(self):
         report = (
-            "Cleanup: curl -X DELETE https://gitea.example/api/v1/repos/o/r/"
+            "- Cleanup mutations: curl -X DELETE https://gitea.example/api/v1/repos/o/r/"
             "issues/9/comments/42"
         )
         result = assess_raw_comment_delete_report(report)
         self.assertTrue(result["block"])
 
-    def test_delete_issue_comment_helper_blocked(self):
+    def test_narrative_comment_delete_not_blocked(self):
-        report = "Used delete_issue_comment script to purge lease comments"
+        report = "\n".join([
+            "Files reviewed: post_merge_cleanup_proof.py (raw comment delete patterns)",
+            "Validation note: never use delete_issue_comment for lease cleanup.",
+            "- Cleanup mutations: none",
+            "- Git ref mutations: none",
+            "- Worktree mutations: none",
+        ])
+        result = assess_raw_comment_delete_report(report)
+        self.assertFalse(result["block"])
+
+    def test_delete_issue_comment_helper_in_ledger_blocked(self):
+        report = "- Cleanup mutations: delete_issue_comment purged lease comments"
         result = assess_raw_comment_delete_report(report)
         self.assertTrue(result["block"])
 
@@ -84,6 +106,41 @@ class TestAuthorizedReconcilerPath(unittest.TestCase):
         result = assess_authorized_reconciler_cleanup_path(_authorized_cleanup_report())
         self.assertFalse(result["block"])
 
+    def test_git_ref_cleanup_without_mcp_tool_blocked(self):
+        report = "\n".join([
+            "## Controller Handoff",
+            "- Cleanup mutations: none",
+            "- Git ref mutations: git branch -D feat/issue-517-test",
+            "- Reconciler capability: prgs-reconciler",
+        ])
+        result = assess_authorized_reconciler_cleanup_path(report)
+        self.assertTrue(result["block"])
+        self.assertTrue(
+            any("Git ref mutations cleanup must name" in r for r in result["reasons"])
+        )
+
+    def test_worktree_cleanup_without_mcp_tool_blocked(self):
+        report = "\n".join([
+            "## Controller Handoff",
+            "- Cleanup mutations: none",
+            "- Worktree mutations: git worktree remove branches/review-pr542",
+            "- Reconciler capability: prgs-reconciler",
+        ])
+        result = assess_authorized_reconciler_cleanup_path(report)
+        self.assertTrue(result["block"])
+        self.assertTrue(
+            any("Worktree mutations cleanup must name" in r for r in result["reasons"])
+        )
+
+    def test_git_fetch_in_git_ref_mutations_not_cleanup_claim(self):
+        report = "\n".join([
+            "## Controller Handoff",
+            "- Cleanup mutations: none",
+            "- Git ref mutations: git fetch prgs master",
+        ])
+        result = assess_authorized_reconciler_cleanup_path(report)
+        self.assertFalse(result["block"])
+
 
 class TestMergerHandoffGuidance(unittest.TestCase):
     def test_merger_cleanup_without_handoff_blocked(self):
@@ -114,6 +171,33 @@ class TestCompositeVerifier(unittest.TestCase):
         self.assertTrue(result["block"])
         self.assertGreaterEqual(len(result["reasons"]), 2)
 
+    def test_narrative_cleanup_mentions_pass_when_ledgers_none(self):
+        report = "\n".join([
+            "## Review summary",
+            "Validated workflow §28B MCP-native cleanup guidance.",
+            "Files reviewed describe raw git branch -d and delete_issue_comment blocks.",
+            "",
+            "## Controller Handoff",
+            "- Cleanup mutations: none",
+            "- Git ref mutations: none",
+            "- Worktree mutations: none",
+        ])
+        result = assess_mcp_native_cleanup_proof(report)
+        self.assertFalse(result["block"], result["reasons"])
+
+    def test_git_ref_cleanup_bypass_blocked_in_composite(self):
+        report = "\n".join([
+            "## Controller Handoff",
+            "- Cleanup mutations: none",
+            "- Git ref mutations: git branch -D feat/issue-517-test",
+            "- Reconciler capability: prgs-reconciler",
+        ])
+        result = assess_mcp_native_cleanup_proof(report)
+        self.assertTrue(result["block"])
+        self.assertTrue(
+            any("Git ref mutations cleanup must name" in r for r in result["reasons"])
+        )
+
 
 class TestFinalReportValidatorIntegration(unittest.TestCase):
     def test_validator_blocks_raw_branch_delete_in_review_report(self):
@@ -133,6 +217,35 @@ class TestFinalReportValidatorIntegration(unittest.TestCase):
         ]
         self.assertEqual(mcp_blocks, [])
 
+    def test_validator_passes_narrative_cleanup_mentions_with_none_ledgers(self):
+        report = "\n".join([
+            "## Review summary",
+            "Workflow §28B documents raw git push --delete and delete_issue_comment blocks.",
+            "",
+            "## Controller Handoff",
+            "- Cleanup mutations: none",
+            "- Git ref mutations: none",
+            "- Worktree mutations: none",
+        ])
+        result = assess_final_report_validator(report, "review_pr")
+        mcp_blocks = [
+            f for f in result["findings"]
+            if f.get("rule_id") == "shared.mcp_native_cleanup_proof"
+            and f.get("severity") == "block"
+        ]
+        self.assertEqual(mcp_blocks, [])
+
+    def test_validator_blocks_git_ref_cleanup_bypass(self):
+        report = "\n".join([
+            "## Controller Handoff",
+            "- Cleanup mutations: none",
+            "- Git ref mutations: git branch -D feat/issue-517-test",
+            "- Reconciler capability: prgs-reconciler",
+        ])
+        result = assess_final_report_validator(report, "reconcile_already_landed")
+        rules = {f["rule_id"] for f in result["findings"] if f.get("severity") == "block"}
+        self.assertIn("shared.mcp_native_cleanup_proof", rules)
+
 
 if __name__ == "__main__":
     unittest.main()