test: seed review workflow load and pass expected_head_sha in legacy test fixtures

Extend the review workflow load gate with pre-review command classification,
boundary violation tracking, and structured helper-result requirements in
final reports. Adds gitea_record_pre_review_command MCP tool.

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

docs/llm-workflow-runbooks.md

@@ -377,6 +377,37 @@ explicit control-checkout repair.
 
 Portable wording: [`skills/llm-project-workflow/SKILL.md`](../skills/llm-project-workflow/SKILL.md).
 
+### Root checkout guard (#475)
+
+The MCP server enforces a fail-closed **root checkout guard** before author,
+reviewer, and merger mutations when the active workspace is not an isolated
+`branches/...` worktree (reconciler close paths remain exempt per #468).
+
+The guard blocks when the **control checkout** (repository root) is:
+
+- on a non-stable branch (`master` / `main` / `dev` expected),
+- detached HEAD,
+- dirty (tracked edits),
+- or its `HEAD` does not match `prgs/master` when that ref is available.
+
+**Remediation (never auto-reset or stash):**
+
+> Root checkout is not on master. Preserve state, switch root back to master,
+> and use `scripts/worktree-review` or the sanctioned issue worktree flow.
+
+**Recovery after root hijack:**
+
+1. Preserve any in-progress edits (copy paths, note branch name, or commit on a
+   rescue branch from a `branches/...` worktree).
+2. From the repository root: `git checkout master` (or `main` / `dev` per repo
+   policy) and `git fetch prgs && git merge --ff-only prgs/master` when safe.
+3. Confirm `git status` is clean and `git branch --show-current` is `master`.
+4. Resume work only inside `branches/issue-<n>-<slug>` via `gitea_lock_issue` /
+   `git worktree add`.
+
+`branches/...` directories are disposable role worktrees; the root checkout is
+the stable orchestration surface only.
+
 ## Shell Spawn Hard-Stop Rule
 
 Symptom: a shell tool call returns `exit_code: -1` with empty stdout/stderr.

gitea_mcp_server.py

@@ -651,6 +651,7 @@ def verify_preflight_purity(
                     f"{_format_preflight_files(reviewer_delta)}"
                 )
 
+    _enforce_root_checkout_guard(worktree_path)
     _enforce_branches_only_author_mutation(worktree_path)
     _clear_preflight_capability_state()
 
@@ -693,6 +694,27 @@ def _verify_role_mutation_workspace(
     verify_preflight_purity(remote, worktree_path=resolved, task=task)
     return resolved
 
+
+def _enforce_root_checkout_guard(worktree_path: str | None = None) -> None:
+    """#475: fail closed when the stable control checkout is contaminated."""
+    ctx = _resolve_author_mutation_context(worktree_path)
+    canonical_root = ctx["canonical_repo_root"]
+    workspace = ctx["workspace_path"]
+    git_state = issue_lock_worktree.read_worktree_git_state(canonical_root)
+    remote_master_sha = root_checkout_guard.resolve_remote_master_sha(canonical_root)
+    assessment = root_checkout_guard.assess_root_checkout_guard(
+        workspace_path=workspace,
+        canonical_repo_root=canonical_root,
+        current_branch=git_state.get("current_branch"),
+        head_sha=git_state.get("head_sha"),
+        porcelain_status=git_state.get("porcelain_status") or "",
+        remote_master_sha=remote_master_sha,
+        resolved_role=_preflight_resolved_role,
+    )
+    if assessment["block"]:
+        raise RuntimeError(root_checkout_guard.format_root_checkout_guard_error(assessment))
+
+
 from mcp.server.fastmcp import FastMCP  # noqa: E402
 
 from gitea_auth import (  # noqa: E402
@@ -725,6 +747,7 @@ 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 root_checkout_guard  # noqa: E402
 import issue_claim_heartbeat  # noqa: E402
 import issue_work_duplicate_gate  # noqa: E402
 import reviewer_pr_lease  # noqa: E402
@@ -735,6 +758,7 @@ import audit_reconciliation_mode  # noqa: E402
 import review_merge_state_machine  # noqa: E402
 import pr_work_lease  # noqa: E402
 import native_mcp_preference  # noqa: E402
+import worktree_cleanup_audit  # noqa: E402
 
 
 # Keyed issue-lock storage (#443): per remote/org/repo/issue files under
@@ -4654,6 +4678,90 @@ def gitea_scan_already_landed_open_prs(
     }
 
 
+@mcp.tool()
+def gitea_audit_worktree_cleanup(
+    remote: str = "dadeschools",
+    host: str | None = None,
+    org: str | None = None,
+    repo: str | None = None,
+    ttl_hours: float = worktree_cleanup_audit.DEFAULT_TTL_HOURS,
+) -> dict:
+    """Read-only: classify every session-owned worktree under ``branches/`` (#401).
+
+    Audits the local ``branches/`` directory, classifying each worktree as
+    active open PR, active issue work, dirty local, clean stale removable,
+    detached review leftover, or unsafe/unknown. Open PR branch heads are
+    fetched live so a worktree tied to an open PR is never marked removable;
+    the active issue-lock branch is read from the local lock file and treated
+    as active work. Deletes nothing and mutates no Gitea state.
+
+    Fails closed if the live open-PR list cannot be fetched: without it,
+    removability cannot be proven, so no candidates are returned.
+
+    Args:
+        remote: Known instance — 'dadeschools' or 'prgs'.
+        host: Override the Gitea host.
+        org: Override the owner/organization.
+        repo: Override the repository name.
+        ttl_hours: Age (hours) after which a clean issue/conflict-fix
+            worktree becomes stale-removable (default from
+            GITEA_WORKTREE_TTL_HOURS).
+
+    Returns:
+        dict with per-worktree classifications, counts, removable
+        candidates, and the ``git worktree list`` verification proof.
+    """
+    read_block = _profile_operation_gate("gitea.read")
+    if read_block:
+        return {
+            "success": False,
+            "performed": False,
+            "reasons": read_block,
+            "permission_report": _permission_block_report("gitea.read"),
+        }
+
+    try:
+        h, o, r = _resolve(remote, host, org, repo)
+        auth = _auth(h)
+        open_prs = api_get_all(f"{repo_api_url(h, o, r)}/pulls?state=open", auth)
+    except Exception as exc:
+        return {
+            "success": False,
+            "performed": False,
+            "open_pr_state_verified": False,
+            "reasons": [
+                "could not fetch live open PRs; removability unverified "
+                f"(fail closed): {_redact(str(exc))}"
+            ],
+        }
+
+    open_pr_branches = {
+        str((pr.get("head") or {}).get("ref"))
+        for pr in open_prs
+        if (pr.get("head") or {}).get("ref")
+    }
+
+    active_issue_branches: set[str] = set()
+    lock = merged_cleanup_reconcile.read_issue_lock(ISSUE_LOCK_FILE)
+    if lock and lock.get("branch_name"):
+        active_issue_branches.add(str(lock["branch_name"]).strip())
+
+    report = worktree_cleanup_audit.audit_branches_directory(
+        PROJECT_ROOT,
+        open_pr_branches=open_pr_branches,
+        active_issue_branches=active_issue_branches,
+        now=datetime.now(timezone.utc),
+        ttl_hours=ttl_hours,
+    )
+    return {
+        "success": True,
+        "performed": False,
+        "open_pr_state_verified": True,
+        "task_mode": "work-issue",
+        **report,
+    }
+
+
 @mcp.tool()
 def gitea_reconcile_already_landed_pr(
     pr_number: int,

root_checkout_guard.py

@@ -0,0 +1,133 @@
+"""Root checkout guard (#475).
+
+The project root checkout is the stable control checkout on master/prgs/master.
+Author/reviewer/merge flows must fail closed when the control checkout is
+contaminated (wrong branch, detached HEAD, dirty, or HEAD behind/ahead of
+prgs/master). Isolated ``branches/...`` worktrees remain allowed.
+"""
+
+from __future__ import annotations
+
+import os
+import subprocess
+
+from author_mutation_worktree import is_path_under_branches
+from reviewer_worktree import parse_dirty_tracked_files
+
+REMEDIATION = (
+    "Root checkout is not on master. Preserve state, switch root back to master, "
+    "and use scripts/worktree-review or the sanctioned issue worktree flow."
+)
+
+BASE_BRANCHES = frozenset({"master", "main", "dev"})
+REMOTE_MASTER_REFS = ("prgs/master", "refs/remotes/prgs/master")
+
+
+def resolve_remote_master_sha(
+    canonical_repo_root: str,
+    *,
+    remote_refs: tuple[str, ...] | None = None,
+) -> str | None:
+    """Return the commit SHA for the tracking master ref when available."""
+    root = (canonical_repo_root or "").strip()
+    if not root:
+        return None
+    for ref in remote_refs or REMOTE_MASTER_REFS:
+        res = subprocess.run(
+            ["git", "-C", root, "rev-parse", "--verify", ref],
+            capture_output=True,
+            text=True,
+            check=False,
+        )
+        if res.returncode == 0:
+            sha = (res.stdout or "").strip()
+            if sha:
+                return sha
+    return None
+
+
+resolve_tracking_master_sha = resolve_remote_master_sha
+
+
+def assess_root_checkout_guard(
+    *,
+    workspace_path: str,
+    canonical_repo_root: str,
+    current_branch: str | None,
+    head_sha: str | None,
+    porcelain_status: str,
+    remote_master_sha: str | None,
+    resolved_role: str | None = None,
+) -> dict:
+    """Fail closed when the control checkout is not clean master/prgs/master."""
+    reasons: list[str] = []
+    root = os.path.realpath(canonical_repo_root)
+    workspace = os.path.realpath(workspace_path)
+    branch = (current_branch or "").strip()
+    dirty_files = parse_dirty_tracked_files(porcelain_status)
+
+    if resolved_role == "reconciler":
+        return _assessment(True, [], root, workspace, branch, head_sha, dirty_files)
+
+    if resolved_role != "merger" and is_path_under_branches(workspace, root):
+        return _assessment(True, [], root, workspace, branch, head_sha, dirty_files)
+
+    if dirty_files:
+        reasons.append(
+            "control checkout has tracked local edits before role work "
+            f"(dirty files: {', '.join(dirty_files)})"
+        )
+
+    if not branch:
+        reasons.append("control checkout is detached HEAD; expected branch 'master'")
+    elif branch not in BASE_BRANCHES:
+        reasons.append(
+            f"control checkout branch '{branch}' is not a stable base branch "
+            f"({'/'.join(sorted(BASE_BRANCHES))})"
+        )
+
+    if remote_master_sha and head_sha and head_sha != remote_master_sha:
+        reasons.append(
+            "control checkout HEAD does not match prgs/master "
+            f"(HEAD {head_sha[:12]}, prgs/master {remote_master_sha[:12]})"
+        )
+
+    proven = not reasons
+    return _assessment(proven, reasons, root, workspace, branch or None, head_sha, dirty_files)
+
+
+def format_root_checkout_guard_error(assessment: dict) -> str:
+    """Single RuntimeError message for MCP preflight gates."""
+    root = assessment.get("canonical_repo_root") or "(unknown)"
+    workspace = assessment.get("workspace_path") or "(unknown)"
+    reasons = "; ".join(assessment.get("reasons") or ["unknown root checkout violation"])
+    return (
+        f"Root checkout guard (#475): {reasons}. "
+        f"canonical repository root: {root}; workspace: {workspace}. "
+        f"{REMEDIATION}"
+    )
+
+
+def _assessment(
+    proven: bool,
+    reasons: list[str],
+    canonical_repo_root: str,
+    workspace_path: str,
+    current_branch: str | None,
+    head_sha: str | None,
+    dirty_files: list[str],
+) -> dict:
+    return {
+        "proven": proven,
+        "block": not proven,
+        "reasons": reasons,
+        "canonical_repo_root": canonical_repo_root,
+        "workspace_path": workspace_path,
+        "current_branch": current_branch,
+        "head_sha": head_sha,
+        "dirty_files": dirty_files,
+        "remediation": REMEDIATION,
+    }
+
+
+assess_root_checkout = assess_root_checkout_guard

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

@@ -909,6 +909,16 @@ Confirm:
 
 Clean only the session-owned `branches/` review worktree if the project workflow explicitly allows cleanup.
 
+Review, baseline, and merge-simulation worktrees created during this run are
+transient and are removed automatically at successful completion once they are
+clean, carry no open PR, and hold no active lease (#401). Use
+`gitea_audit_worktree_cleanup` (read-only) to classify `branches/` entries; only
+`clean_stale_removable` and `detached_review_leftover` may be removed, one-by-one,
+after `git worktree list` proof plus per-worktree proof of path, branch/HEAD,
+clean/dirty status, no active PR/lease, and the removal result. Dirty,
+active-PR, active-issue, and leased worktrees are never deleted automatically; a
+failed removal must be reported with the leftover path and reason.
+
 Do not delete or mutate unrelated branches/worktrees.
 
 Do not touch the main checkout except to update the stable branch after merge if explicitly allowed by the workflow.

skills/llm-project-workflow/workflows/work-issue.md

@@ -699,6 +699,39 @@ Do not update the main checkout unless the canonical workflow explicitly allows
 
 Any cleanup is a mutation and must be reported.
 
+### 22A. Session-owned worktree cleanup and TTL (#401)
+
+Every session-owned worktree created under `branches/` has ownership metadata:
+path, workflow type, issue number, PR number, branch/head SHA, creator
+identity/profile, created timestamp, last-used timestamp, and cleanup
+eligibility.
+
+Cleanup is classification-driven. `gitea_audit_worktree_cleanup` (read-only)
+classifies every `branches/` entry as exactly one of:
+
+* `active_open_pr` — branch has an open PR; never auto-removed.
+* `active_issue_work` — active claim/lease or fresh issue worktree; never
+  auto-removed.
+* `dirty_local_worktree` — uncommitted changes; never auto-removed.
+* `clean_stale_removable` — clean, no PR, no lease; removable.
+* `detached_review_leftover` — clean detached review/baseline/merge-simulation
+  worktree; removable.
+* `unsafe_unknown` — protected base checkout or unknown workflow type; never
+  auto-removed.
+
+Only `clean_stale_removable` and `detached_review_leftover` may be removed, and
+only one-by-one after `git worktree list` proof plus per-worktree proof of:
+worktree path, branch/HEAD, clean/dirty status, no active PR/lease, and the
+removal result. Review, baseline, and merge-simulation worktrees are removed
+automatically at successful workflow completion; issue/conflict-fix worktrees
+are removed only after their TTL (`GITEA_WORKTREE_TTL_HOURS`, default 24h)
+expires and no lock/lease is held.
+
+Dirty, active-PR, active-issue, and leased worktrees are never deleted
+automatically. If a removal fails, the final report must list the leftover
+worktree path and the reason. Include the `git worktree list` output as final
+cleanup verification.
+
 ## 23. Recovery handoff rules
 
 If blocked, produce a recovery handoff with:

tests/test_audit.py

@@ -294,11 +294,13 @@ class TestGatedToolAudit(_AuditWiringBase):
         super().setUp()
         from tests.test_mcp_server import _init_reviewer_session, _install_owned_reviewer_lease
         import reviewer_pr_lease
+        import review_workflow_boundary
         import review_workflow_load
 
         # Session init clears any prior session lease (#407) and loads workflow (#389).
         _init_reviewer_session("prgs")
         self.addCleanup(review_workflow_load.clear_review_workflow_load)
+        self.addCleanup(review_workflow_boundary.clear_pre_review_commands)
         self._lease_patch = _install_owned_reviewer_lease(8)
         self._lease_patch.start()
         self._auth_identity_patch = patch(

tests/test_create_issue_workspace_guard.py

@@ -38,8 +38,18 @@ class TestCreateIssueWorkspaceGuard(unittest.TestCase):
     @patch("gitea_mcp_server.role_session_router.check_author_mutation_after_reviewer_stop", return_value=(True, []))
     @patch("gitea_mcp_server.api_request")
     @patch("gitea_mcp_server.api_get_all", return_value=[])
-    @patch("gitea_mcp_server.issue_lock_worktree.read_worktree_git_state", return_value={"current_branch": "feat/issue-1"})
+    @patch("gitea_mcp_server.root_checkout_guard.resolve_remote_master_sha", return_value="a" * 40)
-    def test_create_issue_stable_checkout_rejected(self, _git, _get_all, mock_api, _role, _ns, _prof, _auth):
+    @patch(
+        "gitea_mcp_server.issue_lock_worktree.read_worktree_git_state",
+        return_value={
+            "current_branch": "master",
+            "head_sha": "a" * 40,
+            "porcelain_status": "",
+        },
+    )
+    def test_create_issue_stable_checkout_rejected(
+        self, _git, _remote_sha, _get_all, mock_api, _role, _ns, _prof, _auth,
+    ):
         # Without worktree_path/env hints, workspace resolves to PROJECT_ROOT. When that
         # path is the stable control checkout (not under branches/), mutation must fail.
         with patch.object(srv, "PROJECT_ROOT", CONTROL_CHECKOUT_ROOT):

tests/test_root_checkout_guard.py

@@ -0,0 +1,161 @@
+"""Tests for root checkout guard (#475)."""
+
+from __future__ import annotations
+
+import os
+import sys
+import unittest
+from pathlib import Path
+from unittest.mock import patch
+
+sys.path.insert(0, str(Path(__file__).resolve().parent.parent))
+
+import gitea_mcp_server as srv  # noqa: E402
+import root_checkout_guard as rcg  # noqa: E402
+
+CONTROL_ROOT = str(Path(__file__).resolve().parents[3])
+BRANCHES_WORKTREE = str(Path(__file__).resolve().parents[1])
+MASTER_SHA = "a" * 40
+OTHER_SHA = "b" * 40
+
+
+class TestAssessRootCheckoutGuard(unittest.TestCase):
+    def _assess(self, **kwargs):
+        defaults = {
+            "workspace_path": CONTROL_ROOT,
+            "canonical_repo_root": CONTROL_ROOT,
+            "current_branch": "master",
+            "head_sha": MASTER_SHA,
+            "porcelain_status": "",
+            "remote_master_sha": MASTER_SHA,
+            "resolved_role": "author",
+        }
+        defaults.update(kwargs)
+        return rcg.assess_root_checkout_guard(**defaults)
+
+    def test_clean_master_control_checkout_allowed(self):
+        result = self._assess()
+        self.assertTrue(result["proven"])
+        self.assertFalse(result["block"])
+
+    def test_branches_worktree_allowed_for_author(self):
+        result = self._assess(
+            workspace_path=BRANCHES_WORKTREE,
+            current_branch="feat/issue-475-root-checkout-guard",
+            head_sha=OTHER_SHA,
+            resolved_role="author",
+        )
+        self.assertTrue(result["proven"])
+
+    def test_branches_worktree_allowed_for_reviewer(self):
+        result = self._assess(
+            workspace_path=f"{CONTROL_ROOT}/branches/review-pr-1",
+            current_branch="review-pr-1",
+            resolved_role="reviewer",
+        )
+        self.assertTrue(result["proven"])
+
+    def test_reconciler_always_allowed(self):
+        result = self._assess(
+            current_branch="feat/some-branch",
+            head_sha=OTHER_SHA,
+            porcelain_status=" M gitea_mcp_server.py\n",
+            resolved_role="reconciler",
+        )
+        self.assertTrue(result["proven"])
+
+    def test_feature_branch_on_control_checkout_blocked(self):
+        result = self._assess(
+            current_branch="feat/issue-99-example",
+            head_sha=OTHER_SHA,
+        )
+        self.assertTrue(result["block"])
+        self.assertIn("not a stable base branch", result["reasons"][0])
+
+    def test_detached_head_blocked(self):
+        result = self._assess(current_branch=None)
+        self.assertTrue(result["block"])
+        self.assertIn("detached HEAD", result["reasons"][0])
+
+    def test_dirty_control_checkout_blocked(self):
+        result = self._assess(porcelain_status=" M gitea_mcp_server.py\n")
+        self.assertTrue(result["block"])
+        self.assertIn("tracked local edits", result["reasons"][0])
+
+    def test_head_behind_prgs_master_blocked(self):
+        result = self._assess(
+            head_sha=OTHER_SHA,
+            remote_master_sha=MASTER_SHA,
+        )
+        self.assertTrue(result["block"])
+        self.assertIn("does not match prgs/master", result["reasons"][0])
+
+    def test_merger_requires_clean_control_checkout(self):
+        result = self._assess(
+            workspace_path=BRANCHES_WORKTREE,
+            current_branch="feat/issue-475-root-checkout-guard",
+            head_sha=OTHER_SHA,
+            resolved_role="merger",
+        )
+        self.assertTrue(result["block"])
+
+
+class TestVerifyPreflightRootGuardIntegration(unittest.TestCase):
+    def setUp(self):
+        srv._preflight_whoami_called = True
+        srv._preflight_capability_called = True
+        srv._preflight_resolved_role = "reviewer"
+        srv._preflight_whoami_violation = False
+        srv._preflight_capability_violation = False
+        self._orig_in_test = srv._preflight_in_test_mode
+        srv._preflight_in_test_mode = lambda: False
+
+    def tearDown(self):
+        srv._preflight_in_test_mode = self._orig_in_test
+
+    @patch("gitea_mcp_server._get_workspace_porcelain", return_value="")
+    @patch("gitea_mcp_server.root_checkout_guard.resolve_remote_master_sha", return_value=MASTER_SHA)
+    @patch("gitea_mcp_server.issue_lock_worktree.read_worktree_git_state")
+    @patch("gitea_mcp_server._resolve_author_mutation_context")
+    def test_reviewer_from_contaminated_root_blocked(
+        self, mock_ctx, mock_git, _remote_sha, _porcelain,
+    ):
+        srv._preflight_capability_baseline_porcelain = ""
+        mock_ctx.return_value = {
+            "workspace_path": CONTROL_ROOT,
+            "canonical_repo_root": CONTROL_ROOT,
+            "process_project_root": CONTROL_ROOT,
+        }
+        mock_git.return_value = {
+            "current_branch": "feat/hijacked-root",
+            "head_sha": OTHER_SHA,
+            "porcelain_status": "",
+        }
+        with self.assertRaises(RuntimeError) as ctx:
+            srv.verify_preflight_purity("prgs", worktree_path=CONTROL_ROOT)
+        self.assertIn("Root checkout guard (#475)", str(ctx.exception))
+        self.assertIn(rcg.REMEDIATION, str(ctx.exception))
+
+    @patch("gitea_mcp_server._get_workspace_porcelain", return_value="")
+    @patch("gitea_mcp_server.root_checkout_guard.resolve_remote_master_sha", return_value=MASTER_SHA)
+    @patch("gitea_mcp_server.issue_lock_worktree.read_worktree_git_state")
+    @patch("gitea_mcp_server._resolve_author_mutation_context")
+    def test_reviewer_from_branches_worktree_allowed(
+        self, mock_ctx, mock_git, _remote_sha, _porcelain,
+    ):
+        srv._preflight_capability_baseline_porcelain = ""
+        mock_ctx.return_value = {
+            "workspace_path": BRANCHES_WORKTREE,
+            "canonical_repo_root": CONTROL_ROOT,
+            "process_project_root": BRANCHES_WORKTREE,
+        }
+        mock_git.return_value = {
+            "current_branch": "feat/issue-475-root-checkout-guard",
+            "head_sha": OTHER_SHA,
+            "porcelain_status": "",
+        }
+        srv.verify_preflight_purity("prgs", worktree_path=BRANCHES_WORKTREE)
+
+
+if __name__ == "__main__":
+    unittest.main()

tests/test_workspace_guard_alignment.py

@@ -126,17 +126,37 @@ class TestRuntimeContextGuardAlignment(unittest.TestCase):
             with mock.patch.dict("os.environ", {"GITEA_TEST_PORCELAIN": ""}, clear=False):
                 srv.verify_preflight_purity(worktree_path=BRANCHES_WORKTREE)
 
-    def test_stable_checkout_still_rejected(self):
+    @mock.patch("gitea_mcp_server.root_checkout_guard.resolve_remote_master_sha", return_value="a" * 40)
+    @mock.patch(
+        "gitea_mcp_server.issue_lock_worktree.read_worktree_git_state",
+        return_value={
+            "current_branch": "master",
+            "head_sha": "a" * 40,
+            "porcelain_status": "",
+        },
+    )
+    def test_stable_checkout_still_rejected(self, _git, _remote_sha):
         with mock.patch.object(srv, "PROJECT_ROOT", CONTROL_ROOT):
             with mock.patch.dict("os.environ", {"GITEA_TEST_PORCELAIN": ""}, clear=False):
                 with self.assertRaises(RuntimeError) as ctx:
                     srv.verify_preflight_purity()
         self.assertIn("stable control checkout", str(ctx.exception))
 
+    @mock.patch("gitea_mcp_server.root_checkout_guard.resolve_remote_master_sha", return_value="a" * 40)
+    @mock.patch(
+        "gitea_mcp_server.issue_lock_worktree.read_worktree_git_state",
+        return_value={
+            "current_branch": "master",
+            "head_sha": "a" * 40,
+            "porcelain_status": "",
+        },
+    )
     @mock.patch("os.path.isdir", return_value=True)
     @mock.patch("os.path.exists", return_value=True)
     @mock.patch("subprocess.run")
-    def test_non_branches_worktree_rejected(self, mock_run, *_exists):
+    def test_non_branches_worktree_rejected(
+        self, mock_run, mock_exists, mock_isdir, _git, _remote_sha,
+    ):
         outside = "/tmp/outside-repo-checkout"
         mock_run.return_value = MagicMock(
             returncode=0,

tests/test_worktree_cleanup_audit.py

@@ -0,0 +1,383 @@
+"""Tests for session-owned worktree cleanup audit and TTL enforcement (#401)."""
+
+import sys
+import unittest
+from datetime import datetime, timedelta, timezone
+from unittest.mock import patch
+
+sys.path.insert(0, str(__import__("pathlib").Path(__file__).resolve().parent.parent))
+
+import worktree_cleanup_audit as wca  # noqa: E402
+
+
+NOW = datetime(2026, 7, 7, 12, 0, 0, tzinfo=timezone.utc)
+
+
+def _iso(dt):
+    return dt.isoformat().replace("+00:00", "Z")
+
+
+class TestWorkflowTypeInference(unittest.TestCase):
+    def test_review_pr_path(self):
+        self.assertEqual(
+            wca.infer_workflow_type("branches/review-pr376", "review-pr376"),
+            wca.WORKFLOW_REVIEW,
+        )
+
+    def test_baseline_path(self):
+        self.assertEqual(
+            wca.infer_workflow_type("branches/baseline-master-issue-401"),
+            wca.WORKFLOW_BASELINE,
+        )
+
+    def test_merge_simulation_path(self):
+        self.assertEqual(
+            wca.infer_workflow_type("branches/merge-sim-pr380"),
+            wca.WORKFLOW_MERGE_SIMULATION,
+        )
+
+    def test_issue_work_branch(self):
+        self.assertEqual(
+            wca.infer_workflow_type(
+                "branches/issue-401-worktree", "feat/issue-401-worktree"
+            ),
+            wca.WORKFLOW_ISSUE_WORK,
+        )
+
+    def test_conflict_fix_path(self):
+        self.assertEqual(
+            wca.infer_workflow_type("branches/conflict-fix-pr376"),
+            wca.WORKFLOW_CONFLICT_FIX,
+        )
+
+    def test_unknown_path(self):
+        self.assertEqual(
+            wca.infer_workflow_type("branches/scratchpad"), wca.WORKFLOW_UNKNOWN
+        )
+
+
+class TestMetadata(unittest.TestCase):
+    def test_metadata_fields_present(self):
+        meta = wca.build_worktree_metadata(
+            path="branches/issue-401-worktree",
+            branch="feat/issue-401-worktree",
+            head_sha="abc123",
+            creator="jcwalker3",
+            profile="prgs-author",
+            created_at=_iso(NOW),
+            last_used_at=_iso(NOW),
+        )
+        for field in (
+            "path",
+            "workflow_type",
+            "issue_number",
+            "pr_number",
+            "branch",
+            "head_sha",
+            "creator",
+            "profile",
+            "created_at",
+            "last_used_at",
+            "cleanup_eligibility",
+        ):
+            self.assertIn(field, meta)
+        self.assertEqual(meta["issue_number"], 401)
+        self.assertEqual(meta["workflow_type"], wca.WORKFLOW_ISSUE_WORK)
+
+    def test_review_metadata_auto_removable_flag(self):
+        meta = wca.build_worktree_metadata(path="branches/review-pr42")
+        self.assertTrue(meta["auto_remove_on_success"])
+
+
+class TestTTL(unittest.TestCase):
+    def test_expired(self):
+        old = _iso(NOW - timedelta(hours=48))
+        self.assertTrue(wca.is_ttl_expired(last_used_at=old, now=NOW, ttl_hours=24))
+
+    def test_not_expired(self):
+        recent = _iso(NOW - timedelta(hours=1))
+        self.assertFalse(
+            wca.is_ttl_expired(last_used_at=recent, now=NOW, ttl_hours=24)
+        )
+
+    def test_unknown_timestamp_fails_safe(self):
+        self.assertFalse(wca.is_ttl_expired(last_used_at=None, now=NOW))
+        self.assertFalse(
+            wca.is_ttl_expired(last_used_at="not-a-date", now=NOW)
+        )
+
+
+class TestClassification(unittest.TestCase):
+    def test_successful_review_cleanup(self):
+        # Scenario 1: clean review worktree -> removable.
+        cls = wca.classify_worktree(
+            workflow_type=wca.WORKFLOW_REVIEW, is_dirty=False
+        )
+        self.assertEqual(cls, wca.CLASS_CLEAN_STALE_REMOVABLE)
+        self.assertTrue(wca.is_removable(cls))
+
+    def test_dirty_worktree_preserved(self):
+        # Scenario 3: dirty worktree is never removable.
+        cls = wca.classify_worktree(
+            workflow_type=wca.WORKFLOW_REVIEW, is_dirty=True
+        )
+        self.assertEqual(cls, wca.CLASS_DIRTY_LOCAL)
+        self.assertFalse(wca.is_removable(cls))
+
+    def test_active_pr_worktree_preserved(self):
+        # Scenario 4: open PR wins over an otherwise-removable review worktree.
+        cls = wca.classify_worktree(
+            workflow_type=wca.WORKFLOW_REVIEW,
+            is_dirty=False,
+            has_open_pr=True,
+        )
+        self.assertEqual(cls, wca.CLASS_ACTIVE_OPEN_PR)
+        self.assertFalse(wca.is_removable(cls))
+
+    def test_stale_clean_issue_worktree_removable(self):
+        # Scenario 5: clean issue worktree, TTL expired, no lock -> removable.
+        cls = wca.classify_worktree(
+            workflow_type=wca.WORKFLOW_ISSUE_WORK,
+            is_dirty=False,
+            ttl_expired=True,
+        )
+        self.assertEqual(cls, wca.CLASS_CLEAN_STALE_REMOVABLE)
+        self.assertTrue(wca.is_removable(cls))
+
+    def test_fresh_clean_issue_worktree_preserved(self):
+        # Clean issue worktree not yet TTL-expired stays active.
+        cls = wca.classify_worktree(
+            workflow_type=wca.WORKFLOW_ISSUE_WORK,
+            is_dirty=False,
+            ttl_expired=False,
+        )
+        self.assertEqual(cls, wca.CLASS_ACTIVE_ISSUE_WORK)
+        self.assertFalse(wca.is_removable(cls))
+
+    def test_detached_review_worktree_classified(self):
+        # Scenario 6: detached review worktree -> detached_review_leftover.
+        cls = wca.classify_worktree(
+            workflow_type=wca.WORKFLOW_REVIEW,
+            is_dirty=False,
+            is_detached=True,
+        )
+        self.assertEqual(cls, wca.CLASS_DETACHED_REVIEW_LEFTOVER)
+        self.assertTrue(wca.is_removable(cls))
+
+    def test_ttl_expired_but_dirty_preserved(self):
+        # Scenario 7: dirty wins over TTL expiry.
+        cls = wca.classify_worktree(
+            workflow_type=wca.WORKFLOW_ISSUE_WORK,
+            is_dirty=True,
+            ttl_expired=True,
+        )
+        self.assertEqual(cls, wca.CLASS_DIRTY_LOCAL)
+        self.assertFalse(wca.is_removable(cls))
+
+    def test_lease_protected_worktree_preserved(self):
+        # Scenario 8: active lease is never removable.
+        cls = wca.classify_worktree(
+            workflow_type=wca.WORKFLOW_REVIEW,
+            is_dirty=False,
+            has_active_lease=True,
+        )
+        self.assertEqual(cls, wca.CLASS_ACTIVE_ISSUE_WORK)
+        self.assertFalse(wca.is_removable(cls))
+
+    def test_active_issue_lock_preserved(self):
+        cls = wca.classify_worktree(
+            workflow_type=wca.WORKFLOW_ISSUE_WORK,
+            is_dirty=False,
+            has_active_issue_lock=True,
+            ttl_expired=True,
+        )
+        self.assertEqual(cls, wca.CLASS_ACTIVE_ISSUE_WORK)
+        self.assertFalse(wca.is_removable(cls))
+
+    def test_protected_base_worktree_never_removable(self):
+        cls = wca.classify_worktree(
+            workflow_type=wca.WORKFLOW_ISSUE_WORK,
+            is_dirty=False,
+            is_protected=True,
+            ttl_expired=True,
+        )
+        self.assertEqual(cls, wca.CLASS_UNSAFE_UNKNOWN)
+        self.assertFalse(wca.is_removable(cls))
+
+    def test_unknown_workflow_type_unsafe(self):
+        cls = wca.classify_worktree(
+            workflow_type=wca.WORKFLOW_UNKNOWN, is_dirty=False, ttl_expired=True
+        )
+        self.assertEqual(cls, wca.CLASS_UNSAFE_UNKNOWN)
+        self.assertFalse(wca.is_removable(cls))
+
+
+class TestRemovalDecision(unittest.TestCase):
+    def test_safe_removal_proof(self):
+        decision = wca.assess_worktree_removal(
+            path="branches/review-pr42",
+            branch="review-pr42",
+            head_sha="abc123",
+            is_dirty=False,
+            has_open_pr=False,
+            has_active_lease=False,
+            classification=wca.CLASS_CLEAN_STALE_REMOVABLE,
+        )
+        self.assertTrue(decision["safe_to_remove"])
+        self.assertEqual(decision["block_reasons"], [])
+        self.assertTrue(decision["clean"])
+        self.assertTrue(decision["no_active_pr"])
+        self.assertTrue(decision["no_active_lease"])
+
+    def test_dirty_blocks_removal(self):
+        decision = wca.assess_worktree_removal(
+            path="branches/review-pr42",
+            branch="review-pr42",
+            head_sha="abc123",
+            is_dirty=True,
+            has_open_pr=False,
+            has_active_lease=False,
+            classification=wca.CLASS_DIRTY_LOCAL,
+        )
+        self.assertFalse(decision["safe_to_remove"])
+        self.assertIn("worktree has uncommitted changes", decision["block_reasons"])
+
+    def test_open_pr_and_lease_block_removal(self):
+        decision = wca.assess_worktree_removal(
+            path="branches/review-pr42",
+            branch="review-pr42",
+            head_sha="abc123",
+            is_dirty=False,
+            has_open_pr=True,
+            has_active_lease=True,
+            classification=wca.CLASS_ACTIVE_OPEN_PR,
+        )
+        self.assertFalse(decision["safe_to_remove"])
+        self.assertIn("worktree branch has an open PR", decision["block_reasons"])
+        self.assertIn("worktree has an active lease", decision["block_reasons"])
+
+
+class TestSuccessCleanupPlan(unittest.TestCase):
+    def test_review_worktree_removed_at_success(self):
+        # Scenario 1 end-to-end: clean review worktree removed at success.
+        meta = wca.build_worktree_metadata(path="branches/review-pr42")
+        plan = wca.plan_success_cleanup(
+            metadata=meta, is_dirty=False, has_open_pr=False, has_active_lease=False
+        )
+        self.assertTrue(plan["remove"])
+
+    def test_failed_review_leaves_worktree_reported(self):
+        # Scenario 2: dirty review worktree preserved and reported at failure.
+        meta = wca.build_worktree_metadata(path="branches/review-pr42")
+        plan = wca.plan_success_cleanup(
+            metadata=meta, is_dirty=True, has_open_pr=False, has_active_lease=False
+        )
+        self.assertFalse(plan["remove"])
+        self.assertIn("uncommitted", plan["reason"])
+        report = wca.cleanup_failure_report(meta["path"], plan["reason"])
+        self.assertFalse(report["removed"])
+        self.assertEqual(report["path"], "branches/review-pr42")
+
+    def test_issue_worktree_preserved_by_policy(self):
+        meta = wca.build_worktree_metadata(
+            path="branches/issue-401-worktree", branch="feat/issue-401-worktree"
+        )
+        plan = wca.plan_success_cleanup(
+            metadata=meta, is_dirty=False, has_open_pr=False, has_active_lease=False
+        )
+        self.assertFalse(plan["remove"])
+        self.assertIn("preserved by policy", plan["reason"])
+
+
+class TestPorcelainParser(unittest.TestCase):
+    def test_parse_branch_and_detached(self):
+        text = (
+            "worktree /repo\n"
+            "HEAD 1111111111111111111111111111111111111111\n"
+            "branch refs/heads/master\n"
+            "\n"
+            "worktree /repo/branches/review-pr42\n"
+            "HEAD 2222222222222222222222222222222222222222\n"
+            "detached\n"
+        )
+        entries = wca.parse_worktree_porcelain(text)
+        self.assertEqual(len(entries), 2)
+        self.assertEqual(entries[0]["branch"], "master")
+        self.assertFalse(entries[0]["detached"])
+        self.assertIsNone(entries[1]["branch"])
+        self.assertTrue(entries[1]["detached"])
+
+
+class TestAuditReportAccuracy(unittest.TestCase):
+    """Scenario 9: cleanup report accuracy over a mixed branches/ directory."""
+
+    PORCELAIN = (
+        "worktree /repo\n"
+        "HEAD 1111111111111111111111111111111111111111\n"
+        "branch refs/heads/master\n"
+        "\n"
+        "worktree /repo/branches/review-pr42\n"
+        "HEAD 2222222222222222222222222222222222222222\n"
+        "branch refs/heads/review-pr42\n"
+        "\n"
+        "worktree /repo/branches/issue-400-open-pr\n"
+        "HEAD 3333333333333333333333333333333333333333\n"
+        "branch refs/heads/feat/issue-400-open-pr\n"
+        "\n"
+        "worktree /repo/branches/issue-401-dirty\n"
+        "HEAD 4444444444444444444444444444444444444444\n"
+        "branch refs/heads/feat/issue-401-dirty\n"
+    )
+
+    def _fake_dirty(self, path):
+        if path.endswith("issue-401-dirty"):
+            return {"exists": True, "dirty": True, "dirty_files": [" M x.py"]}
+        return {"exists": True, "dirty": False, "dirty_files": []}
+
+    def test_mixed_directory_classified(self):
+        with patch.object(
+            wca,
+            "list_worktrees",
+            return_value=wca.parse_worktree_porcelain(self.PORCELAIN),
+        ), patch.object(
+            wca, "read_worktree_dirty", side_effect=self._fake_dirty
+        ), patch.object(
+            wca, "git_worktree_list", return_value="(mocked)"
+        ):
+            report = wca.audit_branches_directory(
+                "/repo",
+                open_pr_branches={"feat/issue-400-open-pr"},
+            )
+
+        by_path = {wt["path"]: wt for wt in report["worktrees"]}
+        # main checkout on master -> protected -> unsafe/unknown, not removable
+        self.assertEqual(
+            by_path["/repo"]["classification"], wca.CLASS_UNSAFE_UNKNOWN
+        )
+        # clean review worktree -> removable
+        self.assertEqual(
+            by_path["/repo/branches/review-pr42"]["classification"],
+            wca.CLASS_CLEAN_STALE_REMOVABLE,
+        )
+        # open PR branch -> preserved
+        self.assertEqual(
+            by_path["/repo/branches/issue-400-open-pr"]["classification"],
+            wca.CLASS_ACTIVE_OPEN_PR,
+        )
+        # dirty worktree -> preserved
+        self.assertEqual(
+            by_path["/repo/branches/issue-401-dirty"]["classification"],
+            wca.CLASS_DIRTY_LOCAL,
+        )
+        # exactly one removable candidate (the clean review worktree)
+        self.assertEqual(report["removable_count"], 1)
+        self.assertEqual(
+            report["removable_candidates"][0]["path"],
+            "/repo/branches/review-pr42",
+        )
+        self.assertEqual(report["total"], 4)
+        self.assertEqual(report["git_worktree_list"], "(mocked)")
+
+
+if __name__ == "__main__":
+    unittest.main()

worktree_cleanup_audit.py

@@ -0,0 +1,480 @@
+"""Session-owned worktree cleanup audit and TTL enforcement (#401).
+
+LLM workflows create many session-owned worktrees under ``branches/``
+(review, baseline, merge-simulation, issue, and conflict-fix worktrees).
+When a run stops early, races a sibling session, hits a validation failure,
+or loses shell/cwd state, those worktrees are left behind and later workflow
+decisions get harder and riskier.
+
+This module provides:
+
+* ``build_worktree_metadata`` — ownership/purpose metadata for a worktree.
+* ``classify_worktree`` — safety-first classification into the audit
+  vocabulary (active open PR, active issue work, dirty, clean stale
+  removable, detached review leftover, unsafe/unknown).
+* ``assess_worktree_removal`` — a per-worktree removal decision with an
+  explicit proof and block reasons.
+* git-shelling helpers (``list_worktrees``, ``read_worktree_dirty``,
+  ``git_worktree_list``, ``remove_worktree``) and ``audit_branches_directory``
+  that classify every entry under ``branches/``.
+
+Pure assessment functions take explicit state so they are unit-testable
+without a filesystem or network. Only the thin git helpers shell out, and
+removal is only ever executed after ``assess_worktree_removal`` proves the
+worktree is safe to delete.
+"""
+
+from __future__ import annotations
+
+import os
+import re
+import subprocess
+from datetime import datetime, timezone
+from typing import Any
+
+PROTECTED_BRANCHES = frozenset({"master", "main", "dev"})
+DEFAULT_TTL_HOURS = float(os.environ.get("GITEA_WORKTREE_TTL_HOURS", "24") or 24)
+
+# Workflow types that can create session-owned worktrees.
+WORKFLOW_REVIEW = "review"
+WORKFLOW_BASELINE = "baseline"
+WORKFLOW_MERGE_SIMULATION = "merge_simulation"
+WORKFLOW_ISSUE_WORK = "issue_work"
+WORKFLOW_CONFLICT_FIX = "conflict_fix"
+WORKFLOW_UNKNOWN = "unknown"
+
+# Workflow types whose worktrees are transient and removed automatically at
+# successful workflow completion (acceptance criterion 2).
+AUTO_REMOVE_ON_SUCCESS = frozenset(
+    {WORKFLOW_REVIEW, WORKFLOW_BASELINE, WORKFLOW_MERGE_SIMULATION}
+)
+
+# Cleanup-audit classification vocabulary (acceptance criterion 4).
+CLASS_ACTIVE_OPEN_PR = "active_open_pr"
+CLASS_ACTIVE_ISSUE_WORK = "active_issue_work"
+CLASS_DIRTY_LOCAL = "dirty_local_worktree"
+CLASS_CLEAN_STALE_REMOVABLE = "clean_stale_removable"
+CLASS_DETACHED_REVIEW_LEFTOVER = "detached_review_leftover"
+CLASS_UNSAFE_UNKNOWN = "unsafe_unknown"
+
+# Only these two classifications may ever be removed automatically.
+REMOVABLE_CLASSES = frozenset(
+    {CLASS_CLEAN_STALE_REMOVABLE, CLASS_DETACHED_REVIEW_LEFTOVER}
+)
+
+_ISSUE_REF_RE = re.compile(r"issue-(\d+)", re.IGNORECASE)
+_ISSUE_BRANCH_PREFIXES = ("feat/", "fix/", "docs/", "chore/")
+
+
+def infer_workflow_type(path: str | None, branch: str | None = None) -> str:
+    """Infer the creating workflow type from a worktree path or branch name."""
+    text = f"{path or ''} {branch or ''}".lower()
+    if "baseline" in text:
+        return WORKFLOW_BASELINE
+    if "merge-sim" in text or "merge_sim" in text or "mergesim" in text:
+        return WORKFLOW_MERGE_SIMULATION
+    if "review" in text or "review-pr" in text:
+        return WORKFLOW_REVIEW
+    if "conflict" in text:
+        return WORKFLOW_CONFLICT_FIX
+    if "issue-" in text or (branch or "").startswith(_ISSUE_BRANCH_PREFIXES):
+        return WORKFLOW_ISSUE_WORK
+    return WORKFLOW_UNKNOWN
+
+
+def _extract_issue_number(path: str | None, branch: str | None) -> int | None:
+    match = _ISSUE_REF_RE.search(f"{path or ''} {branch or ''}")
+    return int(match.group(1)) if match else None
+
+
+def build_worktree_metadata(
+    *,
+    path: str,
+    branch: str | None = None,
+    head_sha: str | None = None,
+    workflow_type: str | None = None,
+    issue_number: int | None = None,
+    pr_number: int | None = None,
+    creator: str | None = None,
+    profile: str | None = None,
+    created_at: str | None = None,
+    last_used_at: str | None = None,
+    cleanup_eligibility: str | None = None,
+) -> dict[str, Any]:
+    """Return ownership/purpose metadata for a session-owned worktree.
+
+    Covers acceptance criterion 1: path, workflow type, issue/PR number,
+    branch/head SHA, creator identity/profile, created and last-used
+    timestamps, and cleanup eligibility.
+    """
+    wt = workflow_type or infer_workflow_type(path, branch)
+    issue = issue_number
+    if issue is None:
+        issue = _extract_issue_number(path, branch)
+    return {
+        "path": path,
+        "workflow_type": wt,
+        "issue_number": issue,
+        "pr_number": pr_number,
+        "branch": branch,
+        "head_sha": head_sha,
+        "creator": creator,
+        "profile": profile,
+        "created_at": created_at,
+        "last_used_at": last_used_at,
+        "auto_remove_on_success": wt in AUTO_REMOVE_ON_SUCCESS,
+        "cleanup_eligibility": cleanup_eligibility,
+    }
+
+
+def _parse_timestamp(value: str | None) -> datetime | None:
+    if not value:
+        return None
+    text = str(value).strip()
+    if not text:
+        return None
+    if text.endswith("Z"):
+        text = text[:-1] + "+00:00"
+    try:
+        parsed = datetime.fromisoformat(text)
+    except ValueError:
+        return None
+    if parsed.tzinfo is None:
+        return parsed.replace(tzinfo=timezone.utc)
+    return parsed
+
+
+def is_ttl_expired(
+    *,
+    last_used_at: str | None,
+    now: datetime | str | None,
+    ttl_hours: float = DEFAULT_TTL_HOURS,
+) -> bool:
+    """Return True only when the age is known and exceeds ``ttl_hours``.
+
+    Unknown or unparseable timestamps fail safe (not expired) so a worktree
+    is never treated as removable merely because its age is unknown.
+    """
+    last = _parse_timestamp(last_used_at)
+    now_dt = now if isinstance(now, datetime) else _parse_timestamp(now)
+    if last is None or now_dt is None:
+        return False
+    if now_dt.tzinfo is None:
+        now_dt = now_dt.replace(tzinfo=timezone.utc)
+    return (now_dt - last).total_seconds() > ttl_hours * 3600.0
+
+
+def classify_worktree(
+    *,
+    workflow_type: str,
+    is_dirty: bool,
+    has_open_pr: bool = False,
+    has_active_lease: bool = False,
+    has_active_issue_lock: bool = False,
+    is_detached: bool = False,
+    branch_gone: bool = False,
+    ttl_expired: bool = False,
+    is_protected: bool = False,
+    metadata_known: bool = True,
+) -> str:
+    """Classify a worktree, safety-first: any preservation signal wins.
+
+    Dirty, open-PR, leased, active-lock, protected, and unknown states are
+    all non-removable and are checked before any removable classification,
+    so nothing removable can shadow a preservation signal (criteria 6-8).
+    """
+    if is_protected:
+        # The main checkout / a protected base branch is never removable.
+        return CLASS_UNSAFE_UNKNOWN
+    if is_dirty:
+        return CLASS_DIRTY_LOCAL  # never auto-deleted (criterion 6)
+    if has_open_pr:
+        return CLASS_ACTIVE_OPEN_PR  # never auto-deleted (criterion 7)
+    if has_active_lease:
+        return CLASS_ACTIVE_ISSUE_WORK  # never auto-deleted (criterion 8)
+    if has_active_issue_lock:
+        return CLASS_ACTIVE_ISSUE_WORK
+    if not metadata_known or workflow_type == WORKFLOW_UNKNOWN:
+        return CLASS_UNSAFE_UNKNOWN  # never auto-deleted without proof
+
+    # Clean, no PR, no lease, no lock, known workflow type.
+    if workflow_type in AUTO_REMOVE_ON_SUCCESS:
+        if is_detached or branch_gone:
+            return CLASS_DETACHED_REVIEW_LEFTOVER
+        return CLASS_CLEAN_STALE_REMOVABLE
+    # issue_work / conflict_fix: only removable once the TTL has expired.
+    if ttl_expired:
+        return CLASS_CLEAN_STALE_REMOVABLE
+    return CLASS_ACTIVE_ISSUE_WORK
+
+
+def is_removable(classification: str) -> bool:
+    """Return True only for the two auto-removable classifications."""
+    return classification in REMOVABLE_CLASSES
+
+
+def assess_worktree_removal(
+    *,
+    path: str,
+    branch: str | None,
+    head_sha: str | None,
+    is_dirty: bool,
+    has_open_pr: bool,
+    has_active_lease: bool,
+    classification: str,
+) -> dict[str, Any]:
+    """Return a per-worktree removal decision with proof (criterion 9).
+
+    A worktree is only safe to remove when it is clean, has no active PR,
+    has no active lease, and its classification is auto-removable.
+    """
+    block_reasons: list[str] = []
+    if is_dirty:
+        block_reasons.append("worktree has uncommitted changes")
+    if has_open_pr:
+        block_reasons.append("worktree branch has an open PR")
+    if has_active_lease:
+        block_reasons.append("worktree has an active lease")
+    if not is_removable(classification):
+        block_reasons.append(
+            f"classification '{classification}' is not auto-removable"
+        )
+    return {
+        "path": path,
+        "branch": branch,
+        "head_sha": head_sha,
+        "classification": classification,
+        "clean": not is_dirty,
+        "no_active_pr": not has_open_pr,
+        "no_active_lease": not has_active_lease,
+        "safe_to_remove": not block_reasons,
+        "block_reasons": block_reasons,
+    }
+
+
+def plan_success_cleanup(
+    *,
+    metadata: dict[str, Any],
+    is_dirty: bool,
+    has_open_pr: bool,
+    has_active_lease: bool,
+) -> dict[str, Any]:
+    """Decide whether a just-completed worktree is removed at success.
+
+    Review/baseline/merge-simulation worktrees are removed automatically at
+    successful completion (criterion 2); everything else is preserved and
+    reported. Dirty/PR/leased worktrees are always preserved (criteria 6-8).
+    """
+    workflow_type = metadata.get("workflow_type", WORKFLOW_UNKNOWN)
+    if not metadata.get("auto_remove_on_success"):
+        return {
+            "remove": False,
+            "reason": f"workflow type '{workflow_type}' is preserved by policy",
+        }
+    classification = classify_worktree(
+        workflow_type=workflow_type,
+        is_dirty=is_dirty,
+        has_open_pr=has_open_pr,
+        has_active_lease=has_active_lease,
+    )
+    decision = assess_worktree_removal(
+        path=metadata.get("path", ""),
+        branch=metadata.get("branch"),
+        head_sha=metadata.get("head_sha"),
+        is_dirty=is_dirty,
+        has_open_pr=has_open_pr,
+        has_active_lease=has_active_lease,
+        classification=classification,
+    )
+    return {
+        "remove": decision["safe_to_remove"],
+        "reason": "clean transient worktree removable at success completion"
+        if decision["safe_to_remove"]
+        else "; ".join(decision["block_reasons"]),
+        "classification": classification,
+        "decision": decision,
+    }
+
+
+def cleanup_failure_report(path: str, reason: str) -> dict[str, Any]:
+    """Structured leftover-worktree record for the final report (criterion 3)."""
+    return {"path": path, "removed": False, "reason": reason}
+
+
+# --------------------------------------------------------------------------
+# git-shelling helpers (only these touch the filesystem)
+# --------------------------------------------------------------------------
+
+
+def parse_worktree_porcelain(text: str) -> list[dict[str, Any]]:
+    """Parse ``git worktree list --porcelain`` output into entries."""
+    entries: list[dict[str, Any]] = []
+    current: dict[str, Any] = {}
+    for raw in (text or "").splitlines():
+        line = raw.rstrip("\n")
+        if not line:
+            if current:
+                entries.append(current)
+                current = {}
+            continue
+        if line.startswith("worktree "):
+            if current:
+                entries.append(current)
+            current = {
+                "path": line[len("worktree ") :].strip(),
+                "head": None,
+                "branch": None,
+                "detached": False,
+                "bare": False,
+            }
+        elif line.startswith("HEAD "):
+            current["head"] = line[len("HEAD ") :].strip()
+        elif line.startswith("branch "):
+            ref = line[len("branch ") :].strip()
+            current["branch"] = ref.replace("refs/heads/", "", 1)
+        elif line == "detached":
+            current["detached"] = True
+        elif line == "bare":
+            current["bare"] = True
+    if current:
+        entries.append(current)
+    return entries
+
+
+def list_worktrees(project_root: str) -> list[dict[str, Any]]:
+    """Return parsed ``git worktree list`` entries for ``project_root``."""
+    result = subprocess.run(
+        ["git", "-C", project_root, "worktree", "list", "--porcelain"],
+        capture_output=True,
+        text=True,
+        check=False,
+    )
+    if result.returncode != 0:
+        return []
+    return parse_worktree_porcelain(result.stdout)
+
+
+def git_worktree_list(project_root: str) -> str:
+    """Return plain ``git worktree list`` output for final verification (criterion 10)."""
+    result = subprocess.run(
+        ["git", "-C", project_root, "worktree", "list"],
+        capture_output=True,
+        text=True,
+        check=False,
+    )
+    return (result.stdout or "").strip()
+
+
+def read_worktree_dirty(path: str) -> dict[str, Any]:
+    """Return dirty state for a worktree path via ``git status --porcelain``."""
+    if not path or not os.path.isdir(path):
+        return {"exists": False, "dirty": None, "dirty_files": []}
+    result = subprocess.run(
+        ["git", "-C", path, "status", "--porcelain"],
+        capture_output=True,
+        text=True,
+        check=False,
+    )
+    files = [ln for ln in (result.stdout or "").splitlines() if ln.strip()]
+    return {"exists": True, "dirty": bool(files), "dirty_files": files}
+
+
+def remove_worktree(project_root: str, path: str) -> dict[str, Any]:
+    """Remove a single worktree via ``git worktree remove`` (no ``--force``)."""
+    result = subprocess.run(
+        ["git", "-C", project_root, "worktree", "remove", path],
+        capture_output=True,
+        text=True,
+        check=False,
+    )
+    ok = result.returncode == 0
+    return {
+        "path": path,
+        "removed": ok,
+        "reason": None if ok else (result.stderr or "git worktree remove failed").strip(),
+    }
+
+
+def _is_under_branches(project_root: str, path: str) -> bool:
+    branches_root = os.path.join(os.path.abspath(project_root), "branches")
+    return os.path.abspath(path or "").startswith(branches_root + os.sep)
+
+
+def audit_branches_directory(
+    project_root: str,
+    *,
+    open_pr_branches: set[str] | None = None,
+    leased_branches: set[str] | None = None,
+    active_issue_branches: set[str] | None = None,
+    now: datetime | str | None = None,
+    ttl_hours: float = DEFAULT_TTL_HOURS,
+) -> dict[str, Any]:
+    """Classify every session-owned worktree under ``branches/``.
+
+    Read-only: shells out to git for discovery and dirty state, then applies
+    the pure classifier. Returns per-worktree classifications, counts, the
+    list of removable candidates, and the ``git worktree list`` proof.
+    """
+    open_pr_branches = open_pr_branches or set()
+    leased_branches = leased_branches or set()
+    active_issue_branches = active_issue_branches or set()
+
+    worktrees: list[dict[str, Any]] = []
+    for entry in list_worktrees(project_root):
+        path = entry.get("path") or ""
+        branch = entry.get("branch")
+        is_protected = (branch in PROTECTED_BRANCHES) or not _is_under_branches(
+            project_root, path
+        )
+        dirty_state = read_worktree_dirty(path)
+        is_dirty = bool(dirty_state.get("dirty"))
+        metadata = build_worktree_metadata(
+            path=path, branch=branch, head_sha=entry.get("head")
+        )
+        has_open_pr = bool(branch) and branch in open_pr_branches
+        has_active_lease = bool(branch) and branch in leased_branches
+        has_active_lock = bool(branch) and branch in active_issue_branches
+        ttl_expired = is_ttl_expired(
+            last_used_at=metadata.get("last_used_at"), now=now, ttl_hours=ttl_hours
+        )
+        classification = classify_worktree(
+            workflow_type=metadata["workflow_type"],
+            is_dirty=is_dirty,
+            has_open_pr=has_open_pr,
+            has_active_lease=has_active_lease,
+            has_active_issue_lock=has_active_lock,
+            is_detached=bool(entry.get("detached")),
+            branch_gone=branch is None and not entry.get("detached"),
+            ttl_expired=ttl_expired,
+            is_protected=is_protected,
+        )
+        metadata["cleanup_eligibility"] = classification
+        worktrees.append(
+            {
+                **metadata,
+                "detached": bool(entry.get("detached")),
+                "dirty": is_dirty,
+                "dirty_files": dirty_state.get("dirty_files", []),
+                "has_open_pr": has_open_pr,
+                "has_active_lease": has_active_lease,
+                "has_active_issue_lock": has_active_lock,
+                "is_protected": is_protected,
+                "classification": classification,
+                "removable": is_removable(classification),
+            }
+        )
+
+    counts: dict[str, int] = {}
+    for wt in worktrees:
+        counts[wt["classification"]] = counts.get(wt["classification"], 0) + 1
+    removable = [wt for wt in worktrees if wt["removable"]]
+
+    return {
+        "project_root": project_root,
+        "worktrees": worktrees,
+        "counts": counts,
+        "removable_candidates": removable,
+        "removable_count": len(removable),
+        "total": len(worktrees),
+        "git_worktree_list": git_worktree_list(project_root),
+    }