trusty-mpm 1.8.2

//! Session manager: CRUD, spawning, and reconciliation.
//!
//! Why: the daemon needs a single authoritative component that creates,
//! tracks, and reconciles managed tmux sessions. Centralising all of that
//! logic here keeps the HTTP handlers thin and makes the manager unit-testable
//! through the [`ManagedTmuxDriver`] trait seam.
//! What: [`SessionManager`] wraps a [`SessionStore`] and a [`ManagedTmuxDriver`]
//! and provides `create`, `list`, `get`, `send_input`, `stop`,
//! `mark_runtime_exited_stopped` (non-destructive counterpart to `stop`,
//! #2023 A), `resume`, and `decommission`.
//! `mark_reactivated` (the in-place counterpart to `resume`, #2023 C) lives in
//! the sibling `reactivate.rs`, `reconcile_on_boot` lives in the sibling
//! `reconcile.rs` (#2379), and the post-creation field setters
//! (`mark_errored`, `set_workspace`, `set_pending_decision`, `set_source_id`,
//! `set_deliverable_id`, `clear_pending_decision`) live in the sibling
//! `setters.rs` (#7087) — all extracted to keep this file under the 500-SLOC
//! cap.
//! [`ReconcileReport`] describes what the reconciliation pass found.
//! [`ManagedError`] is the module's error type.
//! Test: `manager_create_record`, `manager_stop_keeps_workspace`,
//! `manager_resume_respawns`, `manager_decommission_removes_workspace`,
//! `manager_reconcile_gone_tmux_yields_stopped` in tests.rs.

use std::collections::{HashMap, HashSet};
use std::path::{Path, PathBuf};
use std::sync::Arc;
use std::sync::atomic::AtomicU64;

use chrono::Utc;
use thiserror::Error;
use tokio::sync::RwLock;
use tracing::{info, warn};

use crate::core::names::SessionNameError;
use crate::core::sm::control::Submit;

use super::record::{ManagedSessionId, ManagedSessionState, SessionRecord};
use super::resume_workdir;
use super::slots::SlotRegistry;
use super::store::{SessionStore, StoreError};

/// One session's cached `(root, palace_id, index_ids)` residency derivation.
/// See `residency_state.rs`.
type ResidencyCacheEntry = (PathBuf, Option<String>, Vec<String>);

/// Errors produced by the session manager.
///
/// Why: HTTP handlers dispatch on error variants to choose status codes;
/// a typed enum keeps that mapping clean and avoids stringly-typed matching.
/// What: one variant per failure mode: tmux problems, missing sessions,
/// store I/O, miscellaneous I/O errors, and invalid state transitions.
/// Test: `ManagedError` variants are exercised by the manager unit tests.
// #8372: non_exhaustive, so a new failure mode is not an API break.
#[derive(Debug, Error)]
#[non_exhaustive]
pub enum ManagedError {
    /// tmux was unavailable or a tmux operation failed.
    #[error("tmux error: {0}")]
    TmuxUnavailable(String),

    /// The requested session id was not present in the store.
    #[error("session not found: {0}")]
    SessionNotFound(String),

    /// The session store operation failed.
    #[error("store error: {0}")]
    Store(#[from] StoreError),

    /// A miscellaneous I/O error.
    #[error("I/O error: {0}")]
    Io(#[from] std::io::Error),

    /// A name derived from the cwd hint collided with an existing session.
    #[error("name already in use: {0} — use `tm session ls` to find it")]
    NameCollision(String),

    /// The operation is not valid for the current session state.
    #[error("invalid state transition for session {0}: {1}")]
    InvalidState(String, String),

    /// The #8942 kill floor refused to signal or kill a tmux session by name.
    ///
    /// Why: typed, so a teardown aborts before it touches the workspace or
    /// the record instead of reading the refusal as "already gone".
    /// What: carries the floor's refusal text, which names the requester.
    /// Test: `an_undeterminable_floor_aborts_stop_and_decommission_of_an_ordinary_session`.
    #[error("kill refused: {0}")]
    KillRefused(String),

    /// Another path is already resuming this session (#8233 item 4).
    ///
    /// Why: distinct from [`InvalidState`](Self::InvalidState), which says the
    /// RECORD forbids a resume. This says the record permits one and someone
    /// else got there first, so the caller's correct response is to leave the
    /// session alone — the supervisor must not append an error to a record the
    /// operator's resume is still working on, and a second launch line must
    /// never reach the pane. Typed so the supervisor can tell this apart from
    /// a real failure instead of matching on message text.
    /// What: carries the session id. Raised only by
    /// [`SessionManager::begin_resume`](super::SessionManager::begin_resume).
    /// Test: `a_second_concurrent_resume_of_one_session_is_refused`.
    #[error("a resume is already in flight for session {0}; not starting a second one")]
    ResumeInFlight(String),

    /// An auto-resume failed AND has already recorded the failure (#8233).
    ///
    /// Why: `resume_auto` marks the record errored itself, which APPENDS
    /// `[error: …]` to `record.task`. The supervisor's generic failure arm then
    /// marked it errored a second time, so one failed auto-resume left TWO
    /// notes on the task the next relaunch hands to the runtime. Typed rather
    /// than string-matched: the poller must be able to tell "already recorded"
    /// from "failed before anything was written" without reading the message.
    /// What: carries the failure text, which is also what the record now holds.
    /// Test: `one_failed_auto_resume_appends_exactly_one_error_note`.
    #[error("{0}")]
    AutoResumeRecorded(String),

    /// Adoption was requested for a tmux session that does not exist on the host.
    ///
    /// Why: adoption CONNECTS to a pre-existing, unmanaged pane — there is nothing
    /// to drive if the pane is absent. This is the inverse of `NameCollision`:
    /// `create` fails when a name exists, `adopt_existing` fails when it does NOT.
    #[error("tmux session does not exist: {0} — adoption requires a live pane")]
    TmuxSessionMissing(String),

    /// Adoption was requested for a tmux session this store already tracks.
    ///
    /// Why: re-adopting a session the manager already owns would create a second,
    /// conflicting record for the same pane. The operator should drive the existing
    /// record instead.
    #[error("tmux session already adopted/registered: {0}")]
    AlreadyAdopted(String),

    /// A session-name derivation failure (currently: all 99 `tm-<leaf>-NN`
    /// serials for a project are in use).
    ///
    /// Why (#1955, renamed in the #1966 review follow-up): the serial-numbered
    /// naming scheme caps at two digits per project; this surfaces
    /// [`SessionNameError`] through the same typed-error seam as every other
    /// create-path failure instead of stringly-typed-wrapping it into
    /// [`TmuxUnavailable`](Self::TmuxUnavailable). Named `SessionName` (not
    /// `NameSerialExhausted`, its original name) and given a generic message
    /// ("session name error", not "serial exhausted") because of the `#[from]`
    /// below: [`SessionNameError`] currently has exactly one variant
    /// ([`SessionNameError::SerialExhausted`]), but `#[from]` auto-converts
    /// ANY future variant into this one — a name/message naming one specific
    /// variant would silently mislabel a later, unrelated `SessionNameError`
    /// variant.
    #[error("session name error: {0}")]
    SessionName(#[from] SessionNameError),

    /// A session's resume workdir is gone during `resume` (#2250, #8551).
    ///
    /// Why: prior to #2250, `resume()`'s recreate branch handed
    /// `workspace_path` straight to tmux with no existence check — a
    /// removed/stale worktree silently rooted the recreated pane at `$HOME`,
    /// discarding the project-tier `.claude/` skills/persona/MCP config that
    /// lives only under the real workspace.
    /// [`super::resume_workdir::resolve_existing_workdir`] now existence-checks
    /// every candidate. A recorded `workspace_path` decides alone: when it is
    /// gone this error fires even if `cwd` exists (#8551). `cwd` is a candidate
    /// only when no workspace is recorded. Failing loudly beats silently
    /// spawning a pane at `$HOME` or in the main checkout.
    /// What: `(session_id, path)` — `path` is the most-informative candidate
    /// considered (`workspace_path` if set, else `cwd`), surfaced in the error
    /// message so the operator knows exactly which directory vanished.
    #[error(
        "workspace directory {1} no longer exists; cannot resume session {0} — the worktree may have been removed"
    )]
    WorkspaceMissing(String, String),

    /// The session's recorded `pane_id` no longer exists, though its tmux
    /// SESSION is still alive (sibling-window hijack, follow-up to #2456).
    ///
    /// Why: `resume`'s reuse branch previously trusted `session_exists` alone
    /// to mean "the recorded pane survived" — but a tmux session stays alive
    /// as long as ANY pane/window in it is open, including a sibling that has
    /// nothing to do with this record. Recreating in that situation via the
    /// existing `kill_session` + `create_and_verify_pane` path would destroy
    /// the sibling too; silently respawning via a session-scoped target would
    /// land the runtime in the sibling instead of failing. Neither is safe,
    /// so this variant surfaces the ambiguity to the operator explicitly
    /// rather than guessing.
    /// What: `(session_id, pane_id)` — the missing pane's id, surfaced in the
    /// error message so the operator knows exactly what vanished.
    // #8935: the advice names delete as record-only; it never tears down the
    // sibling's session.
    #[error(
        "recorded pane {1} for session {0} no longer exists, but its tmux session is still \
         alive (likely a sibling window) — refusing to respawn into an unrelated active pane; \
         close the sibling window and delete/recreate this session (delete removes the record \
         only and never kills that tmux session), or manually verify pane state with \
         `tmux list-panes`"
    )]
    PaneGone(String, String),

    /// A `caller`-identified decommission refused because the target's
    /// worktree has a KNOWN, non-ownerless owner that disagrees with the
    /// caller (#3649). `(caller, owner, target)`.
    #[error("session {0} refused to decommission {2}'s worktree — owned by session {1}")]
    WorktreeOwnerMismatch(ManagedSessionId, ManagedSessionId, ManagedSessionId),

    /// Decommission refused because an `Active` session OTHER than the target
    /// also claims the target's workspace directory (#3764). `(target,
    /// foreign_active_session)`.
    ///
    /// Why: the #3715 incident's precursor was exactly this — two `Active`
    /// records canonicalizing to one worktree path (#1744) — with no caller
    /// needing to identify itself at all, which is why this check runs
    /// unconditionally rather than only when `caller` is `Some(..)` like
    /// [`WorktreeOwnerMismatch`](Self::WorktreeOwnerMismatch). Removing the
    /// directory while a sibling record still reads it as `Active` would
    /// silently destroy that sibling's live work.
    #[error(
        "refusing to remove session {0}'s workspace — session {1} is also Active with a \
         workspace or cwd that resolves to the same directory (#3764); stop or decommission \
         session {1} first"
    )]
    ForeignActiveWorkspaceClaim(ManagedSessionId, ManagedSessionId),
}

// [`ManagedTmuxDriver`] lives in `driver.rs` (issue #1955 SLOC split — the
// trait's default-impl doc comments alone were ~130 lines, which pushed this
// file over the 500-SLOC production cap once the serial-numbered naming
// rework added a new error variant and helper method). Re-exported here so
// existing `super::manager::ManagedTmuxDriver` import paths keep resolving.
pub use super::driver::ManagedTmuxDriver;

/// Summary of what a reconciliation pass found and changed.
///
/// Why: operators and the daemon start-up log need to know how many sessions
/// were re-adopted and how many were marked stopped after a restart.
/// What: counts of re-adopted (live) and stopped (gone) sessions, plus the
/// tmux names of sessions that were unknown to the store before reconciliation.
/// Test: `manager_reconcile_gone_tmux_yields_stopped` in tests.rs.
#[derive(Debug, Default)]
pub struct ReconcileReport {
    /// tmux session names that were live and re-adopted into the store.
    pub adopted: Vec<String>,
    /// Session ids whose tmux session was gone; marked Stopped (resumable).
    pub stopped: Vec<String>,
    /// Managed tmux sessions (`tm-`/`tmpm-`/`trusty-mpm-`) that the store did not know about.
    pub external_adopted: Vec<String>,
    /// Session ids of terminal (`Decommissioned`/`Deleted`) records whose
    /// stale `pending_decision`/`proposed_default` were cleared this boot
    /// (#4400 backfill for rows tombstoned before the decommission-path fix).
    pub stale_decisions_cleared: Vec<String>,
    /// Live managed tmux names reconciliation DECLINED to adopt because the
    /// pane's working directory could not be resolved (#6118).
    ///
    /// Why: these panes are the ones the daemon deliberately leaves untracked,
    /// so the boot log has to name them — otherwise a pane simply stops
    /// appearing in `tm ls` with nothing said about where it went.
    pub adoption_declined: Vec<String>,
    /// Live tmux names reconciliation REFUSED to adopt because they are in the
    /// test-owned namespace
    /// [`trusty_common::session_naming::RESERVED_TEST_PREFIX`] (#6116).
    ///
    /// Why: separate from [`adoption_declined`](Self::adoption_declined)
    /// because the two say different things — a declined pane is one whose cwd
    /// would not resolve, a refused one is a leaked test session the daemon
    /// will not track no matter what its pane says.
    pub reserved_test_refused: Vec<String>,
    /// Session ids of adopted reserved-test-namespace records tombstoned this
    /// boot, whether their pane was live or gone (#6116).
    ///
    /// Why: such a record was adopted by a daemon build predating the refusal
    /// above. Left alone it either sat in the picker as a `Stopped` row nothing
    /// could resume, or — with a live pane — fed the re-adopt/reap/resume loop
    /// this module's doc describes.
    pub reserved_test_swept: Vec<String>,
}

/// Manages the lifecycle of daemon-owned tmux sessions.
///
/// Why: a single, persistent component that creates named tmux sessions,
/// tracks them in a crash-recoverable store, and reconciles live tmux state
/// with stored records on restart is the heart of the session-manager MVP.
/// What: wraps a [`SessionStore`] behind an async `RwLock` and a
/// [`ManagedTmuxDriver`] behind an `Arc`; all public methods are `async`
/// so the HTTP handlers can await them directly.
/// Test: `manager_create_record`, `manager_stop_keeps_workspace`,
/// `manager_resume_respawns`, `manager_decommission_removes_workspace`.
pub struct SessionManager {
    /// Persisted session store; `pub(crate)` for test helpers that need to
    /// seed internal state without going through the public API.
    pub(crate) store: Arc<RwLock<SessionStore>>,
    /// tmux driver; `pub(crate)` for the decommission / adopt sibling modules.
    pub(crate) tmux: Arc<dyn ManagedTmuxDriver>,
    data_dir: PathBuf,
    /// Stable `tm ls` slot numbering (#3034) — in-memory ONLY, deliberately
    /// never persisted. See [`SlotRegistry`]'s doc for why a fresh instance
    /// per daemon process is exactly the "reset numbering on restart"
    /// requirement. `pub(crate)` for the sibling `numbering.rs`.
    pub(crate) slots: RwLock<SlotRegistry>,
    /// #6568: per-session auto-resume flap counters, persisted beside the store
    /// because the reaper and the poller run in separate processes.
    /// `pub(crate)` for the sibling `resume_breaker.rs`, which owns every read.
    pub(crate) resume_breaker: RwLock<super::resume_breaker::ResumeBreakerStore>,
    /// #6568: the breaker's window/threshold, read once at construction.
    pub(crate) resume_breaker_cfg: super::resume_breaker::ResumeBreakerConfig,
    /// #7087: monotonic residency-set version. In-memory only, like `slots` —
    /// a fresh daemon process has nothing to compare a first pull against, so
    /// there is nothing to persist. Bumped by `create`/`stop`/`adopt`/
    /// `decommission`; read (never bumped) by `mpm.residency.active`. See
    /// `residency_state.rs`.
    pub(crate) residency_generation: AtomicU64,
    /// #7087: per-session `(root, palace_id, index_ids)` derivation cache the
    /// residency route reuses across polls, avoiding a repeat
    /// `resolve_palace_slug`/`derive_project_index_id` (both filesystem/`git`
    /// touching) for a session whose root has not changed. In-memory only,
    /// like `slots` and `residency_generation` above. See `residency_state.rs`.
    pub(crate) residency_cache: RwLock<HashMap<ManagedSessionId, ResidencyCacheEntry>>,
    /// #8233: the runtime relaunch the AUTOMATIC resume paths use. Installed
    /// once by the daemon; `None` everywhere else, which keeps `resume_auto`
    /// behaving exactly as it did. See `relaunch.rs`.
    pub(crate) relauncher:
        std::sync::OnceLock<std::sync::Arc<dyn super::relaunch::RuntimeRelauncher>>,
    /// #8233 item 4: session ids with a resume in flight. In-memory only, like
    /// `slots` — it guards two callers inside ONE daemon process (the operator
    /// path and the supervisor tick), and a fresh process has no resume in
    /// flight to remember. See `resume_in_flight.rs`.
    pub(crate) resume_in_flight: super::resume_in_flight::InFlightSet,
    /// #7771: reads the Claude session registry that proves an owner
    /// session ended. Installed by the host daemon and supervisor only; see
    /// `worktree_claude_registry.rs`.
    pub(crate) claude_registry:
        std::sync::OnceLock<super::worktree_claude_registry::RegistryReader>,
}

impl std::fmt::Debug for SessionManager {
    /// Why: `DaemonState` derives `Debug` and now holds a `SessionManager`, but
    /// the `Arc<dyn ManagedTmuxDriver>` field is not `Debug`, so the derive
    /// cannot be used. What: prints only the data_dir (the tmux driver and store
    /// are runtime handles with no useful debug form). Test: compile-time only.
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        f.debug_struct("SessionManager")
            .field("data_dir", &self.data_dir)
            .finish_non_exhaustive()
    }
}

impl SessionManager {
    /// Construct a session manager, loading the session store from `data_dir`.
    ///
    /// Why: the store must be loaded once at daemon start so that subsequent
    /// operations see prior state.
    /// What: calls [`SessionStore::load`] and wraps the result in an `Arc<RwLock>`.
    /// Test: used by every manager test via `make_manager`.
    pub async fn new(
        data_dir: &Path,
        tmux: Arc<dyn ManagedTmuxDriver>,
    ) -> Result<Self, ManagedError> {
        let store = SessionStore::load(data_dir).await?;
        Ok(Self {
            store: Arc::new(RwLock::new(store)),
            tmux,
            data_dir: data_dir.to_owned(),
            slots: RwLock::new(SlotRegistry::new()),
            // #6568: a missing or unreadable sidecar starts empty — see
            // `ResumeBreakerStore::load` for why that can only ever delay a
            // park. Every later read reloads from disk, because the process
            // that writes the stamp is not the one that reads it.
            resume_breaker: RwLock::new(
                super::resume_breaker::ResumeBreakerStore::load(data_dir).await,
            ),
            resume_breaker_cfg: super::resume_breaker::ResumeBreakerConfig::from_env(),
            residency_generation: AtomicU64::new(0),
            residency_cache: RwLock::new(HashMap::new()),
            relauncher: std::sync::OnceLock::new(),
            // #8233 item 4: empty at construction; every entry is added and
            // removed by a `ResumeInFlightGuard` within one resume call.
            resume_in_flight: std::sync::Arc::new(std::sync::Mutex::new(HashSet::new())),
            claude_registry: std::sync::OnceLock::new(),
        })
    }

    /// Create a new managed session, spawn the tmux host, and persist the record.
    ///
    /// Why: `POST /api/v1/sessions/managed` needs the full create-name-record-spawn
    /// flow in one operation so the HTTP handler stays thin.
    /// What: derives the tmux name via [`Self::resolve_session_name`]
    /// (`tm-<project-leaf>-NN`, #1955), creates the tmux session via the driver,
    /// persists a [`SessionRecord`] in state `Provisioning`, and returns it.
    /// The runtime backend defaults to [`crate::runtime::RuntimeKind::ClaudeCode`]
    /// so callers that do not care about the backend keep the pre-#1203 behavior.
    /// Test: `manager_create_record`.
    pub async fn create(
        &self,
        task: String,
        cwd: Option<PathBuf>,
        name_hint: Option<String>,
        workspace_path: Option<PathBuf>,
        repo_url: Option<String>,
        branch: Option<String>,
    ) -> Result<SessionRecord, ManagedError> {
        self.create_with_id(
            ManagedSessionId::new(),
            task,
            cwd,
            name_hint,
            workspace_path,
            repo_url,
            branch,
            crate::runtime::RuntimeKind::default(),
            false,
            false,
        )
        .await
    }

    /// Inject an answer to the session's pending decision.
    ///
    /// Why: the calling agentic process resolves pending decisions by calling
    /// POST /sessions/{id}/answer; this method persists the answer and clears
    /// pending_decision/proposed_default so they are not re-surfaced. This
    /// route has the exact same sibling-window hijack risk #2468 fixed for
    /// `inject`/`observe`: a session-scoped `send_line` lands in whichever
    /// pane tmux currently considers active, not necessarily the pane the
    /// pending decision was raised in.
    /// What: looks up the record, proves its pane with [`Self::owned_pane`]
    /// (#9101: a record with no pane id or no matching tmux server is
    /// refused), sends the answer text via `send_line_to_pane`, clears the
    /// pending fields, and persists.
    /// Test: `manager_answer_decision` in tests;
    /// `answer_decision_targets_pane_when_pane_id_known`,
    /// `answer_decision_refuses_when_stored_pane_gone`,
    /// `answer_decision_refuses_a_legacy_record_without_pane_id`
    /// in `pane_scoped_tests.rs`.
    pub async fn answer_decision(
        &self,
        id: &ManagedSessionId,
        answer: &str,
    ) -> Result<(), ManagedError> {
        let mut record = self.get(id).await?;
        let pane = self.owned_pane(id, &record)?;
        self.tmux
            .send_line_to_pane(&record.tmux_name, &pane, answer)
            .map_err(|e| ManagedError::TmuxUnavailable(e.to_string()))?;
        record.pending_decision = None;
        record.proposed_default = None;
        record.last_activity_at = Some(Utc::now());
        self.store.write().await.upsert(record).await?;
        Ok(())
    }

    /// Look up a session by its managed id.
    ///
    /// Why: the HTTP GET and activity handlers need a typed, async lookup. Since
    /// #1219 the lookup must also reflect writes made by another process (the
    /// supervisor) to the shared store, so it reloads-on-read first. That reload
    /// mutates the in-memory map, hence a write lock rather than a read lock. A
    /// transient reload error must NOT manifest as a false "session not found":
    /// if the id is still present in the last-known in-memory map we return that
    /// record (slightly stale) instead of failing the lookup — only a genuinely
    /// absent id yields `SessionNotFound`.
    /// What: acquires a write lock, attempts [`SessionStore::reload_if_changed`];
    /// on reload success the freshly-reloaded map is consulted, on reload failure
    /// the last-known map is consulted. Either way the lookup uses
    /// [`SessionStore::cached_get`], so a reload error degrades to stale-but-present
    /// rather than "gone".
    /// Test: `manager_create_record`, `manager_get_reflects_out_of_process_write`,
    /// `manager_get_returns_last_known_on_reload_error`.
    pub async fn get(&self, id: &ManagedSessionId) -> Result<SessionRecord, ManagedError> {
        let mut guard = self.store.write().await;
        if let Err(e) = guard.reload_if_changed().await {
            // Reload failed (transient I/O): do NOT surface as "not found". Fall
            // through to the last-known in-memory record if we have it.
            warn!(id = %id, "session get: reload failed: {e}; using last-known record");
        }
        guard.cached_get(id).map_err(|e| match e {
            StoreError::NotFound(k) => ManagedError::SessionNotFound(k),
            other => ManagedError::Store(other),
        })
    }

    /// Return all managed sessions.
    ///
    /// Why: `GET /api/v1/sessions/managed` returns the full list, and (since
    /// #1219) must reflect any out-of-process write before answering. Crucially, a
    /// transient reload I/O error (e.g. an NFS hiccup or a momentarily unreadable
    /// file) must NOT make the endpoint report ZERO sessions — that would mislead
    /// the supervisor/operator into thinking the fleet is empty and could trigger
    /// spurious re-provisioning. The in-memory map already holds the last-known
    /// set, so a reload failure degrades to "slightly stale", never "fleet empty".
    /// What: acquires a write lock and delegates to [`SessionStore::all`], which
    /// reloads from disk first if the backing file changed. On a reload error it
    /// logs and falls back to the ACTUAL last-known in-memory set
    /// ([`SessionStore::cached_all`]) rather than an empty list.
    /// Test: `manager_get_reflects_out_of_process_write`,
    /// `manager_list_returns_last_known_on_reload_error`.
    pub async fn list(&self) -> Vec<SessionRecord> {
        let mut guard = self.store.write().await;
        match guard.all().await {
            Ok(records) => records,
            Err(e) => {
                // #5007: the fallback stays — a transient read failure must not
                // report an empty fleet — but it is no longer silent. See
                // `store_health::log_reload_fallback`.
                let last_known = guard.cached_all();
                super::store_health::log_reload_fallback(&e, last_known.len());
                last_known
            }
        }
    }

    // `numbered_snapshot` (#3034) lives in the sibling `numbering.rs` — see
    // this file's module doc for why sibling extraction keeps this file
    // under the 500-SLOC cap (the same pattern `reactivate.rs`/`reconcile.rs`
    // already use for `mark_reactivated`/`reconcile_on_boot`).

    /// Inject text into a live session's tmux pane (Enter-submit variant),
    /// state- and modal-gated (issue #3591).
    ///
    /// Why: `POST /api/v1/sessions/managed/{id}/send`, `tm sessions send`, and
    /// the MCP `session_send` tool all let the operator or automation feed
    /// arbitrary text into a running session without attaching. Before #3591
    /// the only guard was [`Self::inject`]'s Stopped/Decommissioned check —
    /// `Provisioning` (pane exists, Claude Code hasn't finished booting) and
    /// `Errored` (runtime crashed back to a bare shell) were NOT refused, so
    /// `Submit::Enter`'s literal-then-Enter dispatch would type the message
    /// into a bare shell and press Enter, **executing it as a shell command**.
    /// [`super::task_inject::SessionManager::check_send_input_ready`](crate::session_manager::SessionManager::check_send_input_ready) (kept
    /// in that sibling file, alongside the readiness/modal machinery it
    /// reuses, to stay under THIS file's 500-SLOC production cap) now runs
    /// first: a state guard refusing `Provisioning`/`Errored` (closes the
    /// shell-execution risk outright — the priority fix), and — for
    /// `RuntimeKind::ClaudeCode` sessions only — the SAME modal-detection
    /// machinery [`Self::inject_task_when_ready`] uses for spawn-time
    /// `--task` delivery, catching an `Active` session whose pane is showing
    /// a first-run onboarding/trust modal that would otherwise silently
    /// swallow the send. Deliberately NOT reused: `task_inject`'s
    /// `runtime_ready`-based half of [`super::task_inject::PaneReadiness`]
    /// (the "is the `claude` PID up" check) — it was tried and reverted
    /// after it broke three hermetic tests built on
    /// `DaemonState::with_root_isolated_managed` (whose `FakeNoopTmuxDriver`
    /// makes `runtime_ready` `false` by construction), proof it is too blunt
    /// an instrument outside the narrow, bounded, just-spawned poll loop
    /// `inject_task_when_ready` uses it in. See `PaneReadiness`'s doc for the
    /// full account. `RuntimeKind::Tcode` sessions skip the modal probe
    /// entirely (mirrors `should_inject_task`'s existing ClaudeCode-only
    /// scoping): its markers are Claude-Code-specific onboarding copy.
    /// What: fail-fast, not block-and-poll — unlike spawn-time injection
    /// (which owns the session and can afford to wait out a first-run modal),
    /// `send_input` is called from an interactive CLI and from MCP tools
    /// where a caller blocked for up to ~60s on someone else's pane is bad
    /// UX; a caller that gets an immediate, precise "still provisioning" or
    /// "showing a modal" error can decide to retry itself. This also
    /// strengthens what a caller gets back on success: previously
    /// `sent: true` (the MCP `session_send` wrapper,
    /// `crates/trusty-mpm/src/daemon/mcp_session.rs`) meant only "the two
    /// tmux subprocesses exited 0"; now reaching that path additionally means
    /// the pane was NOT `Provisioning`/`Errored` and (for ClaudeCode) showed
    /// no known blocking modal immediately beforehand — the existing field's
    /// meaning is strengthened for free, no wire format change needed (#3168
    /// scopes the full delivery-ack layer). Delegates the actual dispatch to
    /// `inject(id, text, Submit::Enter)` so the pane-alive check and
    /// `send_line` dispatch stay the single source of truth for the wire
    /// mechanics (unchanged — `Submit::Enter`'s literal-then-Enter dispatch
    /// is issue #1461's deliberate, unit-tested behavior).
    /// Test: `manager_send_input`,
    /// `manager_send_input_rejected_for_stopped_and_decommissioned`,
    /// `manager_send_input_rejected_for_provisioning_and_errored`,
    /// `manager_send_input_rejected_when_pane_shows_blocking_modal`,
    /// `manager_send_input_skips_modal_probe_for_tcode` (the last three in
    /// the sibling `send_input_gate_tests.rs`).
    pub async fn send_input(&self, id: &ManagedSessionId, text: &str) -> Result<(), ManagedError> {
        self.check_send_input_ready(id).await?;
        self.inject(id, text, Submit::Enter).await
    }

    /// Inject text into a live session, committed per [`Submit`] (#1461).
    ///
    /// Why: the harness-agnostic `inject_text` verb needs ONE manager helper that
    /// dispatches the three keystroke intents onto the [`ManagedTmuxDriver`] seam
    /// so the `SessionControl` impl stays a thin mapping. It reuses the same
    /// Stopped/Decommissioned guard as [`send_input`](Self::send_input) so input
    /// is never sent to a dead pane.
    /// What: looks up the record, rejects Stopped/Decommissioned sessions, then
    /// proves the record's pane with [`Self::owned_pane`] (issue #2468's
    /// sibling-window hijack; #9101: a pane id from before a tmux server
    /// restart, or a record with no pane id, is refused) before dispatching
    /// to that pane: [`Submit::Enter`] → `send_line_to_pane` (literal +
    /// Enter), [`Submit::NoSubmit`] → `send_keys_literal_to_pane` (literal
    /// only), [`Submit::Interrupt`] → `send_interrupt_to_pane` (Ctrl-C).
    /// Bumps `last_activity_at` ONLY for
    /// `Enter`/`NoSubmit`: an `Interrupt` (Ctrl-C) is a STOP signal, not
    /// forward progress, so treating it as activity would mislead the
    /// idle/orphan-GC reconciliation into believing a stalled session is
    /// still working.
    /// Test: `inject_dispatch_enter_sends_literal_then_enter`,
    /// `inject_dispatch_nosubmit_sends_literal_only`,
    /// `inject_dispatch_interrupt_sends_ctrl_c` in tests/session_control_api.rs;
    /// `inject_targets_pane_when_pane_id_known`,
    /// `inject_refuses_when_stored_pane_gone`,
    /// `inject_refuses_a_legacy_record_without_pane_id`
    /// in `pane_scoped_tests.rs`;
    /// `a_stale_record_after_a_server_restart_never_sends_keys`.
    pub async fn inject(
        &self,
        id: &ManagedSessionId,
        text: &str,
        submit: Submit,
    ) -> Result<(), ManagedError> {
        self.inject_into_owned(id, text, submit).await.map(drop)
    }

    /// [`Self::inject`], answering the session name and the pane the gate
    /// proved, so a caller reads back the same pane (#9101).
    pub(super) async fn inject_into_owned(
        &self,
        id: &ManagedSessionId,
        text: &str,
        submit: Submit,
    ) -> Result<(String, String), ManagedError> {
        let mut record = self.get(id).await?;
        if matches!(
            record.state,
            ManagedSessionState::Stopped | ManagedSessionState::Decommissioned
        ) {
            return Err(ManagedError::TmuxUnavailable(format!(
                "session {} is {}; cannot inject input",
                record.tmux_name, record.state
            )));
        }
        let pane = self.owned_pane(id, &record)?;
        let name = record.tmux_name.as_str();
        let result = match submit {
            Submit::Enter => self.tmux.send_line_to_pane(name, &pane, text),
            Submit::NoSubmit => self.tmux.send_keys_literal_to_pane(name, &pane, text),
            Submit::Interrupt => self.tmux.send_interrupt_to_pane(name, &pane),
        };
        result.map_err(|e| ManagedError::TmuxUnavailable(e.to_string()))?;
        // Interrupt is a STOP signal, not activity — do not bump last_activity_at.
        if matches!(submit, Submit::Enter | Submit::NoSubmit) {
            record.last_activity_at = Some(Utc::now());
        }
        let name = record.tmux_name.clone();
        self.store.write().await.upsert(record).await?;
        Ok((name, pane))
    }

    /// Observe a session's raw surface — LLM-FREE (#1461).
    ///
    /// Why: every harness needs a cheap, deterministic read of what a session is
    /// actually showing (pane + liveness + any pending escalation) WITHOUT an LLM
    /// key. This bundles the three reads the managed activity route already does
    /// into one manager helper so `SessionControl::observe` is a thin mapping.
    /// It proves the record's pane with [`Self::owned_pane`] first (issue
    /// #2468's sibling-window hijack; #9101: a pane id from before a tmux
    /// server restart names another session's pane).
    /// What: captures the last `lines` rows of the proven pane via
    /// `capture_pane`, probes `runtime_active` via `session_exists`, and reads
    /// the record's `pending_decision` / `proposed_default`. Never calls the
    /// LLM.
    /// Test: `observe_returns_raw_pane_without_llm`, `observe_reports_runtime_active`,
    /// `observe_captures_pane_scoped_when_pane_id_known`,
    /// `observe_refuses_when_stored_pane_gone` in `pane_scoped_tests.rs`;
    /// `a_stale_record_after_a_server_restart_never_reads_the_pane`.
    pub async fn observe(
        &self,
        id: &ManagedSessionId,
        lines: usize,
    ) -> Result<crate::core::sm::control::RawObservation, ManagedError> {
        let record = self.get(id).await?;
        let pane = self.owned_pane(id, &record)?;
        let raw_pane = self
            .tmux
            .capture_pane(&record.tmux_name, &pane, lines)
            .unwrap_or_default();
        let runtime_active = self.tmux.session_exists(&record.tmux_name);
        Ok(crate::core::sm::control::RawObservation {
            raw_pane,
            runtime_active,
            pending_decision: record.pending_decision,
            proposed_default: record.proposed_default,
        })
    }

    /// Mark a runtime-exited session `Stopped` WITHOUT killing its tmux pane (#2023 A).
    ///
    /// Why: `stop` is the EXPLICIT-stop contract (`tm session stop`, HTTP
    /// stop route, MCP `sessions.stop`) — a human or client asked to tear the
    /// runtime down, so killing the pane is correct there. The runtime-exit
    /// reaper ([`crate::daemon::runtime_reap::stop_runtime_exited`]) is a
    /// DIFFERENT event: the inner `claude` process exited on its own and the
    /// tmux pane already fell back to a bare login shell — the pane is not
    /// misbehaving, it is simply idle. Routing that self-healing transition
    /// through `stop` (and therefore `graceful_terminate_runtime` /
    /// `kill_session`) killed a pane the user may still have attached, or
    /// wanted to glance at, purely because the daemon noticed the runtime was
    /// gone ~60s earlier than the human did. This method gives the reaper its
    /// own non-destructive path: same record transition, no pane teardown.
    ///
    /// Scope: this ONLY preserves the pane — it marks `Stopped` and leaves the
    /// tmux session/pane alive so an operator can still `tmux attach` and look
    /// at the trailing output, or manually re-launch `claude` in that same
    /// pane. It does NOT provide in-place runtime reuse by itself: today's
    /// `tm session resume` / the guided-picker Restart path unconditionally
    /// kills any surviving tmux session and creates a fresh one
    /// (`Self::resume`, below — see the `kill_session` + `create_session` pair
    /// there), so resuming a session marked `Stopped` by this method still
    /// tears the preserved pane down. True bare-`tm`-in-pane relaunch reusing
    /// the surviving shell is the scope of #2023 component C, not this method.
    ///
    /// NOTE — auto-resume supervisors: `tm supervisor --auto-resume`
    /// (`supervisor::poller::run_tick`) auto-resumes a `Stopped` record whose
    /// stop nobody asked for, and `resume()` kills the preserved pane as
    /// described above. This method records
    /// [`StopCause::Unexpected`](crate::session_manager::record::StopCause::Unexpected), so its
    /// records ARE that kind — the "pane left alive" guarantee holds for
    /// `auto_resume = false` / interactive deployments, and under an auto-resume
    /// supervisor the session is still revived (and its idle pane replaced) by
    /// that mode's own design. What #6194 changed is the other half: a stop
    /// somebody DID ask for now records
    /// [`StopCause::Deliberate`](crate::session_manager::record::StopCause::Deliberate) and is left
    /// down. Reconciling supervisor-revive vs. preserve-on-exit semantics for
    /// this path remains a follow-up (#2026).
    ///
    /// What: loads the record, captures a pane snapshot the same way `stop`
    /// does (best-effort — the pane usually still exists, it is just an idle
    /// shell), sets `state = Stopped`, and persists. Deliberately never calls
    /// [`Self::graceful_terminate_runtime`] or `kill_session` — the tmux
    /// session and its pane are left exactly as they are. #9101: the snapshot
    /// and the `TM_MANAGED_SESSION_ID` publish run only on a pane
    /// [`Self::owned_pane`] proves.
    /// Test: `a_stale_record_after_a_server_restart_never_publishes_its_id_on_reap`,
    /// `an_unreadable_pane_identity_reaps_the_record_without_reading_the_pane`,
    /// `stop_runtime_exited_transitions_active_to_stopped` (in
    /// `daemon::runtime_reap`) asserts the record becomes `Stopped`;
    /// `stop_runtime_exited_does_not_kill_pane` (same module) asserts
    /// `kill_session` is never invoked on the fake driver;
    /// `mark_runtime_exited_stopped_rejects_concurrently_decommissioned`
    /// (this module's tests) asserts the CAS guard below;
    /// `generation_increments_across_mark_runtime_exited_stopped` in
    /// `daemon::managed_routes::residency`'s route tests asserts the #7087
    /// residency bump;
    /// `the_reaper_leaves_a_session_whose_resume_is_in_flight_alone` asserts the
    /// #8233 in-flight refusal below;
    /// `list_not_blocked_by_stalled_runtime_exit_capture_9034` and
    /// `a_record_changed_during_the_runtime_exit_capture_is_not_written_9034`
    /// assert the #9034 guard release and re-check.
    ///
    /// CAS guard (#2453 review finding 3): a `decommission`/`stop` that lands
    /// between this function's read and its write must win, never be
    /// overwritten by a `Stopped` copy of the stale read. #9034: no store
    /// guard is held across the tmux calls (pane gate, capture, pane-id read)
    /// or the scrollback write, because one stalled tmux subprocess queued
    /// every `list()`/`get()` behind it. The record is read under one guard,
    /// then re-read under a fresh guard before the write. A record that is
    /// gone, no longer `Active`, or changed in any field since the first read
    /// is refused with the same error a reap of it would get, and nothing is
    /// written: no upsert, no breaker death, no residency bump. A racing reap
    /// may still refresh the best-effort scrollback file on disk.
    pub async fn mark_runtime_exited_stopped(
        &self,
        id: &ManagedSessionId,
    ) -> Result<SessionRecord, ManagedError> {
        // #8233: a resume writes `Active` before the runtime exists, so for the
        // whole rest of that resume this reconcile would see "Active, no
        // runtime" and stop the record under the path that is still working on
        // it — handing the next supervisor tick a `Stopped` record to launch a
        // SECOND time. The claim is not a lock this caller can take (it must
        // not queue behind the resume, it must decline), so it reads it.
        if self.is_resume_in_flight(id) {
            return Err(ManagedError::ResumeInFlight(id.to_string()));
        }
        // #9034: read under a short guard; the tmux work below runs with none.
        let observed = {
            let mut guard = self.store.write().await;
            super::runtime_exit_guard::active_for_runtime_exit(&mut guard, id).await?
        };
        let mut record = observed.clone();
        // #9101: the scrollback read and the id publish below reach the pane
        // only when it is proven this record's; the Stopped transition is
        // record-only and runs either way.
        let owned = self.owned_pane(id, &record).ok();
        if let Some(pane) = owned.as_deref() {
            super::snapshot::capture_into_pane(&mut record, &*self.tmux, Some(pane)).await;
        }
        // #2453 review finding 1 (round 3 — round 2's re-derive-on-every-call
        // approach was proven UNSOUND): `get_pane_id` shells out to `tmux
        // display-message -t <SESSION_NAME> -p '#{pane_id}'`, which is
        // SESSION-scoped — tmux resolves it to the session's CURRENTLY
        // ACTIVE window's pane, not necessarily the pane this reconcile is
        // about. Verified against a live tmux 3.6b: `display-message -t
        // testsess` reads `%0` with one window, then reads `%1` after `tmux
        // new-window -t testsess` (which auto-activates the new window) —
        // the SAME session-scoped query now resolves to a DIFFERENT pane.
        // Since this function runs on every runtime-exit reconcile (the
        // periodic reaper, the #2455 SessionEnd hook, and the #2453
        // reconcile-then-reactivate route), a sibling window merely being
        // tmux-active at reconcile time would silently OVERWRITE a
        // known-good `pane_id` with the active sibling's — reopening the
        // cross-pane hijack via a new vector AND breaking the legitimate
        // relaunch for the pane that actually exited. Backfill-only-when-
        // `None` (chosen over re-verifying the stored pane_id via a
        // pane-scoped `display-message -t <pane_id>` query, #2456 review's
        // alternative option) is simpler and sufficient: it still heals a
        // legacy record that never had a `pane_id` captured (created before
        // this field existed, where session-scoped ambiguity does not yet
        // matter — no earlier capture exists to protect), but a
        // known-good id, once captured at spawn/adopt time, is NEVER
        // re-derived here again.
        // #9101: backfill the pane id only. A by-name read reaches whatever
        // session holds the name now, so it never pairs the pane with a
        // server: the record stays unverifiable and every pane gate refuses.
        if record.pane_id.is_none() {
            record.pane_id = self.tmux.get_pane_id(&record.tmux_name);
        }
        // #9034: re-check under a fresh guard; a failed re-check writes nothing.
        let mut guard = self.store.write().await;
        if self.is_resume_in_flight(id) {
            return Err(ManagedError::ResumeInFlight(id.to_string()));
        }
        let current = super::runtime_exit_guard::active_for_runtime_exit(&mut guard, id).await?;
        if !super::runtime_exit_guard::unchanged(&observed, &current) {
            return Err(ManagedError::InvalidState(
                id.to_string(),
                "cannot mark runtime-exited-stopped: the record changed while its pane \
                 was read — a concurrent operation already updated it"
                    .into(),
            ));
        }
        record.state = ManagedSessionState::Stopped;
        // #6194: nothing asked for this stop — the runtime exited on its own —
        // so relaunching it is the self-healing action auto-resume exists for.
        // #6568: unless relaunching it has already been tried and has already
        // failed to change anything K times in a row — see
        // `SessionManager::runtime_exit_stop_cause` in `resume_breaker.rs`.
        record.stop_cause = Some(self.runtime_exit_stop_cause(id, &record.tmux_name).await);
        guard.upsert(record.clone()).await?;
        drop(guard);
        // #7087: Active -> Stopped leaves the active-project set, exactly as
        // `stop` does — the reaper reaches it without going through `stop`.
        self.bump_residency_generation();
        info!(
            id = %id,
            name = %record.tmux_name,
            "runtime-reap: managed session marked Stopped (pane left alive, #2023)"
        );
        // #2157 item 3: heal stale panes — durably publish TM_MANAGED_SESSION_ID
        // into this session's tmux environment right when the runtime is
        // confirmed to have exited (the pane is now an idle shell). This covers
        // panes that never got the durable publish at spawn time (created by a
        // pre-#2157 build, or whose set-environment call failed at spawn), so a
        // LATER bare `tm` run inside this pane can still resolve the id via
        // `tmux show-environment` even though the process-env export never
        // landed. Best-effort — never fails the reap. #9101: never into a
        // session whose pane is not proven this record's.
        if owned.is_some()
            && let Err(e) = self.tmux.set_environment(
                &record.tmux_name,
                "TM_MANAGED_SESSION_ID",
                &id.to_string(),
            )
        {
            warn!(
                id = %id,
                name = %record.tmux_name,
                "runtime-reap: tmux set-environment heal failed (non-fatal): {e}"
            );
        }
        Ok(record)
    }

    /// Resume a stopped session, reusing its tmux pane when one already survives.
    ///
    /// Why: a session ENDURES until decommissioned; after `stop` the workspace
    /// directory is still on disk and `resume` brings the runtime back without
    /// re-cloning. Prior to #2148 this UNCONDITIONALLY killed any surviving tmux
    /// session and created a brand-new one, even when the pane was left alive on
    /// purpose (e.g. [`Self::mark_runtime_exited_stopped`], #2023 A, which marks a
    /// session `Stopped` WITHOUT touching its pane so the operator can glance at
    /// trailing output or keep a live `tmux attach`). Every caller of `resume`
    /// (CLI restart, HTTP `/resume`, MCP `sessions.resume`, the auto-resume
    /// supervisor) inherited that destructiveness, dropping the operator into a
    /// freshly recreated pane instead of the one they were already looking at.
    /// What: validates the session is `Stopped` or `Errored`, resolves the
    /// workdir via [`resume_workdir::resolve_existing_workdir`] (#2250, #8551 —
    /// a recorded workspace decides alone and `cwd` is a candidate only when
    /// none is recorded; erroring with [`ManagedError::WorkspaceMissing`]
    /// rather than handing a stale/removed path to tmux), then branches on
    /// [`ManagedTmuxDriver::session_exists_checked`] — a probe that cannot
    /// reach tmux refuses the resume rather than falling into the destructive
    /// recreate branch below (#5859): if the tmux SESSION is STILL
    /// alive, it is (usually) reused as-is — no `kill_session`, no
    /// `create_session` — because the caller (e.g. `resume_managed`)
    /// re-spawns the runtime via `RuntimeAdapter::spawn_resume`, which now
    /// types the resume command straight into the record's OWN recorded
    /// `pane_id` (sibling-window hijack fix, follow-up to #2456) rather than a
    /// session-scoped target that tmux could resolve to any other pane. Before
    /// trusting that reuse, [`Self::owned_pane`] confirms that the SPECIFIC
    /// recorded pane — not just some pane in the session — is still there, on
    /// the tmux server the record captured it on (#9101: a record with no
    /// pane id or server identity, or from before a server restart, refuses
    /// and adopts nothing): `session_exists` alone cannot tell "the recorded pane survived"
    /// apart from "a sibling window opened after it was closed is keeping the
    /// session alive". A confirmed-gone recorded pane fails loudly with
    /// [`ManagedError::PaneGone`] rather than either (a) silently respawning
    /// into the unrelated active sibling, or (b) killing the whole session
    /// (which would destroy the sibling too) to recreate it. If the whole
    /// session is gone: the kill-floor gate (no kill by name, #9101), then
    /// [`resume_workdir::create_and_verify_pane`], which creates the fresh
    /// session AND verifies (#2250) the pane actually landed at the resolved
    /// workdir — tmux `-c <dir>` can exit 0 while silently falling back to
    /// `$HOME`, which this catches and fails loudly on rather than typing the
    /// resume command into a mis-rooted pane; the freshly created pane's id is
    /// then captured with its tmux server to refresh the stale `pane_id`, so
    /// the resumed record is ownable again (#9004). Either way (reuse or
    /// recreate) the record is marked `Active` and persisted.
    /// Test: `manager_resume_respawns_in_existing_workspace` (`tests.rs`) —
    /// asserts a new `create_session` call is issued when no pane survives (the
    /// `stop()` path, which kills the pane); `resume_reattach_tests.rs`'s
    /// `manager_resume_reuses_live_pane_without_recreate` — asserts NEITHER
    /// `kill_session` NOR `create_session` fires when the pane survives (the
    /// `mark_runtime_exited_stopped` path, #2148); `resume_reattach_tests.rs`'s
    /// `manager_resume_errors_when_recreated_pane_cwd_mismatches` — asserts a
    /// pane-cwd mismatch on the recreate path fails loudly (#2250);
    /// `resume_reattach_tests.rs`'s
    /// `resume_refuses_when_stored_pane_gone_but_session_alive` — asserts a
    /// confirmed-gone recorded pane refuses with `PaneGone` and never falls
    /// through to a session-scoped reuse or a session-wide kill/recreate;
    /// `a_stale_record_after_a_server_restart_is_never_adopted_by_resume`,
    /// `a_resume_that_recreates_the_pane_makes_the_record_ownable` (#9101);
    /// `liveness_tests.rs`'s `resume_refuses_when_the_tmux_probe_fails` —
    /// asserts an unobservable probe refuses instead of killing the pane
    /// (#5859); `generation_increments_across_resume` in
    /// `daemon::managed_routes::residency`'s route tests — asserts the #7087
    /// residency bump.
    ///
    /// #6568: this is the OPERATOR entry point. It forgives the auto-resume flap
    /// streak, because a person resuming a session by hand is saying the cause
    /// is addressed and the session deserves a fresh budget. The automatic
    /// callers — the supervisor poller and the boot-reconcile tail — go through
    /// [`Self::resume_auto`] instead (in the sibling `resume_breaker` module),
    /// which stamps the attempt without forgiving anything.
    pub async fn resume(&self, id: &ManagedSessionId) -> Result<SessionRecord, ManagedError> {
        // #8233 item 4: held across the whole resume, so the supervisor tick
        // cannot enter this session's `resume_auto` mid-flight. Dropped on
        // every exit path below, including the `?`.
        let _in_flight = self.begin_resume(id)?;
        let record = self.resume_inner(id).await?;
        self.note_operator_resume(id).await;
        Ok(record)
    }

    /// The shared resume body — see [`Self::resume`] for the full contract.
    pub(crate) async fn resume_inner(
        &self,
        id: &ManagedSessionId,
    ) -> Result<SessionRecord, ManagedError> {
        let mut record = self.get(id).await?;
        match record.state {
            ManagedSessionState::Stopped | ManagedSessionState::Errored => {}
            ref s => {
                return Err(ManagedError::InvalidState(
                    id.to_string(),
                    format!(
                        "cannot resume a session in state '{s}'; only Stopped or Errored sessions can be resumed"
                    ),
                ));
            }
        }
        // #8942: the Architect is relaunched by `tm fleet init`, never here.
        super::supervisor::refuse_protected(&record, super::supervisor::ProtectedVerb::Resume)?;

        // #3823: guarantee the tmux SERVER exists before the FIRST tmux call
        // below (`session_exists_checked`'s `list-sessions` probe), so a cold
        // socket is STARTED rather than reported as a probe failure the
        // #5859 guard below would (correctly) refuse to resume through. Also redundant
        // (cheap no-op `tmux start-server`) with the recreate branch's own
        // eventual #3386/#3722 `ensure_server_up` when the pane needs
        // rebuilding — accepted for the same reason as the create paths'
        // identical round-trip: correctness over one extra idempotent spawn.
        self.tmux.ensure_server_up()?;

        // Prefer last_cwd → workspace_path → cwd (#1816), each existence-checked
        // on disk (#2250 — workspace_path and cwd previously were NOT, so a
        // stale/removed worktree silently rooted the recreated pane at $HOME).
        // Errors loudly via WorkspaceMissing when none of the three remain.
        // #8551: a recorded workspace that is gone refuses; no fallback to cwd.
        let workdir = resume_workdir::resolve_existing_workdir(id, &record)
            .await?
            .to_string_lossy()
            .to_string();

        // #2148: a pane that survived the runtime exit (e.g.
        // `mark_runtime_exited_stopped`, #2023 A) must be REUSED, not destroyed.
        // Only recreate the tmux session when no live pane remains.
        //
        // #5859: `session_exists_checked` + `?`, not `session_exists`. The else
        // branch below KILLS the session and rebuilds the pane, so an
        // undeterminable probe answering `false` destroyed a live pane. Refuse
        // the resume instead — the operator can retry once tmux answers.
        if self.tmux.session_exists_checked(&record.tmux_name)? {
            // Sibling-window hijack fix (follow-up to #2456): `session_exists`
            // only proves SOME pane in this tmux session is alive — it says
            // nothing about whether it is the SPECIFIC pane this record is
            // bound to. #9101: `owned_pane` also proves the pane is on the
            // server the record captured it on, so a stale record never adopts
            // a restarted server's pane; an `Owned` record's server already
            // matches, so there is nothing to re-capture here.
            self.owned_pane(id, &record)?;
            info!(
                id = %id,
                name = %record.tmux_name,
                workdir = %workdir,
                "managed session resumed: re-attached to live pane (#2148, no recreate)"
            );
        } else {
            // #8942: a name the floor cannot clear refuses the whole resume.
            // #9101: no kill by name here. The probe just proved no session
            // carries the name, so a kill could only reach one that took the
            // name since. The create below is exclusive, so such a session
            // fails it instead of being attached to.
            self.kill_gate(&record.tmux_name, "SessionManager::resume")?;

            // Create a fresh tmux session rooted at the EXISTING workspace, then
            // verify tmux didn't silently fall back to $HOME (#2250). No
            // re-clone — workspace_path is reused as-is.
            resume_workdir::create_and_verify_pane(
                self.tmux.as_ref(),
                &record.tmux_name,
                &workdir,
            )?;
            // #2453 review finding 1 (round 2): the recreated pane is a BRAND
            // NEW tmux pane — the record's previously-captured `pane_id` now
            // refers to the DESTROYED pane and must be refreshed, or the
            // bare-`tm` in-pane relaunch's pane-identity gate would never
            // match this session again until the next runtime-exit reconcile
            // heals it. Best-effort — `None` on failure, consistent with
            // every other `get_pane_id` call site.
            // #9004: the new pane with the server it was read on.
            (record.pane_id, record.tmux_server) =
                super::pane_identity::capture(self.tmux.as_ref(), &record.tmux_name);
            info!(
                id = %id,
                name = %record.tmux_name,
                workdir = %workdir,
                "managed session resumed: recreated pane"
            );
        }

        record.state = ManagedSessionState::Active;
        record.last_activity_at = Some(Utc::now());
        // #6194: the session is running again, so the reason it last stopped no
        // longer describes it — a later crash must be auto-resumable even if
        // this resume followed a deliberate stop.
        record.stop_cause = None;
        self.store.write().await.upsert(record.clone()).await?;
        // #7087: Stopped/Errored -> Active re-enters the active-project set.
        self.bump_residency_generation();
        Ok(record)
    }

    /// Return the data directory the store is backed by.
    ///
    /// Why: tests need to inspect the data directory; callers constructing
    /// the store path need this for the store file location.
    /// What: returns the data_dir captured at construction.
    /// Test: used implicitly by store tests.
    pub fn data_dir(&self) -> &Path {
        &self.data_dir
    }

    /// Collect every tmux session name the store knows about.
    ///
    /// Why: the orphan-GC must never reap a session this store tracks. It treats
    /// the store's `tmux_name`s as a protected set — including `Decommissioned`
    /// tombstones, so a name the store has *any* record of is never mistaken for
    /// an untracked orphan (fail-closed: better to keep a stray than risk
    /// reaping a tracked one).
    /// What: reads all records and returns the set of their `tmux_name`s. A store
    /// read error is **propagated as `Err`** — it is NOT swallowed into an empty
    /// set. An empty (but successful) read means "the store tracks nothing"; an
    /// `Err` means "we could not determine what the store tracks". Conflating the
    /// two would fail OPEN: a session known ONLY to this store (absent from the
    /// legacy registry) would look untracked → idle → reapable, violating the
    /// GC's fail-closed contract. The caller
    /// ([`crate::daemon::state::DaemonState::gather_tracked_names`]) turns this
    /// `Err` into a *degraded* [`crate::daemon::orphan_gc::TrackedNames`] snapshot
    /// that makes the sweep skip its reap phase entirely.
    /// Test: `manager_known_tmux_names_collects_all` (happy path) in tests.rs;
    /// the error path is exercised end-to-end by `gather_tracked_names` degraded
    /// tests and by `run_sweep` against a degraded snapshot.
    ///
    /// Note: this is a read-only scan but still takes the store's WRITE lock,
    /// because [`SessionStore::all`] requires `&mut self`: it calls
    /// `reload_if_changed()` first (since #1219) to pick up records another
    /// process wrote, which mutates the in-memory map. Using `read()` would not
    /// compile, and the read-only `cached_all()` alternative would skip that
    /// reload — risking a stale set that omits a just-registered session and so
    /// lets the orphan-GC mistake it for an untracked orphan. The write lock is
    /// the fail-closed choice here; the lock is held only for the brief reload.
    pub async fn known_tmux_names(
        &self,
    ) -> Result<std::collections::HashSet<String>, ManagedError> {
        let records = self.store.write().await.all().await?;
        Ok(records.into_iter().map(|r| r.tmux_name).collect())
    }

    /// Gather the session names currently "in use" for per-project serial
    /// allocation (issue #1955).
    ///
    /// Why: [`crate::core::names::build_session_name`] must know which
    /// `tm-<leaf>-NN` serials are taken so it can pick the lowest free one. A
    /// DECOMMISSIONED record's serial must be free for immediate reuse (the
    /// ticket's gap-reuse requirement — "sessions 01,02,03 exist, 02 gets
    /// decommissioned, next new session reuses 02"), so this deliberately
    /// excludes them. This is DIFFERENT from [`Self::known_tmux_names`], which
    /// protects ALL records — including decommissioned tombstones — from the
    /// orphan-GC, a stricter safety purpose where "was this ever ours" matters
    /// more than "is this serial still occupied".
    /// What: unions the live tmux session list (covers adopted/foreign
    /// sessions not yet reflected in the store) with every NON-decommissioned
    /// store record's `tmux_name`. Takes the store's WRITE lock rather than a
    /// read lock for the same reason as [`Self::known_tmux_names`]:
    /// [`SessionStore::all`] requires `&mut self` (it calls
    /// `reload_if_changed()` first to pick up records another process wrote),
    /// so a read-only guard would not compile and the read-only
    /// `cached_all()` alternative would risk a stale serial set.
    /// Test: `manager_serial_reuses_decommissioned_gap` in tests.rs.
    pub(crate) async fn names_for_serial_allocation(&self) -> Result<Vec<String>, ManagedError> {
        // #3886: propagate, never re-wrap — `list_sessions` already returns a
        // `ManagedError`, so `TmuxUnavailable(e.to_string())` doubled its own
        // `tmux error:` prefix in the user-facing provisioning failure.
        let mut names: Vec<String> = self.tmux.list_sessions()?;
        let records = self.store.write().await.all().await?;
        names.extend(
            records
                .into_iter()
                .filter(|r| r.state != ManagedSessionState::Decommissioned)
                .map(|r| r.tmux_name),
        );
        Ok(names)
    }

    /// Return a clone of the shared tmux driver Arc.
    ///
    /// Why: the spawn handler needs to hand the driver to `ClaudeCodeAdapter`
    /// without duplicating the Arc lookup.
    /// What: clones the `Arc<dyn ManagedTmuxDriver>` stored at construction.
    /// Test: used in handler_spawn_wires_provision_and_spawn.
    pub fn tmux_driver(&self) -> Arc<dyn ManagedTmuxDriver> {
        self.tmux.clone()
    }

    /// Capture the last `lines` of the session's RECORDED harness pane.
    ///
    /// Why: activity classification (`managed_routes/activity`), the MCP
    /// `session_activity` read (`mcp_session`), the CLI transcript
    /// (`sm_stdio/control`, `bin/tm/commands/meta/launch`), and the supervisor
    /// idle classifier (`supervisor/poller`) all read "the session's pane" to
    /// decide state or show output. A session-scoped `capture` resolves to
    /// whichever pane tmux considers ACTIVE — which need not be the harness pane
    /// this record owns (an operator's sibling window keeps the session alive
    /// but steals the active-pane slot). That silently fed wrong-pane content to
    /// every one of those consumers and could misclassify a busy session as idle
    /// (issue #2515). This shares the exact pane-scoping [`Self::observe`] uses.
    /// What: proves the record's pane with [`Self::owned_pane`] first
    /// (`PaneGone` for a pane confirmed gone while the session lives on;
    /// #9101: `InvalidState` for a record with no pane id, or whose pane is
    /// not on the server it captured it on), then captures the last `lines`
    /// rows of that pane via `capture_pane`. Every consumer already degrades a
    /// capture error to empty/no-verdict, so a refusal fails closed.
    /// Test: `capture_pane_targets_recorded_pane_when_pane_id_known`,
    /// `capture_pane_refuses_when_stored_pane_gone`,
    /// `capture_pane_refuses_a_legacy_record_without_pane_id` in
    /// `pane_scoped_tests.rs`;
    /// `a_stale_record_after_a_server_restart_never_reads_the_pane`.
    pub async fn capture_pane(
        &self,
        id: &ManagedSessionId,
        lines: usize,
    ) -> Result<String, ManagedError> {
        let record = self.get(id).await?;
        let pane = self.owned_pane(id, &record)?;
        self.tmux
            .capture_pane(&record.tmux_name, &pane, lines)
            .map_err(|e| ManagedError::TmuxUnavailable(e.to_string()))
    }

    /// Capture the recorded harness pane for the LIVE managed session whose tmux
    /// session name is `tmux_name`, pane-scoped when a `pane_id` is tracked.
    ///
    /// Why: the idle reaper (`daemon::mod::DaemonVerdictProvider::verdict`) knows
    /// a session only by its tmux name and captured via the raw driver's
    /// session-scoped `capture`, which tmux resolves to whichever pane is ACTIVE
    /// — a sibling operator window could make a busy session read as idle and be
    /// auto-stopped, or a quiet one read as busy (issue #2515). Unlike
    /// [`Self::capture_pane`] the reaper has no `ManagedSessionId`, so this
    /// resolves the record from the tmux name first.
    /// What: selects the record whose `tmux_name` matches and whose state is NOT
    /// `Decommissioned` (a `tmux_name` is recycled after decommission, so a naive
    /// match can hit a stale tombstone — same guard as
    /// `daemon::api::claude_config_routes::select_restart_pane_id`, #2514),
    /// preferring the most recently created on the narrow provisioning-race
    /// overlap. A record's pane is captured only when [`Self::owned_pane`]
    /// proves it (#9101); any refusal is `None` (no verdict — fail closed),
    /// including a record with no `pane_id`. A tmux name with no live record
    /// has no identity to check and falls back to the session-scoped
    /// `capture`, exactly as before #2515. `None` on any capture error or
    /// empty output so the caller skips the poll.
    /// Test: `capture_pane_by_tmux_name_targets_recorded_pane`,
    /// `capture_pane_by_tmux_name_refuses_when_pane_gone`,
    /// `capture_pane_by_tmux_name_without_a_record_falls_back_to_session` in
    /// `pane_scoped_tests.rs`;
    /// `a_stale_record_after_a_server_restart_never_reads_the_pane`.
    pub async fn capture_pane_by_tmux_name(&self, tmux_name: &str, lines: usize) -> Option<String> {
        let record = self
            .list()
            .await
            .into_iter()
            .filter(|r| r.tmux_name == tmux_name)
            .filter(|r| !matches!(r.state, ManagedSessionState::Decommissioned))
            .max_by_key(|r| r.created_at);
        let text = match record {
            Some(record) => {
                let pane = self.owned_pane(&record.id, &record).ok()?;
                self.tmux.capture_pane(tmux_name, &pane, lines).ok()?
            }
            None => self.tmux.capture(tmux_name, lines).ok()?,
        };
        (!text.is_empty()).then_some(text)
    }
}