trusty-mpm 1.8.2

//! Unit tests for the session manager.
//!
//! Why: tests in a separate file keep manager.rs under the 500 SLOC production
//! cap while the 1500 SLOC test cap gives the test suite room to grow. The
//! `manager_adopt_existing_*` tests were extracted to `adopt_existing_tests.rs`
//! (issue #2468, same reason `reload_error_tests.rs` was split out — staying
//! under this file's own 1500-SLOC test cap after the pane-scoped inject/observe
//! `FakeTmuxDriver` additions).
//! What: full lifecycle tests for create, stop (keep workspace), resume
//! (re-spawn in existing workspace), decommission (remove workspace from disk),
//! send_input, reconcile (gone tmux → Stopped), answer_decision,
//! and the env-scrub command convention.
//! Test: this file IS the test module; run with `cargo test -p trusty-mpm`.

use std::collections::HashMap;
use std::path::PathBuf;
use std::sync::Mutex;

use chrono::Utc;
use tempfile::TempDir;

use super::manager::{ManagedError, ManagedTmuxDriver, SessionManager};
use super::record::{ManagedSessionId, ManagedSessionState, SessionRecord};

use std::sync::Arc;

/// RAII guard restoring `$HOME` on drop (including panic) — mirrors the
/// identical pattern in `core::session_launch::tests::EnvVarGuard` and
/// siblings (each module needs its own copy; `pub(super)`/module-private
/// visibility does not cross sibling module trees — see those sites' docs).
///
/// Why (#3965): a test that stands up a workspace and creates a session record
/// for it reaches `~/.claude.json` through code that resolves the REAL process
/// `$HOME`, not the test's own root, so it writes into the operator's real
/// `~/.claude.json` unless `$HOME` is pinned. Pairs with
/// `#[serial_test::serial]`.
/// Test: used by `spawn_session_tmux_cwd_is_workspace`.
struct HomeGuard(Option<String>);
impl Drop for HomeGuard {
    fn drop(&mut self) {
        // SAFETY: paired with `#[serial_test::serial]` — no other thread
        // reads/writes the environment concurrently.
        match self.0 {
            Some(ref p) => unsafe { std::env::set_var("HOME", p) },
            None => unsafe { std::env::remove_var("HOME") },
        }
    }
}

/// Point `$HOME` at `home` for the duration of the caller's scope. Callers
/// MUST be `#[serial_test::serial]` — see [`HomeGuard`].
fn set_home(home: &std::path::Path) -> HomeGuard {
    let prior = std::env::var("HOME").ok();
    // SAFETY: serialized via `#[serial_test::serial]`.
    unsafe { std::env::set_var("HOME", home) };
    HomeGuard(prior)
}

/// #9004: the one tmux server instance [`FakeTmuxDriver`] reports.
pub const FAKE_TMUX_SERVER: &str = "1:1";

/// #9101: the `$N` id the fake's server gives every session.
pub const FAKE_SESSION_ID: &str = "$0";

/// A fake tmux driver for unit testing.
///
/// Why: the manager must be testable without a real tmux binary; this
/// implementation records calls and allows the test to control which
/// sessions appear to exist.
/// What: stores created sessions in a mutex-guarded map; `session_exists`
/// consults the map; all operations record their call. `create_cwd_calls`
/// records `(session_name, workdir)` pairs so tests can assert the correct
/// cwd was passed to `tmux new-session -c`. `graceful_stop_calls` records
/// `(session_name, Option<u32>)` pairs for graceful-stop assertions.
/// Test: used by every manager unit test.
pub struct FakeTmuxDriver {
    sessions: Mutex<HashMap<String, String>>,
    pub send_calls: Mutex<Vec<(String, String)>>,
    pub kill_calls: Mutex<Vec<String>>,
    pub capture_responses: Mutex<HashMap<String, String>>,
    /// Names to return from `list_sessions`.
    pub seeded_names: Mutex<Vec<String>>,
    /// Records `(session_name, workdir)` for every `create_session` call.
    ///
    /// Why: regression guard — tests assert that the cwd passed to
    /// `tmux new-session` equals the provisioned workspace path, never $HOME.
    pub create_cwd_calls: Mutex<Vec<(String, String)>>,
    /// Records `(session_name, claude_pid)` for every `graceful_stop` call.
    pub graceful_stop_calls: Mutex<Vec<(String, Option<u32>)>>,
    /// Records the session name for every `send_interrupt` (Ctrl-C) call.
    ///
    /// Why: the CLI graceful-drain path (#1975) signals the runtime via
    /// `signal_terminate`, which — with no PID known in unit tests — falls back to
    /// `send_interrupt`. Recording it lets `graceful_terminate_runtime_signals_then_kills`
    /// assert the signal fired before the pane was reclaimed.
    pub interrupt_calls: Mutex<Vec<String>>,
    /// Controllable `get_pane_cwd` response (#2250) — `None` by default,
    /// matching the trait's "cannot verify" default; set to `Some(path)` to
    /// simulate a real driver reporting the pane's actual cwd, including a
    /// deliberate mismatch against the requested workdir.
    pub pane_cwd_override: Mutex<Option<PathBuf>>,
    /// Controllable `get_pane_id` response (sibling-window hijack fix,
    /// follow-up to #2456) — `None` by default, matching the trait's "cannot
    /// verify" default (this fake never captures a real pane_id at
    /// `create_session` time); set to `Some(id)` to simulate a real driver
    /// having captured one, so tests can exercise the pane-scoped resume
    /// respawn / pane-gone-refusal paths.
    pub pane_id_override: Mutex<Option<String>>,
    /// Controllable `pane_exists` response (sibling-window hijack fix,
    /// follow-up to #2456) — `None` by default, which falls through to the
    /// trait's optimistic `true` default (matches every existing test's
    /// assumption that a reused pane is still there); set to `Some(false)`
    /// to simulate the recorded pane having been closed while a sibling
    /// window keeps the tmux session alive.
    pub pane_exists_override: Mutex<Option<bool>>,
    /// #9004: the session the last pane lookup named, for `pane_identity`.
    pane_session: Mutex<Option<String>>,
    /// Controllable `pane_claude` answer (#8942); `Absent` when unset.
    pub pane_claude_override: Mutex<Option<crate::core::process::PaneClaude>>,
    /// Records every `send_line_to_pane` call as `(session_name, pane_id,
    /// text)` (sibling-window hijack fix, follow-up to #2456).
    pub pane_send_calls: Mutex<Vec<(String, String, String)>>,
    /// Records every `send_keys_literal_to_pane` call as `(session_name,
    /// pane_id, text)` (issue #2468).
    pub pane_literal_calls: Mutex<Vec<(String, String, String)>>,
    /// Records every `send_interrupt_to_pane` call as `(session_name,
    /// pane_id)` (issue #2468).
    pub pane_interrupt_calls: Mutex<Vec<(String, String)>>,
    /// Records every `capture_pane` call as `(session_name, pane_id)`; the
    /// returned body is looked up from `capture_responses` BY PANE ID (not
    /// session name), so a test can assert the pane-scoped path returns
    /// pane-owned content distinct from a session-keyed response (#2468).
    pub pane_capture_calls: Mutex<Vec<(String, String)>>,
    /// Records every `rename_session` call as `(old_name, new_name)`.
    pub rename_calls: Mutex<Vec<(String, String)>>,
    /// Names to report from `attached_session_names` (seed to simulate a
    /// client being attached to a session for `tm ls` reconciliation tests).
    pub attached_names: Mutex<Vec<String>>,
    /// Count of `ensure_server_up` calls (#3823) — asserts the
    /// server-up guard actually ran, and exactly once per create/resume.
    pub ensure_server_up_calls: Mutex<u32>,
    /// When `true`, `ensure_server_up` fails with `TmuxUnavailable` —
    /// simulates a machine where tmux has never run and the server cannot be
    /// started (#3823).
    pub ensure_server_up_should_fail: Mutex<bool>,
    /// When `true`, `list_sessions` fails with the EXACT cold-host stderr
    /// from #3886 until `ensure_server_up` has been called at least once —
    /// i.e. the socket directory exists but the socket does not, so every
    /// tmux call that is not `start-server` errors out. Reproduces the
    /// in-project "launch new session" failure on a fresh host.
    pub cold_tmux_host: Mutex<bool>,
    /// When `true`, `list_sessions` fails unconditionally with an
    /// already-typed `ManagedError::TmuxUnavailable` — the shape that let
    /// #3886's callers double the `tmux error:` prefix by re-wrapping it.
    pub list_sessions_should_fail: Mutex<bool>,
    /// Count of `apply_scrollback_options` calls (#6469) — asserts the boot
    /// reconcile re-asserts tm's server globals on a tmux server it did not
    /// start (a tmux-resurrect restore).
    pub scrollback_option_calls: Mutex<u32>,
    /// #9101: the store a rename test watches. `pane_identity` records in
    /// `probed_under_guard` whether that store's lock was held when it ran.
    pub watched_store: Mutex<Option<Arc<tokio::sync::RwLock<super::store::SessionStore>>>>,
    /// #9101: whether an ownership probe ran with `watched_store` locked.
    pub probed_under_guard: Mutex<bool>,
}

impl FakeTmuxDriver {
    pub fn new() -> Arc<Self> {
        Arc::new(Self {
            sessions: Mutex::new(HashMap::new()),
            send_calls: Mutex::new(Vec::new()),
            kill_calls: Mutex::new(Vec::new()),
            capture_responses: Mutex::new(HashMap::new()),
            seeded_names: Mutex::new(Vec::new()),
            create_cwd_calls: Mutex::new(Vec::new()),
            graceful_stop_calls: Mutex::new(Vec::new()),
            interrupt_calls: Mutex::new(Vec::new()),
            pane_cwd_override: Mutex::new(None),
            pane_id_override: Mutex::new(None),
            pane_exists_override: Mutex::new(None),
            pane_session: Mutex::new(None),
            pane_claude_override: Mutex::new(None),
            pane_send_calls: Mutex::new(Vec::new()),
            pane_literal_calls: Mutex::new(Vec::new()),
            pane_interrupt_calls: Mutex::new(Vec::new()),
            pane_capture_calls: Mutex::new(Vec::new()),
            rename_calls: Mutex::new(Vec::new()),
            attached_names: Mutex::new(Vec::new()),
            ensure_server_up_calls: Mutex::new(0),
            ensure_server_up_should_fail: Mutex::new(false),
            cold_tmux_host: Mutex::new(false),
            list_sessions_should_fail: Mutex::new(false),
            scrollback_option_calls: Mutex::new(0),
            watched_store: Mutex::new(None),
            probed_under_guard: Mutex::new(false),
        })
    }
}

impl ManagedTmuxDriver for FakeTmuxDriver {
    /// Records the call and optionally fails, simulating a fresh machine
    /// where the tmux server cannot be started (#3823).
    fn ensure_server_up(&self) -> Result<(), ManagedError> {
        *self.ensure_server_up_calls.lock().unwrap() += 1;
        if *self.ensure_server_up_should_fail.lock().unwrap() {
            return Err(ManagedError::TmuxUnavailable(
                "fake: tmux server absent and could not be started".into(),
            ));
        }
        Ok(())
    }

    /// Records the call so `reconcile_on_boot_reapplies_the_scrollback_server_options`
    /// can assert the daemon re-asserts tm's tmux globals at boot (#6469).
    fn apply_scrollback_options(&self) {
        *self.scrollback_option_calls.lock().unwrap() += 1;
    }

    fn create_session(&self, name: &str, workdir: &str) -> Result<(), ManagedError> {
        self.sessions
            .lock()
            .unwrap()
            .insert(name.to_owned(), workdir.to_owned());
        self.create_cwd_calls
            .lock()
            .unwrap()
            .push((name.to_owned(), workdir.to_owned()));
        Ok(())
    }

    fn kill_session(&self, name: &str) -> Result<(), ManagedError> {
        self.kill_calls.lock().unwrap().push(name.to_owned());
        self.sessions.lock().unwrap().remove(name);
        Ok(())
    }

    /// Rename a live session in the in-memory map and record the call, so a
    /// rename test can assert the tmux entity followed the record's new name.
    fn rename_session(&self, old: &str, new: &str) -> Result<(), ManagedError> {
        self.rename_calls
            .lock()
            .unwrap()
            .push((old.to_owned(), new.to_owned()));
        let mut sessions = self.sessions.lock().unwrap();
        if let Some(workdir) = sessions.remove(old) {
            sessions.insert(new.to_owned(), workdir);
        }
        Ok(())
    }

    /// Report the seeded attached-session names (empty by default), so a
    /// `tm ls` reconciliation test can simulate a client being attached.
    fn attached_session_names(&self) -> Vec<String> {
        self.attached_names.lock().unwrap().clone()
    }

    fn send_line(&self, name: &str, text: &str) -> Result<(), ManagedError> {
        self.send_calls
            .lock()
            .unwrap()
            .push((name.to_owned(), text.to_owned()));
        Ok(())
    }

    fn capture(&self, name: &str, _lines: usize) -> Result<String, ManagedError> {
        Ok(self
            .capture_responses
            .lock()
            .unwrap()
            .get(name)
            .cloned()
            .unwrap_or_default())
    }

    /// Honours the two #3886 failure simulations before answering: an
    /// unconditional typed failure, and the cold-host socket-absent failure
    /// that only `ensure_server_up` can clear.
    fn list_sessions(&self) -> Result<Vec<String>, ManagedError> {
        if *self.list_sessions_should_fail.lock().unwrap() {
            return Err(ManagedError::TmuxUnavailable(
                "protocol error: tmux list-sessions: fake failure".into(),
            ));
        }
        if *self.cold_tmux_host.lock().unwrap() && *self.ensure_server_up_calls.lock().unwrap() == 0
        {
            return Err(ManagedError::TmuxUnavailable(
                "protocol error: tmux list-sessions: error connecting to \
                 /private/tmp/tmux-501/default (No such file or directory)"
                    .into(),
            ));
        }
        let mut names: Vec<String> = self.sessions.lock().unwrap().keys().cloned().collect();
        // Also include seeded names (for reconcile tests that seed live sessions
        // without going through create_session).
        names.extend(self.seeded_names.lock().unwrap().iter().cloned());
        Ok(names)
    }

    /// Override graceful_stop to avoid the real 2-second sleep in unit tests.
    ///
    /// Why: the default impl blocks for 2 s which would make the test suite
    /// unacceptably slow; the fake records the call and delegates to kill_session
    /// synchronously — no sleep, no real signals.
    /// What: records `(name, claude_pid)` and calls `kill_session`. No SIGTERM.
    /// Test: `graceful_stop_sends_sigterm_then_kill`,
    /// `graceful_stop_skips_sigterm_when_no_pid`.
    fn graceful_stop(&self, name: &str, claude_pid: Option<u32>) -> Result<(), ManagedError> {
        self.graceful_stop_calls
            .lock()
            .unwrap()
            .push((name.to_owned(), claude_pid));
        self.kill_session(name)
    }

    /// Record Ctrl-C interrupts so the graceful-drain path (#1975) is observable.
    ///
    /// Why: the default `send_interrupt` fails loudly (`TmuxUnavailable`) to catch
    /// drivers that silently drop interrupts; the fake instead records the call and
    /// succeeds so `graceful_terminate_runtime`'s signal phase can be asserted.
    /// What: records `name` and returns `Ok`.
    fn send_interrupt(&self, name: &str) -> Result<(), ManagedError> {
        self.interrupt_calls.lock().unwrap().push(name.to_owned());
        Ok(())
    }

    /// Report the controllable `pane_cwd_override` (#2250) instead of the
    /// trait's silent `None` default, so tests can simulate a real driver
    /// confirming (or disagreeing with) the pane's actual cwd.
    fn get_pane_cwd(&self, _name: &str) -> Option<PathBuf> {
        self.pane_cwd_override.lock().unwrap().clone()
    }

    fn pane_claude(&self, _name: &str) -> crate::core::process::PaneClaude {
        self.pane_claude_override
            .lock()
            .unwrap()
            .unwrap_or(crate::core::process::PaneClaude::Absent)
    }

    /// Report the controllable `pane_id_override` (sibling-window hijack fix,
    /// follow-up to #2456) instead of the trait's silent `None` default, so
    /// tests can simulate a real driver having captured a `pane_id` at
    /// `create_session` time.
    fn get_pane_id(&self, name: &str) -> Option<String> {
        *self.pane_session.lock().unwrap() = Some(name.to_owned());
        self.pane_id_override.lock().unwrap().clone()
    }

    /// Report the controllable `pane_exists_override` (sibling-window hijack
    /// fix, follow-up to #2456) when set; otherwise fall through to the
    /// trait's optimistic `true` default — matches every existing test's
    /// implicit assumption that a reused pane is still there.
    fn pane_exists(&self, name: &str, _pane_id: &str) -> bool {
        *self.pane_session.lock().unwrap() = Some(name.to_owned());
        self.pane_exists_override.lock().unwrap().unwrap_or(true)
    }

    /// #9004: one fake server, [`FAKE_TMUX_SERVER`]; the pane sits in the
    /// session the last pane lookup named.
    fn pane_identity(
        &self,
        pane_id: &str,
    ) -> Result<super::pane_identity::PaneIdentity, ManagedError> {
        if let Some(store) = self.watched_store.lock().unwrap().as_ref()
            && store.try_write().is_err()
        {
            *self.probed_under_guard.lock().unwrap() = true;
        }
        Ok(super::pane_identity::PaneIdentity {
            pane_id: pane_id.to_owned(),
            session_id: FAKE_SESSION_ID.into(),
            server: FAKE_TMUX_SERVER.into(),
            session_name: self
                .pane_session
                .lock()
                .unwrap()
                .clone()
                .unwrap_or_default(),
        })
    }

    /// #9004: recorded in `kill_calls` as the `$N` id, never the name
    /// (#9101), so a test can tell a kill by id from a kill by name.
    fn kill_session_id(&self, name: &str, session_id: &str) -> Result<(), ManagedError> {
        self.kill_calls.lock().unwrap().push(session_id.to_owned());
        self.sessions.lock().unwrap().remove(name);
        Ok(())
    }

    /// #9101: recorded in `rename_calls` as `(session_id, new)`; the
    /// in-memory session `name` moves to `new`.
    fn rename_session_id(
        &self,
        name: &str,
        session_id: &str,
        new: &str,
    ) -> Result<(), ManagedError> {
        self.rename_calls
            .lock()
            .unwrap()
            .push((session_id.to_owned(), new.to_owned()));
        let mut sessions = self.sessions.lock().unwrap();
        if let Some(workdir) = sessions.remove(name) {
            sessions.insert(new.to_owned(), workdir);
        }
        Ok(())
    }

    /// Records `(name, pane_id, text)` instead of delegating to `send_line`
    /// (the trait default) — a test asserting `pane_send_calls` got the call
    /// and `send_calls` stayed empty is exactly what proves the manager chose
    /// the pane-scoped path over the session-scoped one.
    fn send_line_to_pane(&self, name: &str, pane_id: &str, text: &str) -> Result<(), ManagedError> {
        self.pane_send_calls.lock().unwrap().push((
            name.to_owned(),
            pane_id.to_owned(),
            text.to_owned(),
        ));
        Ok(())
    }

    /// Overrides the trait default (which errors "not implemented") so
    /// `inject`'s `Submit::NoSubmit` legacy (no `pane_id`) path is testable;
    /// records `(name, text)` into `send_calls` — the SAME session-scoped log
    /// `send_line` uses — so a legacy-record test can assert a session-scoped
    /// call landed without caring which literal-vs-Enter primitive it was.
    fn send_keys_literal(&self, name: &str, text: &str) -> Result<(), ManagedError> {
        self.send_calls
            .lock()
            .unwrap()
            .push((name.to_owned(), text.to_owned()));
        Ok(())
    }

    /// Records `(name, pane_id, text)` (issue #2468) — proves `inject`'s
    /// `Submit::NoSubmit` dispatch chose the pane-scoped path.
    fn send_keys_literal_to_pane(
        &self,
        name: &str,
        pane_id: &str,
        text: &str,
    ) -> Result<(), ManagedError> {
        self.pane_literal_calls.lock().unwrap().push((
            name.to_owned(),
            pane_id.to_owned(),
            text.to_owned(),
        ));
        Ok(())
    }

    /// Records `(name, pane_id)` (issue #2468) — proves `inject`'s
    /// `Submit::Interrupt` dispatch chose the pane-scoped path.
    fn send_interrupt_to_pane(&self, name: &str, pane_id: &str) -> Result<(), ManagedError> {
        self.pane_interrupt_calls
            .lock()
            .unwrap()
            .push((name.to_owned(), pane_id.to_owned()));
        Ok(())
    }

    /// Records `(name, pane_id)` and returns the response keyed BY PANE ID —
    /// deliberately a different keyspace than `capture`'s session-name
    /// lookup, so a test can prove `observe` returned pane-owned content
    /// rather than session-scoped content (issue #2468).
    fn capture_pane(
        &self,
        name: &str,
        pane_id: &str,
        _lines: usize,
    ) -> Result<String, ManagedError> {
        self.pane_capture_calls
            .lock()
            .unwrap()
            .push((name.to_owned(), pane_id.to_owned()));
        Ok(self
            .capture_responses
            .lock()
            .unwrap()
            .get(pane_id)
            .cloned()
            .unwrap_or_default())
    }
}

pub(super) async fn make_manager(dir: &TempDir) -> (SessionManager, Arc<FakeTmuxDriver>) {
    let fake = FakeTmuxDriver::new();
    let mgr = SessionManager::new(dir.path(), fake.clone())
        .await
        .expect("manager");
    (mgr, fake)
}

/// A [`FakeTmuxDriver`] whose new sessions carry pane `%1`, which its
/// `pane_exists` reports present, so a decommission proves the live session
/// is the record's own (#8935: an unproven one refuses).
pub(super) fn fake_with_pane() -> Arc<FakeTmuxDriver> {
    let fake = FakeTmuxDriver::new();
    *fake.pane_id_override.lock().unwrap() = Some("%1".into());
    fake
}

/// Bind `id`'s record to pane `%1`, which [`FakeTmuxDriver`] reports present
/// unless overridden, so a decommission proves the live session is the
/// record's own (#8935). For records written by [`seed_record`].
pub(super) async fn bind_pane(mgr: &SessionManager, id: &ManagedSessionId) {
    let mut record = mgr.get(id).await.expect("record");
    record.pane_id = Some("%1".into());
    record.tmux_server = Some(FAKE_TMUX_SERVER.into()); // #9004
    mgr.store
        .write()
        .await
        .upsert(record)
        .await
        .expect("bind pane");
}

/// [`make_manager`] over [`fake_with_pane`].
pub(super) async fn make_manager_with_pane(dir: &TempDir) -> (SessionManager, Arc<FakeTmuxDriver>) {
    let fake = fake_with_pane();
    let mgr = SessionManager::new(dir.path(), fake.clone())
        .await
        .expect("manager");
    (mgr, fake)
}

#[tokio::test]
async fn manager_create_record() {
    let dir = crate::test_support::hermetic_temp_dir();
    let (mgr, _fake) = make_manager(&dir).await;

    let record = mgr
        .create(
            "implement OAuth2".into(),
            Some(PathBuf::from("/tmp/wt1")),
            None,
            None,
            None,
            None,
        )
        .await
        .expect("create");

    assert!(
        record.tmux_name.starts_with("tm-"),
        "tmux_name has prefix: {}",
        record.tmux_name
    );
    assert_eq!(record.state, ManagedSessionState::Provisioning);
    assert_eq!(record.task, "implement OAuth2");

    let listed = mgr.list().await;
    assert_eq!(listed.len(), 1);
    assert_eq!(listed[0].id, record.id);
}

/// `known_tmux_names` must surface every stored session's tmux name so the
/// orphan-GC treats them all as protected (never an untracked orphan).
///
/// Why: the GC's safety depends on the store reporting its tracked names
/// completely; a name missing here would make a live, tracked session look like
/// an orphan and risk a false kill.
/// What: creates two sessions and asserts both their tmux names appear in the
/// returned set.
/// Test: this function IS the test.
#[tokio::test]
async fn manager_known_tmux_names_collects_all() {
    let dir = crate::test_support::hermetic_temp_dir();
    let (mgr, _fake) = make_manager(&dir).await;

    let r1 = mgr
        .create(
            "task one".into(),
            Some(PathBuf::from("/tmp/k1")),
            None,
            None,
            None,
            None,
        )
        .await
        .expect("create 1");
    let r2 = mgr
        .create(
            "task two".into(),
            Some(PathBuf::from("/tmp/k2")),
            None,
            None,
            None,
            None,
        )
        .await
        .expect("create 2");

    let names = mgr
        .known_tmux_names()
        .await
        .expect("store read should succeed on the happy path");
    assert!(names.contains(&r1.tmux_name), "missing {}", r1.tmux_name);
    assert!(names.contains(&r2.tmux_name), "missing {}", r2.tmux_name);
    assert_eq!(names.len(), 2);
}

/// A SUCCESSFUL create must DISARM the ownership guard so the freshly-created
/// tmux session is NOT reaped at the end of the request scope (#1453).
///
/// Why: the create path owns the new tmux session via a `TmuxSessionGuard` until
/// the record is persisted, then hands ownership to the store by disarming. If
/// disarm were ever dropped from the happy path, every created session would be
/// killed the instant `create_with_id` returned — a catastrophic regression.
/// This test pins disarm to the success path: no `kill_session` must occur.
/// What: creates a session through the manager (which persists successfully via
/// the temp store) and asserts the fake driver recorded ZERO kill calls.
/// Test: this function IS the test.
#[tokio::test]
async fn manager_create_success_does_not_reap_session() {
    let dir = crate::test_support::hermetic_temp_dir();
    let (mgr, fake) = make_manager(&dir).await;

    let record = mgr
        .create(
            "keep me alive".into(),
            Some(PathBuf::from("/tmp/keepalive")),
            None,
            None,
            None,
            None,
        )
        .await
        .expect("create");

    assert!(
        fake.kill_calls.lock().unwrap().is_empty(),
        "a successful create must NOT kill the session it created (guard disarmed); \
         kills seen: {:?}",
        fake.kill_calls.lock().unwrap()
    );
    // And the session is still tracked.
    assert_eq!(mgr.get(&record.id).await.unwrap().id, record.id);
}

/// When the store write FAILS, the orphaned tmux session must be reaped
/// (#1453, #1457).
///
/// Why: this is the exact failure that produced 159 orphans (#1452). If
/// `create_session` succeeds but the subsequent `store.upsert` fails, the tmux
/// session has no owner in the registry — the armed `TmuxSessionGuard` must drop
/// and kill it so it cannot accumulate as an orphan. #1457 additionally logs the
/// rollback at warn so it is visible in the daemon's stderr; this test pins the
/// reap behaviour the log narrates.
/// What: makes the store's DIRECTORY read-only so the staging write inside
/// `save()` fails, drives a create, asserts the create returns a `Store` error
/// AND that the fake driver recorded a `kill_session` for the exact tmux name
/// that was created.
///
/// #5007 changed the sabotage, not the behaviour under test. It used to replace
/// `sessions.json` with a non-empty DIRECTORY so the rename failed — but a read
/// error is no longer swallowed as "empty store", so that sabotage now fails
/// the reload at the top of `upsert`, before `create_session` has made anything
/// to orphan. A read-only parent leaves the read succeeding (the file is simply
/// absent) and fails only the WRITE, which is the sequence this test exists to
/// cover.
/// Test: this function IS the test.
#[tokio::test]
async fn manager_create_store_failure_reaps_orphan() {
    use std::os::unix::fs::PermissionsExt;

    let dir = crate::test_support::hermetic_temp_dir();
    let (mgr, fake) = make_manager(&dir).await;

    // Sabotage the store: make its directory read-only, so `save()`'s staging
    // write fails with EACCES AFTER `create_session` already created the tmux
    // session — the orphan case.
    let restore = std::fs::metadata(dir.path())
        .expect("stat data dir")
        .permissions();
    std::fs::set_permissions(dir.path(), std::fs::Permissions::from_mode(0o500))
        .expect("make the store directory read-only");

    let err = mgr
        .create(
            "doomed".into(),
            Some(PathBuf::from("/tmp/doomed-ws")),
            None,
            None,
            None,
            None,
        )
        .await
        .expect_err("create must fail when the store write fails");

    // Restore permissions immediately so the temp dir can be cleaned up.
    std::fs::set_permissions(dir.path(), restore).expect("restore permissions");

    assert!(
        matches!(err, ManagedError::Store(_)),
        "store-write failure must surface as a Store error, got {err:?}"
    );

    // The orphaned tmux session must have been reaped by the guard's Drop.
    let kills = fake.kill_calls.lock().unwrap();
    assert_eq!(
        kills.len(),
        1,
        "exactly one orphaned tmux session must be reaped; kills: {kills:?}"
    );
    assert!(
        kills[0].starts_with("tm-"),
        "the reaped session is the one just created (tm- prefix): {}",
        kills[0]
    );
}

// `manager_naming_convention` (full `tm-<leaf>-NN` format assertion) lives in
// `naming_tests.rs` alongside the rest of the #1955 naming-scheme coverage —
// this file is at the 1500-SLOC test cap (see naming_tests.rs's module doc).

#[tokio::test]
async fn manager_name_hint_overrides() {
    let dir = crate::test_support::hermetic_temp_dir();
    let (mgr, _fake) = make_manager(&dir).await;

    let record = mgr
        .create(
            "task".into(),
            None,
            Some("ticket-1234".into()),
            None,
            None,
            None,
        )
        .await
        .expect("create");

    assert_eq!(record.tmux_name, "tm-ticket-1234-01");
}

/// `stop` must kill the runtime but KEEP the workspace directory and record,
/// setting state to `Stopped` (not `Dead` or any terminal state).
///
/// Why: a session ENDURES beyond its running runtime; stop is non-destructive.
/// What: creates a session with a temp workspace dir, stops it, asserts state
/// is `Stopped`, workspace dir still exists, tmux kill was called.
/// Test: this function IS the test.
#[tokio::test]
async fn manager_stop_keeps_workspace() {
    let dir = crate::test_support::hermetic_temp_dir();
    let workspace_dir = crate::test_support::hermetic_temp_dir();
    let (mgr, fake) = make_manager(&dir).await;
    // #8935: a teardown kills only a session proved to hold the record's pane.
    *fake.pane_id_override.lock().unwrap() = Some("%1".into());

    let record = mgr
        .create(
            "task".into(),
            Some(workspace_dir.path().to_owned()),
            None,
            Some(workspace_dir.path().to_owned()),
            None,
            None,
        )
        .await
        .expect("create");

    let stopped = mgr.stop(&record.id).await.expect("stop");

    // State must be Stopped (runtime gone) not Dead (which implied loss).
    assert_eq!(stopped.state, ManagedSessionState::Stopped);

    // tmux session must have been killed — #9101: by its session id.
    assert_eq!(
        fake.kill_calls.lock().unwrap().as_slice(),
        [FAKE_SESSION_ID.to_string()]
    );

    // Workspace directory must STILL EXIST on disk.
    assert!(
        workspace_dir.path().exists(),
        "workspace dir must survive a stop; it is still on disk for resume"
    );

    // workspace_path field must still be set in the persisted record.
    let after = mgr.get(&record.id).await.unwrap();
    assert_eq!(after.state, ManagedSessionState::Stopped);
    assert!(
        after.workspace_path.is_some(),
        "workspace_path must be preserved in the record after stop"
    );
}

/// `resume` must create a NEW tmux session rooted at the EXISTING workspace,
/// NOT re-clone the repository.
///
/// Why: workspace is provisioned once; resume only re-spawns the runtime.
/// What: creates a session with a workspace dir, stops it, resumes it, and
/// asserts: (a) a second `create_session` call was issued, (b) its cwd equals
/// the original workspace_path, (c) state transitions to Active.
/// Test: this function IS the test.
#[tokio::test]
async fn manager_resume_respawns_in_existing_workspace() {
    let dir = crate::test_support::hermetic_temp_dir();
    let workspace_dir = crate::test_support::hermetic_temp_dir();
    let (mgr, fake) = make_manager(&dir).await;
    // #8935: a teardown kills only a session proved to hold the record's pane.
    *fake.pane_id_override.lock().unwrap() = Some("%1".into());

    let workspace_path = workspace_dir.path().to_owned();

    let record = mgr
        .create(
            "task".into(),
            Some(workspace_path.clone()),
            Some("my-session".into()),
            Some(workspace_path.clone()),
            Some("https://github.com/owner/repo".into()),
            Some("main".into()),
        )
        .await
        .expect("create");

    // Stop the session.
    mgr.stop(&record.id).await.expect("stop");

    // Record the create_session call count before resume.
    let creates_before = fake.create_cwd_calls.lock().unwrap().len();

    // Resume the session.
    let resumed = mgr.resume(&record.id).await.expect("resume");

    // State must be Active.
    assert_eq!(resumed.state, ManagedSessionState::Active);

    // A NEW create_session must have been issued.
    // Drop the lock guard before the next await point.
    let (create_len, resume_cwd) = {
        let create_calls = fake.create_cwd_calls.lock().unwrap();
        let len = create_calls.len();
        let cwd = create_calls
            .get(creates_before)
            .map(|c| c.1.clone())
            .unwrap_or_default();
        (len, cwd)
    };
    assert!(
        create_len > creates_before,
        "resume must issue a new tmux create_session call"
    );

    // The new create_session must use the EXISTING workspace as cwd.
    assert_eq!(
        resume_cwd,
        workspace_path.to_string_lossy().to_string(),
        "resume must use the EXISTING workspace path as cwd, not re-clone"
    );

    // workspace_path field must still be set (no re-clone).
    let after = mgr.get(&record.id).await.unwrap();
    assert_eq!(
        after.workspace_path.as_deref(),
        Some(workspace_path.as_path()),
        "workspace_path must be preserved after resume (no re-clone)"
    );
}

/// `decommission` removes an SM-OWNED workspace directory and sets state to
/// `Decommissioned`, but keeps a tombstone record.
///
/// Why: decommission is the ONLY teardown that removes disk artifacts for
/// SM-provisioned (clone-based) sessions. The workspace must only be deleted
/// when `workspace_owned = true` AND the path is inside the managed root —
/// this test exercises the owned path by pointing the workspace inside a temp
/// dir that serves as the managed root (#1511).
/// What: creates a session, marks it workspace_owned=true, uses a temp dir as
/// the managed root (via env override), decommissions it, asserts the workspace
/// dir is gone from disk and the record state is `Decommissioned` with
/// `workspace_path = None`.
/// Test: this function IS the test.
#[tokio::test]
async fn manager_decommission_removes_workspace() {
    let dir = crate::test_support::hermetic_temp_dir();
    // #8935: the record's pane proves the live session is its own.
    let (mgr, _fake) = make_manager_with_pane(&dir).await;

    // Build a workspace path INSIDE a temp "managed root" dir so the
    // path-containment guard passes. `decommission_with_root` is called with
    // the managed root injected directly — no env var mutation required.
    let managed_root = crate::test_support::hermetic_temp_dir();
    let workspace_path = managed_root
        .path()
        .join("owner")
        .join("repo")
        .join("abc-session-id");
    std::fs::create_dir_all(&workspace_path).unwrap();
    // Write a file so we can verify the dir was removed. #8663: build output,
    // since a user file would (correctly) keep the workspace.
    std::fs::create_dir_all(workspace_path.join("target")).unwrap();
    std::fs::write(workspace_path.join("target/sentinel.txt"), "exists").unwrap();

    // Create the session with `owned=true` (atomic: mirrors what `spawn_managed`
    // does for clone-provisioned workspaces after Fix 1).
    let record = mgr
        .create_with_id(
            ManagedSessionId::new(),
            "task".into(),
            Some(workspace_path.clone()),
            None,
            Some(workspace_path.clone()),
            None,
            None,
            crate::runtime::RuntimeKind::default(),
            false,
            true, // owned: SM provisioned via clone
        )
        .await
        .expect("create");

    // Decommission using the injectable root — no unsafe env mutation.
    let (tombstone, workspace_removed) = mgr
        .decommission_with_root(&record.id, managed_root.path(), None)
        .await
        .expect("decommission");

    // decommission_with_root must report that it actually removed the directory.
    assert!(
        workspace_removed,
        "workspace_removed must be true for an owned, existing workspace"
    );

    // State must be Decommissioned.
    assert_eq!(tombstone.state, ManagedSessionState::Decommissioned);

    // workspace_path must be cleared in the tombstone record.
    assert!(
        tombstone.workspace_path.is_none(),
        "workspace_path must be None after decommission (workspace was deleted)"
    );

    // Workspace directory MUST be gone from disk.
    // Note: TempDir's Drop won't fail if the dir is already removed.
    assert!(
        !workspace_path.exists(),
        "workspace directory must be removed from disk after decommission"
    );

    // Tombstone record must still be queryable (for `ls` history).
    let after = mgr.get(&record.id).await.unwrap();
    assert_eq!(after.state, ManagedSessionState::Decommissioned);
    assert!(after.workspace_path.is_none());
}

/// `decommission_record_only` (owner request 2026-07-29, critic HIGH finding
/// #1) NEVER removes filesystem content — a remount-race stand-in: the
/// workspace directory genuinely EXISTS (and holds real content) at call
/// time, exactly as it would if a network/removable volume remounted between
/// the `tm ls` auto-prune's listing-time probe and this call. The directory
/// and its content must survive untouched, even though the record IS
/// tombstoned to `Decommissioned`.
///
/// Why: this is the auto-prune sweep's actual safety net — [`is_unresumable`]
/// can only observe "absent right now," never "will stay absent," so the
/// call that ACTS on that observation must be structurally incapable of
/// deleting anything, not merely re-check-and-hope.
/// What: same fixture as [`manager_decommission_removes_workspace`] (a real,
/// owned, populated workspace dir), but calls `decommission_with_root` with
/// `record_only = true`. Asserts `workspace_removed == false`, the directory
/// AND its sentinel file still exist on disk, the tombstone is
/// `Decommissioned`, and — because the path is STILL on disk — the record's
/// `workspace_path` is retained (not nulled), mirroring the #4344
/// dirty-retained boundary.
/// Test: this function IS the test.
#[tokio::test]
async fn decommission_record_only_never_removes_existing_workspace() {
    let dir = crate::test_support::hermetic_temp_dir();
    let (mgr, _fake) = make_manager(&dir).await;

    let managed_root = crate::test_support::hermetic_temp_dir();
    let workspace_path = managed_root
        .path()
        .join("owner")
        .join("repo")
        .join("remount-race-session-id");
    std::fs::create_dir_all(&workspace_path).unwrap();
    std::fs::write(workspace_path.join("sentinel.txt"), "still here").unwrap();

    let record = mgr
        .create_with_id(
            ManagedSessionId::new(),
            "task".into(),
            Some(workspace_path.clone()),
            None,
            Some(workspace_path.clone()),
            None,
            None,
            crate::runtime::RuntimeKind::default(),
            false,
            true, // owned: SM provisioned via clone
        )
        .await
        .expect("create");

    let (tombstone, workspace_removed) = mgr
        .decommission_record_only(&record.id)
        .await
        .expect("record-only decommission");

    assert!(
        !workspace_removed,
        "record_only must never report a removal"
    );
    assert_eq!(tombstone.state, ManagedSessionState::Decommissioned);
    assert!(
        workspace_path.exists(),
        "the workspace directory must survive a record-only decommission"
    );
    assert!(
        workspace_path.join("sentinel.txt").exists(),
        "real content inside the workspace must survive untouched"
    );
    assert_eq!(
        tombstone.workspace_path.as_deref(),
        Some(workspace_path.as_path()),
        "workspace_path must be RETAINED (not nulled) since the directory is \
         still genuinely on disk — mirrors the #4344 dirty-retained boundary"
    );
}

/// 🔴 A record-only decommission must not touch the RUNTIME either (#4702).
///
/// Why: `record_only` was named and documented as "never touches the
/// filesystem," and every caller — including the shipped `tm ls` auto-prune
/// (#4384) — read that as "safe to run unattended on every listing." It was
/// only half true. `graceful_terminate_runtime` sat ABOVE the `record_only`
/// guard in `decommission_with_root` and ran unconditionally: it SIGTERMed the
/// claude process and `kill_session`ed the pane. Its self-guard is
/// `session_exists(name)` — plain LIVE-TMUX NAME MEMBERSHIP, never the record's
/// captured `pane_id` — so it killed whatever live session happened to carry
/// this record's name, routing around the pane-scoped containment #3714 added
/// for display. The sibling test above proves the filesystem half and passed
/// throughout; it never seeded a live tmux session, so it could not see this.
///
/// What: seeds `FakeTmuxDriver::seeded_names` with the record's `tmux_name` so
/// `session_exists` reports it LIVE — the exact precondition that made the old
/// code signal and kill — then drives `decommission_with_root(…, record_only =
/// true)`. Asserts on the DRIVER'S CALL LOG (`interrupt_calls`, `kill_calls`,
/// `graceful_stop_calls` all empty), not on an observable outcome: a test that
/// checked only "the record is tombstoned" would pass against the buggy code.
/// Test: this function IS the test.
#[tokio::test]
async fn decommission_record_only_never_touches_the_runtime() {
    let dir = crate::test_support::hermetic_temp_dir();
    let (mgr, fake) = make_manager(&dir).await;

    let managed_root = crate::test_support::hermetic_temp_dir();
    let workspace_path = managed_root.path().join("owner").join("repo").join("live");
    std::fs::create_dir_all(&workspace_path).unwrap();

    let record = mgr
        .create_with_id(
            ManagedSessionId::new(),
            "task".into(),
            Some(workspace_path.clone()),
            None,
            Some(workspace_path.clone()),
            None,
            None,
            crate::runtime::RuntimeKind::default(),
            false,
            true,
        )
        .await
        .expect("create");

    // Make the record's name resolve LIVE, so `graceful_terminate_runtime`'s
    // `session_exists` fast-path does NOT short-circuit. Without this the test
    // would pass vacuously against the buggy code.
    fake.seeded_names
        .lock()
        .unwrap()
        .push(record.tmux_name.clone());
    assert!(
        fake.session_exists(&record.tmux_name),
        "precondition: the driver must report this session LIVE, otherwise the \
         helper's own fast-path makes this test vacuous"
    );

    mgr.decommission_record_only(&record.id)
        .await
        .expect("record-only decommission");

    // #8935: the teardown's Ctrl-C now targets the record's pane.
    assert!(
        fake.interrupt_calls.lock().unwrap().is_empty()
            && fake.pane_interrupt_calls.lock().unwrap().is_empty(),
        "record-only must never signal the runtime; got {:?}",
        fake.interrupt_calls.lock().unwrap()
    );
    assert!(
        fake.kill_calls.lock().unwrap().is_empty(),
        "record-only must never reclaim the pane — this is the call that killed \
         a live session sharing the record's name; got {:?}",
        fake.kill_calls.lock().unwrap()
    );
    assert!(
        fake.graceful_stop_calls.lock().unwrap().is_empty(),
        "record-only must never issue a graceful-stop; got {:?}",
        fake.graceful_stop_calls.lock().unwrap()
    );
}

/// 🔴 Why the record-only path may not reuse `decommission_with_root`'s
/// search-index cleanup (PR #4725 review round 2).
///
/// Why: `decommission_with_root` derives its index id with
/// `disposable_workspace_index_id`, whose own doc states it MUST run before any
/// removal, because `resolve_project_root` walks UP for a `.git` marker — once
/// the workspace is gone the walk keeps climbing and lands on an ancestor. The
/// auto-prune sweep only ever fires for a workspace that is ALREADY gone, so it
/// can never satisfy that precondition; reusing the shared teardown would issue
/// a cross-daemon `DELETE /indexes/{id}` against the PARENT PROJECT's index.
/// This test pins that hazard directly so nobody "simplifies" the record-only
/// path back onto the shared function.
///
/// What: one git repo (the parent project) containing a `.worktrees/<leaf>`
/// path that does NOT exist on disk — the exact shape of a GC'd session
/// worktree. Asserts the derived id equals the id derived for the PARENT ROOT,
/// i.e. deleting it would destroy the parent project's index.
/// Test: this function IS the test.
#[test]
fn disposable_index_id_for_a_removed_worktree_resolves_to_the_parent_project() {
    let dir = crate::test_support::hermetic_temp_dir();
    let project_root = dir.path().join("my-project");
    std::fs::create_dir_all(project_root.join(".git")).unwrap();

    // The session worktree that auto-prune would be acting on: already gone.
    let gone_worktree = project_root.join(".worktrees").join("tm-my-project-01");
    assert!(
        !gone_worktree.exists(),
        "precondition: the worktree must be absent, which is when auto-prune fires"
    );

    let derived = super::search_gc::disposable_workspace_index_id(Some(&gone_worktree), true)
        .expect("an owned workspace always derives some id");
    let parent_id = trusty_common::derive_index_id(&project_root);

    assert_eq!(
        derived, parent_id,
        "a removed worktree derives the PARENT project's index id — this is why \
         the record-only path must never run the shared teardown's index cleanup"
    );
}

/// 🔴 The enumeration test: `decommission_record_only` has NO side effect
/// outside the record store (PR #4725 review round 2).
///
/// Why: two review rounds each found a different ungated destructive effect in
/// `decommission_with_root` (#4728's runtime kill, then the search-index
/// delete). The response was to stop threading a flag through a multi-effect
/// function and give the sweep its own single-effect one. This test is the
/// standing guard on that decision: it drives the record-only path against a
/// fixture where EVERY effect in `decommission_with_root`'s side-effect table
/// would be observable, and asserts none of them fired.
///
/// What: a real, populated, SM-owned workspace, plus a live tmux session
/// carrying the record's name. Asserts the directory and its content survive
/// (effects 2–3), no runtime signal or kill was issued (effect 1), and the
/// record IS tombstoned with its pointer retained because the directory is
/// still there (effects 5–7). Effect 4 (the cross-daemon index DELETE) is
/// covered structurally — this function contains no `search_gc` call at all —
/// and by the sibling test above, which pins why calling it here would be
/// wrong. It is deliberately NOT exercised against the real resolver: on a
/// machine with a live trusty-search daemon that would issue a genuine DELETE,
/// which is the exact hazard under review.
/// Test: this function IS the test.
#[tokio::test]
async fn decommission_record_only_has_no_side_effects_beyond_the_store() {
    let dir = crate::test_support::hermetic_temp_dir();
    let (mgr, fake) = make_manager(&dir).await;

    let managed_root = crate::test_support::hermetic_temp_dir();
    let workspace_path = managed_root.path().join("owner").join("repo").join("sess");
    std::fs::create_dir_all(&workspace_path).unwrap();
    std::fs::write(workspace_path.join("uncommitted.txt"), "real work").unwrap();

    let record = mgr
        .create_with_id(
            ManagedSessionId::new(),
            "task".into(),
            Some(workspace_path.clone()),
            None,
            Some(workspace_path.clone()),
            None,
            None,
            crate::runtime::RuntimeKind::default(),
            false,
            true, // owned — the full path WOULD remove_dir_all this
        )
        .await
        .expect("create");

    fake.seeded_names
        .lock()
        .unwrap()
        .push(record.tmux_name.clone());

    let (tombstone, workspace_removed) = mgr
        .decommission_record_only(&record.id)
        .await
        .expect("record-only");

    assert!(!workspace_removed);
    assert!(
        workspace_path.join("uncommitted.txt").exists(),
        "effects 2-3: real content must survive"
    );
    assert!(
        fake.kill_calls.lock().unwrap().is_empty()
            && fake.interrupt_calls.lock().unwrap().is_empty()
            && fake.pane_interrupt_calls.lock().unwrap().is_empty()
            && fake.graceful_stop_calls.lock().unwrap().is_empty(),
        "effect 1: the runtime must be untouched"
    );
    assert_eq!(tombstone.state, ManagedSessionState::Decommissioned);
    assert_eq!(
        tombstone.workspace_path.as_deref(),
        Some(workspace_path.as_path()),
        "effect 5: a workspace still on disk keeps its pointer (#4344)"
    );
}

/// The full (non-record-only) decommission still tears the runtime down
/// (#1975) — the #4702 gate must not silently disable it for everyone.
///
/// Why: gating `graceful_terminate_runtime` behind `!record_only` is only
/// correct if the normal path is unaffected. Without this test the gate could
/// be inverted or over-applied and nothing would notice.
/// What: identical fixture to
/// [`decommission_record_only_never_touches_the_runtime`], but `record_only =
/// false`. Asserts the pane IS reclaimed.
/// Test: this function IS the test.
#[tokio::test]
async fn decommission_full_still_terminates_the_runtime() {
    let dir = crate::test_support::hermetic_temp_dir();
    let (mgr, fake) = make_manager(&dir).await;
    // #8935: a teardown kills only a session proved to hold the record's pane.
    *fake.pane_id_override.lock().unwrap() = Some("%1".into());

    let managed_root = crate::test_support::hermetic_temp_dir();
    let workspace_path = managed_root.path().join("owner").join("repo").join("full");
    std::fs::create_dir_all(&workspace_path).unwrap();

    let record = mgr
        .create_with_id(
            ManagedSessionId::new(),
            "task".into(),
            Some(workspace_path.clone()),
            None,
            Some(workspace_path.clone()),
            None,
            None,
            crate::runtime::RuntimeKind::default(),
            false,
            true,
        )
        .await
        .expect("create");

    fake.seeded_names
        .lock()
        .unwrap()
        .push(record.tmux_name.clone());

    mgr.decommission_with_root(&record.id, managed_root.path(), None)
        .await
        .expect("full decommission");

    // #9101: by session id, which the fake records instead of the name.
    assert_eq!(
        *fake.kill_calls.lock().unwrap(),
        vec![FAKE_SESSION_ID.to_string()],
        "the FULL decommission path must still reclaim the pane (#1975)"
    );
}

// Build a minimal Active session record for reconcile tests (avoids repeating
// 20-field struct literals when only tmux_name / task / ws_path vary).
// `pub(super)` since #6116: the sibling `naming_tests` reconcile coverage seeds
// the same shape, and a second copy of a 27-field literal is one field addition
// away from drifting. Widened to `pub(crate)` by #7259, for the same reason:
// the doctor's worktree-disk regression seeds a tombstoned record and would
// otherwise carry a third copy.
pub(crate) fn make_active_test_record(tmux_name: &str, task: &str, ws_path: &str) -> SessionRecord {
    SessionRecord {
        id: ManagedSessionId::new(),
        tmux_name: tmux_name.into(),
        cwd: PathBuf::from("/tmp"),
        task: task.into(),
        state: ManagedSessionState::Active,
        created_at: Utc::now(),
        last_activity_at: None,
        workspace_path: Some(PathBuf::from(ws_path)),
        repo_url: None,
        branch: None,
        pending_decision: None,
        proposed_default: None,
        correlation: Default::default(),
        runtime: Default::default(),
        ephemeral: false,
        workspace_owned: false,
        source_id: None,
        claude_session_id: None,
        scrollback_path: None,
        last_cwd: None,
        deliverable_id: None,
        pane_id: None,
        tmux_server: None,
        injection_status: Default::default(),
        worktree_owner: None,
        terminal_at: None,
        stop_cause: None,
        kind: Default::default(),
    }
}

/// Reconciliation with a gone tmux session must yield `Stopped` (resumable),
/// not `Orphaned` or `Dead` (both imply the session is lost).
///
/// Why: a gone tmux after a daemon restart means the RUNTIME stopped, not
/// the SESSION. The record and workspace are intact and resumable.
/// What: seeds a live tmux session for one record and no live session for
/// another (simulating reboot), runs reconcile, asserts: live → Active,
/// gone → Stopped (not Orphaned).
/// Test: this function IS the test.
/// #6469: the boot reconcile re-asserts tm's tmux server globals.
///
/// Why: a tmux server restart with tmux-continuum restore recreates every
/// `tm-*` session through resurrect's own bare `new-session`, which never
/// passes through `create_managed_session` — so the server carries none of the
/// `history-limit`/`mouse`/`alternate-screen` options tm specifies, and every
/// restored pane sits on the factory 2000-line scrollback. Boot reconcile is
/// the one pass that runs against a server tm did not start.
/// What: runs `reconcile_on_boot` against a fake driver that counts the calls,
/// and asserts the options were applied exactly once for the pass.
/// Test: this function IS the test.
#[tokio::test]
async fn reconcile_on_boot_reapplies_the_scrollback_server_options() {
    let dir = crate::test_support::hermetic_temp_dir();
    let fake = FakeTmuxDriver::new();
    fake.seeded_names
        .lock()
        .unwrap()
        .push("tm-restored-01".into());
    let mgr = SessionManager::new(dir.path(), fake.clone()).await.unwrap();

    mgr.reconcile_on_boot(false).await.expect("reconcile");

    assert_eq!(
        *fake.scrollback_option_calls.lock().unwrap(),
        1,
        "boot reconcile must re-apply tm's tmux server globals (#6469) — a \
         tmux-resurrect restore leaves them unset"
    );
}

#[tokio::test]
async fn manager_reconcile_gone_tmux_yields_stopped() {
    let dir = crate::test_support::hermetic_temp_dir();
    let fake = FakeTmuxDriver::new();

    // Seed a live tmux session.
    fake.seeded_names
        .lock()
        .unwrap()
        .push("tmpm-live-session".into());

    let mgr = SessionManager::new(dir.path(), fake.clone()).await.unwrap();

    let live_record = make_active_test_record("tmpm-live-session", "live task", "/tmp/live-ws");
    // A record whose tmux session will NOT be found (simulating reboot).
    let rebooted_record =
        make_active_test_record("tmpm-rebooted-session", "rebooted task", "/tmp/rebooted-ws");
    {
        let mut store = mgr.store.write().await;
        store.upsert(live_record.clone()).await.unwrap();
        store.upsert(rebooted_record.clone()).await.unwrap();
    }

    let report = mgr.reconcile_on_boot(false).await.expect("reconcile");
    assert!(report.adopted.contains(&"tmpm-live-session".to_string()));
    // The gone session must be in the `stopped` list, NOT `orphaned`.
    assert!(
        report.stopped.contains(&rebooted_record.id.to_string()),
        "gone session must be in report.stopped; report: {:?}",
        report
    );

    // Live session → Active.
    let live = mgr.get(&live_record.id).await.unwrap();
    assert_eq!(live.state, ManagedSessionState::Active);

    // Gone session → Stopped (RESUMABLE), workspace_path preserved.
    let rebooted = mgr.get(&rebooted_record.id).await.unwrap();
    assert_eq!(
        rebooted.state,
        ManagedSessionState::Stopped,
        "gone-tmux session must be Stopped (resumable), not Orphaned or Dead"
    );
    assert!(
        rebooted.workspace_path.is_some(),
        "workspace_path must be preserved after reconcile→Stopped"
    );
}

/// Decommissioned tombstones are not touched by reconciliation.
///
/// Why: a decommissioned session has no workspace and no tmux; reconciliation
/// must not try to resurrect or re-stop it.
/// Test: this function IS the test.
#[tokio::test]
async fn manager_reconcile_skips_decommissioned() {
    let dir = crate::test_support::hermetic_temp_dir();
    let fake = FakeTmuxDriver::new();
    let mgr = SessionManager::new(dir.path(), fake.clone()).await.unwrap();

    let tombstone = SessionRecord {
        id: ManagedSessionId::new(),
        tmux_name: "tmpm-decomm".into(),
        cwd: PathBuf::from("/tmp"),
        task: "done task".into(),
        state: ManagedSessionState::Decommissioned,
        created_at: Utc::now(),
        last_activity_at: None,
        workspace_path: None,
        repo_url: None,
        branch: None,
        pending_decision: None,
        proposed_default: None,
        correlation: Default::default(),
        runtime: Default::default(),
        ephemeral: false,
        workspace_owned: false,
        source_id: None,
        claude_session_id: None,
        scrollback_path: None,
        last_cwd: None,
        deliverable_id: None,
        pane_id: None,
        tmux_server: None,
        injection_status: Default::default(),
        worktree_owner: None,
        terminal_at: None,
        stop_cause: None,
        kind: Default::default(),
    };
    {
        let mut store = mgr.store.write().await;
        store.upsert(tombstone.clone()).await.unwrap();
    }

    let report = mgr.reconcile_on_boot(false).await.expect("reconcile");

    // Tombstone must not appear in adopted or stopped lists.
    assert!(!report.adopted.contains(&tombstone.tmux_name));
    assert!(!report.stopped.contains(&tombstone.id.to_string()));

    // State must remain Decommissioned after reconcile.
    let after = mgr.get(&tombstone.id).await.unwrap();
    assert_eq!(after.state, ManagedSessionState::Decommissioned);
}

/// A soft-deleted (`Deleted`) record must survive daemon boot/reconcile
/// without being resurrected — the same terminal-tombstone guarantee as
/// `Decommissioned`.
///
/// Why: `reconcile_on_boot` used to hand-roll
/// `matches!(record.state, ManagedSessionState::Decommissioned)` instead of
/// asking the record's own `is_terminal()`, so a `Deleted` record (no live
/// tmux session, by definition) fell through to the "gone" branch and was
/// resurrected to `Stopped` on every daemon boot. `is_terminal()` covers
/// BOTH `Decommissioned` and `Deleted`, so this is the regression test for
/// the second terminal variant the hand-rolled check missed.
/// Test: this function IS the test.
#[tokio::test]
async fn manager_reconcile_skips_deleted() {
    let dir = crate::test_support::hermetic_temp_dir();
    let fake = FakeTmuxDriver::new();
    let mgr = SessionManager::new(dir.path(), fake.clone()).await.unwrap();

    let deleted = SessionRecord {
        id: ManagedSessionId::new(),
        tmux_name: "tmpm-deleted".into(),
        cwd: PathBuf::from("/tmp"),
        task: "deleted task".into(),
        state: ManagedSessionState::Deleted,
        created_at: Utc::now(),
        last_activity_at: None,
        workspace_path: None,
        repo_url: None,
        branch: None,
        pending_decision: None,
        proposed_default: None,
        correlation: Default::default(),
        runtime: Default::default(),
        ephemeral: false,
        workspace_owned: false,
        source_id: None,
        claude_session_id: None,
        scrollback_path: None,
        last_cwd: None,
        deliverable_id: None,
        pane_id: None,
        tmux_server: None,
        injection_status: Default::default(),
        worktree_owner: None,
        terminal_at: None,
        stop_cause: None,
        kind: Default::default(),
    };
    {
        let mut store = mgr.store.write().await;
        store.upsert(deleted.clone()).await.unwrap();
    }

    let report = mgr.reconcile_on_boot(false).await.expect("reconcile");

    // Tombstone must not appear in adopted or stopped lists — a resurrection
    // would show up as this record's id in `report.stopped`.
    assert!(!report.adopted.contains(&deleted.tmux_name));
    assert!(
        !report.stopped.contains(&deleted.id.to_string()),
        "a Deleted record must never be resurrected to Stopped by reconcile; report: {:?}",
        report
    );

    // State must remain Deleted after reconcile — never resurrected.
    let after = mgr.get(&deleted.id).await.unwrap();
    assert_eq!(
        after.state,
        ManagedSessionState::Deleted,
        "a Deleted record must survive daemon boot/reconcile without being resurrected"
    );
}

/// #4400 backfill: a terminal (`Decommissioned`) record persisted BEFORE the
/// decommission-path fix landed — still carrying a stale `pending_decision`/
/// `proposed_default` — must have both cleared by the next boot reconcile,
/// and the clear must be reported in `ReconcileReport::stale_decisions_cleared`.
///
/// Why: `supervisor_status`'s human-confirmation queue must not accumulate
/// phantom T4 gates from rows that predate the fix; the reconcile-on-boot
/// sweep is the one-time backfill mechanism for those existing rows.
/// What: seeds a Decommissioned record with `pending_decision` set, runs
/// `reconcile_on_boot`, asserts both fields are cleared, the state stays
/// Decommissioned (not resurrected), and the id appears in the report.
#[tokio::test]
async fn manager_reconcile_backfills_stale_pending_decision_on_terminal_record() {
    let dir = crate::test_support::hermetic_temp_dir();
    let fake = FakeTmuxDriver::new();
    let mgr = SessionManager::new(dir.path(), fake.clone()).await.unwrap();

    let stale = SessionRecord {
        id: ManagedSessionId::new(),
        tmux_name: "tmpm-stale-decision".into(),
        cwd: PathBuf::from("/tmp"),
        task: "throwaway verification - will be decommissioned immediately".into(),
        state: ManagedSessionState::Decommissioned,
        created_at: Utc::now(),
        last_activity_at: None,
        workspace_path: None,
        repo_url: None,
        branch: None,
        pending_decision: Some(
            "T4: irreversible or security-sensitive operation; human confirmation required".into(),
        ),
        proposed_default: Some("use cursor".into()),
        correlation: Default::default(),
        runtime: Default::default(),
        ephemeral: false,
        workspace_owned: false,
        source_id: None,
        claude_session_id: None,
        scrollback_path: None,
        last_cwd: None,
        deliverable_id: None,
        pane_id: None,
        tmux_server: None,
        injection_status: Default::default(),
        worktree_owner: None,
        terminal_at: None,
        stop_cause: None,
        kind: Default::default(),
    };
    {
        let mut store = mgr.store.write().await;
        store.upsert(stale.clone()).await.unwrap();
    }

    let report = mgr.reconcile_on_boot(false).await.expect("reconcile");
    assert!(
        report
            .stale_decisions_cleared
            .contains(&stale.id.to_string()),
        "backfill must report the cleared record id; report: {:?}",
        report
    );

    let after = mgr.get(&stale.id).await.unwrap();
    assert_eq!(
        after.state,
        ManagedSessionState::Decommissioned,
        "backfill must not resurrect a terminal record's state"
    );
    assert!(
        after.pending_decision.is_none(),
        "stale pending_decision must be cleared by the boot backfill"
    );
    assert!(
        after.proposed_default.is_none(),
        "stale proposed_default must be cleared by the boot backfill"
    );
}

#[tokio::test]
async fn manager_send_input() {
    let dir = crate::test_support::hermetic_temp_dir();
    // #9101: a send needs a pane the record owns.
    let (mgr, fake) = make_manager_with_pane(&dir).await;

    let record = mgr
        .create(
            "task".into(),
            Some(PathBuf::from("/tmp/x")),
            None,
            None,
            None,
            None,
        )
        .await
        .expect("create");

    // Transition to Active so send_input does not reject.
    {
        let mut store = mgr.store.write().await;
        let mut r = store.get(&record.id).await.unwrap();
        r.state = ManagedSessionState::Active;
        store.upsert(r).await.unwrap();
    }

    mgr.send_input(&record.id, "hello from test")
        .await
        .expect("send");
    let calls = fake.pane_send_calls.lock().unwrap();
    assert!(calls.iter().any(|(_, _, text)| text == "hello from test"));
}

/// send_input must be rejected for Stopped and Decommissioned sessions.
///
/// Why: those states mean the tmux session is gone; sending would fail silently.
/// Test: this function IS the test.
#[tokio::test]
async fn manager_send_input_rejected_for_stopped_and_decommissioned() {
    let dir = crate::test_support::hermetic_temp_dir();
    let (mgr, _fake) = make_manager(&dir).await;

    let record = mgr
        .create(
            "task".into(),
            Some(PathBuf::from("/tmp/x")),
            None,
            None,
            None,
            None,
        )
        .await
        .expect("create");

    // Test Stopped rejection.
    {
        let mut store = mgr.store.write().await;
        let mut r = store.get(&record.id).await.unwrap();
        r.state = ManagedSessionState::Stopped;
        store.upsert(r).await.unwrap();
    }
    let result = mgr.send_input(&record.id, "test").await;
    assert!(result.is_err(), "send_input must fail for Stopped sessions");

    // Test Decommissioned rejection.
    {
        let mut store = mgr.store.write().await;
        let mut r = store.get(&record.id).await.unwrap();
        r.state = ManagedSessionState::Decommissioned;
        store.upsert(r).await.unwrap();
    }
    let result = mgr.send_input(&record.id, "test").await;
    assert!(
        result.is_err(),
        "send_input must fail for Decommissioned sessions"
    );
}

// `manager_send_input_rejected_for_provisioning_and_errored`,
// `manager_send_input_rejected_when_pane_shows_blocking_modal`, and
// `manager_send_input_skips_modal_probe_for_tcode` (issue #3591) live in the
// sibling `send_input_gate_tests.rs` — this file is at the 1500-SLOC test
// cap, so new coverage for the `send_input` readiness/modal gate goes there
// (mirrors the `restart_tests.rs` / `decommission_worktree_tests.rs`
// precedent for keeping this file under its limit).

#[tokio::test]
async fn manager_env_scrub_command_sent() {
    // Verify that the spawn sends `env -u ANTHROPIC_API_KEY claude`.
    // The actual send is in ClaudeCodeAdapter, but we can verify the
    // convention here: the command must not reference ANTHROPIC_API_KEY
    // without the `env -u` prefix.
    let dir = crate::test_support::hermetic_temp_dir();
    let (mgr, fake) = make_manager_with_pane(&dir).await;

    let record = mgr
        .create(
            "task".into(),
            Some(PathBuf::from("/tmp/x")),
            None,
            None,
            None,
            None,
        )
        .await
        .expect("create");

    // Simulate ClaudeCodeAdapter::spawn sending the scrubbed command.
    let scrubbed_cmd = "env -u ANTHROPIC_API_KEY claude";
    mgr.send_input(
        &{
            let mut store = mgr.store.write().await;
            let mut r = store.get(&record.id).await.unwrap();
            r.state = ManagedSessionState::Active;
            store.upsert(r).await.unwrap();
            record.id
        },
        scrubbed_cmd,
    )
    .await
    .expect("send");

    let calls = fake.pane_send_calls.lock().unwrap();
    let found = calls
        .iter()
        .any(|(_, _, cmd)| cmd.contains("env -u ANTHROPIC_API_KEY claude"));
    assert!(found, "env scrub command must be sent; calls: {calls:?}");
}

#[tokio::test]
async fn manager_answer_decision() {
    let dir = crate::test_support::hermetic_temp_dir();
    let (mgr, fake) = make_manager_with_pane(&dir).await;

    let record = mgr
        .create(
            "task".into(),
            Some(PathBuf::from("/tmp/x")),
            None,
            None,
            None,
            None,
        )
        .await
        .expect("create");

    // Seed a pending decision so answer_decision has something to clear.
    {
        let mut store = mgr.store.write().await;
        let mut r = store.get(&record.id).await.unwrap();
        r.pending_decision = Some("merge or rebase?".into());
        r.proposed_default = Some("rebase".into());
        store.upsert(r).await.unwrap();
    }

    mgr.answer_decision(&record.id, "rebase")
        .await
        .expect("answer");

    // The answer must be injected into the pane. Compute the assertion into
    // an owned bool so the mutex guard is released before the next `.await`.
    let injected = {
        let calls = fake.pane_send_calls.lock().unwrap();
        calls.iter().any(|(_, _, text)| text == "rebase")
    };
    assert!(injected);

    // pending_decision / proposed_default must be cleared.
    let after = mgr.get(&record.id).await.unwrap();
    assert!(after.pending_decision.is_none());
    assert!(after.proposed_default.is_none());
}

/// Regression guard: the tmux session must be created with the provisioned
/// workspace path as its cwd, never with $HOME.
///
/// Why: before the fix, `spawn_session` called `mgr.create()` with `cwd = None`,
/// which fell back to `dirs::home_dir()` ($HOME). The tmux session was therefore
/// rooted at $HOME and claude opened there instead of the isolated workspace.
/// What: simulates the `spawn_session` handler sequence — pre-generate id,
/// provision (FakeGitBackend creates the directory), then `create_with_id` with
/// `cwd = Some(workspace_path)`. Asserts the recorded cwd equals the workspace
/// path and is NOT the home directory.
/// Test: this function IS the test.
#[tokio::test]
#[serial_test::serial]
async fn spawn_session_tmux_cwd_is_workspace() {
    use crate::session_manager::record::ManagedSessionId;

    let store_dir = crate::test_support::hermetic_temp_dir();
    let workspace_root = crate::test_support::hermetic_temp_dir();
    // #3965: `#[serial]` + `$HOME` override — see `HomeGuard` above.
    let _home = set_home(workspace_root.path());
    let fake = FakeTmuxDriver::new();
    let mgr = SessionManager::new(store_dir.path(), fake.clone())
        .await
        .expect("manager");

    // Pre-generate the session id (as the fixed spawn_session handler does).
    let session_id = ManagedSessionId::new();

    // #6000 / ADR-0055: trusty-mpm provisions no workspace any more, so the
    // spawn handler is handed a directory that already exists. Stand one up.
    let workspace_path = workspace_root.path().join(session_id.to_string());
    std::fs::create_dir_all(&workspace_path).expect("workspace dir");

    // Create with the provisioned workspace as cwd — this is the fixed order.
    let record = mgr
        .create_with_id(
            session_id,
            "task".into(),
            Some(workspace_path.clone()),
            None,
            Some(workspace_path.clone()),
            Some("https://github.com/owner/repo".into()),
            Some("main".into()),
            crate::runtime::RuntimeKind::default(),
            false,
            false,
        )
        .await
        .expect("create_with_id");

    // The tmux session must have been created with cwd = workspace_path.
    let cwd_calls = fake.create_cwd_calls.lock().unwrap();
    assert_eq!(
        cwd_calls.len(),
        1,
        "exactly one tmux session must be created"
    );
    let (session_name, cwd) = &cwd_calls[0];
    assert_eq!(
        session_name, &record.tmux_name,
        "session name must match the record"
    );
    assert_eq!(
        cwd,
        &workspace_path.to_string_lossy().to_string(),
        "tmux session cwd must equal the provisioned workspace path"
    );

    // Must NOT be $HOME.
    let home = dirs::home_dir()
        .map(|h| h.to_string_lossy().to_string())
        .unwrap_or_default();
    assert_ne!(
        cwd, &home,
        "tmux session cwd must NOT be $HOME (workspace-isolation regression)"
    );

    // Must NOT be /tmp (generic fallback).
    assert_ne!(
        cwd, "/tmp",
        "tmux session cwd must NOT be /tmp (workspace-isolation regression)"
    );

    // workspace_path must be within workspace_root.
    assert!(
        workspace_path.starts_with(workspace_root.path()),
        "workspace must be under the mpm workspace root"
    );
}

/// `create` defaults the runtime to claude-code (unchanged pre-#1203 behavior).
///
/// Why: every existing caller of `create` must keep getting the Claude Code
/// backend so #1203 introduces no behavior change for the default path.
/// Test: this function IS the test.
#[tokio::test]
async fn manager_create_defaults_runtime_to_claude_code() {
    let dir = crate::test_support::hermetic_temp_dir();
    let (mgr, _fake) = make_manager(&dir).await;

    let record = mgr
        .create(
            "task".into(),
            Some(PathBuf::from("/tmp/wt-d")),
            None,
            None,
            None,
            None,
        )
        .await
        .expect("create");

    assert_eq!(record.runtime, crate::runtime::RuntimeKind::ClaudeCode);
    // It must survive the round-trip through the store.
    let reloaded = mgr.get(&record.id).await.expect("get");
    assert_eq!(reloaded.runtime, crate::runtime::RuntimeKind::ClaudeCode);
}

/// `create_with_id` persists the caller-selected runtime on the record.
///
/// Why: #1203 — a tcode session must carry `runtime = Tcode` so `resume`
/// re-spawns the SAME backend; this asserts the field is stored and reloaded.
/// Test: this function IS the test.
#[tokio::test]
async fn manager_create_persists_runtime() {
    let dir = crate::test_support::hermetic_temp_dir();
    let (mgr, _fake) = make_manager(&dir).await;

    let record = mgr
        .create_with_id(
            ManagedSessionId::new(),
            "task".into(),
            Some(PathBuf::from("/tmp/wt-t")),
            None,
            None,
            None,
            None,
            crate::runtime::RuntimeKind::Tcode,
            false,
            false,
        )
        .await
        .expect("create_with_id");

    assert_eq!(record.runtime, crate::runtime::RuntimeKind::Tcode);
    let reloaded = mgr.get(&record.id).await.expect("get");
    assert_eq!(
        reloaded.runtime,
        crate::runtime::RuntimeKind::Tcode,
        "runtime must survive persistence so resume re-spawns the same backend"
    );
}

/// Why: #1219 — the daemon and the supervisor each own a `SessionManager` over
/// the SAME on-disk `sessions.json`. When the supervisor writes a state change
/// (e.g. auto-resume flips `stopped` → `active`), the daemon's manager MUST
/// reflect that transition on its next read; previously it served stale state
/// from its load-once in-memory map forever. This test simulates the supervisor
/// as a second, independent `SessionManager` over the same data dir, writes a
/// state change through it, and asserts the first manager's `get` returns the
/// NEW state — proving reload-on-read.
/// What: builds two managers over one temp data dir, creates+stops a session via
/// manager A (so both managers' file is seeded), then resumes via manager B
/// (out-of-process write to disk), then asserts manager A's `get` returns
/// `Active`, not the stale `Stopped` it last held in memory.
/// Test: this test.
#[tokio::test]
async fn manager_get_reflects_out_of_process_write() {
    let dir = crate::test_support::hermetic_temp_dir();
    // #2250: resume() now existence-checks cwd/workspace_path before handing
    // either to tmux, so the workspace must be a REAL directory (a fake
    // "/tmp/wt-shared" placeholder made `mgr_b.resume` below fail loudly with
    // `WorkspaceMissing` — correct new behavior, but orthogonal to what this
    // test actually verifies: cross-manager reload-on-read).
    let workspace_dir = crate::test_support::hermetic_temp_dir();

    // Manager A = the daemon's view; Manager B = the supervisor's view.
    // Both point at the same data dir / sessions.json.
    let (mgr_a, _fake_a) = make_manager(&dir).await;
    let (mgr_b, fake_b) = make_manager(&dir).await;

    // Create + stop a session via A. The record is now `Stopped` on disk.
    let record = mgr_a
        .create(
            "shared-state task".into(),
            Some(workspace_dir.path().to_owned()),
            None,
            None,
            None,
            None,
        )
        .await
        .expect("create");
    let id = record.id;
    mgr_a.stop(&id).await.expect("stop");

    // A reads the session: it now holds `Stopped` in its in-memory map.
    let before = mgr_a.get(&id).await.expect("get before");
    assert_eq!(
        before.state,
        ManagedSessionState::Stopped,
        "precondition: manager A sees the session as Stopped"
    );

    // The supervisor (manager B) resumes the session out of A's process. This
    // writes `Active` to the shared sessions.json. B reloads-on-read first, so
    // it sees the Stopped record A persisted, then transitions it to Active.
    // The fake tmux driver must report the session as NOT existing so resume's
    // kill-stale path is a no-op, then must accept the create_session call.
    fake_b.seeded_names.lock().unwrap().clear();
    mgr_b.resume(&id).await.expect("supervisor resume");

    // The daemon (manager A) reads again. WITHOUT reload-on-read this returns the
    // stale `Stopped`; WITH it, A re-reads the file and returns `Active`.
    let after = mgr_a.get(&id).await.expect("get after");
    assert_eq!(
        after.state,
        ManagedSessionState::Active,
        "manager A must reflect the out-of-process resume written by manager B"
    );

    // `list` must also reflect the cross-process write.
    let listed = mgr_a.list().await;
    let found = listed
        .iter()
        .find(|r| r.id == id)
        .expect("session present in list");
    assert_eq!(
        found.state,
        ManagedSessionState::Active,
        "manager A's list must also reflect the out-of-process write"
    );
}

/// Adopting an EXISTING (driver-reports-live) tmux session must register a
/// durable `Active` record carrying the supplied cwd/task/runtime (#1433).
///
/// Why: the explicit adopt path connects to a pane that already exists — it must
/// NOT call `create_session` (no new pane) and must persist a queryable record.
/// What: seeds a live tmux name on the fake driver, adopts it, and asserts the
/// record is `Active`, NOT created via `create_session`, and is retrievable.
/// Test: this function IS the test.
/// Corrupt the manager's backing `sessions.json` so the next reload-on-read
/// fails. Writing garbage (a) changes the file length so `reload_if_changed`
/// detects a change and re-reads, and (b) makes `serde_json::from_str` fail with
/// `StoreError::Serialize` — a faithful stand-in for a transient reload I/O error
/// (NFS hiccup, partial write observed by a reader, etc.).
///
/// `pub(super)`: also used by `reload_error_tests.rs` (#2453 review finding 1,
/// round 2 — extracted from this file to stay under the 1500-SLOC test cap
/// after the `pane_id` field addition).
pub(super) fn corrupt_store_file(mgr: &SessionManager) {
    let path = mgr.data_dir().join("sessions.json");
    std::fs::write(&path, b"{ this is not valid json ]").expect("corrupt store file");
}

/// Why: #1219 follow-up — `list()` must never report an EMPTY fleet because of a
/// transient reload error. The old code returned `Vec::new()` on reload failure
/// (despite a comment claiming "last-known set"), which would mislead the
/// supervisor/operator into thinking every session vanished. This test pins the
/// corrected behavior: a reload error yields the ACTUAL last-known in-memory set.
/// What: creates a session (so the manager holds it in memory and on disk), then
/// corrupts `sessions.json` so the next `list()` reload fails, and asserts
/// `list()` still returns the previously-loaded record rather than an empty Vec.
/// Test: this test.
#[tokio::test]
async fn manager_list_returns_last_known_on_reload_error() {
    let dir = crate::test_support::hermetic_temp_dir();
    let (mgr, _fake) = make_manager(&dir).await;

    let record = mgr
        .create(
            "fleet-visibility task".into(),
            Some(PathBuf::from("/tmp/wt-lastknown")),
            None,
            None,
            None,
            None,
        )
        .await
        .expect("create");
    let id = record.id;

    // Sanity: with a healthy file, list sees the one session.
    assert_eq!(
        mgr.list().await.len(),
        1,
        "precondition: one session listed"
    );

    // Inject a transient reload failure by corrupting the backing file.
    corrupt_store_file(&mgr);

    // The reload now fails — but list() must fall back to the last-known set,
    // NOT report an empty fleet.
    let listed = mgr.list().await;
    assert_eq!(
        listed.len(),
        1,
        "list() must return the last-known set on reload error, not empty: {listed:?}"
    );
    assert_eq!(
        listed[0].id, id,
        "the last-known record must be the one we created"
    );
}

// `manager_get_returns_last_known_on_reload_error` was extracted to
// `reload_error_tests.rs` (#2453 review finding 1 round 2 — keeps this file
// under the 1500-SLOC test cap after the `pane_id` field addition).

// ── #1508: ephemeral tagging, bulk teardown, by-state prune, compaction ─────────

/// Seed a record DIRECTLY into the store with explicit state/ephemeral flags.
///
/// Why: the prune tests need records in arbitrary lifecycle states (Stopped,
/// Decommissioned, …) and ephemeral flags without driving the full create/stop
/// ritual; upserting a hand-built record is the cheapest way to set up the matrix.
/// What: builds a `SessionRecord` with the given id/state/ephemeral and a
/// workspace path that actually exists on disk (so a real decommission can remove
/// it), upserts it, and returns the workspace path for assertions. When `state`
/// is `Active` or `Provisioning`, also registers a LIVE tmux session for
/// `record.tmux_name` on `mgr`'s driver (#2022) — mirroring the real production
/// invariant that a running record has a live tmux session backing it — so every
/// EXISTING caller that seeds an `Active`/`Provisioning` record to mean "this one
/// is really running" keeps working under the new tmux-probe-based
/// [`super::prune::is_running`] guard. A test that specifically needs a STALE
/// `Active` record (state says running, tmux is actually gone — the #2022 bug
/// scenario) should seed normally and then kill the session itself via
/// `fake.kill_session(&format!("tmpm-seed-{id}"))`.
/// Test: used by the prune/delete tests below.
pub(super) async fn seed_record(
    mgr: &SessionManager,
    root: &TempDir,
    id: ManagedSessionId,
    state: ManagedSessionState,
    ephemeral: bool,
) -> PathBuf {
    let ws = root.path().join(format!("ws-{id}"));
    // Decommissioned tombstones carry NO workspace (it was already removed); every
    // other state keeps a real on-disk dir so a teardown can remove it.
    let workspace_path = if state == ManagedSessionState::Decommissioned {
        None
    } else {
        std::fs::create_dir_all(&ws).expect("mk ws");
        Some(ws.clone())
    };
    let tmux_name = format!("tmpm-seed-{id}");
    let seeds_live_tmux = matches!(
        state,
        ManagedSessionState::Active | ManagedSessionState::Provisioning
    );
    let record = SessionRecord {
        id,
        tmux_name: tmux_name.clone(),
        cwd: root.path().to_path_buf(),
        task: "seed".into(),
        state,
        created_at: Utc::now(),
        last_activity_at: None,
        workspace_path,
        repo_url: None,
        branch: None,
        pending_decision: None,
        proposed_default: None,
        correlation: Default::default(),
        runtime: Default::default(),
        ephemeral,
        workspace_owned: false,
        source_id: None,
        claude_session_id: None,
        scrollback_path: None,
        last_cwd: None,
        deliverable_id: None,
        pane_id: None,
        tmux_server: None,
        injection_status: Default::default(),
        worktree_owner: None,
        terminal_at: None,
        stop_cause: None,
        kind: Default::default(),
    };
    if seeds_live_tmux {
        mgr.tmux
            .create_session(&tmux_name, &root.path().to_string_lossy())
            .expect("seed: register live tmux session");
    }
    mgr.store
        .write()
        .await
        .upsert(record)
        .await
        .expect("seed upsert");
    ws
}

/// `create_with_id` persists the caller-supplied `ephemeral` flag (#1508).
///
/// Why: the flag is the foundation of the whole feature — it must round-trip
/// through the create path onto the persisted record.
/// What: creates one session with `ephemeral=true` and one with `false`, then
/// reads each back and asserts the flag survived.
/// Test: this function IS the test.
#[tokio::test]
async fn manager_create_persists_ephemeral_flag() {
    let dir = crate::test_support::hermetic_temp_dir();
    let (mgr, _fake) = make_manager(&dir).await;

    let eph = mgr
        .create_with_id(
            ManagedSessionId::new(),
            "ephemeral task".into(),
            Some(PathBuf::from("/tmp/eph")),
            None,
            None,
            None,
            None,
            crate::runtime::RuntimeKind::default(),
            true,
            false,
        )
        .await
        .expect("create ephemeral");
    let durable = mgr
        .create_with_id(
            ManagedSessionId::new(),
            "durable task".into(),
            Some(PathBuf::from("/tmp/dur")),
            None,
            None,
            None,
            None,
            crate::runtime::RuntimeKind::default(),
            false,
            false,
        )
        .await
        .expect("create durable");

    assert!(
        mgr.get(&eph.id).await.unwrap().ephemeral,
        "ephemeral flag persisted"
    );
    assert!(
        !mgr.get(&durable.id).await.unwrap().ephemeral,
        "durable stays false"
    );
}

/// `decommission_all_ephemeral` tears down ONLY ephemeral sessions (#1508).
///
/// Why: the core safety invariant — REAL (non-ephemeral) sessions must never be
/// touched by the bulk-teardown path.
/// What: seeds two ephemeral (Active + Stopped) and two durable (Active + Stopped)
/// sessions, runs the bulk teardown, and asserts only the two ephemeral records
/// became Decommissioned while the two durable records are untouched.
/// Test: this function IS the test.
#[tokio::test]
async fn decommission_all_ephemeral_ignores_non_ephemeral() {
    let dir = crate::test_support::hermetic_temp_dir();
    let (mgr, _fake) = make_manager(&dir).await;

    let eph_active = ManagedSessionId::new();
    let eph_stopped = ManagedSessionId::new();
    let dur_active = ManagedSessionId::new();
    let dur_stopped = ManagedSessionId::new();
    seed_record(&mgr, &dir, eph_active, ManagedSessionState::Active, true).await;
    // #8935: the record's pane proves the live session is its own.
    bind_pane(&mgr, &eph_active).await;
    seed_record(&mgr, &dir, eph_stopped, ManagedSessionState::Stopped, true).await;
    seed_record(&mgr, &dir, dur_active, ManagedSessionState::Active, false).await;
    seed_record(&mgr, &dir, dur_stopped, ManagedSessionState::Stopped, false).await;

    let count = mgr
        .decommission_all_ephemeral()
        .await
        .expect("bulk teardown");
    assert_eq!(count, 2, "exactly the two ephemeral sessions are torn down");

    assert_eq!(
        mgr.get(&eph_active).await.unwrap().state,
        ManagedSessionState::Decommissioned
    );
    assert_eq!(
        mgr.get(&eph_stopped).await.unwrap().state,
        ManagedSessionState::Decommissioned
    );
    // Durable sessions are untouched.
    assert_eq!(
        mgr.get(&dur_active).await.unwrap().state,
        ManagedSessionState::Active
    );
    assert_eq!(
        mgr.get(&dur_stopped).await.unwrap().state,
        ManagedSessionState::Stopped
    );
}

/// The by-state Stopped prune NEVER touches a running (Active) session (#1508).
///
/// Why: clearing legacy stopped/decommissioned records must not risk reaping a
/// live session. `include_active=false` is the fail-closed default.
/// What: seeds an Active and a Stopped session, prunes `Stopped`, asserts only the
/// Stopped one is decommissioned and the Active one is left running.
/// Test: this function IS the test.
#[tokio::test]
async fn prune_by_state_never_touches_active() {
    let dir = crate::test_support::hermetic_temp_dir();
    let (mgr, _fake) = make_manager(&dir).await;

    let active = ManagedSessionId::new();
    let stopped = ManagedSessionId::new();
    seed_record(&mgr, &dir, active, ManagedSessionState::Active, false).await;
    seed_record(&mgr, &dir, stopped, ManagedSessionState::Stopped, false).await;

    let outcome = mgr
        .prune_managed(
            crate::session_manager::PruneFilter::Stopped,
            false,
            false,
            None,
            None,
        )
        .await
        .expect("prune stopped");
    assert_eq!(outcome.count(), 1, "only the Stopped session is pruned");
    assert_eq!(
        mgr.get(&active).await.unwrap().state,
        ManagedSessionState::Active,
        "the Active session must be untouched"
    );
    assert_eq!(
        mgr.get(&stopped).await.unwrap().state,
        ManagedSessionState::Decommissioned
    );
}

/// The Decommissioned prune COMPACTS the store (removes tombstones) (#1508).
///
/// Why: tombstones accumulated unbounded; the compaction pass must actually delete
/// them from sessions.json so the file stops growing.
/// What: seeds two Decommissioned tombstones + one Stopped session, prunes
/// `Decommissioned`, and asserts both tombstones are GONE from the store while the
/// Stopped session remains.
/// Test: this function IS the test.
#[tokio::test]
async fn prune_decommissioned_compacts() {
    let dir = crate::test_support::hermetic_temp_dir();
    let (mgr, _fake) = make_manager(&dir).await;

    let t1 = ManagedSessionId::new();
    let t2 = ManagedSessionId::new();
    let stopped = ManagedSessionId::new();
    seed_record(&mgr, &dir, t1, ManagedSessionState::Decommissioned, false).await;
    seed_record(&mgr, &dir, t2, ManagedSessionState::Decommissioned, false).await;
    seed_record(&mgr, &dir, stopped, ManagedSessionState::Stopped, false).await;

    let outcome = mgr
        .prune_managed(
            crate::session_manager::PruneFilter::Decommissioned,
            false,
            false,
            None,
            None,
        )
        .await
        .expect("compact");
    assert_eq!(outcome.count(), 2, "both tombstones compacted");
    assert!(
        outcome
            .sessions
            .iter()
            .all(|s| s.action == crate::session_manager::PruneAction::Removed),
        "decommissioned prune reports Removed"
    );

    // Both tombstones are GONE from the store; the Stopped record survives.
    assert!(matches!(
        mgr.get(&t1).await,
        Err(ManagedError::SessionNotFound(_))
    ));
    assert!(matches!(
        mgr.get(&t2).await,
        Err(ManagedError::SessionNotFound(_))
    ));
    assert_eq!(
        mgr.list().await.len(),
        1,
        "only the Stopped session remains"
    );
}

/// The Deleted prune COMPACTS the store (removes `--deleted--` tombstones) (#2012).
///
/// Why: `tm sessions delete` now SOFT-deletes (marks `--deleted--`, keeps the
/// record); operators need a permanent-removal path, and `prune --state deleted`
/// is it. A `Deleted` record is a terminal tombstone, so prune must COMPACT it
/// (remove), not re-run a decommission teardown.
/// What: seeds two `Deleted` tombstones + one `Stopped` session, prunes
/// `Deleted`, and asserts both tombstones are GONE while the Stopped survives,
/// each reported as `Removed`.
/// Test: this function IS the test.
#[tokio::test]
async fn prune_deleted_compacts() {
    let dir = crate::test_support::hermetic_temp_dir();
    let (mgr, _fake) = make_manager(&dir).await;

    let d1 = ManagedSessionId::new();
    let d2 = ManagedSessionId::new();
    let stopped = ManagedSessionId::new();
    seed_record(&mgr, &dir, d1, ManagedSessionState::Deleted, false).await;
    seed_record(&mgr, &dir, d2, ManagedSessionState::Deleted, false).await;
    seed_record(&mgr, &dir, stopped, ManagedSessionState::Stopped, false).await;

    let outcome = mgr
        .prune_managed(
            crate::session_manager::PruneFilter::Deleted,
            false,
            false,
            None,
            None,
        )
        .await
        .expect("compact deleted");
    assert_eq!(outcome.count(), 2, "both deleted tombstones compacted");
    assert!(
        outcome
            .sessions
            .iter()
            .all(|s| s.action == crate::session_manager::PruneAction::Removed),
        "deleted prune reports Removed"
    );
    assert!(matches!(
        mgr.get(&d1).await,
        Err(ManagedError::SessionNotFound(_))
    ));
    assert_eq!(
        mgr.list().await.len(),
        1,
        "only the Stopped session remains"
    );
}

/// `All` targets every NON-running record (#1508).
///
/// Why: the legacy purge needs ONE sweep that tears down stopped/errored/ephemeral
/// AND compacts decommissioned, while leaving running sessions alone.
/// What: seeds Active + Stopped + Errored + Decommissioned, prunes `All`, and
/// asserts the Active is untouched, Stopped/Errored became Decommissioned, and the
/// pre-existing tombstone was removed.
/// Test: this function IS the test.
#[tokio::test]
async fn prune_all_targets_non_running() {
    let dir = crate::test_support::hermetic_temp_dir();
    let (mgr, _fake) = make_manager(&dir).await;

    let active = ManagedSessionId::new();
    let stopped = ManagedSessionId::new();
    let errored = ManagedSessionId::new();
    let tomb = ManagedSessionId::new();
    seed_record(&mgr, &dir, active, ManagedSessionState::Active, false).await;
    seed_record(&mgr, &dir, stopped, ManagedSessionState::Stopped, false).await;
    seed_record(&mgr, &dir, errored, ManagedSessionState::Errored, false).await;
    seed_record(&mgr, &dir, tomb, ManagedSessionState::Decommissioned, false).await;

    let outcome = mgr
        .prune_managed(
            crate::session_manager::PruneFilter::All,
            false,
            false,
            None,
            None,
        )
        .await
        .expect("prune all");
    assert_eq!(
        outcome.count(),
        3,
        "stopped + errored + tombstone (not active)"
    );

    assert_eq!(
        mgr.get(&active).await.unwrap().state,
        ManagedSessionState::Active,
        "running session is never touched by All"
    );
    assert_eq!(
        mgr.get(&stopped).await.unwrap().state,
        ManagedSessionState::Decommissioned
    );
    assert_eq!(
        mgr.get(&errored).await.unwrap().state,
        ManagedSessionState::Decommissioned
    );
    assert!(matches!(
        mgr.get(&tomb).await,
        Err(ManagedError::SessionNotFound(_))
    ));
}

/// A dry-run reports candidates WITHOUT mutating anything (#1508).
///
/// Why: the operator must be able to preview a legacy purge before destroying
/// records. `--dry-run` must be side-effect free.
/// What: seeds a Stopped session, prunes `Stopped` with `dry_run=true`, asserts the
/// outcome lists it but the record is STILL Stopped afterward.
/// Test: this function IS the test.
#[tokio::test]
async fn prune_dry_run_reports_without_mutating() {
    let dir = crate::test_support::hermetic_temp_dir();
    let (mgr, _fake) = make_manager(&dir).await;

    let stopped = ManagedSessionId::new();
    seed_record(&mgr, &dir, stopped, ManagedSessionState::Stopped, false).await;

    let outcome = mgr
        .prune_managed(
            crate::session_manager::PruneFilter::Stopped,
            true,
            false,
            None,
            None,
        )
        .await
        .expect("dry run");
    assert!(outcome.dry_run, "outcome flagged dry_run");
    assert_eq!(outcome.count(), 1, "candidate reported");
    // The record must be UNCHANGED after a dry run.
    assert_eq!(
        mgr.get(&stopped).await.unwrap().state,
        ManagedSessionState::Stopped,
        "dry run must not mutate the record"
    );
}

/// `PruneFilter::parse` round-trips and rejects garbage (#1508).
///
/// Why: the CLI/HTTP/MCP surfaces all parse the same spellings; a typo must be a
/// clear error, not a silent default.
/// What: parses every valid spelling (asserting `as_str` round-trips) and asserts
/// an unknown value errors.
/// Test: this function IS the test.
#[test]
fn prune_filter_parse_round_trip() {
    use crate::session_manager::PruneFilter;
    for f in [
        PruneFilter::Ephemeral,
        PruneFilter::Stopped,
        PruneFilter::Decommissioned,
        PruneFilter::Deleted,
        PruneFilter::All,
        // #6118: the unresolvable-workspace selector round-trips like the rest.
        PruneFilter::Unresolvable,
    ] {
        assert_eq!(PruneFilter::parse(f.as_str()).unwrap(), f);
    }
    assert_eq!(
        PruneFilter::parse("EPHEMERAL ").unwrap(),
        PruneFilter::Ephemeral
    );
    assert!(PruneFilter::parse("bogus").is_err());
}

/// `PruneOutcome`/`PruneAction` serialize to the wire shape the HTTP+MCP surfaces
/// expect (#1508).
///
/// Why: the dry-run/report JSON must carry `dry_run`, `filter`, and per-session
/// `action` so callers can render a precise preview; a serde regression would
/// silently change the wire contract.
/// What: builds an outcome, serializes it, and asserts the key fields/strings.
/// Test: this function IS the test.
#[tokio::test]
async fn prune_outcome_serializes() {
    let dir = crate::test_support::hermetic_temp_dir();
    let (mgr, _fake) = make_manager(&dir).await;
    let id = ManagedSessionId::new();
    seed_record(&mgr, &dir, id, ManagedSessionState::Stopped, true).await;

    let outcome = mgr
        .prune_managed(
            crate::session_manager::PruneFilter::Ephemeral,
            true,
            false,
            None,
            None,
        )
        .await
        .expect("dry run");
    let v = serde_json::to_value(&outcome).expect("serialize outcome");
    assert_eq!(v["dry_run"], serde_json::json!(true));
    assert_eq!(v["filter"], serde_json::json!("ephemeral"));
    assert_eq!(
        v["sessions"][0]["action"],
        serde_json::json!("decommissioned")
    );

    // #4344 review (critic MEDIUM): pin the NEW `DecommissionedWorktreeRetained`
    // variant's wire form too, so a future rename/reorder of the enum is
    // caught here rather than only by the `prune_reports_dirty_worktree_retained`
    // integration test in `decommission_worktree_tests.rs`.
    assert_eq!(
        serde_json::to_value(crate::session_manager::PruneAction::DecommissionedWorktreeRetained)
            .expect("serialize action"),
        serde_json::json!("decommissioned_worktree_retained")
    );
    let retained = crate::session_manager::PrunedSession {
        id: "id".into(),
        tmux_name: "tmux".into(),
        state: "stopped".into(),
        action: crate::session_manager::PruneAction::DecommissionedWorktreeRetained,
        retained_workspace_path: Some(PathBuf::from("/tmp/retained-ws")),
    };
    let retained_v = serde_json::to_value(&retained).expect("serialize retained row");
    assert_eq!(
        retained_v["retained_workspace_path"],
        serde_json::json!("/tmp/retained-ws"),
        "retained_workspace_path must serialize as a plain path string when present"
    );
    let clean = crate::session_manager::PrunedSession {
        retained_workspace_path: None,
        ..retained.clone()
    };
    let clean_v = serde_json::to_value(&clean).expect("serialize clean row");
    assert!(
        clean_v.get("retained_workspace_path").is_none(),
        "retained_workspace_path must be OMITTED (not null) when there is nothing retained"
    );
}

/// `compact_record` deletes a tombstone from the store (#1508).
///
/// Why: the single-record compaction primitive must actually remove the record so
/// the age-based reaper / prune can shrink sessions.json.
/// What: seeds a Decommissioned tombstone, compacts it, asserts it is gone.
/// Test: this function IS the test.
#[tokio::test]
async fn compact_record_removes_from_store() {
    let dir = crate::test_support::hermetic_temp_dir();
    let (mgr, _fake) = make_manager(&dir).await;
    let id = ManagedSessionId::new();
    seed_record(&mgr, &dir, id, ManagedSessionState::Decommissioned, false).await;

    mgr.compact_record(&id).await.expect("compact");
    assert!(matches!(
        mgr.get(&id).await,
        Err(ManagedError::SessionNotFound(_))
    ));
}

// #2012: `delete_record` coverage (hard-delete via compact_record, the
// fail-closed running-guard, --force bypass, and the workspace-dir-untouched
// invariant) lives in the sibling `delete_tests` module — this file is at the
// 1500-SLOC test cap, mirroring the `decommission_worktree_tests` /
// `backfill_tests` split pattern documented there.

/// Age-based auto-reap targets ONLY old EPHEMERAL sessions (#1508).
///
/// Why: the backstop must reclaim leaked test sessions older than the threshold
/// WITHOUT ever touching a real (non-ephemeral) session or a young ephemeral one.
/// What: seeds (a) an OLD ephemeral, (b) a YOUNG ephemeral, (c) an OLD durable, and
/// reaps with a 1-hour threshold. Only (a) must be decommissioned.
/// Test: this function IS the test.
#[tokio::test]
async fn reap_aged_ephemeral_picks_old_ephemeral_only() {
    let dir = crate::test_support::hermetic_temp_dir();
    let (mgr, _fake) = make_manager(&dir).await;

    // Helper: seed a record with an explicit created_at + ephemeral flag.
    async fn seed_aged(
        mgr: &SessionManager,
        root: &TempDir,
        id: ManagedSessionId,
        created_at: chrono::DateTime<Utc>,
        ephemeral: bool,
    ) {
        let ws = root.path().join(format!("aged-{id}"));
        std::fs::create_dir_all(&ws).expect("mk ws");
        let record = SessionRecord {
            id,
            tmux_name: format!("tmpm-aged-{id}"),
            cwd: root.path().to_path_buf(),
            task: "aged".into(),
            state: ManagedSessionState::Active,
            created_at,
            last_activity_at: None,
            workspace_path: Some(ws),
            repo_url: None,
            branch: None,
            pending_decision: None,
            proposed_default: None,
            correlation: Default::default(),
            runtime: Default::default(),
            ephemeral,
            workspace_owned: false,
            source_id: None,
            claude_session_id: None,
            scrollback_path: None,
            last_cwd: None,
            deliverable_id: None,
            pane_id: None,
            tmux_server: None,
            injection_status: Default::default(),
            worktree_owner: None,
            terminal_at: None,
            stop_cause: None,
            kind: Default::default(),
        };
        mgr.store.write().await.upsert(record).await.expect("seed");
    }

    let old_eph = ManagedSessionId::new();
    let young_eph = ManagedSessionId::new();
    let old_durable = ManagedSessionId::new();
    let two_hours_ago = Utc::now() - chrono::Duration::hours(2);
    let now = Utc::now();
    seed_aged(&mgr, &dir, old_eph, two_hours_ago, true).await;
    seed_aged(&mgr, &dir, young_eph, now, true).await;
    seed_aged(&mgr, &dir, old_durable, two_hours_ago, false).await;

    let reaped = mgr
        .reap_aged_ephemeral(chrono::Duration::hours(1))
        .await
        .expect("reap");
    assert_eq!(reaped, 1, "only the OLD ephemeral session is reaped");

    assert_eq!(
        mgr.get(&old_eph).await.unwrap().state,
        ManagedSessionState::Decommissioned,
        "old ephemeral was reaped"
    );
    assert_eq!(
        mgr.get(&young_eph).await.unwrap().state,
        ManagedSessionState::Active,
        "young ephemeral is below the age threshold"
    );
    assert_eq!(
        mgr.get(&old_durable).await.unwrap().state,
        ManagedSessionState::Active,
        "a non-ephemeral session is NEVER reaped by age"
    );
}

// ── workspace-ownership guard tests (#1511) ──────────────────────────────────

/// Decommissioning an UNOWNED record (local-path spawn, adopt) does NOT delete
/// the workspace directory — only the session record is tombstoned.
///
/// Why (#1511): this is the core safety property. Before #1511, `decommission`
/// unconditionally `remove_dir_all`'d `workspace_path`, which deleted a live
/// user repo. With the `workspace_owned = false` guard, the directory must be
/// preserved even though decommission completes successfully.
/// What: creates a temp dir as the "workspace", builds an unowned record that
/// points at it, decommissions the session, asserts the dir still exists on
/// disk AND the record state is `Decommissioned`.
/// Test: this function IS the test.
#[tokio::test]
async fn manager_decommission_unowned_skips_deletion() {
    let dir = crate::test_support::hermetic_temp_dir();
    let (mgr, _fake) = make_manager(&dir).await;

    // This dir represents a REAL user repo — it was not created by the SM.
    let real_user_repo = crate::test_support::hermetic_temp_dir();
    let repo_path = real_user_repo.path().to_owned();
    // Write a sentinel file; if decommission deletes the dir this assert fails.
    std::fs::write(repo_path.join("important_file.txt"), "do not delete").unwrap();

    // Create the session record directly in the store, simulating a local-path
    // spawn (#1502) that sets workspace_path to the real directory.
    let id = ManagedSessionId::new();
    let record = SessionRecord {
        id,
        tmux_name: format!("tmpm-local-{id}"),
        cwd: repo_path.clone(),
        task: "local task".into(),
        state: ManagedSessionState::Stopped,
        created_at: Utc::now(),
        last_activity_at: None,
        workspace_path: Some(repo_path.clone()),
        repo_url: None,
        branch: None,
        pending_decision: None,
        proposed_default: None,
        correlation: Default::default(),
        runtime: Default::default(),
        ephemeral: false,
        // workspace_owned = false — the SM did NOT create this directory.
        workspace_owned: false,
        source_id: None,
        claude_session_id: None,
        scrollback_path: None,
        last_cwd: None,
        deliverable_id: None,
        pane_id: None,
        tmux_server: None,
        injection_status: Default::default(),
        worktree_owner: None,
        terminal_at: None,
        stop_cause: None,
        kind: Default::default(),
    };
    mgr.store.write().await.upsert(record).await.unwrap();

    // Decommission must SUCCEED (return Ok) but NOT delete the directory.
    let (tombstone, workspace_removed) = mgr
        .decommission(&id, None)
        .await
        .expect("decommission of an unowned record must succeed (skip deletion, not error)");

    // workspace_removed must be false — we did NOT remove an unowned directory.
    assert!(
        !workspace_removed,
        "workspace_removed must be false for an unowned (adopted/local-path) workspace"
    );

    // The record must be tombstoned.
    assert_eq!(
        tombstone.state,
        ManagedSessionState::Decommissioned,
        "record state must be Decommissioned"
    );

    // The REAL directory must still exist — decommission must not have deleted it.
    assert!(
        repo_path.exists(),
        "the unowned workspace directory must NOT be deleted by decommission (#1511)"
    );
    assert!(
        repo_path.join("important_file.txt").exists(),
        "the sentinel file inside the unowned workspace must still exist"
    );
}

/// `set_workspace_owned` persists the flag so a subsequent `decommission` can
/// read it back and make the correct deletion decision.
///
/// Why (#1511): the clone-provision path calls `set_workspace_owned(id, true)`
/// after creating the tmux session; this test pins that the flag survives the
/// store round-trip and is visible on `get`.
/// What: creates a session with `workspace_owned = false`, calls
/// `set_workspace_owned(id, true)`, reads the record back, asserts the flag is
/// now `true`.
/// Test: this function IS the test.
#[tokio::test]
async fn workspace_owned_flag_round_trips_via_set() {
    let dir = crate::test_support::hermetic_temp_dir();
    let (mgr, _fake) = make_manager(&dir).await;

    let record = mgr
        .create(
            "clone task".into(),
            Some(PathBuf::from("/tmp/ws")),
            None,
            Some(PathBuf::from("/tmp/ws")),
            Some("https://github.com/owner/repo".into()),
            Some("main".into()),
        )
        .await
        .expect("create");

    // Fresh record starts unowned (the default).
    assert!(
        !record.workspace_owned,
        "a freshly created record must default to workspace_owned = false"
    );

    // Simulate what the clone-provision path does.
    mgr.set_workspace_owned(&record.id, true)
        .await
        .expect("set_workspace_owned");

    // Read back and assert.
    let updated = mgr
        .get(&record.id)
        .await
        .expect("get after set_workspace_owned");
    assert!(
        updated.workspace_owned,
        "workspace_owned must be true after set_workspace_owned(true)"
    );
}

// ── Graceful-stop tests (WI-4 PR A, #1595) ──────────────────────────────────

/// `graceful_stop` with a known PID records the call and kills the session.
///
/// Why: regression guard for the graceful-shutdown path — verifies that when
/// a PID is supplied, `graceful_stop` still results in `kill_session` being
/// called (the SIGTERM phase is OS-level and not asserted in unit tests, but
/// the post-signal kill is the mandatory cleanup).
/// What: creates a session, calls `graceful_stop` with a dummy PID, then
/// asserts the name appears in `kill_calls` and `graceful_stop_calls`.
/// Test: this is the test.
#[allow(deprecated)] // #9101: the deprecated name-addressed graceful_stop
#[tokio::test]
async fn fake_driver_graceful_stop_with_pid() {
    let dir = crate::test_support::hermetic_temp_dir();
    let (mgr, fake) = make_manager(&dir).await;

    mgr.create(
        "task".into(),
        Some(PathBuf::from("/tmp/wt-graceful")),
        Some("graceful-test".into()),
        None,
        None,
        None,
    )
    .await
    .expect("create");

    let tmux_name = "tmpm-graceful-test";
    let pid: u32 = 99_999; // dummy PID — SIGTERM will fail with ESRCH; we ignore it
    fake.graceful_stop(tmux_name, Some(pid))
        .expect("graceful_stop");

    let kills = fake.kill_calls.lock().unwrap();
    assert!(
        kills.iter().any(|n| n == tmux_name),
        "kill_session must be called as the final step of graceful_stop"
    );
    let gs_calls = fake.graceful_stop_calls.lock().unwrap();
    assert!(
        gs_calls
            .iter()
            .any(|(n, p)| n == tmux_name && *p == Some(pid)),
        "graceful_stop_calls must record (name, Some(pid))"
    );
}

/// `graceful_stop` with no PID falls back to send_interrupt then kill_session.
///
/// Why: when the PID is unknown the default impl sends a Ctrl-C interrupt
/// (best-effort) and proceeds to kill_session. This test asserts that even
/// without a PID the session is ultimately killed.
/// What: calls `graceful_stop(name, None)` on the fake and asserts that
/// `kill_calls` contains the name and `graceful_stop_calls` records `None`.
/// Test: this is the test.
#[allow(deprecated)] // #9101: the deprecated name-addressed graceful_stop
#[tokio::test]
async fn fake_driver_graceful_stop_without_pid() {
    let dir = crate::test_support::hermetic_temp_dir();
    let (mgr, fake) = make_manager(&dir).await;

    mgr.create(
        "task".into(),
        Some(PathBuf::from("/tmp/wt-nopid")),
        Some("no-pid-test".into()),
        None,
        None,
        None,
    )
    .await
    .expect("create");

    let tmux_name = "tmpm-no-pid-test";
    fake.graceful_stop(tmux_name, None)
        .expect("graceful_stop without pid");

    let kills = fake.kill_calls.lock().unwrap();
    assert!(
        kills.iter().any(|n| n == tmux_name),
        "kill_session must be called even when no PID is known"
    );
    let gs_calls = fake.graceful_stop_calls.lock().unwrap();
    assert!(
        gs_calls.iter().any(|(n, p)| n == tmux_name && p.is_none()),
        "graceful_stop_calls must record (name, None) when no pid supplied"
    );
}

/// `SessionManager::shutdown` signals and kills an owned Active session.
///
/// Why: the trait-level unit tests above call the driver directly, bypassing
/// the `SessionManager::shutdown` orchestration.
/// What: builds a manager with one Active session whose pane `%1` is on the
/// fake's server, calls `mgr.shutdown()`, and asserts one Ctrl-C reached that
/// pane and the session was killed. #9101: by session id after an ownership
/// re-check, no longer by name. The grace window in test builds is 0 s
/// (`SIGTERM_GRACE_SECS = 0` under `#[cfg(test)]`).
/// Test: this is the test.
#[tokio::test]
async fn shutdown_stops_an_owned_active_session() {
    let dir = crate::test_support::hermetic_temp_dir();
    let fake = fake_with_pane();
    let mgr = SessionManager::new(dir.path(), fake.clone())
        .await
        .expect("manager");

    let record = mgr
        .create(
            "shutdown-integration-task".into(),
            Some(PathBuf::from("/tmp/wt-shutdown")),
            Some("shutdown-integ".into()),
            None,
            None,
            None,
        )
        .await
        .expect("create");
    assert_eq!(record.tmux_server.as_deref(), Some(FAKE_TMUX_SERVER));
    {
        let mut store = mgr.store.write().await;
        let mut r = record.clone();
        r.state = ManagedSessionState::Active;
        store.upsert(r).await.expect("upsert active");
    }
    let tmux_name = record.tmux_name.clone();

    mgr.shutdown().await;

    assert_eq!(
        fake.pane_interrupt_calls.lock().unwrap().as_slice(),
        [(tmux_name.clone(), "%1".to_string())],
        "shutdown signals the owned pane once"
    );
    // #9101: the fake records a kill by id as the id, a kill by name as the
    // name, so this tells the two apart.
    assert_eq!(
        fake.kill_calls.lock().unwrap().as_slice(),
        [FAKE_SESSION_ID.to_string()],
        "shutdown must kill every owned Active session by its session id"
    );
}

#[tokio::test]
async fn claude_session_id_persists_on_session() {
    // Why (#1744): set_claude_session_id must survive a store reload so that
    // resume_managed can read the id back after a daemon restart.
    // What: create a session, write the id via hook_sync, reload from disk,
    // assert the id is present on the re-read record.
    let dir = crate::test_support::hermetic_temp_dir();
    let (mgr, _fake) = make_manager(&dir).await;

    let record = mgr
        .create(
            "task".into(),
            Some(PathBuf::from("/tmp/wt")),
            None,
            None,
            None,
            None,
        )
        .await
        .expect("create");

    mgr.set_claude_session_id(&record.id, "uuid-abc-123")
        .await
        .expect("set_claude_session_id");

    let reloaded = mgr.get(&record.id).await.expect("get after set");
    assert_eq!(
        reloaded.claude_session_id.as_deref(),
        Some("uuid-abc-123"),
        "claude_session_id must survive a store reload (#1744)"
    );
}

#[tokio::test]
async fn clear_claude_session_id_if_clears_on_exact_match() {
    // Why (#4337): SessionEnd must be able to un-correlate its OWN ended
    // session so the NEXT SessionStart can freely re-correlate, instead of a
    // stale id blocking re-correlation forever.
    let dir = crate::test_support::hermetic_temp_dir();
    let (mgr, _fake) = make_manager(&dir).await;
    let record = mgr
        .create(
            "task".into(),
            Some(PathBuf::from("/tmp/wt")),
            None,
            None,
            None,
            None,
        )
        .await
        .expect("create");
    mgr.set_claude_session_id(&record.id, "uuid-abc-123")
        .await
        .expect("set_claude_session_id");

    mgr.clear_claude_session_id_if(&record.id, "uuid-abc-123")
        .await
        .expect("clear_claude_session_id_if");

    let reloaded = mgr.get(&record.id).await.expect("get after clear");
    assert_eq!(
        reloaded.claude_session_id, None,
        "an exact-match clear must remove claude_session_id (#4337)"
    );
}

#[tokio::test]
async fn clear_claude_session_id_if_leaves_a_different_id_untouched() {
    // Why (#4337): a SessionEnd for an id that is NO LONGER the stored one
    // (a fresher SessionStart already overwrote it) must never clobber the
    // fresher value — the exact-match guard is the whole point.
    let dir = crate::test_support::hermetic_temp_dir();
    let (mgr, _fake) = make_manager(&dir).await;
    let record = mgr
        .create(
            "task".into(),
            Some(PathBuf::from("/tmp/wt")),
            None,
            None,
            None,
            None,
        )
        .await
        .expect("create");
    mgr.set_claude_session_id(&record.id, "uuid-current")
        .await
        .expect("set_claude_session_id");

    // A stale/unrelated id no longer matches the stored one — must be a no-op.
    mgr.clear_claude_session_id_if(&record.id, "uuid-stale")
        .await
        .expect("clear_claude_session_id_if");

    let reloaded = mgr.get(&record.id).await.expect("get after no-op clear");
    assert_eq!(
        reloaded.claude_session_id.as_deref(),
        Some("uuid-current"),
        "clearing with a non-matching expected id must leave the stored id untouched (#4337)"
    );
}