final_report_validator.py

@@ -568,6 +568,27 @@ def _rule_conflict_fix_push_proof(report_text: str) -> list[dict[str, str]]:
     )
 
 
+def _rule_worktree_cleanup_audit_proof(report_text: str) -> list[dict[str, str]]:
+    from worktree_cleanup_audit import assess_cleanup_audit_final_report
+
+    text = report_text or ""
+    if "cleanup audit" not in text.lower() and "reconciliation table" not in text.lower():
+        return []
+    result = assess_cleanup_audit_final_report(text)
+    if result.get("proven"):
+        return []
+    return _findings_from_reasons(
+        "author.worktree_cleanup_audit_proof",
+        result.get("reasons") or [],
+        field="Worktree cleanup audit",
+        severity="block",
+        safe_next_action=(
+            "include reconciliation table counts, disposition rows, and "
+            "final git worktree list proof"
+        ),
+    )
+
+
 def _rule_reviewer_validation_command(report_text: str) -> list[dict[str, str]]:
     text = report_text or ""
     if not _BARE_PYTEST_RE.search(text):
@@ -1130,6 +1151,7 @@ _RULES_BY_TASK: dict[str, list[Callable[..., list[dict[str, str]]]]] = {
         *_SHARED_ISSUE_LOCK_RULES,
         _rule_reviewer_vague_mutations_none,
         _rule_conflict_fix_push_proof,
+        _rule_worktree_cleanup_audit_proof,
     ],
     "issue_filing": [
         _rule_shared_controller_handoff,

gitea_mcp_server.py

@@ -751,6 +751,7 @@ import issue_claim_heartbeat  # noqa: E402
 import issue_work_duplicate_gate  # noqa: E402
 import reviewer_pr_lease  # noqa: E402
 import merged_cleanup_reconcile  # noqa: E402
+import worktree_cleanup_audit  # noqa: E402
 import reconciler_profile  # noqa: E402
 import reconciliation_workflow  # noqa: E402
 import audit_reconciliation_mode  # noqa: E402
@@ -4326,6 +4327,59 @@ def _remote_branch_exists(h: str, o: str, r: str, auth: str, branch: str) -> boo
         raise
 
 
+@mcp.tool()
+def gitea_capture_branches_worktree_snapshot(
+    open_pr_branches: list[str] | None = None,
+    active_lock_branch: str | None = None,
+    leased_paths: list[str] | None = None,
+    worktree_path: str | None = None,
+) -> dict:
+    """Read-only: capture ``branches/`` and worktree audit snapshot (#404)."""
+    read_block = _profile_operation_gate("gitea.read")
+    if read_block:
+        return {
+            "success": False,
+            "reasons": read_block,
+            "permission_report": _permission_block_report("gitea.read"),
+        }
+    root = PROJECT_ROOT
+    if worktree_path:
+        root = os.path.realpath(os.path.abspath(worktree_path))
+        git_root = _get_git_root(root)
+        if git_root:
+            root = git_root
+    snapshot = worktree_cleanup_audit.capture_branches_worktree_snapshot(
+        root,
+        open_pr_branches=open_pr_branches,
+        active_lock_branch=active_lock_branch,
+        leased_paths=leased_paths,
+    )
+    return {"success": True, "snapshot": snapshot}
+
+
+@mcp.tool()
+def gitea_assess_worktree_cleanup_integrity(
+    before_snapshot: dict,
+    after_snapshot: dict,
+    removals: list[dict] | None = None,
+    explained_missing: dict[str, str] | None = None,
+) -> dict:
+    """Read-only: reconcile cleanup before/after snapshots (#404)."""
+    read_block = _profile_operation_gate("gitea.read")
+    if read_block:
+        return {
+            "integrity_passed": False,
+            "reasons": read_block,
+            "permission_report": _permission_block_report("gitea.read"),
+        }
+    return worktree_cleanup_audit.assess_worktree_cleanup_integrity(
+        before=before_snapshot,
+        after=after_snapshot,
+        removals=removals,
+        explained_missing=explained_missing,
+    )
+
+
 @mcp.tool()
 def gitea_reconcile_merged_cleanups(
     dry_run: bool = True,

skills/llm-project-workflow/templates/worktree-cleanup.md

@@ -26,3 +26,43 @@ Steps:
 
 Handoff: merge confirmed, issue closed, branch+worktree removed, checkout clean.
 ```
+
+## Branches cleanup audit integrity (#404)
+
+Any bulk or multi-path cleanup under `branches/` must capture auditable before/after
+identity for every initial directory and registered worktree. Use
+`worktree_cleanup_audit.capture_cleanup_snapshot` before and after cleanup, record
+every intentional removal in a removal log (path, method, order, timestamp,
+pre-removal proof), then run `reconcile_cleanup_audit` and
+`assess_cleanup_audit_integrity`.
+
+The cleanup report must include a reconciliation table:
+
+* initial count
+* removed count
+* preserved count
+* missing-unexplained count
+* final count
+
+Fail closed when:
+
+* a preserved (active PR, dirty, claim/lease, or unsafe) worktree disappears without
+  a removal log entry or explicit explanation
+* the removal log omits a removed clean-stale path
+* final counts do not reconcile with initial minus removed
+
+If another session removes or mutates a worktree during cleanup, record the path
+under explained missing entries — never treat silent disappearance as success.
+
+## Bulk `branches/` cleanup audit (#404)
+
+Before removing multiple session-owned worktrees:
+
+1. Call `gitea_capture_branches_worktree_snapshot` and record the before snapshot.
+2. Remove only paths classified as `clean_stale_removable` with explicit per-path proof.
+3. Log every removal with path, method, and timestamp/order.
+4. Capture an after snapshot with the same tool.
+5. Call `gitea_assess_worktree_cleanup_integrity` with before, after, and the removal log.
+6. Fail closed when any protected path (active PR, dirty, claim/lease) disappears
+   without an explained state transition.
+7. Final report must include the reconciliation table and `git worktree list` proof.

tests/test_worktree_cleanup_audit.py

@@ -1,4 +1,4 @@
-"""Tests for session-owned worktree cleanup audit and TTL enforcement (#401)."""
+"""Tests for session-owned worktree cleanup audit, TTL, and integrity (#401, #404)."""
 
 import sys
 import unittest
@@ -379,5 +379,154 @@ class TestAuditReportAccuracy(unittest.TestCase):
         self.assertEqual(report["git_worktree_list"], "(mocked)")
 
 
+def _integrity_entry(
+    path: str,
+    *,
+    classification: str,
+    registered: bool = True,
+    preserve: bool | None = None,
+) -> dict:
+    preserve_flag = preserve if preserve is not None else classification in {
+        "active_open_pr",
+        "active_issue_work",
+        "dirty_local_worktree",
+        "unsafe_unknown",
+    }
+    return {
+        "path": path,
+        "classification": classification,
+        "preserve": preserve_flag,
+        "registered_worktree": registered,
+        "worktree_state": {"exists": True, "clean": classification == "clean_stale_removable"},
+        "worktree_record": {"branch": "feat/x"} if registered else None,
+    }
+
+
+def _integrity_snapshot(entries: list[dict]) -> dict:
+    return {"entries": entries}
+
+
+class TestParseWorktreePorcelain(unittest.TestCase):
+    def test_parses_multiple_worktrees(self):
+        text = "\n".join([
+            "worktree /proj/branches/foo",
+            "HEAD abcdef0123456789abcdef0123456789abcdef0",
+            "branch refs/heads/feat/foo",
+            "",
+            "worktree /proj",
+            "HEAD 1111111111111111111111111111111111111111",
+            "branch refs/heads/master",
+        ])
+        parsed = wca.parse_worktree_list_porcelain(text)
+        self.assertEqual(len(parsed), 2)
+        self.assertEqual(parsed[0]["branch"], "feat/foo")
+
+
+class TestClassifyEntry(unittest.TestCase):
+    def test_active_pr_classification(self):
+        result = wca.classify_branches_entry(
+            rel_path="branches/feat-issue-1-x",
+            worktree_record={"branch": "feat/issue-1-x"},
+            worktree_state={"exists": True, "clean": True, "dirty_files": []},
+            open_pr_branches={"feat/issue-1-x"},
+        )
+        self.assertEqual(result, "active_open_pr")
+
+    def test_dirty_classification(self):
+        result = wca.classify_branches_entry(
+            rel_path="branches/dirty-one",
+            worktree_record={"branch": "feat/dirty-one"},
+            worktree_state={"exists": True, "dirty_files": ["a.py"]},
+            open_pr_branches=set(),
+        )
+        self.assertEqual(result, "dirty_local_worktree")
+
+
+class TestCleanupIntegrity(unittest.TestCase):
+    def test_preserved_worktree_remains(self):
+        path = "branches/keep-me"
+        before = _integrity_snapshot([
+            _integrity_entry(path, classification="clean_stale_removable", preserve=False),
+        ])
+        after = _integrity_snapshot([
+            _integrity_entry(path, classification="clean_stale_removable", preserve=False),
+        ])
+        result = wca.assess_worktree_cleanup_integrity(before=before, after=after)
+        self.assertTrue(result["integrity_passed"])
+
+    def test_intentional_removal_passes(self):
+        path = "branches/remove-me"
+        before = _integrity_snapshot([
+            _integrity_entry(path, classification="clean_stale_removable", preserve=False),
+        ])
+        after = _integrity_snapshot([])
+        result = wca.assess_worktree_cleanup_integrity(
+            before=before,
+            after=after,
+            removals=[{
+                "path": path,
+                "method": "git worktree remove",
+                "pre_removal_proof": "clean status",
+            }],
+        )
+        self.assertTrue(result["integrity_passed"])
+
+    def test_dirty_worktree_disappears_fails(self):
+        path = "branches/dirty-wt"
+        before = _integrity_snapshot([_integrity_entry(path, classification="dirty_local_worktree")])
+        after = _integrity_snapshot([])
+        recon = wca.reconcile_cleanup_audit(before, after)
+        result = wca.assess_cleanup_audit_integrity(recon)
+        self.assertFalse(result["proven"])
+
+    def test_active_pr_worktree_disappears_fails(self):
+        path = "branches/review-pr99"
+        before = _integrity_snapshot([_integrity_entry(path, classification="active_open_pr")])
+        after = _integrity_snapshot([])
+        result = wca.assess_worktree_cleanup_integrity(before=before, after=after)
+        self.assertFalse(result["integrity_passed"])
+
+    def test_explained_missing_allowed(self):
+        path = "branches/review-pr382"
+        before = _integrity_snapshot([
+            _integrity_entry(path, classification="detached_review_leftover", preserve=False),
+        ])
+        after = _integrity_snapshot([])
+        result = wca.assess_worktree_cleanup_integrity(
+            before=before,
+            after=after,
+            explained_missing={path: "removed concurrently by sibling session"},
+        )
+        self.assertTrue(result["integrity_passed"])
+
+    def test_removal_log_omits_clean_stale_fails(self):
+        path = "branches/a"
+        before = _integrity_snapshot([
+            _integrity_entry(path, classification="clean_stale_removable", preserve=False),
+        ])
+        after = _integrity_snapshot([])
+        recon = wca.reconcile_cleanup_audit(before, after, removal_log=[])
+        self.assertFalse(recon["removal_log_complete"])
+
+
+class TestCleanupReportProof(unittest.TestCase):
+    def test_complete_report_passes(self):
+        report = "\n".join([
+            "Cleanup audit reconciliation table:",
+            "Initial count: 10",
+            "Removed count: 3",
+            "Preserved count: 7",
+            "Missing-unexplained count: 0",
+            "Final count: 7",
+            "Final verification: git worktree list proof attached",
+        ])
+        result = wca.assess_cleanup_audit_final_report(report)
+        self.assertTrue(result["proven"])
+
+    def test_incomplete_report_fails(self):
+        result = wca.assess_cleanup_audit_final_report("removed some worktrees")
+        self.assertFalse(result["proven"])
+
+
 if __name__ == "__main__":
     unittest.main()

worktree_cleanup_audit.py

@@ -1,4 +1,4 @@
-"""Session-owned worktree cleanup audit and TTL enforcement (#401).
+"""Session-owned worktree cleanup audit, TTL enforcement, and integrity (#401, #404).
 
 LLM workflows create many session-owned worktrees under ``branches/``
 (review, baseline, merge-simulation, issue, and conflict-fix worktrees).
@@ -17,6 +17,8 @@ This module provides:
 * git-shelling helpers (``list_worktrees``, ``read_worktree_dirty``,
   ``git_worktree_list``, ``remove_worktree``) and ``audit_branches_directory``
   that classify every entry under ``branches/``.
+* ``capture_cleanup_snapshot`` / ``reconcile_cleanup_audit`` — before/after
+  reconciliation for bulk cleanup audits (#404).
 
 Pure assessment functions take explicit state so they are unit-testable
 without a filesystem or network. Only the thin git helpers shell out, and
@@ -32,6 +34,9 @@ import subprocess
 from datetime import datetime, timezone
 from typing import Any
 
+from merged_cleanup_reconcile import branch_worktree_folder, read_local_worktree_state
+from reviewer_worktree import parse_dirty_tracked_files
+
 PROTECTED_BRANCHES = frozenset({"master", "main", "dev"})
 DEFAULT_TTL_HOURS = float(os.environ.get("GITEA_WORKTREE_TTL_HOURS", "24") or 24)
 
@@ -478,3 +483,541 @@ def audit_branches_directory(
         "total": len(worktrees),
         "git_worktree_list": git_worktree_list(project_root),
     }
+CLASSIFICATIONS = frozenset({
+    "active_open_pr",
+    "active_issue_work",
+    "dirty_local_worktree",
+    "clean_stale_removable",
+    "detached_review_leftover",
+    "orphan_directory",
+    "unsafe_unknown",
+})
+
+PRESERVE_CLASSIFICATIONS = frozenset({
+    "active_open_pr",
+    "active_issue_work",
+    "dirty_local_worktree",
+    "unsafe_unknown",
+})
+
+DISPOSITIONS = frozenset({
+    "removed_intentionally",
+    "preserved_exists",
+    "preserved_missing_explained",
+    "not_registered_worktree",
+    "unsafe_unknown",
+})
+
+REVIEW_WORKTREE_RE = re.compile(
+    r"branches/(?:review-pr\d+|merge-simulation-pr\d+|review-[\w-]+)",
+    re.IGNORECASE,
+)
+
+
+def normalize_path(path: str) -> str:
+    return os.path.normpath((path or "").strip())
+
+
+def relative_branches_path(project_root: str, path: str) -> str:
+    root = normalize_path(project_root)
+    normalized = normalize_path(path)
+    if normalized.startswith(root + os.sep):
+        return normalized[len(root) + 1 :]
+    return normalized.replace("\\", "/")
+
+
+def parse_worktree_list_porcelain(porcelain: str) -> list[dict[str, Any]]:
+    """Parse ``git worktree list --porcelain`` into worktree records."""
+    entries: list[dict[str, Any]] = []
+    current: dict[str, Any] = {}
+    for raw in (porcelain or "").splitlines():
+        line = raw.strip()
+        if not line:
+            if current:
+                entries.append(current)
+                current = {}
+            continue
+        if line.startswith("worktree "):
+            if current:
+                entries.append(current)
+            current = {"path": line.split(" ", 1)[1].strip()}
+        elif line.startswith("HEAD "):
+            current["head_sha"] = line.split(" ", 1)[1].strip()
+        elif line.startswith("branch "):
+            current["branch"] = line.split(" ", 1)[1].strip().removeprefix("refs/heads/")
+        elif line == "detached":
+            current["detached"] = True
+        elif line == "bare":
+            current["bare"] = True
+    if current:
+        entries.append(current)
+    return entries
+
+
+def list_branches_directories(project_root: str, dir_names: list[str] | None = None) -> list[str]:
+    """Return relative ``branches/<name>`` paths for first-level directories."""
+    branches_root = os.path.join(project_root, "branches")
+    if dir_names is not None:
+        return sorted(
+            f"branches/{name}"
+            for name in dir_names
+            if name and not name.startswith(".")
+        )
+    if not os.path.isdir(branches_root):
+        return []
+    names: list[str] = []
+    for entry in sorted(os.listdir(branches_root)):
+        full = os.path.join(branches_root, entry)
+        if entry.startswith(".") or not os.path.isdir(full):
+            continue
+        names.append(f"branches/{entry}")
+    return names
+
+
+def _untracked_dirty(porcelain: str) -> bool:
+    return any(line.startswith("??") for line in (porcelain or "").splitlines())
+
+
+def classify_branches_entry(
+    *,
+    rel_path: str,
+    worktree_record: dict[str, Any] | None,
+    worktree_state: dict[str, Any] | None,
+    open_pr_branches: set[str] | None = None,
+    active_lock_branches: set[str] | None = None,
+    active_issue_branches: set[str] | None = None,
+) -> str:
+    """Classify a ``branches/`` directory for cleanup policy."""
+    open_pr_branches = open_pr_branches or set()
+    active_lock_branches = active_lock_branches or set()
+    active_issue_branches = active_issue_branches or set()
+    state = worktree_state or {}
+    record = worktree_record or {}
+
+    branch_name = (record.get("branch") or "").strip()
+    folder_name = rel_path.split("/", 1)[-1] if "/" in rel_path else rel_path
+    inferred_branch = folder_name.replace("-", "/") if "/" not in folder_name else folder_name
+
+    candidate_branches = {b for b in (branch_name, inferred_branch) if b}
+    open_folder_names = {branch_worktree_folder(b) for b in open_pr_branches}
+    if folder_name in open_folder_names or any(
+        b in open_pr_branches for b in candidate_branches
+    ):
+        return "active_open_pr"
+    if any(b in active_lock_branches or b in active_issue_branches for b in candidate_branches):
+        return "active_issue_work"
+
+    dirty_tracked = bool(state.get("dirty_files"))
+    dirty_untracked = bool(state.get("dirty_untracked"))
+    if dirty_tracked or dirty_untracked:
+        return "dirty_local_worktree"
+
+    if not record:
+        return "orphan_directory"
+
+    if record.get("detached") and REVIEW_WORKTREE_RE.search(rel_path):
+        return "detached_review_leftover"
+
+    if state.get("exists") and state.get("clean"):
+        return "clean_stale_removable"
+
+    return "unsafe_unknown"
+
+
+def capture_cleanup_snapshot(
+    project_root: str,
+    *,
+    branch_dirs: list[str] | None = None,
+    worktree_porcelain: str | None = None,
+    open_pr_branches: set[str] | None = None,
+    active_lock_branches: set[str] | None = None,
+    active_issue_branches: set[str] | None = None,
+    issue_lock_path: str | None = None,
+) -> dict[str, Any]:
+    """Capture audit snapshot for ``branches/`` dirs and registered worktrees."""
+    root = normalize_path(project_root)
+    rel_dirs = list_branches_directories(root, branch_dirs)
+    worktrees = parse_worktree_list_porcelain(worktree_porcelain or "")
+    worktree_by_rel: dict[str, dict[str, Any]] = {}
+    for wt in worktrees:
+        rel = relative_branches_path(root, wt.get("path") or "")
+        if rel.startswith("branches/"):
+            worktree_by_rel[rel] = wt
+
+    lock_branches = set(active_lock_branches or [])
+    lock = None
+    if issue_lock_path:
+        from merged_cleanup_reconcile import read_issue_lock
+
+        lock = read_issue_lock(issue_lock_path)
+        if lock and lock.get("branch_name"):
+            lock_branches.add(str(lock["branch_name"]))
+
+    entries: list[dict[str, Any]] = []
+    for rel_path in rel_dirs:
+        abs_path = os.path.join(root, rel_path)
+        wt_record = worktree_by_rel.get(rel_path)
+        state = read_local_worktree_state(abs_path) if os.path.isdir(abs_path) else {
+            "exists": False,
+            "clean": None,
+            "dirty_files": [],
+        }
+        if state.get("exists"):
+            status_res = state.get("porcelain_status")
+            if status_res is None and os.path.isdir(abs_path):
+                import subprocess
+
+                proc = subprocess.run(
+                    ["git", "-C", abs_path, "status", "--porcelain"],
+                    capture_output=True,
+                    text=True,
+                    check=False,
+                )
+                status_res = proc.stdout or ""
+            state["dirty_untracked"] = _untracked_dirty(status_res or "")
+            state["dirty_files"] = state.get("dirty_files") or parse_dirty_tracked_files(
+                status_res or ""
+            )
+            state["clean"] = not state["dirty_files"] and not state["dirty_untracked"]
+
+        classification = classify_branches_entry(
+            rel_path=rel_path,
+            worktree_record=wt_record,
+            worktree_state=state,
+            open_pr_branches=open_pr_branches,
+            active_lock_branches=lock_branches,
+            active_issue_branches=active_issue_branches,
+        )
+        entries.append(
+            {
+                "path": rel_path,
+                "absolute_path": abs_path,
+                "classification": classification,
+                "registered_worktree": bool(wt_record),
+                "worktree_record": wt_record or None,
+                "worktree_state": state,
+                "preserve": classification in PRESERVE_CLASSIFICATIONS,
+            }
+        )
+
+    return {
+        "project_root": root,
+        "branch_directory_count": len(rel_dirs),
+        "registered_branches_worktree_count": len(worktree_by_rel),
+        "entries": entries,
+        "worktrees": worktrees,
+    }
+
+
+def _index_snapshot_entries(snapshot: dict[str, Any]) -> dict[str, dict[str, Any]]:
+    return {entry["path"]: entry for entry in snapshot.get("entries") or []}
+
+
+def _removal_paths(removal_log: list[dict[str, Any]] | None) -> dict[str, dict[str, Any]]:
+    indexed: dict[str, dict[str, Any]] = {}
+    for item in removal_log or []:
+        rel = (item.get("path") or "").strip().replace("\\", "/")
+        if rel:
+            indexed[rel] = item
+    return indexed
+
+
+def reconcile_cleanup_audit(
+    before: dict[str, Any],
+    after: dict[str, Any],
+    removal_log: list[dict[str, Any]] | None = None,
+    *,
+    explained_missing: dict[str, str] | None = None,
+    concurrent_mutations: list[str] | None = None,
+) -> dict[str, Any]:
+    """Reconcile before/after snapshots to exactly one disposition per path."""
+    before_index = _index_snapshot_entries(before)
+    after_index = _index_snapshot_entries(after)
+    removals = _removal_paths(removal_log)
+    explained = {
+        (k or "").strip().replace("\\", "/"): (v or "").strip()
+        for k, v in (explained_missing or {}).items()
+    }
+    concurrent = {
+        (p or "").strip().replace("\\", "/")
+        for p in (concurrent_mutations or [])
+    }
+
+    rows: list[dict[str, Any]] = []
+    for path, before_entry in sorted(before_index.items()):
+        after_entry = after_index.get(path)
+        after_exists = bool(after_entry and after_entry.get("worktree_state", {}).get("exists"))
+        classification = before_entry.get("classification") or "unsafe_unknown"
+        preserve = bool(before_entry.get("preserve")) or classification in PRESERVE_CLASSIFICATIONS
+        removal = removals.get(path)
+        explanation = explained.get(path, "")
+
+        if removal:
+            disposition = "removed_intentionally"
+            reasons = []
+        elif after_exists:
+            disposition = "preserved_exists"
+            reasons = []
+        elif explanation:
+            disposition = "preserved_missing_explained"
+            reasons = [explanation]
+        elif not before_entry.get("registered_worktree"):
+            disposition = "not_registered_worktree"
+            reasons = ["directory was not a registered git worktree at audit start"]
+        elif path in concurrent:
+            disposition = "preserved_missing_explained"
+            reasons = ["removed or mutated by another session during cleanup"]
+        elif preserve:
+            disposition = "unsafe_unknown"
+            reasons = [
+                f"preserved classification '{classification}' disappeared without "
+                "removal log or explanation"
+            ]
+        else:
+            disposition = "unsafe_unknown"
+            reasons = [
+                "clean/removable path disappeared without removal log entry"
+            ]
+
+        rows.append(
+            {
+                "path": path,
+                "classification": classification,
+                "preserve": preserve,
+                "disposition": disposition,
+                "removed_intentionally": disposition == "removed_intentionally",
+                "removal_record": removal,
+                "after_exists": after_exists,
+                "reasons": reasons,
+            }
+        )
+
+    for path, removal in removals.items():
+        if path not in before_index:
+            rows.append(
+                {
+                    "path": path,
+                    "classification": "unsafe_unknown",
+                    "preserve": False,
+                    "disposition": "unsafe_unknown",
+                    "removed_intentionally": True,
+                    "removal_record": removal,
+                    "after_exists": path in after_index,
+                    "reasons": ["removal log references path absent from before snapshot"],
+                }
+            )
+
+    counts = {
+        "initial_count": len(before_index),
+        "removed_count": sum(1 for r in rows if r["disposition"] == "removed_intentionally"),
+        "preserved_count": sum(1 for r in rows if r["disposition"] == "preserved_exists"),
+        "missing_unexplained_count": sum(
+            1
+            for r in rows
+            if r["disposition"] == "unsafe_unknown"
+            and r.get("preserve")
+            and not r.get("after_exists")
+        ),
+        "missing_explained_count": sum(
+            1 for r in rows if r["disposition"] == "preserved_missing_explained"
+        ),
+        "final_count": len(after_index),
+        "orphan_directory_count": sum(
+            1 for r in rows if r["disposition"] == "not_registered_worktree"
+        ),
+    }
+    expected_final = (
+        counts["initial_count"]
+        - counts["removed_count"]
+        - counts["missing_explained_count"]
+    )
+    counts["count_reconciles"] = counts["final_count"] == expected_final
+
+    return {
+        "rows": rows,
+        "counts": counts,
+        "removal_log_complete": _removal_log_complete(before_index, after_index, removals),
+    }
+
+
+def _removal_log_complete(
+    before_index: dict[str, dict[str, Any]],
+    after_index: dict[str, dict[str, Any]],
+    removals: dict[str, dict[str, Any]],
+) -> bool:
+    for path, before_entry in before_index.items():
+        if path in after_index:
+            continue
+        classification = before_entry.get("classification") or ""
+        if classification == "clean_stale_removable" and path not in removals:
+            return False
+    return True
+
+
+def assess_cleanup_audit_integrity(reconciliation: dict[str, Any]) -> dict[str, Any]:
+    """Fail closed when preserved worktrees vanish or counts do not reconcile."""
+    reasons: list[str] = []
+    counts = reconciliation.get("counts") or {}
+
+    if counts.get("missing_unexplained_count"):
+        reasons.append(
+            f"{counts['missing_unexplained_count']} preserved worktree(s) missing "
+            "without explanation"
+        )
+
+    for row in reconciliation.get("rows") or []:
+        if not row.get("preserve") or row.get("after_exists"):
+            continue
+        if row.get("disposition") == "removed_intentionally":
+            continue
+        message = (
+            f"preserved worktree {row.get('path')} missing "
+            f"({row.get('disposition')})"
+        )
+        if message not in reasons:
+            reasons.append(message)
+        for item in row.get("reasons") or []:
+            if item not in reasons:
+                reasons.append(item)
+
+    if not counts.get("count_reconciles"):
+        reasons.append(
+            "final directory count does not reconcile with initial minus removed "
+            f"(initial={counts.get('initial_count')}, removed={counts.get('removed_count')}, "
+            f"final={counts.get('final_count')})"
+        )
+
+    if reconciliation.get("removal_log_complete") is False:
+        reasons.append("removal log omits one or more removed clean-stale worktrees")
+
+    for row in reconciliation.get("rows") or []:
+        removal = row.get("removal_record") or {}
+        if row.get("disposition") == "removed_intentionally":
+            if not removal.get("method"):
+                reasons.append(f"removal log for {row.get('path')} missing method")
+            if not removal.get("pre_removal_proof"):
+                reasons.append(f"removal log for {row.get('path')} missing pre-removal proof")
+
+    block = bool(reasons)
+    return {
+        "block": block,
+        "proven": not block,
+        "reasons": reasons,
+        "counts": counts,
+        "safe_next_action": (
+            "capture before/after snapshots, record every removal with proof, and "
+            "explain any preserved path that disappears"
+            if block
+            else "proceed"
+        ),
+    }
+
+
+def build_cleanup_reconciliation_table(reconciliation: dict[str, Any]) -> dict[str, Any]:
+    """Return the operator-facing reconciliation summary table."""
+    counts = dict(reconciliation.get("counts") or {})
+    return {
+        "initial_count": counts.get("initial_count", 0),
+        "removed_count": counts.get("removed_count", 0),
+        "preserved_count": counts.get("preserved_count", 0),
+        "missing_unexplained_count": counts.get("missing_unexplained_count", 0),
+        "missing_explained_count": counts.get("missing_explained_count", 0),
+        "final_count": counts.get("final_count", 0),
+        "count_reconciles": counts.get("count_reconciles", False),
+    }
+
+
+def _read_worktree_porcelain(project_root: str) -> str:
+    import subprocess
+
+    try:
+        res = subprocess.run(
+            ["git", "-C", project_root, "worktree", "list", "--porcelain"],
+            capture_output=True,
+            text=True,
+            check=False,
+        )
+    except OSError:
+        return ""
+    return res.stdout if res.returncode == 0 else ""
+
+
+def capture_branches_worktree_snapshot(
+    project_root: str,
+    *,
+    open_pr_branches: list[str] | None = None,
+    active_lock_branch: str | None = None,
+    leased_paths: list[str] | None = None,
+    issue_lock_path: str | None = None,
+) -> dict[str, Any]:
+    """MCP-facing snapshot capture for live ``branches/`` cleanup audits."""
+    root = normalize_path(project_root)
+    lock_branches = set()
+    if active_lock_branch:
+        lock_branches.add(active_lock_branch)
+    issue_branches = set()
+    for path in leased_paths or []:
+        rel = relative_branches_path(root, path)
+        if rel.startswith("branches/"):
+            issue_branches.add(rel.split("/", 1)[-1].replace("-", "/"))
+    return capture_cleanup_snapshot(
+        root,
+        worktree_porcelain=_read_worktree_porcelain(root),
+        open_pr_branches=set(open_pr_branches or []),
+        active_lock_branches=lock_branches,
+        active_issue_branches=issue_branches,
+        issue_lock_path=issue_lock_path,
+    )
+
+
+def assess_worktree_cleanup_integrity(
+    *,
+    before: dict[str, Any],
+    after: dict[str, Any],
+    removals: list[dict[str, Any]] | None = None,
+    explained_missing: dict[str, str] | None = None,
+) -> dict[str, Any]:
+    """MCP-facing integrity assessment over before/after cleanup snapshots."""
+    reconciliation = reconcile_cleanup_audit(
+        before,
+        after,
+        removals,
+        explained_missing=explained_missing,
+    )
+    integrity = assess_cleanup_audit_integrity(reconciliation)
+    return {
+        **integrity,
+        "integrity_passed": integrity.get("proven", False),
+        "reconciliation": reconciliation,
+        "reconciliation_table": build_cleanup_reconciliation_table(reconciliation),
+        "rows": reconciliation.get("rows") or [],
+    }
+
+
+_RECON_INITIAL_RE = re.compile(r"initial count\s*:\s*(\d+)", re.I)
+_RECON_REMOVED_RE = re.compile(r"removed count\s*:\s*(\d+)", re.I)
+_RECON_PRESERVED_RE = re.compile(r"preserved count\s*:\s*(\d+)", re.I)
+_RECON_MISSING_RE = re.compile(r"missing-unexplained count\s*:\s*(\d+)", re.I)
+_RECON_FINAL_RE = re.compile(r"final count\s*:\s*(\d+)", re.I)
+_WORKTREE_LIST_RE = re.compile(r"git worktree list|worktree list proof", re.I)
+
+
+def assess_cleanup_audit_final_report(report_text: str) -> dict[str, Any]:
+    """Validate cleanup final report includes reconciliation proof (#404)."""
+    text = report_text or ""
+    reasons: list[str] = []
+    for pattern in (
+        _RECON_INITIAL_RE,
+        _RECON_REMOVED_RE,
+        _RECON_PRESERVED_RE,
+        _RECON_MISSING_RE,
+        _RECON_FINAL_RE,
+    ):
+        if not pattern.search(text):
+            reasons.append(
+                f"cleanup report missing field matching /{pattern.pattern}/"
+            )
+    if not _WORKTREE_LIST_RE.search(text):
+        reasons.append("final verification missing git worktree list proof")
+    proven = not reasons
+    return {"proven": proven, "block": not proven, "reasons": reasons}