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,359 @@
+"""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,
+)
+_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,
+)
+_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 _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:
+        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 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: "
+            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 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: "
+            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 ""
+    claiming = _cleanup_claiming_ledger_fields(text)
+    if not claiming:
+        return {
+            "proven": True,
+            "block": False,
+            "reasons": [],
+            "safe_next_action": "proceed",
+        }
+
+    reasons: list[str] = []
+    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 "
+            "(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(
+            scoped_cleanup_mutation_ledger_text(text)
+        ),
+        "raw_comment_commands": raw_comment_delete_commands(
+            scoped_cleanup_mutation_ledger_text(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,251 @@
+"""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_in_ledger_blocked(self):
+        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)
+        self.assertFalse(result["block"])
+
+
+class TestRawCommentDeleteBlocked(unittest.TestCase):
+    def test_curl_delete_comment_in_ledger_blocked(self):
+        report = (
+            "- 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_narrative_comment_delete_not_blocked(self):
+        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"])
+
+    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"])
+
+    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):
+        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)
+
+    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):
+        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, [])
+
+    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()