fix: resolve conflicts for PR #490

Merge prgs/master; unify preflight read-survival tests (#469 + #470 contract).

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

docs/llm-workflow-runbooks.md

@@ -492,6 +492,25 @@ 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,27 +492,6 @@ 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,
     *,
@@ -1117,7 +1096,6 @@ _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,6 +165,7 @@ _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
@@ -379,11 +380,30 @@ def assess_preflight_status(worktree_path: str | None = None) -> dict:
     }
 
 
-def record_preflight_check(type_name: str, resolved_role: str | None = None):
+def _clear_preflight_capability_state() -> 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
+    global _preflight_resolved_role, _preflight_resolved_task
     global _preflight_whoami_baseline_porcelain, _preflight_capability_baseline_porcelain
     global _preflight_whoami_violation_files, _preflight_capability_violation_files
     global _preflight_reviewer_violation_files
@@ -391,13 +411,24 @@ def record_preflight_check(type_name: str, resolved_role: str | None = None):
     current = _get_workspace_porcelain()
 
     if type_name == "whoami":
-        # Fresh whoami restarts the capability step and re-evaluates violations
-        # instead of replaying a sticky record (#252).
-        _preflight_capability_called = False
-        _preflight_capability_violation = False
-        _preflight_capability_violation_files = []
-        _preflight_capability_baseline_porcelain = None
-        _preflight_reviewer_violation_files = []
+        # Re-evaluate whoami violations instead of replaying sticky state (#252).
+        # Interleaved read-only whoami must not clear a valid capability proof (#469).
+        saved_capability = None
+        preserve_capability = (
+            _preflight_capability_called
+            and _preflight_whoami_called
+            and not _preflight_whoami_violation
+        )
+        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)
@@ -405,6 +436,20 @@ def record_preflight_check(type_name: str, resolved_role: str | None = None):
         _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)
@@ -414,6 +459,8 @@ def record_preflight_check(type_name: str, resolved_role: str | None = None):
         _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:
@@ -439,7 +486,11 @@ def _enforce_branches_only_author_mutation(worktree_path: str | None = None) ->
         )
 
 
-def verify_preflight_purity(remote: str | None = None, worktree_path: str | None = None):
+def verify_preflight_purity(
+    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
 
@@ -456,7 +507,27 @@ def verify_preflight_purity(remote: str | None = None, worktree_path: str | None
         )
     if not _preflight_capability_called:
         raise RuntimeError(
-            "Pre-flight order violation: Task capability (gitea_resolve_task_capability) has not been resolved (fail closed)"
+            preflight_contract.format_missing_capability_error(task)
+        )
+    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)
@@ -513,6 +584,7 @@ def verify_preflight_purity(remote: str | None = None, worktree_path: str | None
                 )
 
     _enforce_branches_only_author_mutation(worktree_path)
+    _clear_preflight_capability_state()
 
 from mcp.server.fastmcp import FastMCP  # noqa: E402
 
@@ -544,11 +616,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
@@ -1229,7 +1301,7 @@ def gitea_create_issue(
     )
     if blocked:
         return blocked
-    verify_preflight_purity(remote, worktree_path=worktree_path)
+    verify_preflight_purity(remote, worktree_path=worktree_path, task="create_issue")
     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(
@@ -1362,7 +1434,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)
+    verify_preflight_purity(remote, worktree_path=resolved_worktree, task="lock_issue")
     lock_assessment = issue_lock_worktree.assess_issue_lock_worktree(
         worktree_path=resolved_worktree,
         current_branch=git_state.get("current_branch"),
@@ -1586,7 +1658,7 @@ def gitea_create_pr(
     )
     if blocked:
         return blocked
-    verify_preflight_purity(remote, worktree_path=worktree_path)
+    verify_preflight_purity(remote, worktree_path=worktree_path, task="create_pr")
     h, o, r = _resolve(remote, host, org, repo)
 
     # ── Issue Lock Validation (Issue #194 / #196 / #443) ──
@@ -2652,7 +2724,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)
+    verify_preflight_purity(remote, task="review_pr")
     action = (action or "").strip().lower()
     result = {
         "requested_action": action,
@@ -3159,12 +3231,12 @@ def gitea_edit_pr(
     if not payload:
         raise ValueError("At least one field to edit (title, body, state, base) must be provided.")
 
-    verify_preflight_purity(remote)
+    closing = payload.get("state") == "closed"
+    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:
@@ -3409,7 +3481,7 @@ def gitea_commit_files(
             branch="",
         )
 
-    verify_preflight_purity(remote)
+    verify_preflight_purity(remote, task="commit_files")
     processed_files, source_proofs = _prepare_commit_payload_files(files)
 
     h, o, r = _resolve(remote, host, org, repo)
@@ -3510,7 +3582,7 @@ def gitea_merge_pr(
         reasons/gates passed or blocked, and merge result / merge commit if
         available. Never secrets.
     """
-    verify_preflight_purity(remote)
+    verify_preflight_purity(remote, task="merge_pr")
     do = (do or "").strip().lower()
     result = {
         "performed": False,
@@ -4039,7 +4111,7 @@ def gitea_delete_branch(
             "permission_report": _permission_block_report("gitea.branch.delete"),
         }
 
-    verify_preflight_purity(remote)
+    verify_preflight_purity(remote, task="delete_branch")
     h, o, r = _resolve(remote, host, org, repo)
     auth = _auth(h)
     import urllib.parse
@@ -4070,59 +4142,6 @@ 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,
@@ -4201,7 +4220,7 @@ def gitea_reconcile_merged_cleanups(
         report["executed"] = False
         return {"success": True, "performed": False, **report}
 
-    verify_preflight_purity(remote)
+    verify_preflight_purity(remote, task="reconcile_merged_cleanups")
     actions: list[dict] = []
     for entry in report.get("entries") or []:
         head_branch = entry.get("head_branch") or ""
@@ -4515,7 +4534,7 @@ def gitea_reconcile_already_landed_pr(
         )
         return result
 
-    verify_preflight_purity(remote)
+    verify_preflight_purity(remote, task="reconcile_already_landed_pr")
 
     if post_comment and comment_body.strip():
         comment_block = _profile_operation_gate("gitea.pr.comment")
@@ -4618,7 +4637,7 @@ def gitea_close_issue(
         task_capability_map.required_permission("close_issue"))
     if blocked:
         return blocked
-    verify_preflight_purity(remote)
+    verify_preflight_purity(remote, task="close_issue")
     h, o, r = _resolve(remote, host, org, repo)
     auth = _auth(h)
     url = f"{repo_api_url(h, o, r)}/issues/{issue_number}"
@@ -5046,7 +5065,7 @@ def gitea_acquire_reviewer_pr_lease(
             "permission_report": _permission_block_report("gitea.pr.comment"),
         }
 
-    verify_preflight_purity(remote)
+    verify_preflight_purity(remote, task="review_pr")
     h, o, r = _resolve(remote, host, org, repo)
     auth = _auth(h)
     profile = get_profile()
@@ -5147,7 +5166,7 @@ def gitea_heartbeat_reviewer_pr_lease(
             ],
         }
 
-    verify_preflight_purity(remote)
+    verify_preflight_purity(remote, task="review_pr")
     h, o, r = _resolve(remote, host, org, repo)
     auth = _auth(h)
     body = reviewer_pr_lease.format_lease_body(
@@ -5323,7 +5342,7 @@ def gitea_create_issue_comment(
         (permission blocks also carry a structured 'permission_report',
         #142).
     """
-    verify_preflight_purity(remote)
+    verify_preflight_purity(remote, task="comment_issue")
     gate_reasons = _profile_operation_gate("gitea.issue.comment")
     reasons = list(gate_reasons)
     if not (body or "").strip():
@@ -6857,7 +6876,7 @@ def gitea_mark_issue(
         task_capability_map.required_permission("mark_issue"))
     if blocked:
         return blocked
-    verify_preflight_purity(remote, worktree_path=worktree_path)
+    verify_preflight_purity(remote, worktree_path=worktree_path, task="mark_issue")
     h, o, r = _resolve(remote, host, org, repo)
     auth = _auth(h)
     base = repo_api_url(h, o, r)
@@ -6935,7 +6954,7 @@ def gitea_post_heartbeat(
         task_capability_map.required_permission("post_heartbeat"))
     if blocked:
         return blocked
-    verify_preflight_purity(remote)
+    verify_preflight_purity(remote, task="post_heartbeat")
     active_profile = profile or get_profile().get("profile_name")
     body = issue_claim_heartbeat.format_heartbeat_body(
         kind="progress",
@@ -6974,7 +6993,9 @@ def gitea_acquire_conflict_fix_lease(
         task_capability_map.required_permission("comment_issue"))
     if blocked:
         return blocked
-    verify_preflight_purity(remote, worktree_path=worktree_path)
+    verify_preflight_purity(
+        remote, worktree_path=worktree_path, task="comment_issue"
+    )
     comments = _list_pr_lease_comments(
         pr_number,
         remote=remote,
@@ -7178,7 +7199,7 @@ def gitea_cleanup_stale_claims(
         task_capability_map.required_permission("cleanup_stale_claims"))
     if blocked:
         return blocked
-    verify_preflight_purity(remote)
+    verify_preflight_purity(remote, task="cleanup_stale_claims")
 
     h, o, r = _resolve(remote, host, org, repo)
     auth = _auth(h)
@@ -7332,7 +7353,7 @@ def gitea_set_issue_labels(
         task_capability_map.required_permission("set_issue_labels"))
     if blocked:
         return blocked
-    verify_preflight_purity(remote)
+    verify_preflight_purity(remote, task="set_issue_labels")
     h, o, r = _resolve(remote, host, org, repo)
     auth = _auth(h)
     base = repo_api_url(h, o, r)
@@ -7569,7 +7590,7 @@ def gitea_resolve_task_capability(
                 "exact_safe_next_action": next_safe_action,
             }
 
-    record_preflight_check("capability", required_role)
+    record_preflight_check("capability", required_role, resolved_task=task)
 
     # Try automatic dispatch switching
     _ensure_matching_profile(required_permission, required_role, remote, host)

preflight_contract.py

@@ -0,0 +1,66 @@
+"""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,43 +26,3 @@ 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

@@ -0,0 +1,159 @@
+"""#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

@@ -1,164 +0,0 @@
-#!/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

@@ -1,554 +0,0 @@
-"""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}