trusty-mpm 1.7.0

//! Decommission and workspace-ownership methods for the session manager (#1511).
//!
//! Why: `decommission` and the companion `set_workspace_owned` are extracted
//! from `manager.rs` to keep that file under the 500-SLOC production cap,
//! mirroring the pattern used by `adopt.rs` and `prune.rs`. The decommission
//! logic is also a natural home for the ownership-tracking primitive.
//! What: public [`SessionManager::decommission`] (full teardown with the #1511
//! dual guard), internal [`SessionManager::decommission_with_root`] (injectable
//! managed-root for test isolation without env mutation), and
//! [`SessionManager::set_workspace_owned`] (marks a workspace as SM-provisioned).
//! Test: `manager_decommission_removes_workspace`,
//! `manager_decommission_unowned_skips_deletion`,
//! `workspace_owned_flag_round_trips_via_set` in `super::tests`.

use std::path::Path;

use chrono::Utc;

use tracing::{info, warn};

use crate::core::trusty_tools_config::{TrustyToolsConfig, workspace_root};

use super::manager::{ManagedError, SessionManager};
use super::record::{ManagedSessionId, ManagedSessionState, SessionRecord};
use super::search_gc;
use super::workspace_guard::{foreign_active_claim, is_safe_to_remove};
use super::worktree_protection;
use super::worktree_registry;
use super::worktree_safety::{DirtyWorktree, inspect_dirt, worktree_remove_command};

/// Sentinel file written by [`create_session_worktree`] into every SM-created
/// per-session git worktree (#1845 item 5).
///
/// Why: `is_session_worktree` identifies worktrees by the `.worktrees` parent-name
/// convention, but a user-owned directory that is a direct child of a `.worktrees/`
/// directory would be misclassified and deleted. The sentinel provides an explicit
/// SM-ownership marker so `remove_session_worktree` can distinguish TM-created dirs
/// from user-owned ones without relying solely on the naming convention. The convention
/// check is kept as a fallback for worktrees created before this sentinel was
/// introduced (backward-compatibility).
/// What: a zero-byte file named `.trusty-mpm-worktree` written at the root of every
/// SM-created worktree by [`create_session_worktree`] immediately after git creates it.
/// Test: `sentinel_gates_worktree_removal` in the sibling `decommission_tests`.
pub(crate) const WORKTREE_SENTINEL_FILE: &str = ".trusty-mpm-worktree";

/// Directory name (relative to a base git checkout) under which NEW SM-created
/// per-session git worktrees are created.
///
/// Why: both the in-project spawn path (`daemon::managed_routes::inproject`)
/// and the clone-based shared-base-checkout path (#1935,
/// `provisioner::workspace`) nest per-session worktrees one level under a
/// shared base checkout; naming the segment once (rather than repeating the
/// `".worktrees"` string literal at each call site) keeps the convention
/// singular and greppable. #5204 turned that constant into a resolver so the
/// name is configurable — and so `trusty-search`'s indexing exclusion and
/// `trusty-memory`'s workstream attribution resolve the SAME value.
/// What: [`crate::core::trusty_tools_config::worktrees_dirname`] over the
/// loaded host config — **env > config > `.worktrees`**. Use this ONLY where a
/// worktree is being created or a path is being built; every "is this a
/// worktree base?" question goes to [`worktree_dir_names`] instead.
/// Test: `is_session_worktree_detects_dot_worktrees_component`,
/// `worktrees_dirname_delegates_to_the_shared_resolver`.
pub(crate) fn worktrees_dirname() -> String {
    crate::core::trusty_tools_config::worktrees_dirname(
        &crate::core::trusty_tools_config::TrustyToolsConfig::load(),
    )
}

/// The resolved worktree-base names for DETECTION — configured plus built-in.
///
/// Why: detection is deliberately a superset of creation (#5204). An operator
/// who retargets the base still has worktrees on disk under `.worktrees`; if
/// detection narrowed to the configured name alone, those would stop being
/// recognised as session worktrees — decommission would refuse to remove them,
/// prune would skip them, and trusty-search would start indexing every one.
/// What: a [`trusty_common::workspace_layout::WorktreeDirNames`] whose `matches`
/// accepts the configured name OR `.worktrees`.
/// Test: `is_session_worktree_detects_dot_worktrees_component`.
pub(crate) fn worktree_dir_names() -> trusty_common::workspace_layout::WorktreeDirNames {
    trusty_common::workspace_layout::WorktreeDirNames::from_configured(
        crate::core::trusty_tools_config::TrustyToolsConfig::load()
            .worktrees_dirname
            .as_deref(),
    )
}

/// Timeout for the blocking `git worktree remove` subprocess (#1845 item 4).
///
/// Why: `std::process::Command` is synchronous and has no built-in timeout. A git
/// process that hangs (e.g. waiting for a network mount or a file lock) would
/// block the daemon's async executor indefinitely when called from `decommission`,
/// making the daemon unresponsive for the duration. A 30-second bound converts
/// a hung git call into a clean timeout log entry and a conservative `false` return.
/// What: a [`std::time::Duration`] of 30 seconds passed to `tokio::time::timeout`
/// wrapping the `spawn_blocking` that runs `remove_session_worktree`.
/// Test: `git_worktree_remove_timeout_is_bounded_constant`.
pub(crate) const GIT_WORKTREE_REMOVE_TIMEOUT: std::time::Duration =
    std::time::Duration::from_secs(30);

/// True when `path` is an SM-created per-session git worktree (#1840).
///
/// Why: in-project sessions create their workspace at
/// `<base>/.worktrees/<session-id>/` with `workspace_owned = false` — they do
/// NOT own the base clone, but they DO own their worktree slice. The standard
/// `workspace_owned` guard therefore skips removal entirely, leaving orphaned
/// worktree directories and stale git worktree refs. This predicate identifies
/// the SM-worktree pattern so decommission can take targeted worktree-removal
/// action for `workspace_owned = false` sessions.
/// What: returns `true` when the path's immediate parent directory is named
/// `.worktrees` — i.e. the path is `<base>/.worktrees/<session-id>`. Checking
/// only the immediate parent (not any ancestor) prevents false positives for
/// paths like `<base>/.worktrees/deep/nested` where `.worktrees` is a grandparent.
/// Test: `is_session_worktree_detects_dot_worktrees_component`.
///
/// Visibility (#2033, widened #2508): `pub(crate)` — reused by
/// `session_manager::search_gc` (same "disposable workspace?" rule for the
/// search-index lifecycle) and by `core::agent_reset_workspace` (the same
/// rule gates which session workspaces `--reset-agents-workspaces` is
/// allowed to force-recompose files into — a local-path/adopted session's
/// real, long-lived checkout must never be mistaken for a disposable
/// SM-provisioned one, the #1511 incident class). Keeping this in exactly one
/// place means the filesystem-removal guard, the search-index GC guard, and
/// the agent-reset guard can never diverge.
pub(crate) fn is_session_worktree(path: &Path) -> bool {
    // #5204: detection matches the configured base OR the built-in `.worktrees`,
    // so retargeting never orphans worktrees already on disk.
    is_session_worktree_with(path, &worktree_dir_names())
}

/// [`is_session_worktree`] against ALREADY-RESOLVED base names.
///
/// Why: [`worktree_dir_names`] calls `TrustyToolsConfig::load()`, which re-reads
/// and re-parses config files on every call and re-emits its unknown-key warning
/// (#5207). That is fine once, but a caller asking the question per item in a
/// loop pays it per item — and on a repeating timer it also multiplies that
/// one-shot warning into a per-item, per-tick log flood. `WorktreeDirNames`'s own
/// docs prescribe the remedy: resolve once at the top of a scan and pass the
/// value down (#5204). #5327's retention sweep is the first such loop caller.
/// What: the pure path-shape predicate — is the immediate parent a base name.
/// [`is_session_worktree`] is this function plus a per-call resolve.
/// Test: `is_session_worktree_detects_dot_worktrees_component` (through the
/// delegating wrapper), `workspace_needs_protection_covers_a_session_worktree`.
pub(crate) fn is_session_worktree_with(
    path: &Path,
    names: &trusty_common::workspace_layout::WorktreeDirNames,
) -> bool {
    path.parent()
        .and_then(|p| p.file_name())
        .and_then(|n| n.to_str())
        .map(|n| names.matches(n))
        .unwrap_or(false)
}

/// May trusty-mpm remove this worktree directory at all (#6561)?
///
/// Why: this is the OWNERSHIP question — "is this ours to delete", asked before
/// any of the safety questions about whether deleting it right now would lose
/// work. It existed twice, inline: once as [`remove_session_worktree`]'s
/// two-tier sentinel/convention gate and once as
/// [`super::worktree_reclaim::tm_provisioned`], whose whole doc is about
/// applying "exactly the predicate the remover applies". Two copies of one rule
/// is the drift this repository's common-entry-point convention forbids, and the
/// drift had already happened in the other direction: the classifier's copy said
/// the harness `.claude/worktrees/` store could never be removed, while
/// [`super::super::daemon::services::agent_worktree_reap`] removes trees from
/// exactly that store on every agent exit.
///
/// What: three tiers, any of which admits the path.
///
/// 1. **Sentinel** — the path carries [`WORKTREE_SENTINEL_FILE`]. This is the
///    explicit marker every SM-created worktree gets, and since #4311 it is also
///    what the delegation tracker writes into a registered agent worktree.
/// 2. **Convention** — [`is_session_worktree`], the `.worktrees/<name>` shape,
///    kept for worktrees created before the sentinel existed.
/// 3. **Harness agent store (#6561)** — [`is_harness_agent_worktree`], the
///    `<checkout>/.claude/worktrees/<name>` shape Claude Code's
///    `isolation: "worktree"` dispatch creates.
///
/// **This grants no permission to delete anything.** It answers ownership only —
/// "is this directory a kind trusty-mpm may act on at all" — and every gate that
/// decides whether a removal is SAFE stands after it, untouched: the merged-PR
/// requirement, the #4091 dirty-work guard
/// ([`super::worktree_safety::inspect_dirt`], which fails toward dirty and
/// refuses on an unpushed commit), the live-session check, the agent-ownership
/// refusal (#5661), and the per-candidate re-read immediately before each delete.
///
/// **Tier 3 in particular decides nothing on its own (#6561 critic round).**
/// [`super::worktree_reclaim::agent_ownership_blocks`] refuses EVERY agent-store
/// path whose sentinel does not name an agent the delegation registry can
/// resolve, absent sentinels included — so tier 3 can never be the reason such a
/// tree is deleted. Its effect is to keep this predicate and
/// `super::worktree_reclaim::tm_provisioned` a single rule rather than two
/// copies that had already drifted: the classifier's copy called the harness
/// store out of scope while
/// [`super::super::daemon::services::agent_worktree_reap`] removes trees from
/// it on every agent exit.
/// Test: `removal_permitted_admits_all_three_tiers`,
/// `removal_permitted_refuses_a_user_directory`,
/// `tm_provisioned_matches_the_removers_own_predicate`,
/// `an_unattributed_agent_store_worktree_is_never_reclaimable`.
pub(crate) fn removal_permitted(path: &Path) -> bool {
    path.join(WORKTREE_SENTINEL_FILE).exists()
        || is_session_worktree(path)
        // #6561: the harness's own isolation store, which the agent-worktree
        // reaper already removes from.
        || super::worktree_ownership::is_harness_agent_worktree(path)
}

/// What happened to a worktree the removal path was asked to delete (#4732).
///
/// Why: the caller has to be able to tell "gone" from "still on disk, and here
/// is why" — and it has to be told the reason, not left to find it in a `warn!`
/// nobody reads. The previous `bool` could express neither: `false` conflated a
/// deliberate refusal, an I/O error, and a timeout, and the reason existed only
/// as a log line inside the function.
/// What: [`Removed`](Self::Removed) — the directory is gone;
/// [`Kept`](Self::Kept) — it is still on disk, carrying an operator-facing
/// reason.
/// Test: `remove_session_worktree_refuses_a_git_locked_worktree`,
/// `remove_cleans_up_a_directory_no_repository_claims`.
#[derive(Debug, Clone, PartialEq, Eq)]
pub(crate) enum WorktreeRemoval {
    /// The directory is gone — git removed it, it was already absent, or an
    /// unclaimed trusty-mpm-owned directory was removed directly.
    Removed,
    /// The directory is still on disk. The string says why, for the operator.
    Kept(String),
}

impl WorktreeRemoval {
    /// `true` only when the directory is gone.
    pub(super) fn removed(&self) -> bool {
        matches!(self, Self::Removed)
    }

    /// The operator-facing reason the directory was kept, if it was.
    pub(super) fn reason(&self) -> Option<&str> {
        match self {
            Self::Removed => None,
            Self::Kept(reason) => Some(reason),
        }
    }
}

/// Remove an in-project per-session git worktree via `git worktree remove --force`.
///
/// Why (#1840): `remove_dir_all` alone leaves the git ref (`session/<id>`) and
/// the git worktree entry (`.git/worktrees/<id>`) in the base clone, polluting
/// `git worktree list` and `git branch` output. `git worktree remove --force`
/// prunes both the directory AND the ref atomically, restoring a clean state.
/// What: runs `git -C <repo-root> worktree remove --force <path>` where
/// `<repo-root>` is the checkout git itself reports as owning `path`'s
/// worktree registry ([`super::worktree_registry::registry_root_for`], #4207 —
/// previously guessed as the grandparent directory). Also runs
/// `git -C <repo-root> worktree prune` and
/// `git -C <repo-root> branch -D session/<leaf>` on success to clear stale git
/// refs and the session branch, where `<leaf>` is the last component of `path`
/// (the worktree dir name) and the `session/` prefix matches EXACTLY what
/// `inproject::create_session_worktree` creates (issue #2032 fix — before
/// this, the missing prefix meant the branch delete always targeted a
/// nonexistent ref and silently no-opped, leaking every session's branch).
/// Works identically for both pre-#2032 UUID-named leaves and the new
/// semantic-tmux-name leaves, since both share the `session/<leaf>`
/// convention. Branch deletion is best-effort — "not found" is silently
/// ignored since older sessions may not have a branch. OsStr-safe path args
/// avoid lossy UTF-8 coercion (#1840).
/// Idempotent: if `path` is already absent, returns
/// [`Removed`](WorktreeRemoval::Removed).
///
/// # 🔴 Git declining is a REFUSAL, not a failure (#4732)
///
/// This function used to fall through to `std::fs::remove_dir_all` on ANY
/// non-zero exit, and on any failure to resolve the owning checkout. Git exits
/// 128 for every fatal condition, so that fall-through deleted the exact things
/// git was protecting — most sharply, `git worktree lock`, whose only mechanism
/// for saying "leave this alone" is a 128 exit. Locking a worktree to protect
/// it was what got it deleted. A stale worktree pointer and an unreadable
/// `.git` produced the same outcome while their working trees, uncommitted work
/// included, were entirely intact.
///
/// Every git failure is now classified by
/// [`super::worktree_protection`], which answers three states and refuses on
/// two of them. The `remove_dir_all` fallback survives for EXACTLY ONE case:
///
/// > `path` has passed the trusty-mpm ownership gate above, and git has
/// > POSITIVELY established that it holds no state there — the path carries no
/// > `.git` entry, and either no repository exists above it at all, or the
/// > owning repository's `git worktree list` does not name it.
///
/// That is a leftover directory: a worktree git already pruned, or one whose
/// creation never completed registration. There is no ref to prune and nothing
/// for git to protect, so a direct removal is the only way to clean it up.
/// Every other outcome — including "git could not be asked" — keeps the
/// directory and returns [`Kept`](WorktreeRemoval::Kept) with the reason.
///
/// Test: `is_session_worktree_absent_path_is_noop`,
/// `remove_session_worktree_refuses_a_git_locked_worktree`,
/// `remove_refuses_a_stale_worktree_pointer`,
/// `remove_refuses_an_unreadable_git_entry`,
/// `remove_refuses_a_worktree_with_a_broken_git_file`,
/// `remove_cleans_up_a_directory_no_repository_claims`,
/// `remove_cleans_up_an_unregistered_leftover_inside_a_repo`,
/// `remove_still_removes_a_healthy_worktree`; integration coverage via the
/// decommission round-trip tests that set up real git worktrees.
/// #7885: `reason` names WHICH route asked for this removal. Every route ended
/// here already, and each logged differently or not at all — two merged
/// worktrees lost their tracked `crates/` subtree on 2026-09-14 and no log named
/// either path or session, so the mechanism could not be identified even after
/// the fact. It is a required argument rather than a defaulted one because a
/// route that cannot say why it is deleting is the case that went unrecorded.
pub(super) fn remove_session_worktree(path: &Path, reason: &str) -> WorktreeRemoval {
    remove_session_worktree_guarded(path, reason, &|| None)
}

/// [`remove_session_worktree`] with one last refusal asked inside the audit
/// window (#7652 critic round).
///
/// Why: the merged-PR reclaim's claim and owner answers go stale across its own
/// dirt probe, which takes seconds on a large tree, and `--force` follows.
/// What: `guard` runs after the ownership gate and the audit attempt line, and
/// immediately before `git worktree remove --force`. `Some(reason)` keeps the
/// tree, and the audit outcome line records that refusal.
/// Test: `worktree_7652_an_owner_back_after_the_dirt_check_is_refused`.
pub(super) fn remove_session_worktree_guarded(
    path: &Path,
    reason: &str,
    guard: &dyn Fn() -> Option<String>,
) -> WorktreeRemoval {
    if !path.exists() {
        // Already gone — either removed by a concurrent decommission or by a
        // previous partial run. Treat as success (idempotent removal).
        return WorktreeRemoval::Removed;
    }

    // Data-safety gate (#1845 item 5): prefer the SM ownership sentinel over the
    // naming-convention check. Every SM-created worktree has a `.trusty-mpm-worktree`
    // sentinel written by `create_session_worktree`. If the sentinel is ABSENT:
    //   • and the path IS under `.worktrees/` → backward-compat (pre-sentinel worktree);
    //     proceed with a WARN so operators know the sentinel is missing.
    //   • and the path is NOT under `.worktrees/` → NOT a SM worktree; refuse removal.
    // This two-tier check is conservative: it avoids deleting user-owned directories
    // that happen to sit under a `.worktrees/` parent.
    let sentinel = path.join(WORKTREE_SENTINEL_FILE);
    if !sentinel.exists() {
        // #6561: one predicate, shared with the reclaim classifier that proposes
        // these candidates — see `removal_permitted`.
        if !removal_permitted(path) {
            warn!(
                path = %path.display(),
                sentinel = WORKTREE_SENTINEL_FILE,
                "decommission: refusing worktree removal — no SM ownership sentinel \
                 and path is not under .worktrees/; skipping conservatively"
            );
            // #5204: name the configured base in the message the operator reads.
            return WorktreeRemoval::Kept(format!(
                "no trusty-mpm ownership sentinel ({WORKTREE_SENTINEL_FILE}) and the path \
                 is not under {}/ or the harness agent-worktree store",
                worktrees_dirname()
            ));
        }
        warn!(
            path = %path.display(),
            sentinel = WORKTREE_SENTINEL_FILE,
            "decommission: sentinel absent; falling back to convention check \
             (backward-compat with pre-sentinel worktrees)"
        );
    }

    // #7885: audited HERE — past the ownership gate, so it describes a removal
    // that is actually attempted, and around every destructive branch in
    // `remove_registered_worktree`, so no route can delete unlogged.
    // #7885 critic round: an attempt line before and an outcome line after, so
    // a refused removal never reads as a deletion.
    super::worktree_removal_audit::audited_removal(path, reason, || {
        // #7652 critic round: the caller's last refusal, after the attempt line.
        if let Some(refusal) = guard() {
            return WorktreeRemoval::Kept(format!("refused immediately before removal: {refusal}"));
        }
        remove_registered_worktree(path)
    })
}

/// Ask git to remove a worktree that already passed the ownership gate (#4207,
/// #4732).
///
/// Why: split from [`remove_session_worktree`] so the removal audit wraps ONE
/// call rather than being repeated on each of this function's five return arms
/// (#7885).
/// What: registry resolution, `git worktree remove --force`, and the two
/// [`worktree_protection`]-gated fallbacks, unchanged by the split.
/// Test: `remove_session_worktree_refuses_a_git_locked_worktree`,
/// `worktree_7885_a_refused_removal_is_never_audited_as_a_deletion`.
fn remove_registered_worktree(path: &Path) -> WorktreeRemoval {
    // #4207: ask git which checkout owns this worktree's registry instead of
    // guessing that it is the grandparent directory. The grandparent rule held
    // only for the two shapes it was written against; a worktree registered to
    // the parent repo but living under `.base/.worktrees/` resolved to `.base`,
    // which disowns it, so `git worktree remove` there could never succeed.
    let repo_root = match super::worktree_registry::registry_root_for(path) {
        Some(r) => r,
        None => {
            // #4732: a `None` here is NOT "no git repository owns this path".
            // `registry_root_for` also returns `None` when git could not
            // resolve one — which is exactly what a worktree with a stale or
            // unreadable `.git` pointer answers, while its working tree and
            // uncommitted work are entirely intact. Classify before deleting.
            let verdict = worktree_protection::protection_without_registry_root(path);
            if let Some(reason) = verdict.refusal() {
                warn!(
                    path = %path.display(),
                    "decommission: refusing worktree removal — {reason} (#4732)"
                );
                return WorktreeRemoval::Kept(reason.to_string());
            }
            // The one surviving fallback case — see this function's doc.
            warn!(
                path = %path.display(),
                "decommission: no git repository claims this path — removing the \
                 directory directly (no worktree ref to prune)"
            );
            return remove_unclaimed_directory(path);
        }
    };
    let repo_root = repo_root.as_path();
    // Step 1: git worktree remove --force <path> (run from repo root).
    // #6391: through the hardened builder, which keeps the #1840 OsStr-safe
    // Path args and additionally strips the env vars that would point
    // `git worktree` at a different repository than `-C` names.
    let out = worktree_remove_command(repo_root, path).output();
    match out {
        Ok(o) if o.status.success() => {
            info!(path = %path.display(), "decommission: git worktree removed (incl. ref)");
            prune_refs_after_removal(repo_root, path);
            WorktreeRemoval::Removed
        }
        Ok(o) => {
            // #4732: git exits 128 for every fatal condition, and `git worktree
            // lock` uses that exit to REFUSE. Classify the reason instead of
            // reading every non-zero exit as permission to delete by hand.
            let stderr = String::from_utf8_lossy(&o.stderr);
            let verdict =
                worktree_protection::protection_after_failed_removal(path, repo_root, &stderr);
            if let Some(reason) = verdict.refusal() {
                warn!(
                    path = %path.display(),
                    "decommission: git worktree remove --force declined ({}) and the \
                     directory must be preserved — {reason} (#4732)",
                    o.status
                );
                return WorktreeRemoval::Kept(reason.to_string());
            }
            // The one surviving fallback case — see this function's doc.
            warn!(
                path = %path.display(),
                "decommission: git holds no worktree state at this path ({}: {stderr}); \
                 removing the leftover directory directly",
                o.status
            );
            remove_unclaimed_directory(path)
        }
        Err(e) => {
            // #4732: git could not be run at all, so nothing is known about
            // what it may be protecting. An unanswerable probe is never a
            // licence to delete.
            let reason = format!("git could not be run to remove the worktree: {e}");
            warn!(
                path = %path.display(),
                "decommission: refusing worktree removal — {reason} (#4732)"
            );
            WorktreeRemoval::Kept(reason)
        }
    }
}

/// Remove a directory git has POSITIVELY disclaimed (#4732).
///
/// Why: named so the one legitimate `remove_dir_all` in this module has exactly
/// one call shape and is greppable. Reaching it requires a
/// [`worktree_protection`] verdict of "no git state at or claiming this path" —
/// never a bare non-zero exit, and never an unanswerable probe. See
/// [`remove_session_worktree`]'s doc for the full statement of that case.
/// What: `std::fs::remove_dir_all`, mapped onto [`WorktreeRemoval`].
/// Test: `remove_cleans_up_a_directory_no_repository_claims`,
/// `remove_cleans_up_an_unregistered_leftover_inside_a_repo`.
fn remove_unclaimed_directory(path: &Path) -> WorktreeRemoval {
    match std::fs::remove_dir_all(path) {
        Ok(()) => WorktreeRemoval::Removed,
        Err(e) => {
            warn!(path = %path.display(), "decommission: remove_dir_all failed: {e}");
            WorktreeRemoval::Kept(format!("removing the directory failed: {e}"))
        }
    }
}

/// Clear the git refs a successful `git worktree remove` leaves behind (#1840,
/// #2032).
///
/// Why: split out of [`remove_session_worktree`] when #4732 turned that
/// function's tail into a three-way classification — the ref cleanup is
/// best-effort bookkeeping that runs only after git itself removed the
/// worktree, and interleaving it with the safety classification made both
/// harder to read.
/// What: `git worktree prune`, then `git branch -D session/<leaf>`. Both are
/// best-effort; a failure leaves a stale ref in git's output, never data loss.
/// Test: `manager_decommission_removes_real_git_worktree` asserts both effects.
fn prune_refs_after_removal(repo_root: &Path, path: &Path) {
    {
        // Step 2: git worktree prune to clear any stale git worktree refs.
        // Best-effort: a failure here is a minor annoyance (stale ref in git output),
        // not a correctness failure.
        // #7171: through the shared entry point — this runs on every session
        // teardown across the fleet, one of the storm-trigger commands named
        // by the incident.
        let prune_out = trusty_common::git::command_in(repo_root)
            .args(["worktree", "prune"])
            .output();
        if let Err(e) = prune_out {
            warn!(root = %repo_root.display(), "decommission: git worktree prune failed: {e}");
        }

        // Step 3: delete the session branch ref (if any). #2032 FIX: the branch
        // `create_session_worktree` actually creates is `session/<worktree-leaf>`
        // (see `crate::core::worktree_naming::worktree_branch_for` — the SAME
        // convention `daemon::managed_routes::inproject::create_session_worktree`
        // uses), NOT the bare leaf name. Before this fix the missing `session/`
        // prefix meant `git branch -D <leaf>` always targeted a nonexistent
        // branch and silently fell into the "not found" debug-log path below —
        // session branches were NEVER actually cleaned up. This works
        // identically for both OLD (raw-UUID-named, pre-#2032) and NEW
        // (semantic-tmux-name) worktree leaves, since both used/use the same
        // `session/<leaf>` convention for the branch name. Ignore "not found"
        // — the branch may not exist for older sessions that never created one
        // (#1840). Uses `core::worktree_naming` (unconditionally compiled),
        // NOT `daemon::managed_routes::inproject` (feature = "daemon"), so
        // this module keeps compiling with the `daemon` feature disabled.
        if let Some(session_name) = path.file_name().and_then(|n| n.to_str()) {
            let branch = crate::core::worktree_naming::worktree_branch_for(session_name);
            let branch_out = std::process::Command::new("git")
                .arg("-C")
                .arg(repo_root)
                .args(["branch", "-D"])
                .arg(&branch)
                .output();
            match branch_out {
                Ok(o) if o.status.success() => {
                    info!(
                        path = %path.display(),
                        "decommission: git branch -D {:?} (session ref cleaned)",
                        branch
                    );
                }
                Ok(o) => {
                    // Branch not found is expected for sessions that never created one.
                    tracing::debug!(
                        path = %path.display(),
                        "decommission: git branch -D {:?} not needed: {}",
                        branch,
                        String::from_utf8_lossy(&o.stderr).trim()
                    );
                }
                Err(e) => {
                    warn!(
                        path = %path.display(),
                        "decommission: git branch -D failed to spawn: {e}"
                    );
                }
            }
        }
    }
}

/// The base checkout whose worktree registry a decommission will strand (#5949).
///
/// Why: an SM-OWNED workspace is removed with `remove_dir_all`, which git never
/// learns about — the entry survives in the base checkout's worktree list until
/// something prunes it. The client layer already repaired this for the routed
/// HTTP paths, so the MCP tool and the idle reaper — both of which call
/// `decommission` in-process, below that layer — accumulated stale entries
/// instead, the reaper unattended. Answering the question here means every
/// caller inherits the repair regardless of transport.
/// What: `None` unless the record names an owned workspace that git reports a
/// DIFFERENT checkout as owning. A workspace that is its own checkout root has
/// no external registry to repair, and an unowned one is removed by
/// [`remove_session_worktree`], which prunes for itself. Must be called BEFORE
/// the removal: git can only answer from a directory that still exists.
/// Test: `decommission_prunes_the_base_repo_worktree_registry`,
/// `registry_root_to_repair_ignores_a_standalone_owned_clone`.
pub(super) fn registry_root_to_repair(record: &SessionRecord) -> Option<std::path::PathBuf> {
    if !record.workspace_owned {
        return None;
    }
    let ws = record.workspace_path.as_deref()?;
    let root = worktree_registry::registry_root_for(ws)?;
    // git reports a resolved path; the record's may still carry a symlinked
    // spelling of the same directory (every macOS temp path does), so compare
    // both in canonical form or the two would never look equal.
    let ws_resolved = std::fs::canonicalize(ws).unwrap_or_else(|_| ws.to_path_buf());
    (root != ws_resolved).then_some(root)
}

/// Clear the stale registry entry a removed worktree leaves behind (#5949).
///
/// Why: git keeps reporting a worktree whose directory was deleted out from
/// under it, and every consumer of that listing — this crate's own reclaim
/// sweep included — then reasons about a directory that is not there.
/// What: runs the prune subcommand in `root`. Best-effort: a failure leaves a
/// stale entry in git's output, never data loss, so it is logged and the
/// teardown that already succeeded continues.
/// Test: `decommission_prunes_the_base_repo_worktree_registry`.
fn prune_worktree_registry(root: &Path) {
    // #7171: through the shared entry point — same storm-trigger command as
    // `prune_refs_after_removal` above.
    match trusty_common::git::command_in(root)
        .args(["worktree", "prune"])
        .output()
    {
        Ok(out) if out.status.success() => {
            info!(root = %root.display(), "decommission: worktree registry pruned");
        }
        Ok(out) => warn!(
            root = %root.display(),
            "decommission: worktree prune exited {}: {}",
            out.status,
            String::from_utf8_lossy(&out.stderr).trim()
        ),
        Err(e) => warn!(
            root = %root.display(),
            "decommission: worktree prune failed to spawn: {e}"
        ),
    }
}

impl SessionManager {
    /// Decommission a session: stop the runtime, remove the workspace from disk
    /// (ONLY if the SM provisioned it), and mark the record `Decommissioned`.
    ///
    /// Why: the only full teardown operation. Unlike `stop`, this removes the
    /// workspace directory from disk so no future `resume` is possible. A
    /// tombstone record is kept in the store so `ls` can show history.
    ///
    /// Safety (#1511): `remove_dir_all` is executed ONLY when BOTH conditions hold:
    /// (a) `record.workspace_owned == true` — the SM provisioned (cloned) the
    ///     directory and is the rightful owner; local-path spawn (#1502) and
    ///     `adopt_existing` (#1433) leave `workspace_owned = false` so they are
    ///     NEVER deleted by this path.
    /// (b) `is_safe_to_remove(workspace_path, managed_root)` — the canonicalized
    ///     path is strictly INSIDE the SM's managed workspace root, rejecting any
    ///     path outside it (including `$HOME`, volume roots, and paths with too
    ///     few components). This belt-and-suspenders guard catches stale/incorrect
    ///     `workspace_owned` flags before disk mutation occurs.
    ///
    /// #1840 worktree extension: even when `workspace_owned = false`, if the
    /// workspace path is under a `.worktrees/` directory (an in-project per-session
    /// worktree), the worktree IS removed via `git worktree remove --force`. The
    /// base clone (`<base>/`) is NEVER touched — only the per-session leaf dir.
    ///
    /// When deletion is skipped (unowned non-worktree or unsafe path), decommission
    /// still transitions the record to `Decommissioned` and returns successfully.
    ///
    /// What: delegates to [`decommission_with_root`](Self::decommission_with_root)
    /// with the config-derived managed root so callers remain env-agnostic.
    ///
    /// `caller` (#3649, Option B): `None` means an operator/daemon-internal
    /// caller (CLI, HTTP route, the age-based reaper, bulk prune) — current
    /// authority is preserved unconditionally, matching every pre-#3649
    /// call site. `Some(id)` means a SESSION is asking to decommission
    /// `id`'s target on its own behalf; if the target's worktree has a KNOWN
    /// owner that disagrees with `caller` AND that owner is not provably
    /// ownerless, the call is refused with
    /// [`ManagedError::WorktreeOwnerMismatch`] instead of tearing down a peer
    /// session's worktree out from under it.
    /// Test: `manager_decommission_removes_workspace` — asserts the workspace dir
    /// is gone from disk and the record state is `Decommissioned`.
    /// `manager_decommission_unowned_skips_deletion` — asserts that decommissioning
    /// a local-path/adopt record does NOT delete the directory.
    /// `decommission_owner_gate_refuses_foreign_caller`,
    /// `decommission_owner_gate_allows_terminal_owner` (#3649).
    ///
    /// #3764: BEFORE any of the above, and regardless of `caller`, this also
    /// refuses when a session OTHER than `id` is `Active` with a workspace or
    /// cwd that resolves to the same directory as `id`'s own `workspace_path`
    /// — see [`check_no_foreign_active_claim`](Self::check_no_foreign_active_claim)
    /// and [`ManagedError::ForeignActiveWorkspaceClaim`]. Test:
    /// `decommission_refuses_a_foreign_active_workspace_collision`.
    pub async fn decommission(
        &self,
        id: &ManagedSessionId,
        caller: Option<ManagedSessionId>,
    ) -> Result<(SessionRecord, bool), ManagedError> {
        let config = TrustyToolsConfig::load();
        let managed_root = workspace_root(&config);
        self.decommission_with_root(id, &managed_root, caller).await
    }

    /// Tombstone a record and do NOTHING else — a dedicated, single-effect path,
    /// not a flag threaded through [`decommission_with_root`](Self::decommission_with_root)
    /// (owner request 2026-07-29; rebuilt as a separate function per PR #4725
    /// review round 2).
    ///
    /// Why: this exists for the `tm ls` auto-prune sweep, which runs unattended
    /// on every listing and knows only that a workspace was ABSENT at listing
    /// time. It must therefore be incapable of destroying anything, including on
    /// a remount race. It was originally built as `decommission_with_root(…,
    /// record_only: true)`, and that shape failed twice in review — each time
    /// because a destructive effect in that function was not behind the flag:
    ///
    ///   * round 1 (#4728): `graceful_terminate_runtime` ran above the guard,
    ///     SIGTERMing and `kill_session`ing any live pane whose NAME matched the
    ///     record's;
    ///   * round 2: `delete_search_index_best_effort` ran below every guard,
    ///     issuing a cross-daemon `DELETE /indexes/{id}?delete_data=true`. Worse,
    ///     its target id comes from `disposable_workspace_index_id`, which walks
    ///     UP from the workspace path for a `.git` marker and is documented as
    ///     requiring the workspace to still exist. Auto-prune fires only once the
    ///     workspace is GONE, so that contract is violated by construction and
    ///     the walk resolves to the PARENT PROJECT's index.
    ///
    /// Two rounds, two ungated effects, each found only by executing the path.
    /// A boolean parameter cannot make a multi-effect function safe — it can
    /// only be audited by exercising every branch, and it silently acquires new
    /// effects whenever someone adds one to the shared function. This function
    /// is auditable by reading: its entire body is record mutation plus one
    /// store write. There is no filesystem call, no subprocess, no network
    /// request, and no tmux interaction to gate, so none can be forgotten.
    ///
    /// What: applies the #3649 owner gate (via
    /// [`check_worktree_owner`](Self::check_worktree_owner)), clears
    /// `workspace_path`/`workspace_owned` only when nothing is left on disk
    /// (#4344 — a retained directory must keep its pointer), clears the #4400
    /// pending-decision fields, sets `Decommissioned`, and persists. Returns
    /// `(record, false)` — the `false` is a type-level constant, not a computed
    /// result, because nothing here can remove anything.
    /// Test: `decommission_record_only_never_removes_existing_workspace`,
    /// `decommission_record_only_never_touches_the_runtime`,
    /// `decommission_record_only_has_no_side_effects_beyond_the_store`;
    /// `disposable_index_id_for_a_removed_worktree_resolves_to_the_parent_project`
    /// pins why the shared teardown's index cleanup must not be reused here.
    pub async fn decommission_record_only(
        &self,
        id: &ManagedSessionId,
    ) -> Result<(SessionRecord, bool), ManagedError> {
        let mut record = self.get(id).await?;
        // Daemon-internal sweep: `caller = None` preserves full pre-#3649
        // authority, exactly as before. Kept so this path can never be a wider
        // authority than the teardown it replaces.
        self.check_worktree_owner(&record, None, id).await?;

        // #4344: only blank the pointer when the directory really is gone —
        // otherwise a retained workspace is stranded with no trail back to it.
        let workspace_still_on_disk = record.workspace_path.as_deref().is_some_and(|p| p.exists());
        if !workspace_still_on_disk {
            record.workspace_path = None;
            record.workspace_owned = false;
        }
        // #4400: a decommissioned session is terminal — a pending decision on it
        // would sit in the human-confirmation queue forever.
        record.pending_decision = None;
        record.proposed_default = None;
        // Stamps `terminal_at` as well as `state` — the retention sweep's clock
        // starts here, not at `created_at`.
        record.set_lifecycle_state(ManagedSessionState::Decommissioned, Utc::now());
        self.store.write().await.upsert(record.clone()).await?;
        // #7087: a record-only tombstone leaves the active-project set.
        self.bump_residency_generation();
        // #7087: and a tombstone is terminal — drop its derivation entry so the
        // cache tracks live records, not every session the daemon has served.
        self.residency_cache_evict(id).await;
        info!(id = %id, name = %record.tmux_name, "managed session record tombstoned (record-only)");
        Ok((record, false))
    }

    /// The #3649 worktree-owner gate, shared by the full teardown and the
    /// record-only tombstone.
    ///
    /// Why: extracted so [`decommission_record_only`](Self::decommission_record_only)
    /// can apply the identical authority check without depending on
    /// `decommission_with_root` — the whole point of that function is that it
    /// shares no code path with the destructive one.
    /// What: `None` (operator/daemon-internal) always passes. `Some(id)` is
    /// refused with [`ManagedError::WorktreeOwnerMismatch`] when the target's
    /// worktree has a known owner that disagrees and is not provably ownerless.
    /// Test: `decommission_owner_gate_refuses_foreign_caller`,
    /// `decommission_owner_gate_allows_terminal_owner`.
    async fn check_worktree_owner(
        &self,
        record: &SessionRecord,
        caller: Option<ManagedSessionId>,
        id: &ManagedSessionId,
    ) -> Result<(), ManagedError> {
        if let Some(caller_id) = caller
            && let Some(owner) = self.known_owner_of(record)
            && owner != caller_id
            && !self.resolve_ownerless(owner).await
        {
            return Err(ManagedError::WorktreeOwnerMismatch(caller_id, owner, *id));
        }
        Ok(())
    }

    /// The #3764 session-identity guard: refuse when a session OTHER than
    /// `record` is `Active` with a workspace/cwd that resolves to the same
    /// directory as `record.workspace_path`.
    ///
    /// Why: `check_worktree_owner` above only fires when a caller identifies
    /// itself — every production call site except the bulk age-based reaper
    /// passes `caller: None`, which makes that gate a no-op in practice for
    /// most decommissions. This gate closes the gap: it consults the store
    /// directly and runs on EVERY decommission, because the collision it
    /// detects (#1744: two `Active` records naming one directory) is not
    /// something a caller's own identity has any bearing on — see
    /// `workspace_guard::foreign_active_claim`, which does the actual
    /// comparison.
    /// What: a no-op when `record.workspace_path` is `None` (nothing to
    /// collide on). Otherwise lists every managed session and asks
    /// `foreign_active_claim`; a hit refuses with
    /// [`ManagedError::ForeignActiveWorkspaceClaim`] BEFORE any disk mutation
    /// — this runs ahead of `graceful_terminate_runtime` and every removal
    /// branch below.
    /// Test: `decommission_refuses_a_foreign_active_workspace_collision`,
    /// `decommission_refuses_a_foreign_active_cwd_collision`,
    /// `decommission_allows_a_terminal_sibling_sharing_the_path`,
    /// `decommission_allows_an_uncontested_workspace`.
    async fn check_no_foreign_active_claim(
        &self,
        record: &SessionRecord,
    ) -> Result<(), ManagedError> {
        let Some(ws) = record.workspace_path.as_deref() else {
            return Ok(());
        };
        let records = self.list().await;
        if let Some(other) = foreign_active_claim(ws, &record.id, &records) {
            warn!(
                id = %record.id,
                other = %other,
                workspace = %ws.display(),
                "decommission: refusing — an Active sibling session also claims this \
                 workspace/cwd (#3764, #1744)"
            );
            return Err(ManagedError::ForeignActiveWorkspaceClaim(record.id, other));
        }
        Ok(())
    }

    /// Internal: decommission with an explicit managed root (test seam).
    ///
    /// Why: tests need to inject a temp directory as the managed root to keep the
    /// containment guard working without mutating process-global env vars
    /// (`TRUSTY_MPM_WORKSPACE_ROOT`). Env mutation is thread-unsafe and pollutes
    /// parallel tests; injecting the root avoids that entirely.
    /// What: identical teardown logic as the public `decommission` but resolves the
    /// managed root from the caller-supplied `managed_root` instead of the config.
    ///
    /// 🔴 THIS FUNCTION IS UNCONDITIONALLY DESTRUCTIVE and has no "safe mode".
    /// The `record_only` flag it used to carry is GONE (PR #4725 review round 2);
    /// [`decommission_record_only`](Self::decommission_record_only) is now a
    /// separate function that shares no code path with this one. Read that
    /// function's doc for why. Its side effects, in order, are:
    ///
    /// | # | Effect | Reversible? |
    /// |---|---|---|
    /// | 1 | `graceful_terminate_runtime` — SIGTERM + `kill_session` the pane | no |
    /// | 2 | `remove_session_worktree` — `git worktree remove --force`, `fs::remove_dir_all` fallback, `git worktree prune`, `git branch -D` (dirty-gated, #4344) | no |
    /// | 3 | `fs::remove_dir_all` on an SM-owned workspace (containment-gated) | no |
    /// | 4 | `delete_search_index_best_effort` — cross-daemon `DELETE /indexes/{id}` (never from a test process, #4743) | no |
    /// | 5 | clears `workspace_path`/`workspace_owned` when nothing is on disk | store-only |
    /// | 6 | clears `pending_decision`/`proposed_default` (#4400) | store-only |
    /// | 7 | writes state `Decommissioned` | store-only |
    ///
    /// Effects 1–4 destroy state outside the record store. Any new effect added
    /// here belongs in that table. If a caller needs a subset, give it its own
    /// function rather than a flag — that is the lesson of this PR's two review
    /// rounds, each of which found a different ungated effect.
    ///
    /// Returns `(SessionRecord, workspace_removed)` where `workspace_removed` is
    /// `true` ONLY when `remove_dir_all` actually ran — callers must not infer this
    /// from a post-call filesystem check (TOCTOU: owned workspace already absent
    /// before decommission would give a false-positive filesystem result).
    /// Test: called by `manager_decommission_removes_workspace` (which passes a
    /// TempDir as the managed root, removing the need for `set_var`);
    /// `decommission_full_still_terminates_the_runtime`.
    pub(crate) async fn decommission_with_root(
        &self,
        id: &ManagedSessionId,
        managed_root: &Path,
        caller: Option<ManagedSessionId>,
    ) -> Result<(SessionRecord, bool), ManagedError> {
        self.decommission_with_root_checked(id, managed_root, caller, true)
            .await
    }

    /// Decommission a known DUPLICATE-workspace loser, EXEMPT from the #3764
    /// foreign-active-claim guard (#3764 review fix).
    ///
    /// Why: `dedup_stale_duplicates`'s whole job is to collapse two records
    /// that name the SAME `workspace_path` — one kept, one removed — which is
    /// indistinguishable, by path alone, from the #1744 cross-session
    /// collision `check_no_foreign_active_claim` refuses. Unlike that guard
    /// (which reads only the STORE's `state` field), dedup has already asked
    /// the OS directly: for BOTH the shapes it collapses, it re-verifies
    /// immediately before this call — via a fresh tmux-liveness probe, not a
    /// snapshot taken earlier in the pass (#3764 review fix, both call
    /// sites in `dedup.rs`) — that the LOSER's own tmux session is not live;
    /// the exact-workspace-duplicate (`#3396`) shape additionally rechecks
    /// that the loser is not SM-owned. Either recheck is a stronger and more
    /// accurate liveness signal than a record's persisted `state`. Routing
    /// dedup's call through the ordinary [`decommission`](Self::decommission)
    /// broke `dedup_collapses_the_dead_sibling_when_tmux_answers` and
    /// `reconcile_dedup_collapses_exact_workspace_duplicate_of_live_record` —
    /// the guard is correct to refuse an UNVERIFIED same-path collision, and
    /// wrong to refuse one dedup has already resolved.
    /// What: same teardown as `decommission`, with the foreign-active-claim
    /// check (only) skipped. The #3649 owner gate still runs (`caller: None`
    /// preserves its existing no-op behavior here, matching every other
    /// dedup/reap call site).
    /// Test: `dedup_collapses_the_dead_sibling_when_tmux_answers`,
    /// `reconcile_dedup_collapses_exact_workspace_duplicate_of_live_record`,
    /// `dedup_skips_a_plan_dedup_loser_whose_tmux_went_live_before_act`.
    pub(crate) async fn decommission_dedup_loser(
        &self,
        id: &ManagedSessionId,
    ) -> Result<(SessionRecord, bool), ManagedError> {
        let config = TrustyToolsConfig::load();
        let managed_root = workspace_root(&config);
        self.decommission_with_root_checked(id, &managed_root, None, false)
            .await
    }

    /// The shared body behind [`decommission_with_root`](Self::decommission_with_root)
    /// and [`decommission_dedup_loser`](Self::decommission_dedup_loser).
    ///
    /// `check_foreign_claim` gates ONLY `check_no_foreign_active_claim`
    /// (#3764) — every other guard (the #3649 owner gate, containment, dirty
    /// checks) still runs regardless.
    async fn decommission_with_root_checked(
        &self,
        id: &ManagedSessionId,
        managed_root: &Path,
        caller: Option<ManagedSessionId>,
        check_foreign_claim: bool,
    ) -> Result<(SessionRecord, bool), ManagedError> {
        let mut record = self.get(id).await?;

        // #3649 owner gate: only applies when a SESSION identifies itself as
        // the caller. `None` (operator/daemon-internal) preserves full
        // pre-#3649 authority unconditionally — see the doc above.
        self.check_worktree_owner(&record, caller, id).await?;

        // #3764: unlike the owner gate above, this runs UNCONDITIONALLY for
        // every caller EXCEPT `decommission_dedup_loser` — it does not depend
        // on any caller identifying itself, because the hazard is a STORE
        // inconsistency (two `Active` records naming the same directory,
        // #1744), not a caller impersonating an owner. See
        // `check_no_foreign_active_claim`.
        if check_foreign_claim {
            self.check_no_foreign_active_claim(&record).await?;
        }

        // #2033: derive the trusty-search index id for a disposable workspace
        // (SM-owned clone or in-project worktree — see
        // `search_gc::disposable_workspace_index_id`) BEFORE any removal
        // happens below. This must run first because
        // `trusty_common::resolve_project_root` walks UP from the workspace
        // path looking for a `.git` marker — if the directory is already gone
        // by the time we derive the id, the walk would find the wrong (e.g.
        // shared base clone's) ancestor `.git` and target the WRONG index for
        // deletion. Local-path/adopted sessions (`workspace_owned == false`
        // and not an in-project worktree) return `None` here — their real,
        // long-lived directory keeps its search index, exactly mirroring the
        // filesystem-removal guard below.
        let search_index_id = search_gc::disposable_workspace_index_id(
            record.workspace_path.as_deref(),
            record.workspace_owned,
        );

        // #5949: resolve the checkout that owns this workspace's worktree
        // registry BEFORE anything is removed. `registry_root_for` asks git
        // from the workspace itself, so it can only be answered while the
        // directory still exists — the same ordering contract `search_index_id`
        // above is bound by, for the same reason.
        let registry_root = registry_root_to_repair(&record);

        // Gracefully terminate the runtime before removing the workspace (#1975):
        // SIGTERM the claude process and give it a grace window to flush state,
        // then reclaim the pane — instead of an abrupt `kill_session`. Best-effort:
        // a session whose runtime is already gone still decommissions cleanly —
        // the helper self-guards and is a no-op when the pane is already gone.
        //
        // Effect 1 of the table above. `graceful_terminate_runtime` self-guards
        // on `session_exists(name)` — LIVE-TMUX NAME MEMBERSHIP, not the record's
        // captured `pane_id` (#4728) — so it kills whatever live session carries
        // this name. Acceptable here, where teardown is the caller's stated
        // intent; never acceptable on a listing sweep, which is why the
        // record-only path is a separate function rather than a flag.
        self.graceful_terminate_runtime(&record.tmux_name).await;

        // Effects 2–3. Guard: only remove the workspace directory if the SM
        // provisioned it. Track whether remove_dir_all ACTUALLY RAN (not
        // inferred from filesystem).
        let mut workspace_removed = false;
        if let Some(ref ws) = record.workspace_path {
            if !record.workspace_owned {
                // Unowned workspace (local-path spawn or adopt): never bulk-delete.
                // #1840: EXCEPTION — in-project per-session worktrees live under
                // .worktrees/ and must be cleaned up via `git worktree remove` so
                // the git ref is also pruned.  The base clone directory is NEVER
                // touched — only the leaf worktree path.
                if is_session_worktree(ws) {
                    // Data-safety gate (#4091-style, decommission-side): before
                    // this, `remove_session_worktree` ran `git worktree remove
                    // --force` (falling back to `fs::remove_dir_all`)
                    // unconditionally, with NO dirty check at all — a live
                    // data-loss path for `tm sessions prune --state stopped`
                    // (which calls `decommission` per matching record). Reuse
                    // the same `worktree_safety::inspect_dirt` check the
                    // orphan-worktree sweep (`prune_orphaned_worktrees`)
                    // already uses: refuse (and report) a candidate holding
                    // uncommitted/untracked work or unpushed commits. Runs on
                    // spawn_blocking since it shells out to git, same as the
                    // removal itself; a panicked check is treated as dirty
                    // (fail-safe) rather than a green light to delete.
                    let ws_for_check = ws.clone();
                    let dirt = tokio::task::spawn_blocking(move || inspect_dirt(&ws_for_check))
                        .await
                        .unwrap_or_else(|e| {
                            Some(DirtyWorktree::new(
                                ws,
                                format!("dirty-check task panicked: {e}"),
                                0,
                                0,
                            ))
                        });

                    if let Some(dirt) = dirt {
                        warn!(
                            id = %id,
                            workspace = %ws.display(),
                            reason = %dirt.reason,
                            "decommission: refusing to remove worktree — it holds \
                             unsaved work; leaving it on disk (the record below is \
                             still tombstoned)"
                        );
                        // workspace_removed stays false — nothing was deleted.
                    } else {
                        // Item 4 (#1845): wrap the blocking `git worktree remove`
                        // call in spawn_blocking + tokio::time::timeout so a hung
                        // git process cannot stall the async executor indefinitely.
                        let ws_clone = ws.clone();
                        // #7885: name the route in the audit line.
                        let join = tokio::task::spawn_blocking(move || {
                            remove_session_worktree(
                                &ws_clone,
                                "session decommission: the session ended and its tree is clean",
                            )
                        });
                        let outcome =
                            match tokio::time::timeout(GIT_WORKTREE_REMOVE_TIMEOUT, join).await {
                                Ok(Ok(outcome)) => outcome,
                                Ok(Err(e)) => {
                                    WorktreeRemoval::Kept(format!("the removal task panicked: {e}"))
                                }
                                Err(_elapsed) => WorktreeRemoval::Kept(format!(
                                    "git worktree remove did not finish within {}s; the \
                                     worktree may require manual cleanup",
                                    GIT_WORKTREE_REMOVE_TIMEOUT.as_secs()
                                )),
                            };
                        // #4732: the reason now comes back FROM the remover
                        // rather than being buried in a log line inside it.
                        workspace_removed = outcome.removed();
                        if let Some(reason) = outcome.reason() {
                            warn!(
                                id = %id,
                                workspace = %ws.display(),
                                "decommission: worktree left on disk — {reason}"
                            );
                        }
                    }
                } else {
                    warn!(
                        id = %id,
                        workspace = %ws.display(),
                        "decommission: skipping workspace removal — not SM-owned \
                         (local-path or adopted session); the directory was NOT \
                         created by the session manager"
                    );
                }
            } else {
                // Owned workspace: check existence first so a path that is
                // already gone is not misreported as a containment failure.
                if !ws.exists() {
                    // Benign: the workspace was removed before decommission ran
                    // (e.g. a prior partial teardown). The tombstone is still
                    // written below; no further disk action is needed.
                    // workspace_removed stays false — we did NOT remove it.
                    tracing::debug!(
                        id = %id,
                        workspace = %ws.display(),
                        "decommission: owned workspace already absent — skipping removal"
                    );
                } else {
                    // Workspace exists: apply the belt-and-suspenders
                    // path-containment guard before touching the filesystem.
                    // Only paths that exist but are OUTSIDE the managed root
                    // (or are otherwise unsafe) reach this warning.
                    if !is_safe_to_remove(ws, managed_root) {
                        warn!(
                            id = %id,
                            workspace = %ws.display(),
                            root = %managed_root.display(),
                            "decommission: skipping workspace removal — path fails \
                             containment guard (outside managed root or unsafe path)"
                        );
                    } else {
                        // #7885 critic round: audited like every other removal
                        // route; an I/O failure still propagates below.
                        let mut failure: Option<std::io::Error> = None;
                        super::worktree_removal_audit::audited_removal(
                            ws,
                            "session decommission: owned workspace, containment guard passed",
                            || match std::fs::remove_dir_all(ws) {
                                Ok(()) => WorktreeRemoval::Removed,
                                Err(e) => {
                                    let kept = WorktreeRemoval::Kept(format!(
                                        "removing the workspace failed: {e}"
                                    ));
                                    failure = Some(e);
                                    kept
                                }
                            },
                        );
                        if let Some(e) = failure {
                            return Err(ManagedError::Io(std::io::Error::new(
                                e.kind(),
                                format!("remove workspace {:?}: {e}", ws),
                            )));
                        }
                        workspace_removed = true;
                        info!(
                            id = %id,
                            workspace = %ws.display(),
                            "decommission: owned workspace removed from disk"
                        );
                        // #5949: `remove_dir_all` is invisible to git, so the
                        // base checkout still lists this worktree. Repair it
                        // here — below every caller, in-process ones included.
                        if let Some(ref root) = registry_root {
                            prune_worktree_registry(root);
                        }
                    }
                }
            }
        }

        // Effect 4. #2033: best-effort remove the trusty-search index for a
        // disposed workspace, alongside the worktree/clone directory. Fail-soft:
        // an unreachable/erroring search daemon must never block or fail session
        // teardown — `delete_search_index_best_effort` logs and swallows every
        // failure mode itself.
        //
        // Note the ordering contract that makes this correct ONLY here:
        // `search_index_id` was derived at the top of this function, BEFORE any
        // removal, because `resolve_project_root` walks up for a `.git` marker
        // and would otherwise resolve to the parent project's index. A caller
        // that runs against an already-absent workspace cannot satisfy that
        // contract, which is precisely why the record-only path does not reuse
        // this function (PR #4725 review round 2).
        if let Some(index_id) = search_index_id {
            search_gc::delete_search_index_best_effort(&index_id).await;
        }

        // Tombstone: mark Decommissioned, persist. `workspace_path`/
        // `workspace_owned` are cleared ONLY when nothing is left on disk —
        // every "skip removal" branch above (the dirty-worktree refusal,
        // unowned/local-path, the containment guard) deliberately leaves real
        // content in place, and blanking the pointer here would strand that
        // retained directory with nothing but the transient warn! log line
        // above as a trail back to it (#4344 review). A record whose
        // workspace really was removed (or was already absent) still nulls
        // both fields exactly as before.
        let workspace_still_on_disk = record.workspace_path.as_deref().is_some_and(|p| p.exists());
        if !workspace_still_on_disk {
            record.workspace_path = None;
            record.workspace_owned = false;
        }
        // #4400: a decommissioned session is terminal — no human will ever act
        // on a `pending_decision` raised before teardown, so leaving it set
        // means `supervisor_status`'s human-confirmation queue accumulates
        // phantom gates forever (19-day-old dead entries indistinguishable
        // from a real, live T4 gate). Clear both fields unconditionally on
        // every decommission path (fresh teardown, record-only tombstone,
        // dirty-worktree-refused teardown — all reach this line).
        record.pending_decision = None;
        record.proposed_default = None;
        // Stamps `terminal_at` as well as `state` — the retention sweep's clock
        // starts here, not at `created_at`.
        record.set_lifecycle_state(ManagedSessionState::Decommissioned, Utc::now());
        self.store.write().await.upsert(record.clone()).await?;
        // #7087: a full teardown leaves the active-project set.
        self.bump_residency_generation();
        // #7087: and a tombstone is terminal — drop its derivation entry so the
        // cache tracks live records, not every session the daemon has served.
        self.residency_cache_evict(id).await;
        info!(id = %id, name = %record.tmux_name, "managed session decommissioned");
        Ok((record, workspace_removed))
    }

    /// Mark a session's workspace as SM-owned (provisioned by clone) or unowned.
    ///
    /// Why (#1511): the decommission path must know whether the SM provisioned the
    /// `workspace_path` (and therefore may `remove_dir_all` it) or whether the path
    /// is a real, pre-existing user directory (local-path spawn, adopt) that must
    /// NEVER be deleted. Setting `workspace_owned = true` is the explicit assertion
    /// that this SM cloned the workspace; `false` (the serde default) means "do not
    /// touch this directory on decommission." Callers that use the local-path spawn
    /// or `adopt_existing` path MUST NOT call this method (or call it with `false`).
    /// What: looks up the record, sets `workspace_owned`, and persists.
    /// Test: `workspace_owned_flag_round_trips_via_set` + the decommission guard
    /// tests in `session_manager/tests.rs`.
    pub async fn set_workspace_owned(
        &self,
        id: &ManagedSessionId,
        owned: bool,
    ) -> Result<(), ManagedError> {
        let mut record = self.get(id).await?;
        record.workspace_owned = owned;
        self.store.write().await.upsert(record).await?;
        Ok(())
    }
}