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]>
2026-07-08 13:43:39 -04:00
parent ea51b0196a
commit 749ca04388
2 changed files with 211 additions and 28 deletions
@@ -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."""
-    commands = raw_branch_delete_commands(text)
+    """Fail closed when mutation ledgers cite raw git branch deletion."""
+    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."""
-    commands = raw_comment_delete_commands(text)
+    """Fail closed when mutation ledgers cite raw comment deletion."""
+    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)
-    if not cleanup_match:
-        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"}:
+    claiming = _cleanup_claiming_ledger_fields(text)
+    if not claiming:
        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):
-        reasons.append(
-            "cleanup mutations must name an authorized MCP cleanup tool "
-            f"({', '.join(sorted(AUTHORIZED_CLEANUP_TOOLS))})"
-        )
+    for field_name, body in claiming:
+        if not _AUTHORIZED_TOOL_RE.search(body):
+            reasons.append(
+                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_comment_commands": raw_comment_delete_commands(text),
+        "raw_branch_commands": raw_branch_delete_commands(
+            scoped_cleanup_mutation_ledger_text(text)
+        ),
+        "raw_comment_commands": raw_comment_delete_commands(
+            scoped_cleanup_mutation_ledger_text(text)
+        ),
    }