gitea_mcp_server.py

@@ -547,6 +547,7 @@ import reconciliation_workflow  # 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
@@ -4270,6 +4271,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,

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

@@ -849,6 +849,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

@@ -691,6 +691,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_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),
+    }