fix: resolve conflicts for PR #417

Merge prgs/master into feat/issue-404-worktree-cleanup-audit and keep both
worktree cleanup audit proof (#404) and reviewer validation cwd proof from master.

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

docs/llm-workflow-runbooks.md

@@ -492,25 +492,6 @@ Root-level matches are listed in `.gitignore` so they never get committed.
 `gitea_get_runtime_context` and `gitea_lock_issue` surface **warnings** (not
 hard blocks) when these artifacts are still present.
 
-## Capability preflight lifetime (#470)
-
-After `gitea_resolve_task_capability(task=…)` proves the mutation is allowed,
-interleaved **read-only** calls preserve that proof until a gated mutation
-consumes it:
-
-- Safe reads: `gitea_whoami`, `gitea_view_pr`, `gitea_view_issue`, `gitea_list_*`,
-  `gitea_get_runtime_context`, `gitea_check_pr_eligibility`, and related
-  read-only inventory/eligibility tools (see `preflight_contract.py`).
-- Each mutation consumes the proof once; call `gitea_resolve_task_capability`
-  again immediately before the next mutation on the same task.
-- Resolving capability for a **different** task replaces the prior task binding.
-- Workspace edits before resolve, profile switches, or a dirty whoami baseline
-  invalidate proof (fail closed).
-
-If a mutation fails with “capability has not been resolved” or “task mismatch”,
-re-run `gitea_resolve_task_capability(task="<mutation>")` immediately before
-retrying — do not guess or skip the resolve step.
-
 Implementation work and review work must use separate branch folders. For
 example, an implementation branch might live under
 `branches/fix-issue-123-example`, while a review branch for the resulting PR

final_report_validator.py

@@ -492,6 +492,27 @@ def _rule_reviewer_validation_failure_history(
     ]
 
 
+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_cwd_proof(
     report_text: str,
     *,
@@ -1096,6 +1117,7 @@ _RULES_BY_TASK: dict[str, list[Callable[..., list[dict[str, str]]]]] = {
         _rule_shared_email_disclosure,
         *_SHARED_ISSUE_LOCK_RULES,
         _rule_reviewer_vague_mutations_none,
+        _rule_worktree_cleanup_audit_proof,
         _rule_conflict_fix_push_proof,
     ],
     "issue_filing": [

gitea_mcp_server.py

@@ -165,7 +165,6 @@ _preflight_capability_called = False
 _preflight_whoami_violation = False
 _preflight_capability_violation = False
 _preflight_resolved_role = None
-_preflight_resolved_task: str | None = None
 _process_start_porcelain: str | None = None
 _preflight_whoami_baseline_porcelain: str | None = None
 _preflight_capability_baseline_porcelain: str | None = None
@@ -380,30 +379,11 @@ def assess_preflight_status(worktree_path: str | None = None) -> dict:
     }
 
 
-def _clear_preflight_capability_state() -> None:
+def record_preflight_check(type_name: str, resolved_role: str | None = None):
-    """Drop resolved capability proof (consumed by a mutation or fresh resolve)."""
-    global _preflight_capability_called, _preflight_capability_violation
-    global _preflight_resolved_task
-    global _preflight_capability_baseline_porcelain, _preflight_capability_violation_files
-    global _preflight_reviewer_violation_files
-
-    _preflight_capability_called = False
-    _preflight_capability_violation = False
-    _preflight_capability_violation_files = []
-    _preflight_capability_baseline_porcelain = None
-    _preflight_resolved_task = None
-    _preflight_reviewer_violation_files = []
-
-
-def record_preflight_check(
-    type_name: str,
-    resolved_role: str | None = None,
-    resolved_task: str | None = None,
-):
     """Record a pre-flight check (whoami or capability) with session-scoped deltas."""
     global _preflight_whoami_called, _preflight_capability_called
     global _preflight_whoami_violation, _preflight_capability_violation
-    global _preflight_resolved_role, _preflight_resolved_task
+    global _preflight_resolved_role
     global _preflight_whoami_baseline_porcelain, _preflight_capability_baseline_porcelain
     global _preflight_whoami_violation_files, _preflight_capability_violation_files
     global _preflight_reviewer_violation_files
@@ -411,24 +391,13 @@ def record_preflight_check(
     current = _get_workspace_porcelain()
 
     if type_name == "whoami":
-        # Re-evaluate whoami violations instead of replaying sticky state (#252).
+        # Fresh whoami restarts the capability step and re-evaluates violations
-        # Interleaved read-only whoami must not clear a valid capability proof (#469).
+        # instead of replaying a sticky record (#252).
-        saved_capability = None
+        _preflight_capability_called = False
-        preserve_capability = (
+        _preflight_capability_violation = False
-            _preflight_capability_called
+        _preflight_capability_violation_files = []
-            and _preflight_whoami_called
+        _preflight_capability_baseline_porcelain = None
-            and not _preflight_whoami_violation
+        _preflight_reviewer_violation_files = []
-        )
-        if preserve_capability:
-            saved_capability = (
-                _preflight_capability_violation,
-                list(_preflight_capability_violation_files),
-                _preflight_capability_baseline_porcelain,
-                _preflight_resolved_role,
-                _preflight_resolved_task,
-            )
-        else:
-            _clear_preflight_capability_state()
 
         process_start = _ensure_process_start_porcelain()
         whoami_delta = _new_tracked_changes_since(process_start, current)
@@ -436,20 +405,6 @@ def record_preflight_check(
         _preflight_whoami_violation_files = whoami_delta
         _preflight_whoami_baseline_porcelain = current
         _preflight_whoami_called = True
-
-        if (
-            preserve_capability
-            and saved_capability is not None
-            and not whoami_delta
-        ):
-            (
-                _preflight_capability_violation,
-                _preflight_capability_violation_files,
-                _preflight_capability_baseline_porcelain,
-                _preflight_resolved_role,
-                _preflight_resolved_task,
-            ) = saved_capability
-            _preflight_capability_called = True
     elif type_name == "capability":
         baseline = _preflight_whoami_baseline_porcelain or ""
         capability_delta = _new_tracked_changes_since(baseline, current)
@@ -459,8 +414,6 @@ def record_preflight_check(
         _preflight_capability_called = True
         if resolved_role:
             _preflight_resolved_role = resolved_role
-        if resolved_task:
-            _preflight_resolved_task = resolved_task
 
 
 def _enforce_branches_only_author_mutation(worktree_path: str | None = None) -> None:
@@ -486,11 +439,7 @@ def _enforce_branches_only_author_mutation(worktree_path: str | None = None) ->
         )
 
 
-def verify_preflight_purity(
+def verify_preflight_purity(remote: str | None = None, worktree_path: str | None = None):
-    remote: str | None = None,
-    worktree_path: str | None = None,
-    task: str | None = None,
-):
     """Verify that identity and capability were verified prior to session edits."""
     global _preflight_reviewer_violation_files
 
@@ -507,27 +456,7 @@ def verify_preflight_purity(
         )
     if not _preflight_capability_called:
         raise RuntimeError(
-            preflight_contract.format_missing_capability_error(task)
+            "Pre-flight order violation: Task capability (gitea_resolve_task_capability) has not been resolved (fail closed)"
-        )
-    if (
-        task is not None
-        and _preflight_resolved_task is not None
-        and task != _preflight_resolved_task
-    ):
-        raise RuntimeError(
-            preflight_contract.format_task_mismatch_error(
-                _preflight_resolved_task, task
-            )
-        )
-    if (
-        task is not None
-        and _preflight_resolved_task is not None
-        and task != _preflight_resolved_task
-    ):
-        raise RuntimeError(
-            "Pre-flight task mismatch: "
-            f"resolved '{_preflight_resolved_task}' but mutation requires "
-            f"'{task}' (fail closed)"
         )
 
     ctx = _resolve_author_mutation_context(worktree_path)
@@ -584,7 +513,6 @@ def verify_preflight_purity(
                 )
 
     _enforce_branches_only_author_mutation(worktree_path)
-    _clear_preflight_capability_state()
 
 from mcp.server.fastmcp import FastMCP  # noqa: E402
 
@@ -616,11 +544,11 @@ import stacked_pr_support  # noqa: E402
 import merge_approval_gate  # noqa: E402
 import already_landed_reconcile  # noqa: E402
 import author_mutation_worktree  # noqa: E402
-import preflight_contract  # noqa: E402
 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 review_merge_state_machine  # noqa: E402
@@ -1301,7 +1229,7 @@ def gitea_create_issue(
     )
     if blocked:
         return blocked
-    verify_preflight_purity(remote, worktree_path=worktree_path, task="create_issue")
+    verify_preflight_purity(remote, worktree_path=worktree_path)
     base = repo_api_url(h, o, r)
     open_issues = api_get_all(f"{base}/issues?state=open&type=issues", auth)
     closed_issues = api_get_all(
@@ -1434,7 +1362,7 @@ def gitea_lock_issue(
         )
     else:
         git_state = issue_lock_worktree.read_worktree_git_state(resolved_worktree)
-    verify_preflight_purity(remote, worktree_path=resolved_worktree, task="lock_issue")
+    verify_preflight_purity(remote, worktree_path=resolved_worktree)
     lock_assessment = issue_lock_worktree.assess_issue_lock_worktree(
         worktree_path=resolved_worktree,
         current_branch=git_state.get("current_branch"),
@@ -1658,7 +1586,7 @@ def gitea_create_pr(
     )
     if blocked:
         return blocked
-    verify_preflight_purity(remote, worktree_path=worktree_path, task="create_pr")
+    verify_preflight_purity(remote, worktree_path=worktree_path)
     h, o, r = _resolve(remote, host, org, repo)
 
     # ── Issue Lock Validation (Issue #194 / #196 / #443) ──
@@ -2724,7 +2652,7 @@ def _evaluate_pr_review_submission(
     final_review_decision_ready: bool = False,
 ) -> dict:
     """Shared gate chain for live submit and dry-run review tools."""
-    verify_preflight_purity(remote, task="review_pr")
+    verify_preflight_purity(remote)
     action = (action or "").strip().lower()
     result = {
         "requested_action": action,
@@ -3231,12 +3159,12 @@ def gitea_edit_pr(
     if not payload:
         raise ValueError("At least one field to edit (title, body, state, base) must be provided.")
 
-    closing = payload.get("state") == "closed"
+    verify_preflight_purity(remote)
-    verify_preflight_purity(remote, task="close_pr" if closing else None)
 
     # PR closure is a first-class capability, distinct from retitling or
     # rebasing edits (#216). Gate BEFORE auth/API setup so a blocked close
     # never touches the network.
+    closing = payload.get("state") == "closed"
     if closing:
         gate_reasons = _profile_operation_gate("gitea.pr.close")
         if gate_reasons:
@@ -3481,7 +3409,7 @@ def gitea_commit_files(
             branch="",
         )
 
-    verify_preflight_purity(remote, task="commit_files")
+    verify_preflight_purity(remote)
     processed_files, source_proofs = _prepare_commit_payload_files(files)
 
     h, o, r = _resolve(remote, host, org, repo)
@@ -3582,7 +3510,7 @@ def gitea_merge_pr(
         reasons/gates passed or blocked, and merge result / merge commit if
         available. Never secrets.
     """
-    verify_preflight_purity(remote, task="merge_pr")
+    verify_preflight_purity(remote)
     do = (do or "").strip().lower()
     result = {
         "performed": False,
@@ -4111,7 +4039,7 @@ def gitea_delete_branch(
             "permission_report": _permission_block_report("gitea.branch.delete"),
         }
 
-    verify_preflight_purity(remote, task="delete_branch")
+    verify_preflight_purity(remote)
     h, o, r = _resolve(remote, host, org, repo)
     auth = _auth(h)
     import urllib.parse
@@ -4142,6 +4070,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,
@@ -4220,7 +4201,7 @@ def gitea_reconcile_merged_cleanups(
         report["executed"] = False
         return {"success": True, "performed": False, **report}
 
-    verify_preflight_purity(remote, task="reconcile_merged_cleanups")
+    verify_preflight_purity(remote)
     actions: list[dict] = []
     for entry in report.get("entries") or []:
         head_branch = entry.get("head_branch") or ""
@@ -4534,7 +4515,7 @@ def gitea_reconcile_already_landed_pr(
         )
         return result
 
-    verify_preflight_purity(remote, task="reconcile_already_landed_pr")
+    verify_preflight_purity(remote)
 
     if post_comment and comment_body.strip():
         comment_block = _profile_operation_gate("gitea.pr.comment")
@@ -4637,7 +4618,7 @@ def gitea_close_issue(
         task_capability_map.required_permission("close_issue"))
     if blocked:
         return blocked
-    verify_preflight_purity(remote, task="close_issue")
+    verify_preflight_purity(remote)
     h, o, r = _resolve(remote, host, org, repo)
     auth = _auth(h)
     url = f"{repo_api_url(h, o, r)}/issues/{issue_number}"
@@ -5065,7 +5046,7 @@ def gitea_acquire_reviewer_pr_lease(
             "permission_report": _permission_block_report("gitea.pr.comment"),
         }
 
-    verify_preflight_purity(remote, task="review_pr")
+    verify_preflight_purity(remote)
     h, o, r = _resolve(remote, host, org, repo)
     auth = _auth(h)
     profile = get_profile()
@@ -5166,7 +5147,7 @@ def gitea_heartbeat_reviewer_pr_lease(
             ],
         }
 
-    verify_preflight_purity(remote, task="review_pr")
+    verify_preflight_purity(remote)
     h, o, r = _resolve(remote, host, org, repo)
     auth = _auth(h)
     body = reviewer_pr_lease.format_lease_body(
@@ -5342,7 +5323,7 @@ def gitea_create_issue_comment(
         (permission blocks also carry a structured 'permission_report',
         #142).
     """
-    verify_preflight_purity(remote, task="comment_issue")
+    verify_preflight_purity(remote)
     gate_reasons = _profile_operation_gate("gitea.issue.comment")
     reasons = list(gate_reasons)
     if not (body or "").strip():
@@ -6876,7 +6857,7 @@ def gitea_mark_issue(
         task_capability_map.required_permission("mark_issue"))
     if blocked:
         return blocked
-    verify_preflight_purity(remote, worktree_path=worktree_path, task="mark_issue")
+    verify_preflight_purity(remote, worktree_path=worktree_path)
     h, o, r = _resolve(remote, host, org, repo)
     auth = _auth(h)
     base = repo_api_url(h, o, r)
@@ -6954,7 +6935,7 @@ def gitea_post_heartbeat(
         task_capability_map.required_permission("post_heartbeat"))
     if blocked:
         return blocked
-    verify_preflight_purity(remote, task="post_heartbeat")
+    verify_preflight_purity(remote)
     active_profile = profile or get_profile().get("profile_name")
     body = issue_claim_heartbeat.format_heartbeat_body(
         kind="progress",
@@ -6993,9 +6974,7 @@ def gitea_acquire_conflict_fix_lease(
         task_capability_map.required_permission("comment_issue"))
     if blocked:
         return blocked
-    verify_preflight_purity(
+    verify_preflight_purity(remote, worktree_path=worktree_path)
-        remote, worktree_path=worktree_path, task="comment_issue"
-    )
     comments = _list_pr_lease_comments(
         pr_number,
         remote=remote,
@@ -7199,7 +7178,7 @@ def gitea_cleanup_stale_claims(
         task_capability_map.required_permission("cleanup_stale_claims"))
     if blocked:
         return blocked
-    verify_preflight_purity(remote, task="cleanup_stale_claims")
+    verify_preflight_purity(remote)
 
     h, o, r = _resolve(remote, host, org, repo)
     auth = _auth(h)
@@ -7353,7 +7332,7 @@ def gitea_set_issue_labels(
         task_capability_map.required_permission("set_issue_labels"))
     if blocked:
         return blocked
-    verify_preflight_purity(remote, task="set_issue_labels")
+    verify_preflight_purity(remote)
     h, o, r = _resolve(remote, host, org, repo)
     auth = _auth(h)
     base = repo_api_url(h, o, r)
@@ -7590,7 +7569,7 @@ def gitea_resolve_task_capability(
                 "exact_safe_next_action": next_safe_action,
             }
 
-    record_preflight_check("capability", required_role, resolved_task=task)
+    record_preflight_check("capability", required_role)
 
     # Try automatic dispatch switching
     _ensure_matching_profile(required_permission, required_role, remote, host)

preflight_contract.py

@@ -1,66 +0,0 @@
-"""Capability preflight lifetime contract (#470).
-
-Defines which read-only MCP tools preserve an existing capability proof and the
-canonical sequencing operators must follow before mutations.
-"""
-
-from __future__ import annotations
-
-# Read-only tools that must not invalidate capability proof (#470).
-# gitea_whoami is read-only but re-pins identity; capability is preserved when
-# identity was already verified clean (#469).
-READ_ONLY_PREFLIGHT_TOOLS = frozenset({
-    "gitea_whoami",
-    "gitea_get_authenticated_user",
-    "gitea_get_current_user",
-    "gitea_view_pr",
-    "gitea_view_issue",
-    "gitea_list_prs",
-    "gitea_list_issues",
-    "gitea_list_issue_comments",
-    "gitea_get_runtime_context",
-    "gitea_resolve_task_capability",
-    "gitea_check_pr_eligibility",
-    "gitea_get_pr_review_feedback",
-    "gitea_assess_work_issue_duplicate",
-    "gitea_route_task_session",
-    "gitea_audit_config",
-    "gitea_list_labels",
-    "gitea_get_profile",
-    "gitea_list_profiles",
-})
-
-PREFLIGHT_CONTRACT_SUMMARY = (
-    "Capability preflight is session-scoped per MCP process, profile, and task. "
-    "After gitea_resolve_task_capability(task=...), interleaved read-only calls "
-    "(whoami, view_*, list_*, get_runtime_context, eligibility checks) preserve "
-    "the proof until a gated mutation consumes it. Each mutation consumes the proof "
-    "once; re-resolve immediately before the next mutation. A new resolve for a "
-    "different task replaces the prior task binding. Profile/session changes or "
-    "workspace edits before resolve invalidate proof."
-)
-
-
-def format_missing_capability_error(task: str | None = None) -> str:
-    base = (
-        "Pre-flight order violation: Task capability "
-        "(gitea_resolve_task_capability) has not been resolved (fail closed)"
-    )
-    if task:
-        return (
-            f"{base}. Re-run gitea_resolve_task_capability(task=\"{task}\") "
-            "immediately before this mutation."
-        )
-    return (
-        f"{base}. Re-run gitea_resolve_task_capability for the mutation task "
-        "immediately before acting."
-    )
-
-
-def format_task_mismatch_error(resolved: str, required: str) -> str:
-    return (
-        "Pre-flight task mismatch: "
-        f"resolved '{resolved}' but mutation requires '{required}' (fail closed). "
-        f"Re-run gitea_resolve_task_capability(task=\"{required}\") "
-        "immediately before this mutation."
-    )

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_preflight_read_survival.py

@@ -1,159 +0,0 @@
-"""#469/#470: capability preflight lifetime across safe read-only calls."""
-
-import os
-import unittest
-
-import gitea_mcp_server as mcp_server
-
-
-class TestPreflightReadSurvival(unittest.TestCase):
-    def setUp(self):
-        self.orig_whoami = mcp_server._preflight_whoami_called
-        self.orig_capability = mcp_server._preflight_capability_called
-        self.orig_whoami_violation = mcp_server._preflight_whoami_violation
-        self.orig_capability_violation = mcp_server._preflight_capability_violation
-        self.orig_resolved_role = mcp_server._preflight_resolved_role
-        self.orig_resolved_task = mcp_server._preflight_resolved_task
-        self.orig_process_start = mcp_server._process_start_porcelain
-        self.orig_whoami_baseline = mcp_server._preflight_whoami_baseline_porcelain
-        self.orig_capability_baseline = mcp_server._preflight_capability_baseline_porcelain
-        for key in ("GITEA_TEST_FORCE_DIRTY", "GITEA_TEST_PORCELAIN"):
-            if key in os.environ:
-                del os.environ[key]
-        os.environ["GITEA_TEST_PORCELAIN"] = ""
-        mcp_server._preflight_whoami_called = False
-        mcp_server._preflight_capability_called = False
-        mcp_server._preflight_whoami_violation = False
-        mcp_server._preflight_capability_violation = False
-        mcp_server._preflight_resolved_role = None
-        mcp_server._preflight_resolved_task = None
-        mcp_server._process_start_porcelain = ""
-        mcp_server._preflight_whoami_baseline_porcelain = None
-        mcp_server._preflight_capability_baseline_porcelain = None
-
-    def tearDown(self):
-        mcp_server._preflight_whoami_called = self.orig_whoami
-        mcp_server._preflight_capability_called = self.orig_capability
-        mcp_server._preflight_whoami_violation = self.orig_whoami_violation
-        mcp_server._preflight_capability_violation = self.orig_capability_violation
-        mcp_server._preflight_resolved_role = self.orig_resolved_role
-        mcp_server._preflight_resolved_task = self.orig_resolved_task
-        mcp_server._process_start_porcelain = self.orig_process_start
-        mcp_server._preflight_whoami_baseline_porcelain = self.orig_whoami_baseline
-        mcp_server._preflight_capability_baseline_porcelain = self.orig_capability_baseline
-        for key in ("GITEA_TEST_FORCE_DIRTY", "GITEA_TEST_PORCELAIN"):
-            if key in os.environ:
-                del os.environ[key]
-
-    def test_interleaved_whoami_preserves_capability(self):
-        mcp_server.record_preflight_check("whoami")
-        mcp_server.record_preflight_check(
-            "capability", resolved_role="reconciler", resolved_task="close_pr"
-        )
-        self.assertTrue(mcp_server._preflight_capability_called)
-        self.assertEqual(mcp_server._preflight_resolved_task, "close_pr")
-
-        mcp_server.record_preflight_check("whoami")
-        self.assertTrue(mcp_server._preflight_capability_called)
-        self.assertEqual(mcp_server._preflight_resolved_task, "close_pr")
-
-        mcp_server.verify_preflight_purity(task="close_pr")
-        self.assertFalse(mcp_server._preflight_capability_called)
-
-    def test_missing_capability_still_fails_closed(self):
-        mcp_server.record_preflight_check("whoami")
-        with self.assertRaises(RuntimeError) as ctx:
-            mcp_server.verify_preflight_purity(task="close_pr")
-        self.assertIn("has not been resolved", str(ctx.exception))
-
-    def test_task_mismatch_fails_closed(self):
-        mcp_server.record_preflight_check("whoami")
-        mcp_server.record_preflight_check(
-            "capability", resolved_role="author", resolved_task="create_issue"
-        )
-        with self.assertRaises(RuntimeError) as ctx:
-            mcp_server.verify_preflight_purity(task="close_pr")
-        self.assertIn("task mismatch", str(ctx.exception))
-
-    def test_capability_consumed_after_mutation_gate(self):
-        mcp_server.record_preflight_check("whoami")
-        mcp_server.record_preflight_check(
-            "capability", resolved_role="author", resolved_task="create_issue"
-        )
-        mcp_server.verify_preflight_purity(task="create_issue")
-        with self.assertRaises(RuntimeError) as ctx:
-            mcp_server.verify_preflight_purity(task="create_issue")
-        self.assertIn("has not been resolved", str(ctx.exception))
-
-    def test_whoami_recovery_after_violation_clears_capability(self):
-        os.environ["GITEA_TEST_FORCE_DIRTY"] = "1"
-        mcp_server.record_preflight_check("whoami")
-        self.assertTrue(mcp_server._preflight_whoami_violation)
-
-        del os.environ["GITEA_TEST_FORCE_DIRTY"]
-        os.environ["GITEA_TEST_PORCELAIN"] = ""
-        mcp_server.record_preflight_check("whoami")
-        self.assertFalse(mcp_server._preflight_whoami_violation)
-        self.assertFalse(mcp_server._preflight_capability_called)
-
-        mcp_server.record_preflight_check(
-            "capability", resolved_role="reviewer", resolved_task="review_pr"
-        )
-        mcp_server.verify_preflight_purity(task="review_pr")
-
-    def test_close_pr_sequence_with_interleaved_reads(self):
-        """resolve(close_pr) → whoami/view reads → close_pr gate (#470)."""
-        mcp_server.record_preflight_check("whoami")
-        mcp_server.record_preflight_check(
-            "capability", resolved_role="reconciler", resolved_task="close_pr"
-        )
-        # Simulate live-state revalidation between resolve and mutation.
-        mcp_server.record_preflight_check("whoami")
-        self.assertTrue(mcp_server._preflight_capability_called)
-        self.assertEqual(mcp_server._preflight_resolved_task, "close_pr")
-        mcp_server.verify_preflight_purity(task="close_pr")
-        self.assertFalse(mcp_server._preflight_capability_called)
-
-    def test_fresh_capability_resolve_replaces_prior_task(self):
-        mcp_server.record_preflight_check("whoami")
-        mcp_server.record_preflight_check(
-            "capability", resolved_role="author", resolved_task="create_issue"
-        )
-        mcp_server.record_preflight_check(
-            "capability", resolved_role="reconciler", resolved_task="close_pr"
-        )
-        self.assertEqual(mcp_server._preflight_resolved_task, "close_pr")
-        mcp_server.verify_preflight_purity(task="close_pr")
-
-    def test_missing_capability_error_names_re_resolve(self):
-        mcp_server.record_preflight_check("whoami")
-        with self.assertRaises(RuntimeError) as ctx:
-            mcp_server.verify_preflight_purity(task="close_pr")
-        msg = str(ctx.exception)
-        self.assertIn("gitea_resolve_task_capability", msg)
-        self.assertIn('task="close_pr"', msg)
-
-    def test_task_mismatch_error_names_re_resolve(self):
-        mcp_server.record_preflight_check("whoami")
-        mcp_server.record_preflight_check(
-            "capability", resolved_role="author", resolved_task="create_issue"
-        )
-        with self.assertRaises(RuntimeError) as ctx:
-            mcp_server.verify_preflight_purity(task="close_pr")
-        msg = str(ctx.exception)
-        self.assertIn("task mismatch", msg)
-        self.assertIn('task="close_pr"', msg)
-
-    def test_consumed_capability_error_names_re_resolve(self):
-        mcp_server.record_preflight_check("whoami")
-        mcp_server.record_preflight_check(
-            "capability", resolved_role="reconciler", resolved_task="close_pr"
-        )
-        mcp_server.verify_preflight_purity(task="close_pr")
-        with self.assertRaises(RuntimeError) as ctx:
-            mcp_server.verify_preflight_purity(task="close_pr")
-        self.assertIn('task="close_pr"', str(ctx.exception))
-
-
-if __name__ == "__main__":
-    unittest.main()

tests/test_worktree_cleanup_audit.py

@@ -0,0 +1,164 @@
+#!/usr/bin/env python3
+"""Regression tests for worktree cleanup audit integrity (#404)."""
+
+from __future__ import annotations
+
+import os
+import sys
+import unittest
+
+sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
+
+from worktree_cleanup_audit import (  # noqa: E402
+    assess_cleanup_audit_final_report,
+    assess_cleanup_audit_integrity,
+    assess_worktree_cleanup_integrity,
+    classify_branches_entry,
+    parse_worktree_list_porcelain,
+    reconcile_cleanup_audit,
+)
+
+
+def _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 _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 = 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 = 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 = 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 = _snapshot([_entry(path, classification="clean_stale_removable", preserve=False)])
+        after = _snapshot([_entry(path, classification="clean_stale_removable", preserve=False)])
+        result = assess_worktree_cleanup_integrity(before=before, after=after)
+        self.assertTrue(result["integrity_passed"])
+
+    def test_intentional_removal_passes(self):
+        path = "branches/remove-me"
+        before = _snapshot([_entry(path, classification="clean_stale_removable", preserve=False)])
+        after = _snapshot([])
+        result = 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 = _snapshot([_entry(path, classification="dirty_local_worktree")])
+        after = _snapshot([])
+        recon = reconcile_cleanup_audit(before, after)
+        result = assess_cleanup_audit_integrity(recon)
+        self.assertFalse(result["proven"])
+
+    def test_active_pr_worktree_disappears_fails(self):
+        path = "branches/review-pr99"
+        before = _snapshot([_entry(path, classification="active_open_pr")])
+        after = _snapshot([])
+        result = assess_worktree_cleanup_integrity(before=before, after=after)
+        self.assertFalse(result["integrity_passed"])
+
+    def test_explained_missing_allowed(self):
+        path = "branches/review-pr382"
+        before = _snapshot([
+            _entry(path, classification="detached_review_leftover", preserve=False),
+        ])
+        after = _snapshot([])
+        result = 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 = _snapshot([_entry(path, classification="clean_stale_removable", preserve=False)])
+        after = _snapshot([])
+        recon = 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 = assess_cleanup_audit_final_report(report)
+        self.assertTrue(result["proven"])
+
+    def test_incomplete_report_fails(self):
+        result = assess_cleanup_audit_final_report("removed some worktrees")
+        self.assertFalse(result["proven"])
+
+
+if __name__ == "__main__":
+    unittest.main()

worktree_cleanup_audit.py

@@ -0,0 +1,554 @@
+"""Worktree cleanup audit integrity and reconciliation (#404).
+
+Captures before/after snapshots of ``branches/`` directories and registered
+worktrees, reconciles every initial path to exactly one disposition, and fails
+closed when preserved worktrees disappear without an explicit removal record.
+"""
+
+from __future__ import annotations
+
+import os
+import re
+from typing import Any
+
+from merged_cleanup_reconcile import branch_worktree_folder, read_local_worktree_state
+from reviewer_worktree import parse_dirty_tracked_files
+
+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}