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.

final_report_validator.py

@@ -118,6 +118,22 @@ _TARGET_BRANCH_SHA_RE = re.compile(
     r"target branch sha\s*:\s*[0-9a-f]{40}",
     re.IGNORECASE,
 )
+_WORKFLOW_LOAD_HELPER_RE = re.compile(
+    r"workflow[- ]load helper result\s*:",
+    re.IGNORECASE,
+)
+_WORKFLOW_LOAD_HASH_RE = re.compile(
+    r"workflow[- ]load helper result[\s\S]{0,400}?workflow[_ ]hash\s*:\s*[0-9a-f]{12}",
+    re.IGNORECASE,
+)
+_WORKFLOW_LOAD_BOUNDARY_RE = re.compile(
+    r"workflow[- ]load helper result[\s\S]{0,400}?boundary[_ ]status\s*:\s*(?:clean|violation)",
+    re.IGNORECASE,
+)
+_WORKFLOW_FILE_VIEW_NARRATIVE_RE = re.compile(
+    r"(?:read|viewed|loaded)\s+(?:the\s+)?(?:canonical\s+)?(?:workflow|review-merge-pr\.md)",
+    re.IGNORECASE,
+)
 _FULL_SHA_RE = re.compile(r"\b[0-9a-f]{40}\b", re.IGNORECASE)
 _RECONCILE_STALE_FIELDS = (
     "pr number opened",
@@ -1024,6 +1040,54 @@ def _rule_shared_author_reviewer_same_run(report_text: str) -> list[dict[str, st
     )
 
 
+def _rule_reviewer_workflow_load_boundary(report_text: str) -> list[dict[str, str]]:
+    """#403: require structured workflow-load helper result, not file-view narrative."""
+    if not report_text.strip():
+        return []
+    findings: list[dict[str, str]] = []
+    has_helper = bool(_WORKFLOW_LOAD_HELPER_RE.search(report_text))
+    has_hash = bool(_WORKFLOW_LOAD_HASH_RE.search(report_text))
+    has_boundary = bool(_WORKFLOW_LOAD_BOUNDARY_RE.search(report_text))
+    has_narrative_only = bool(_WORKFLOW_FILE_VIEW_NARRATIVE_RE.search(report_text))
+
+    if has_narrative_only and not has_helper:
+        findings.append(validator_finding(
+            "reviewer.workflow_load_boundary",
+            "block",
+            "Workflow-load helper result",
+            (
+                "canonical workflow file-view narrative without structured "
+                "gitea_load_review_workflow helper result"
+            ),
+            (
+                "include Workflow-load helper result with workflow_hash and "
+                "boundary_status from gitea_load_review_workflow"
+            ),
+        ))
+        return findings
+
+    if has_helper and (not has_hash or not has_boundary):
+        missing = []
+        if not has_hash:
+            missing.append("workflow_hash")
+        if not has_boundary:
+            missing.append("boundary_status")
+        findings.append(validator_finding(
+            "reviewer.workflow_load_boundary",
+            "block",
+            "Workflow-load helper result",
+            (
+                "workflow-load helper result incomplete; missing "
+                + ", ".join(missing)
+            ),
+            (
+                "copy workflow_load_helper_result fields from "
+                "gitea_load_review_workflow into the final report"
+            ),
+        ))
+    return findings
+
+
 def _rule_audit_reconciliation_boundary(report_text: str) -> list[dict[str, str]]:
     from audit_reconciliation_mode import assess_audit_reconciliation_report
 
@@ -1100,6 +1164,7 @@ _RULES_BY_TASK: dict[str, list[Callable[..., list[dict[str, str]]]]] = {
         _rule_reviewer_already_landed_eligible,
         _rule_reviewer_already_landed_state,
         _rule_reviewer_target_branch_freshness,
+        _rule_reviewer_workflow_load_boundary,
         _rule_reviewer_mutation_ledger,
         _rule_reviewer_review_mutation,
         _rule_reviewer_post_merge_cleanup_proof,

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
@@ -714,6 +736,7 @@ import role_session_router  # noqa: E402
 import role_namespace_gate  # noqa: E402
 import task_capability_map  # noqa: E402
 import review_proofs  # noqa: E402
+import review_workflow_boundary  # noqa: E402
 import review_workflow_load  # noqa: E402
 import agent_temp_artifacts
 import issue_lock_worktree  # noqa: E402
@@ -724,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
@@ -734,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
@@ -4653,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,
@@ -6924,14 +7033,42 @@ def gitea_get_runtime_context(
     return result
 
 
+@mcp.tool()
+def gitea_record_pre_review_command(
+    command: str,
+    cwd: str | None = None,
+    classification: str | None = None,
+) -> dict:
+    """Classify and record a command executed before workflow load (#403).
+
+    Read-only with respect to Gitea API. Pre-review inventory/diagnostic commands
+    may be recorded as allowed; boundary violations block reviewer mutations.
+    """
+    recorded = review_workflow_boundary.record_pre_review_command(
+        command,
+        cwd=cwd,
+        project_root=PROJECT_ROOT,
+        classification=classification,
+    )
+    boundary_state = review_workflow_boundary.assess_boundary_status(PROJECT_ROOT)
+    return {
+        "success": True,
+        "recorded": recorded,
+        "boundary_status": boundary_state.get("boundary_status"),
+        "boundary_clean": boundary_state.get("boundary_clean"),
+        "reasons": list(boundary_state.get("reasons") or []),
+    }
+
+
 @mcp.tool()
 def gitea_load_review_workflow(
     prompt_text: str | None = None,
 ) -> dict:
-    """Load and record canonical review-merge workflow proof for this session (#389).
+    """Load and record canonical review-merge workflow proof for this session (#389, #403).
 
     Read-only with respect to Gitea API; records in-process workflow source/hash
-    proof required before reviewer review or merge mutations.
+    proof and session boundary state required before reviewer review or merge
+    mutations.
     """
     try:
         recorded = review_workflow_load.record_review_workflow_load(
@@ -6943,8 +7080,11 @@ def gitea_load_review_workflow(
             "reasons": [str(exc)],
             "recovery_handoff": review_workflow_load.recovery_handoff_without_replay(),
         }
+    boundary_reasons = review_workflow_boundary.boundary_blockers(PROJECT_ROOT)
+    helper = review_workflow_boundary.workflow_load_helper_result(
+        recorded, PROJECT_ROOT)
     return {
-        "success": True,
+        "success": not boundary_reasons,
         "loaded": True,
         "workflow_source": recorded["workflow_source"],
         "task_mode": recorded["task_mode"],
@@ -6956,7 +7096,14 @@ def gitea_load_review_workflow(
             "prompt_conflicts_with_workflow"],
         "prompt_conflict_reasons": recorded.get("prompt_conflict_reasons") or [],
         "workflow_load_proof_present": True,
-        "reasons": [],
+        "boundary_status": recorded.get("boundary_status"),
+        "boundary_clean": recorded.get("boundary_clean"),
+        "workflow_load_helper_result": helper,
+        "reasons": boundary_reasons,
+        "recovery_handoff": (
+            review_workflow_load.recovery_handoff_without_replay()
+            if boundary_reasons else []
+        ),
     }
 
 

review_workflow_boundary.py

@@ -0,0 +1,259 @@
+"""Reviewer session boundary tracking for workflow-load gate (#403).
+
+Pre-review commands executed before ``gitea_load_review_workflow`` must be
+classified. Boundary violations block downstream reviewer mutations even when
+workflow hash proof is present.
+"""
+
+from __future__ import annotations
+
+import os
+import re
+from typing import Any
+
+CLASSIFICATION_READ_ONLY_INVENTORY = "read_only_inventory"
+CLASSIFICATION_DIAGNOSTIC = "diagnostic"
+CLASSIFICATION_BOUNDARY_VIOLATION = "boundary_violation"
+CLASSIFICATION_UNCLASSIFIED = "unclassified"
+
+ALLOWED_CLASSIFICATIONS = frozenset({
+    CLASSIFICATION_READ_ONLY_INVENTORY,
+    CLASSIFICATION_DIAGNOSTIC,
+    CLASSIFICATION_BOUNDARY_VIOLATION,
+    CLASSIFICATION_UNCLASSIFIED,
+})
+
+_PRE_REVIEW_COMMANDS: list[dict[str, Any]] = []
+
+_READ_ONLY_INVENTORY_PATTERNS = (
+    re.compile(
+        r"\bgitea[_-](?:list|view|whoami|get[-_]|resolve[-_]task|check[-_]pr|route[-_]task)",
+        re.I,
+    ),
+    re.compile(r"\bgit\s+(?:fetch|remote\s+update|branch\s+-a|log|show|rev-parse)\b", re.I),
+    re.compile(r"\bgit\s+status\b", re.I),
+    re.compile(r"\bgit\s+worktree\s+list\b", re.I),
+)
+
+_DIAGNOSTIC_PATTERNS = (
+    re.compile(r"\bgit\s+diff(?:\s+--stat)?\b", re.I),
+    re.compile(r"\bwhich\s+pytest\b", re.I),
+    re.compile(r"\bpytest\s+--version\b", re.I),
+)
+
+_BOUNDARY_VIOLATION_PATTERNS: tuple[tuple[re.Pattern[str], str], ...] = (
+    (re.compile(r"\b(?:pytest|python\s+-m\s+pytest|python\s+-m\s+unittest)\b", re.I),
+     "validation command before workflow load"),
+    (re.compile(r"\bprofiles\.json\b", re.I), "local profile config inspection"),
+    (re.compile(r"\bgitea-mcp(?:\.v2-contexts)?\.json\b", re.I),
+     "local Gitea MCP config inspection"),
+    (re.compile(r"\b\.env(?:\.|$|\b)", re.I), "credential file inspection"),
+    (re.compile(r"\bkeychain\b", re.I), "credential store inspection"),
+    (re.compile(r"\bpkill\b", re.I), "MCP repair activity"),
+    (re.compile(r"\b(?:edit|write|modify).{0,40}\bmcp\b", re.I),
+     "MCP config exploration"),
+    (re.compile(r"\bgit\s+(?:add|commit|reset|clean|checkout|merge|rebase|push)\b", re.I),
+     "git mutation before workflow load"),
+)
+
+
+def clear_pre_review_commands() -> None:
+    """Test helper and session reset."""
+    global _PRE_REVIEW_COMMANDS
+    _PRE_REVIEW_COMMANDS = []
+
+
+def pre_review_commands() -> list[dict[str, Any]]:
+    """Return a shallow copy of recorded pre-review commands."""
+    return [dict(entry) for entry in _PRE_REVIEW_COMMANDS]
+
+
+def _normalize_path(path: str | None) -> str:
+    return os.path.realpath(os.path.abspath((path or "").strip() or os.getcwd()))
+
+
+def is_main_checkout_path(cwd: str | None, project_root: str | None) -> bool:
+    """True when *cwd* is the stable control checkout (not under branches/)."""
+    if not project_root:
+        return False
+    root = _normalize_path(project_root)
+    path = _normalize_path(cwd)
+    if path != root:
+        return False
+    marker = f"{os.sep}branches{os.sep}"
+    return marker not in path
+
+
+def classify_pre_review_command(
+    command: str,
+    *,
+    cwd: str | None = None,
+    project_root: str | None = None,
+) -> dict[str, Any]:
+    """Classify a command executed before workflow load."""
+    text = (command or "").strip()
+    path = _normalize_path(cwd)
+    root = _normalize_path(project_root) if project_root else None
+    reasons: list[str] = []
+
+    for pattern, label in _BOUNDARY_VIOLATION_PATTERNS:
+        if pattern.search(text):
+            if label.startswith("validation") and root and not is_main_checkout_path(path, root):
+                continue
+            if label.startswith("git mutation") and root and not is_main_checkout_path(path, root):
+                continue
+            reasons.append(label)
+            return {
+                "command": text,
+                "cwd": path,
+                "classification": CLASSIFICATION_BOUNDARY_VIOLATION,
+                "reasons": reasons,
+            }
+
+    for pattern in _READ_ONLY_INVENTORY_PATTERNS:
+        if pattern.search(text):
+            return {
+                "command": text,
+                "cwd": path,
+                "classification": CLASSIFICATION_READ_ONLY_INVENTORY,
+                "reasons": [],
+            }
+
+    for pattern in _DIAGNOSTIC_PATTERNS:
+        if pattern.search(text):
+            return {
+                "command": text,
+                "cwd": path,
+                "classification": CLASSIFICATION_DIAGNOSTIC,
+                "reasons": [],
+            }
+
+    if root and is_main_checkout_path(path, root):
+        if re.search(r"\b(?:cat|head|less|read)\b", text, re.I):
+            if re.search(r"workflow|skill|runbook", text, re.I):
+                return {
+                    "command": text,
+                    "cwd": path,
+                    "classification": CLASSIFICATION_BOUNDARY_VIOLATION,
+                    "reasons": [
+                        "canonical workflow viewed as local file without "
+                        "gitea_load_review_workflow (narrative load is not proof)"
+                    ],
+                }
+
+    return {
+        "command": text,
+        "cwd": path,
+        "classification": CLASSIFICATION_UNCLASSIFIED,
+        "reasons": [
+            "pre-review command not classified; record via "
+            "gitea_record_pre_review_command before workflow load"
+        ],
+    }
+
+
+def record_pre_review_command(
+    command: str,
+    *,
+    cwd: str | None = None,
+    project_root: str | None = None,
+    classification: str | None = None,
+) -> dict[str, Any]:
+    """Record and classify a pre-review command for the current session."""
+    assessed = classify_pre_review_command(
+        command, cwd=cwd, project_root=project_root)
+    if classification:
+        if classification not in ALLOWED_CLASSIFICATIONS:
+            assessed["classification"] = CLASSIFICATION_UNCLASSIFIED
+            assessed["reasons"] = [
+                f"unknown classification '{classification}'; fail closed"
+            ]
+        else:
+            assessed["classification"] = classification
+            assessed["reasons"] = []
+    entry = {
+        **assessed,
+        "session_pid": os.getpid(),
+    }
+    _PRE_REVIEW_COMMANDS.append(entry)
+    return dict(entry)
+
+
+def assess_boundary_status(project_root: str | None = None) -> dict[str, Any]:
+    """Summarize pre-review boundary state for session proof and reports."""
+    violations = [
+        entry for entry in _PRE_REVIEW_COMMANDS
+        if entry.get("classification") == CLASSIFICATION_BOUNDARY_VIOLATION
+    ]
+    unclassified = [
+        entry for entry in _PRE_REVIEW_COMMANDS
+        if entry.get("classification") == CLASSIFICATION_UNCLASSIFIED
+    ]
+    reasons: list[str] = []
+    for entry in violations:
+        reasons.extend(entry.get("reasons") or [
+            f"boundary violation: {entry.get('command', '')[:80]}"
+        ])
+    for entry in unclassified:
+        reasons.extend(entry.get("reasons") or [
+            "unclassified pre-review command blocks reviewer mutations"
+        ])
+
+    clean = not reasons
+    return {
+        "boundary_status": "clean" if clean else "violation",
+        "boundary_clean": clean,
+        "pre_review_command_count": len(_PRE_REVIEW_COMMANDS),
+        "boundary_violation_count": len(violations),
+        "unclassified_command_count": len(unclassified),
+        "violations": [
+            {
+                "command": v.get("command"),
+                "cwd": v.get("cwd"),
+                "reasons": list(v.get("reasons") or []),
+            }
+            for v in violations
+        ],
+        "reasons": reasons,
+    }
+
+
+def boundary_blockers(project_root: str | None = None) -> list[str]:
+    """Reasons reviewer mutations must fail closed due to boundary state."""
+    status = assess_boundary_status(project_root)
+    if status.get("boundary_clean"):
+        return []
+    return list(status.get("reasons") or [
+        "reviewer session boundary violation before workflow load"
+    ])
+
+
+def workflow_load_helper_result(
+    load: dict | None,
+    project_root: str | None = None,
+) -> dict[str, Any]:
+    """Structured helper result for final reports (#403)."""
+    boundary = assess_boundary_status(project_root)
+    if load is None:
+        return {
+            "workflow_load_proof_present": False,
+            "workflow_source": None,
+            "workflow_hash": None,
+            "final_report_schema_hash": None,
+            "boundary_status": boundary.get("boundary_status"),
+            "boundary_clean": False,
+            "reasons": [
+                "gitea_load_review_workflow helper result missing from report"
+            ],
+        }
+    return {
+        "workflow_load_proof_present": True,
+        "workflow_source": load.get("workflow_source"),
+        "workflow_hash": load.get("workflow_hash"),
+        "final_report_schema_path": load.get("final_report_schema_path"),
+        "final_report_schema_hash": load.get("final_report_schema_hash"),
+        "boundary_status": load.get("boundary_status", boundary.get("boundary_status")),
+        "boundary_clean": bool(load.get("boundary_clean", boundary.get("boundary_clean"))),
+        "pre_review_command_count": boundary.get("pre_review_command_count"),
+        "reasons": [],
+    }

review_workflow_load.py

@@ -1,4 +1,4 @@
-"""Canonical review-merge workflow load proof for reviewer mutations (#389)."""
+"""Canonical review-merge workflow load proof for reviewer mutations (#389, #403)."""
 
 from __future__ import annotations
 
@@ -7,6 +7,8 @@ import os
 import re
 from pathlib import Path
 
+import review_workflow_boundary as boundary
+
 WORKFLOW_REL_PATH = (
     "skills/llm-project-workflow/workflows/review-merge-pr.md"
 )
@@ -95,10 +97,16 @@ def record_review_workflow_load(
     global _REVIEW_WORKFLOW_LOAD
     meta = build_canonical_workflow_metadata(
         project_root, prompt_text=prompt_text)
+    boundary_state = boundary.assess_boundary_status(project_root)
     _REVIEW_WORKFLOW_LOAD = {
         **meta,
         "session_pid": os.getpid(),
         "loaded": True,
+        "boundary_status": boundary_state.get("boundary_status"),
+        "boundary_clean": boundary_state.get("boundary_clean"),
+        "pre_review_command_count": boundary_state.get("pre_review_command_count"),
+        "boundary_violation_count": boundary_state.get("boundary_violation_count"),
+        "boundary_reasons": list(boundary_state.get("reasons") or []),
     }
     return dict(_REVIEW_WORKFLOW_LOAD)
 
@@ -107,6 +115,7 @@ def clear_review_workflow_load() -> None:
     """Test helper and review_pr session reset."""
     global _REVIEW_WORKFLOW_LOAD
     _REVIEW_WORKFLOW_LOAD = None
+    boundary.clear_pre_review_commands()
 
 
 def workflow_load_status(project_root: str | None = None) -> dict:
@@ -125,6 +134,9 @@ def workflow_load_status(project_root: str | None = None) -> dict:
             ],
         }
     reasons = _session_validation_reasons(load, project_root)
+    boundary_reasons = boundary.boundary_blockers(project_root)
+    if boundary_reasons:
+        reasons = list(reasons) + boundary_reasons
     return {
         "workflow_load_proof_present": True,
         "workflow_load_valid": not reasons,
@@ -136,6 +148,10 @@ def workflow_load_status(project_root: str | None = None) -> dict:
         "prompt_conflicts_with_workflow": load.get(
             "prompt_conflicts_with_workflow"),
         "session_pid": load.get("session_pid"),
+        "boundary_status": load.get("boundary_status"),
+        "boundary_clean": load.get("boundary_clean"),
+        "workflow_load_helper_result": boundary.workflow_load_helper_result(
+            load, project_root),
         "reasons": reasons,
     }
 
@@ -179,9 +195,12 @@ def review_workflow_load_blockers(
     project_root: str | None = None,
 ) -> list[str]:
     """Reasons reviewer mutations must fail closed."""
+    boundary_reasons = boundary.boundary_blockers(project_root)
+    if boundary_reasons and _REVIEW_WORKFLOW_LOAD is None:
+        return boundary_reasons
     status = workflow_load_status(project_root)
     if not status.get("workflow_load_proof_present"):
-        return list(status.get("reasons") or [])
+        return list(status.get("reasons") or []) + boundary_reasons
     if not status.get("workflow_load_valid"):
         return list(status.get("reasons") or [])
     return []

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/schemas/review-merge-final-report.md

@@ -63,8 +63,14 @@ Do not use legacy fields: `Pinned reviewed head`, `Scratch worktree used`,
 - Current status:
 - Safe next action:
 - Safety statement:
+- Workflow-load helper result:
 ```
 
+The **Workflow-load helper result** field must carry structured output from
+`gitea_load_review_workflow` (workflow_hash, final_report_schema_hash,
+boundary_status). Narrative claims that workflow files were viewed locally are
+not sufficient (#403).
+
 ### Already-landed handoff overrides
 
 When eligibility class is `ALREADY_LANDED_RECONCILE_REQUIRED`:

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

@@ -36,6 +36,44 @@ If available, load it first and report:
 
 If the canonical workflow cannot be loaded and the project requires it, stop and produce a recovery handoff only.
 
+## 0A. Workflow-load and session boundary anchor (#403)
+
+The MCP gate is the authority — not local file viewing.
+
+Before any reviewer mutation:
+
+1. Record pre-review commands with `gitea_record_pre_review_command` when they
+   are not automatically classified (inventory/diagnostic commands may be
+   recorded explicitly for proof).
+2. Call `gitea_load_review_workflow` to establish workflow hash proof **and**
+   session boundary state in the same in-process session proof.
+3. Do not claim the workflow was loaded by reading
+   `skills/llm-project-workflow/workflows/review-merge-pr.md` as a local file;
+   that narrative does not satisfy the validator.
+
+Allowed before workflow load (classify as `read_only_inventory` or
+`diagnostic`):
+
+* `gitea_whoami`, `gitea_resolve_task_capability`, `gitea_list_prs`,
+  `gitea_view_pr`, `gitea_get_runtime_context`
+* `git fetch` / `git remote update` for inventory
+* `git status`, `git worktree list` (read-only)
+
+Boundary violations (block downstream reviewer mutations even after load):
+
+* validation commands (`pytest`, `python -m unittest`) in the main checkout
+* local profile/credential/config inspection (`profiles.json`, `gitea-mcp.json`,
+  `.env`, keychain dumps)
+* MCP repair (`pkill`, MCP config edits)
+* git mutations before workflow load
+
+Final reports must include a structured **Workflow-load helper result** block
+copied from `gitea_load_review_workflow`, including at minimum:
+
+* `workflow_hash`
+* `final_report_schema_hash`
+* `boundary_status` (`clean` or `violation`)
+
 ## 1. Start with live identity, profile, runtime, and capability checks
 
 Prove:
@@ -871,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,9 +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"})
-    def test_create_issue_stable_checkout_rejected(self, _git, _get_all, mock_api, _role, _ns, _prof, _auth):
+    @patch("gitea_mcp_server.root_checkout_guard.resolve_remote_master_sha", return_value="a" * 40)
+    @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_mcp_server.py

@@ -1057,7 +1057,8 @@ class TestMergePR(unittest.TestCase):
                "GITEA_ALLOWED_OPERATIONS": "read,merge"}
         with patch.dict(os.environ, env, clear=True):
             r = gitea_merge_pr(
-                pr_number=8, confirmation=self._confirm(8), remote="prgs")
+                pr_number=8, confirmation=self._confirm(8), remote="prgs",
+                expected_head_sha=new_sha)
         self.assertFalse(r["performed"])
         self.assertTrue(r.get("approval_visible"))
         self.assertFalse(r.get("approval_at_current_head"))

tests/test_permission_reports.py

@@ -95,8 +95,12 @@ class PermissionReportBase(unittest.TestCase):
         self._dir = tempfile.TemporaryDirectory()
         self.config_path = os.path.join(self._dir.name, "profiles.json")
         self._write_config(CONFIG)
+        import review_workflow_load
+        review_workflow_load.record_review_workflow_load(mcp_server.PROJECT_ROOT)
 
     def tearDown(self):
+        import review_workflow_load
+        review_workflow_load.clear_review_workflow_load()
         self._remotes.stop()
         mcp_server._IDENTITY_CACHE.clear()
         mcp_server.review_workflow_load.clear_review_workflow_load()

tests/test_review_workflow_boundary.py

@@ -0,0 +1,138 @@
+"""Tests for workflow-load session boundary tracking (#403)."""
+
+import os
+import sys
+import unittest
+from unittest.mock import patch
+
+sys.path.insert(0, str(__import__("pathlib").Path(__file__).resolve().parent.parent))
+
+import final_report_validator
+import review_workflow_boundary
+import review_workflow_load
+import mcp_server
+
+
+class TestPreReviewClassification(unittest.TestCase):
+    def setUp(self):
+        review_workflow_boundary.clear_pre_review_commands()
+        review_workflow_load.clear_review_workflow_load()
+
+    def test_inventory_command_allowed(self):
+        result = review_workflow_boundary.classify_pre_review_command(
+            "gitea_list_prs remote=prgs",
+            cwd="/tmp",
+            project_root="/repo/Gitea-Tools",
+        )
+        self.assertEqual(
+            result["classification"],
+            review_workflow_boundary.CLASSIFICATION_READ_ONLY_INVENTORY,
+        )
+
+    def test_main_checkout_pytest_is_boundary_violation(self):
+        root = "/repo/Gitea-Tools"
+        result = review_workflow_boundary.classify_pre_review_command(
+            "python -m pytest tests/",
+            cwd=root,
+            project_root=root,
+        )
+        self.assertEqual(
+            result["classification"],
+            review_workflow_boundary.CLASSIFICATION_BOUNDARY_VIOLATION,
+        )
+
+    def test_profiles_json_inspection_is_boundary_violation(self):
+        result = review_workflow_boundary.classify_pre_review_command(
+            "cat profiles.json",
+            cwd="/repo/Gitea-Tools",
+            project_root="/repo/Gitea-Tools",
+        )
+        self.assertEqual(
+            result["classification"],
+            review_workflow_boundary.CLASSIFICATION_BOUNDARY_VIOLATION,
+        )
+
+
+class TestWorkflowLoadBoundaryGate(unittest.TestCase):
+    def setUp(self):
+        review_workflow_boundary.clear_pre_review_commands()
+        review_workflow_load.clear_review_workflow_load()
+        mcp_server.init_review_decision_lock("prgs", "review_pr")
+        mcp_server.record_preflight_check("whoami")
+        mcp_server.record_preflight_check("capability", "reviewer")
+
+    def _root(self) -> str:
+        return str(__import__("pathlib").Path(__file__).resolve().parent.parent)
+
+    def test_boundary_violation_blocks_mutation_after_load(self):
+        root = self._root()
+        review_workflow_boundary.record_pre_review_command(
+            "python -m pytest tests/",
+            cwd=root,
+            project_root=root,
+        )
+        res = mcp_server.gitea_load_review_workflow()
+        self.assertFalse(res["success"])
+        self.assertEqual(res["boundary_status"], "violation")
+        blocked = mcp_server.gitea_mark_final_review_decision(
+            42, "approve", remote="prgs")
+        self.assertFalse(blocked["marked_ready"])
+        joined = " ".join(blocked["reasons"]).lower()
+        self.assertTrue(
+            "validation" in joined or "boundary" in joined or "workflow" in joined
+        )
+
+    def test_clean_inventory_then_load_passes(self):
+        root = self._root()
+        review_workflow_boundary.record_pre_review_command(
+            "gitea_list_prs remote=prgs",
+            cwd=root,
+            project_root=root,
+        )
+        res = mcp_server.gitea_load_review_workflow()
+        self.assertTrue(res["success"])
+        self.assertEqual(res["boundary_status"], "clean")
+        blockers = review_workflow_load.review_workflow_load_blockers(root)
+        self.assertEqual(blockers, [])
+
+    def test_file_view_narrative_fails_validator_without_helper(self):
+        report = (
+            "## Controller Handoff\n"
+            "- Task: review-merge-pr\n"
+            "- I read the canonical workflow review-merge-pr.md before review.\n"
+        )
+        findings = final_report_validator.assess_final_report_validator(
+            report,
+            task_kind="review_pr",
+        )
+        self.assertTrue(any(
+            f["rule_id"] == "reviewer.workflow_load_boundary"
+            for f in findings.get("findings") or []
+        ))
+
+    def test_helper_result_passes_validator(self):
+        root = self._root()
+        review_workflow_load.record_review_workflow_load(root)
+        helper = review_workflow_boundary.workflow_load_helper_result(
+            review_workflow_load._REVIEW_WORKFLOW_LOAD,
+            root,
+        )
+        report = (
+            "## Controller Handoff\n"
+            "- Task: review-merge-pr\n"
+            f"- Workflow-load helper result: workflow_hash: {helper['workflow_hash']}; "
+            f"boundary_status: {helper['boundary_status']}\n"
+        )
+        findings = final_report_validator.assess_final_report_validator(
+            report,
+            task_kind="review_pr",
+        )
+        boundary_findings = [
+            f for f in (findings.get("findings") or [])
+            if f.get("rule_id") == "reviewer.workflow_load_boundary"
+        ]
+        self.assertEqual(boundary_findings, [])
+
+
+if __name__ == "__main__":
+    unittest.main()

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

@@ -36,10 +36,13 @@ def _lock(mutations=None, correction=False):
 
 
 def _seed(mutations=None, correction=False):
+    import review_workflow_load
+    review_workflow_load.record_review_workflow_load(mcp_server.PROJECT_ROOT)
     mcp_server._save_review_decision_lock(_lock(mutations, correction))
     mcp_server.gitea_load_review_workflow()
 
 
+
 APPROVED_A = {"pr_number": 5, "action": "approve", "review_id": 1,
               "review_state": "approve"}
 RC_A = {"pr_number": 5, "action": "request_changes", "review_id": 2,
@@ -94,6 +97,8 @@ class TestTerminalHardStopReasons(unittest.TestCase):
 
 class TestMergeHardStopWiring(unittest.TestCase):
     def tearDown(self):
+        import review_workflow_load
+        review_workflow_load.clear_review_workflow_load()
         mcp_server._save_review_decision_lock(None)
         mcp_server.review_workflow_load.clear_review_workflow_load()
 
@@ -116,6 +121,8 @@ class TestMergeHardStopWiring(unittest.TestCase):
 
 class TestMarkFinalHardStopWiring(unittest.TestCase):
     def tearDown(self):
+        import review_workflow_load
+        review_workflow_load.clear_review_workflow_load()
         mcp_server._save_review_decision_lock(None)
         mcp_server.review_workflow_load.clear_review_workflow_load()
 
@@ -149,6 +156,8 @@ def _mark(action, pr_number=6, **kwargs):
 
 class TestDuplicateRequestChangesSuppression(unittest.TestCase):
     def tearDown(self):
+        import review_workflow_load
+        review_workflow_load.clear_review_workflow_load()
         mcp_server._save_review_decision_lock(None)
         mcp_server.review_workflow_load.clear_review_workflow_load()
 

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),
+    }