feat: require MCP-native post-merge cleanup proof (Closes #517)

Add mcp_native_cleanup_proof verifier that blocks raw git branch deletion and
raw API comment deletion scripts as cleanup proof, requires authorized
reconciler MCP tools in cleanup mutation ledgers, and separates merge from
cleanup mutations in controller handoffs.

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

final_report_validator.py

@@ -12,6 +12,7 @@ import re
 from typing import Any, Callable
 
 import issue_lock_provenance
+from mcp_native_cleanup_proof import assess_mcp_native_cleanup_proof
 from post_merge_cleanup_proof import assess_post_merge_cleanup_proof
 from review_proofs import (
     HANDOFF_HEADING,
@@ -1158,12 +1159,30 @@ def _rule_reviewer_post_merge_cleanup_proof(report_text: str) -> list[dict[str,
     )
 
 
+def _rule_shared_mcp_native_cleanup_proof(report_text: str) -> list[dict[str, str]]:
+    result = assess_mcp_native_cleanup_proof(report_text)
+    if not result.get("block"):
+        return []
+    return _findings_from_reasons(
+        "shared.mcp_native_cleanup_proof",
+        result.get("reasons") or [],
+        field="Cleanup mutations",
+        severity="block",
+        safe_next_action=result.get("safe_next_action")
+        or "use authorized reconciler MCP cleanup tools; never raw scripts",
+    )
+
+
 _SHARED_ISSUE_LOCK_RULES = (
     _rule_shared_issue_lock_external_state,
     _rule_shared_manual_lock_pr_override,
     _rule_shared_author_reviewer_same_run,
 )
 
+_SHARED_CLEANUP_PROOF_RULES = (
+    _rule_shared_mcp_native_cleanup_proof,
+)
+
 _RULES_BY_TASK: dict[str, list[Callable[..., list[dict[str, str]]]]] = {
     "review_pr": [
         _rule_shared_controller_handoff,
@@ -1189,12 +1208,14 @@ _RULES_BY_TASK: dict[str, list[Callable[..., list[dict[str, str]]]]] = {
         _rule_reviewer_mutation_ledger,
         _rule_reviewer_review_mutation,
         _rule_reviewer_post_merge_cleanup_proof,
+        *_SHARED_CLEANUP_PROOF_RULES,
         _rule_reviewer_stale_head_proof,
     ],
     "reconcile_already_landed": [
         _rule_reconcile_controller_handoff,
         _rule_shared_email_disclosure,
         *_SHARED_ISSUE_LOCK_RULES,
+        *_SHARED_CLEANUP_PROOF_RULES,
         _rule_reconcile_stale_author_fields,
         _rule_reconcile_eligible_reviewed,
         _rule_reconcile_linked_issue_live,

mcp_native_cleanup_proof.py

@@ -0,0 +1,289 @@
+"""MCP-native post-merge cleanup proof verifier (#517).
+
+Post-merge cleanup of leases, comments, branches, and worktrees must be
+proven through explicit MCP tools (or approved helpers), never raw scripts or
+ad hoc git/API commands. Merger/reviewer sessions must hand cleanup to a
+reconciler profile with the right capability proof.
+"""
+
+from __future__ import annotations
+
+import re
+from typing import Any
+
+AUTHORIZED_CLEANUP_TOOLS = frozenset({
+    "gitea_cleanup_post_merge_moot_lease",
+    "gitea_reconcile_merged_cleanups",
+    "gitea_delete_branch",
+    "gitea_cleanup_merged_pr_branch",
+    "gitea_audit_worktree_cleanup",
+    "gitea_capture_branches_worktree_snapshot",
+    "gitea_assess_worktree_cleanup_integrity",
+    "gitea_authorize_reconciliation_cleanup_phase",
+})
+
+_RAW_BRANCH_DELETE_PATTERNS = (
+    re.compile(r"\bgit(?:\s+-C\s+\S+)?\s+branch\s+-[dD]\b[^\n\r]*", re.I),
+    re.compile(r"\bgit(?:\s+-C\s+\S+)?\s+push\b[^\n\r]*\s--delete\b[^\n\r]*", re.I),
+    re.compile(r"\bgit(?:\s+-C\s+\S+)?\s+push\b[^\n\r]*\s:[^\s`]+", re.I),
+)
+
+_RAW_COMMENT_DELETE_PATTERNS = (
+    re.compile(
+        r"(?:curl|wget|httpie)\b[^\n\r]*\b(?:DELETE|delete)\b[^\n\r]*"
+        r"(?:/comments/|issues/\d+/comments)",
+        re.I,
+    ),
+    re.compile(
+        r"\bDELETE\b[^\n\r]*/repos/[^\n\r]*/issues/\d+/comments/\d+",
+        re.I,
+    ),
+    re.compile(
+        r"\b(?:delete_issue_comment|remove_issue_comment|purge_comments?)\b",
+        re.I,
+    ),
+    re.compile(
+        r"\b(?:raw|ad hoc|adhoc)\b[^\n\r]{0,40}\bcomment\b[^\n\r]{0,40}\bdelete",
+        re.I,
+    ),
+)
+
+_CLEANUP_MUTATIONS_RE = re.compile(
+    r"^\s*[-*]?\s*cleanup mutations\s*:\s*(.+)$",
+    re.IGNORECASE | re.MULTILINE,
+)
+_MERGE_MUTATIONS_RE = re.compile(
+    r"^\s*[-*]?\s*merge mutations\s*:\s*(.+)$",
+    re.IGNORECASE | re.MULTILINE,
+)
+_AUTHORIZED_TOOL_RE = re.compile(
+    r"\b(" + "|".join(re.escape(t) for t in sorted(AUTHORIZED_CLEANUP_TOOLS)) + r")\b",
+    re.IGNORECASE,
+)
+_RECONCILER_CAPABILITY_RE = re.compile(
+    r"(?:reconciler|gitea\.branch\.delete|delete_branch|reconcile_merged_cleanups)",
+    re.IGNORECASE,
+)
+_MERGER_CLEANUP_ROLE_RE = re.compile(
+    r"(?:merger|reviewer).{0,80}(?:deleted|removed|cleaned).{0,80}"
+    r"(?:branch|worktree|comment|lease)",
+    re.IGNORECASE,
+)
+_HANDOFF_TO_RECONCILER_RE = re.compile(
+    r"(?:hand(?:ed)? off|defer(?:red)?|next actor).{0,60}reconciler",
+    re.IGNORECASE,
+)
+
+
+def raw_branch_delete_commands(text: str | None) -> list[str]:
+    """Return raw git branch-delete commands cited in *text*."""
+    if not text:
+        return []
+    commands: list[str] = []
+    for pattern in _RAW_BRANCH_DELETE_PATTERNS:
+        commands.extend(match.group(0).strip("` ") for match in pattern.finditer(text))
+    return list(dict.fromkeys(commands))
+
+
+def raw_comment_delete_commands(text: str | None) -> list[str]:
+    """Return raw comment-deletion commands/scripts cited in *text*."""
+    if not text:
+        return []
+    commands: list[str] = []
+    for pattern in _RAW_COMMENT_DELETE_PATTERNS:
+        commands.extend(match.group(0).strip("` ") for match in pattern.finditer(text))
+    return list(dict.fromkeys(commands))
+
+
+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)
+    reasons = [
+        (
+            "raw git branch deletion bypasses MCP branch.delete cleanup gates: "
+            f"{command}"
+        )
+        for command in commands
+    ]
+    return {
+        "proven": not reasons,
+        "block": bool(reasons),
+        "commands": commands,
+        "reasons": reasons,
+        "safe_next_action": (
+            "use gitea_delete_branch, gitea_reconcile_merged_cleanups, or another "
+            "approved MCP cleanup helper with explicit branch.delete capability"
+            if reasons
+            else "proceed"
+        ),
+    }
+
+
+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)
+    reasons = [
+        (
+            "raw comment deletion bypasses MCP lease cleanup gates: "
+            f"{command}"
+        )
+        for command in commands
+    ]
+    return {
+        "proven": not reasons,
+        "block": bool(reasons),
+        "commands": commands,
+        "reasons": reasons,
+        "safe_next_action": (
+            "use gitea_cleanup_post_merge_moot_lease (append-only release comment) "
+            "or hand cleanup to a reconciler session; never delete lease comments"
+            if reasons
+            else "proceed"
+        ),
+    }
+
+
+def assess_merge_cleanup_mutation_separation(text: str | None) -> dict[str, Any]:
+    """Require merge mutations and cleanup mutations in separate ledger fields."""
+    text = text or ""
+    merge_match = _MERGE_MUTATIONS_RE.search(text)
+    cleanup_match = _CLEANUP_MUTATIONS_RE.search(text)
+    reasons: list[str] = []
+
+    if cleanup_match and not merge_match:
+        combined = re.search(
+            r"merge.{0,40}cleanup mutations|cleanup.{0,40}merge mutations",
+            text,
+            re.IGNORECASE,
+        )
+        if combined:
+            reasons.append(
+                "merge and cleanup mutations must use separate 'Merge mutations' "
+                "and 'Cleanup mutations' ledger fields"
+            )
+
+    if merge_match and cleanup_match:
+        merge_val = (merge_match.group(1) or "").strip().lower()
+        cleanup_val = (cleanup_match.group(1) or "").strip().lower()
+        if merge_val == cleanup_val and merge_val not in {"", "none"}:
+            reasons.append(
+                "merge mutations and cleanup mutations must not duplicate the "
+                "same ledger entry"
+            )
+
+    return {
+        "proven": not reasons,
+        "block": bool(reasons),
+        "reasons": reasons,
+        "safe_next_action": (
+            "split merge mutations (gitea_merge_pr) from cleanup mutations "
+            "(reconciler MCP tools) in the controller handoff"
+            if reasons
+            else "proceed"
+        ),
+    }
+
+
+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"}:
+        return {
+            "proven": True,
+            "block": False,
+            "reasons": [],
+            "safe_next_action": "proceed",
+        }
+
+    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))})"
+        )
+    if not _RECONCILER_CAPABILITY_RE.search(text):
+        reasons.append(
+            "post-merge cleanup requires reconciler capability proof "
+            "(reconciler profile, gitea.branch.delete, or reconcile_merged_cleanups)"
+        )
+
+    return {
+        "proven": not reasons,
+        "block": bool(reasons),
+        "reasons": reasons,
+        "safe_next_action": (
+            "hand cleanup to a prgs-reconciler session and cite the exact MCP tool "
+            "plus delete_branch/reconcile_merged_cleanups capability proof"
+            if reasons
+            else "proceed"
+        ),
+    }
+
+
+def assess_merger_cleanup_handoff_guidance(text: str | None) -> dict[str, Any]:
+    """Merger sessions that performed cleanup must hand off to reconciler."""
+    text = text or ""
+    if not _MERGER_CLEANUP_ROLE_RE.search(text):
+        return {
+            "proven": True,
+            "block": False,
+            "reasons": [],
+            "safe_next_action": "proceed",
+        }
+    if _HANDOFF_TO_RECONCILER_RE.search(text):
+        return {
+            "proven": True,
+            "block": False,
+            "reasons": [],
+            "safe_next_action": "proceed",
+        }
+    return {
+        "proven": False,
+        "block": True,
+        "reasons": [
+            "merger/reviewer session performed cleanup but did not hand off to "
+            "reconciler for MCP-native cleanup"
+        ],
+        "safe_next_action": (
+            "merger sessions must end with cleanup handed to prgs-reconciler; "
+            "do not perform ad hoc branch/comment/worktree cleanup as merger"
+        ),
+    }
+
+
+def assess_mcp_native_cleanup_proof(report_text: str | None) -> dict[str, Any]:
+    """Composite #517 verifier for MCP-native post-merge cleanup proof."""
+    text = report_text or ""
+    checks = (
+        assess_raw_branch_delete_report(text),
+        assess_raw_comment_delete_report(text),
+        assess_merge_cleanup_mutation_separation(text),
+        assess_authorized_reconciler_cleanup_path(text),
+        assess_merger_cleanup_handoff_guidance(text),
+    )
+    reasons: list[str] = []
+    safe_next = "proceed"
+    for result in checks:
+        reasons.extend(result.get("reasons") or [])
+        if result.get("block") and result.get("safe_next_action"):
+            safe_next = result["safe_next_action"]
+
+    block = bool(reasons)
+    return {
+        "proven": not block,
+        "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),
+    }

skills/llm-project-workflow/templates/merge-pr.md

@@ -41,7 +41,15 @@ Steps:
 8. Confirm remote master now contains the merge commit (or the expected changes if squash merged).
    *Note: Gitea PR "closed" state is NOT equivalent to "merged". Do not assume a closed PR succeeded without verifying the actual landed changes.*
 
-Then run the cleanup template (worktree-cleanup.md):
+Post-merge cleanup (#517): merger sessions must NOT perform ad hoc cleanup.
+- Record merge mutations separately from cleanup mutations in the controller handoff.
+- Hand cleanup to a `prgs-reconciler` session — never raw `git branch -d`,
+  `git push --delete`, curl/API comment deletion, or local scripts.
+- Reconciler cleanup must cite authorized MCP tools
+  (`gitea_reconcile_merged_cleanups`, `gitea_cleanup_post_merge_moot_lease`,
+  `gitea_delete_branch`, etc.) plus `gitea.branch.delete` capability proof.
+
+Then run the cleanup template (worktree-cleanup.md) in a reconciler session:
 - Verify expected file/commit presence on master (post-merge file-presence verification):
   - Run: git fetch <remote> --prune; git checkout master; git pull <remote> master --ff-only
   - Verify that the expected files added/modified in the PR are present on master (or absent if deleted).

skills/llm-project-workflow/templates/post-merge-cleanup-handoff.md

@@ -0,0 +1,27 @@
+# Template: post-merge cleanup (reconciler only)
+
+Copy, fill the `<...>` fields, and paste as the task prompt. Run only after merge
+is confirmed on remote master. Merger sessions must hand off here — never perform
+this cleanup inline (#517).
+
+```text
+Task: MCP-native post-merge cleanup for PR #<pr> / issue #<n>.
+
+Rules:
+- Active profile must be reconciler (`prgs-reconciler`) with `gitea.branch.delete`.
+- Use MCP tools only — no raw git branch delete, no API comment deletion scripts.
+- Record cleanup mutations separately from merge mutations in the handoff.
+
+Steps:
+1. `gitea_resolve_task_capability(task="reconcile_merged_cleanups", remote=prgs)`
+2. Confirm PR #<pr> merged on <remote>/master.
+3. `gitea_cleanup_post_merge_moot_lease` if a reviewer lease remains (append-only).
+4. `gitea_reconcile_merged_cleanups` for branch/worktree cleanup with dry-run first.
+5. Report authorized cleanup tools used and reconciler capability proof.
+
+Handoff ledger (required fields):
+- Merge mutations: (none — merger already recorded gitea_merge_pr)
+- Cleanup mutations: list exact MCP tools invoked
+- Reconciler capability: profile + gitea.branch.delete proof
+- Next actor: controller acceptance or none
+```

skills/llm-project-workflow/workflows/review-merge-pr.md

@@ -978,6 +978,27 @@ If any gate fails, report:
 
 Skipped cleanup with an exact blocker passes validation. Performed-cleanup claims without the checklist fail validation.
 
+## 28B. MCP-native cleanup only (#517)
+
+Post-merge cleanup of leases, comments, branches, and worktrees must go through
+explicit MCP tools — never raw git, curl/API scripts, or ad hoc helper scripts.
+
+Merger and reviewer sessions must **not** perform cleanup inline. Record:
+
+* **Merge mutations** — only `gitea_merge_pr` (or review mutations for review-only runs)
+* **Cleanup mutations** — only authorized reconciler MCP tools, cited by exact tool name
+
+Authorized cleanup tools include `gitea_reconcile_merged_cleanups`,
+`gitea_cleanup_post_merge_moot_lease`, `gitea_delete_branch`, and
+`gitea_cleanup_merged_pr_branch` (when available). Lease cleanup uses
+append-only release comments — never delete another session's lease comment.
+
+Hand cleanup to a `prgs-reconciler` session with `gitea.branch.delete` capability
+proof. Raw `git branch -d`, `git push --delete`, and comment-deletion API calls
+are blocked in final-report validation.
+
+Template: `templates/post-merge-cleanup-handoff.md`.
+
 ## 29. Recovery handoff rules
 
 If blocked, produce a recovery handoff with:

tests/test_mcp_native_cleanup_proof.py

@@ -0,0 +1,138 @@
+"""Tests for MCP-native post-merge cleanup proof (#517)."""
+import sys
+import unittest
+from pathlib import Path
+
+sys.path.insert(0, str(Path(__file__).resolve().parent.parent))
+
+from final_report_validator import assess_final_report_validator  # noqa: E402
+from mcp_native_cleanup_proof import (  # noqa: E402
+    assess_authorized_reconciler_cleanup_path,
+    assess_mcp_native_cleanup_proof,
+    assess_merger_cleanup_handoff_guidance,
+    assess_raw_branch_delete_report,
+    assess_raw_comment_delete_report,
+)
+
+
+def _authorized_cleanup_report(**overrides):
+    fields = {
+        "Merge mutations": "gitea_merge_pr on PR #100",
+        "Cleanup mutations": (
+            "gitea_reconcile_merged_cleanups deleted remote branch; "
+            "gitea_cleanup_post_merge_moot_lease released lease"
+        ),
+        "Reconciler capability": "prgs-reconciler / gitea.branch.delete resolved",
+        "Next actor": "reconciler session complete",
+    }
+    fields.update(overrides)
+    lines = ["## Controller Handoff", ""]
+    lines.extend(f"- {key}: {value}" for key, value in fields.items())
+    return "\n".join(lines)
+
+
+class TestRawBranchDeleteBlocked(unittest.TestCase):
+    def test_git_branch_d_blocked(self):
+        report = "Cleanup mutations: git branch -d feat/issue-1-x"
+        result = assess_raw_branch_delete_report(report)
+        self.assertTrue(result["block"])
+        self.assertIn("git branch -d", result["commands"][0])
+
+    def test_git_push_delete_blocked(self):
+        report = "Ran git push origin --delete feat/x for cleanup"
+        result = assess_raw_branch_delete_report(report)
+        self.assertTrue(result["block"])
+
+    def test_authorized_mcp_path_passes(self):
+        report = _authorized_cleanup_report()
+        result = assess_raw_branch_delete_report(report)
+        self.assertFalse(result["block"])
+
+
+class TestRawCommentDeleteBlocked(unittest.TestCase):
+    def test_curl_delete_comment_blocked(self):
+        report = (
+            "Cleanup: 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):
+        report = "Used delete_issue_comment script to purge lease comments"
+        result = assess_raw_comment_delete_report(report)
+        self.assertTrue(result["block"])
+
+    def test_moot_lease_append_only_passes(self):
+        report = _authorized_cleanup_report()
+        result = assess_raw_comment_delete_report(report)
+        self.assertFalse(result["block"])
+
+
+class TestAuthorizedReconcilerPath(unittest.TestCase):
+    def test_cleanup_without_mcp_tool_blocked(self):
+        report = "\n".join([
+            "## Controller Handoff",
+            "- Cleanup mutations: removed branch manually after merge",
+            "- Reconciler capability: prgs-reconciler",
+        ])
+        result = assess_authorized_reconciler_cleanup_path(report)
+        self.assertTrue(result["block"])
+        self.assertTrue(any("authorized MCP" in r for r in result["reasons"]))
+
+    def test_authorized_cleanup_succeeds(self):
+        result = assess_authorized_reconciler_cleanup_path(_authorized_cleanup_report())
+        self.assertFalse(result["block"])
+
+
+class TestMergerHandoffGuidance(unittest.TestCase):
+    def test_merger_cleanup_without_handoff_blocked(self):
+        report = "Merger deleted remote branch and removed worktree after merge"
+        result = assess_merger_cleanup_handoff_guidance(report)
+        self.assertTrue(result["block"])
+
+    def test_merger_cleanup_with_reconciler_handoff_passes(self):
+        report = (
+            "Merger deleted remote branch; next actor: reconciler for worktree audit"
+        )
+        result = assess_merger_cleanup_handoff_guidance(report)
+        self.assertFalse(result["block"])
+
+
+class TestCompositeVerifier(unittest.TestCase):
+    def test_full_authorized_report_passes(self):
+        result = assess_mcp_native_cleanup_proof(_authorized_cleanup_report())
+        self.assertFalse(result["block"], result["reasons"])
+
+    def test_raw_bypasses_fail_composite(self):
+        report = "\n".join([
+            "## Controller Handoff",
+            "- Merge mutations: gitea_merge_pr",
+            "- Cleanup mutations: git branch -D feat/x; curl -X DELETE .../comments/1",
+        ])
+        result = assess_mcp_native_cleanup_proof(report)
+        self.assertTrue(result["block"])
+        self.assertGreaterEqual(len(result["reasons"]), 2)
+
+
+class TestFinalReportValidatorIntegration(unittest.TestCase):
+    def test_validator_blocks_raw_branch_delete_in_review_report(self):
+        report = "\n".join([
+            "## Controller Handoff",
+            "- Cleanup mutations: git branch -d feat/issue-517-test",
+        ])
+        result = assess_final_report_validator(report, "review_pr")
+        rules = {f["rule_id"] for f in result["findings"]}
+        self.assertIn("shared.mcp_native_cleanup_proof", rules)
+
+    def test_validator_passes_authorized_cleanup_report(self):
+        result = assess_final_report_validator(_authorized_cleanup_report(), "review_pr")
+        blocked = [f for f in result["findings"] if f.get("severity") == "block"]
+        mcp_blocks = [
+            f for f in blocked if f.get("rule_id") == "shared.mcp_native_cleanup_proof"
+        ]
+        self.assertEqual(mcp_blocks, [])
+
+
+if __name__ == "__main__":
+    unittest.main()