trusty-mpm 1.0.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;

/// 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>>,
    /// 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>,
}

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

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(())
    }

    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())
    }

    fn list_sessions(&self) -> Result<Vec<String>, ManagedError> {
        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()
    }

    /// 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_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_exists_override.lock().unwrap().unwrap_or(true)
    }

    /// 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)
}

#[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 backing `sessions.json` a NON-EMPTY directory so the
/// atomic `rename(tmp, sessions.json)` 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.
/// Test: this function IS the test.
#[tokio::test]
async fn manager_create_store_failure_reaps_orphan() {
    let dir = crate::test_support::hermetic_temp_dir();
    let (mgr, fake) = make_manager(&dir).await;

    // Sabotage the store: replace `sessions.json` (a file after `load`) with a
    // NON-EMPTY directory of the same name. `save()` writes `sessions.json.tmp`
    // then renames it onto `sessions.json`; renaming a file onto a non-empty
    // directory fails on every supported platform, so `upsert` returns an error
    // AFTER `create_session` already created the tmux session — the orphan case.
    let store_path = dir.path().join("sessions.json");
    let _ = std::fs::remove_file(&store_path);
    std::fs::create_dir_all(store_path.join("not-empty")).expect("make sessions.json a dir");

    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");

    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;

    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.
    assert!(fake.kill_calls.lock().unwrap().contains(&record.tmux_name));

    // 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;

    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();
    let (mgr, _fake) = make_manager(&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 sentinel file so we can verify the dir was removed.
    std::fs::write(workspace_path.join("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());
}

// Build a minimal Active session record for reconcile tests (avoids repeating
// 20-field struct literals when only tmux_name / task / ws_path vary).
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,
        injection_status: Default::default(),
        worktree_owner: None,
    }
}

/// 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.
#[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,
        injection_status: Default::default(),
        worktree_owner: None,
    };
    {
        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);
}

#[tokio::test]
async fn manager_send_input() {
    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");

    // 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.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 fake = FakeTmuxDriver::new();
    let mgr = SessionManager::new(dir.path(), fake.clone()).await.unwrap();

    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.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(&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.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]
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();
    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();

    // Provision using FakeGitBackend (creates the workspace directory on disk).
    let provisioner = crate::provisioner::WorkspaceProvisioner::without_prepare(
        crate::provisioner::FakeGitBackend::new(),
        workspace_root.path().to_owned(),
    );
    let prepared = provisioner
        .provision(&session_id, "https://github.com/owner/repo", "main", "task")
        .expect("provision");

    let workspace_path = prepared.path.clone();

    // 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,
        injection_status: Default::default(),
        worktree_owner: None,
    };
    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;
    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,
        )
        .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,
        )
        .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,
        )
        .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)
        .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,
        )
        .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,
    ] {
        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,
        )
        .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")
    );
}

/// `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,
            injection_status: Default::default(),
            worktree_owner: None,
        };
        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,
        injection_status: Default::default(),
        worktree_owner: None,
    };
    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.
#[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.
#[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` exercises the real shutdown → graceful_stop path.
///
/// Why: the two trait-level unit tests above call `fake.graceful_stop` directly,
/// bypassing the `SessionManager::shutdown` orchestration. This test exercises
/// the FULL integration path: create an Active session, call `mgr.shutdown()`,
/// assert that the FakeTmuxDriver recorded a graceful_stop call for that session.
/// What: builds a manager with one Active session, calls `mgr.shutdown()`, then
/// asserts both the graceful_stop and kill_session calls were recorded. The grace
/// window in test builds is 0 s (`SIGTERM_GRACE_SECS = 0` under `#[cfg(test)]`)
/// so the test completes without sleeping.
/// Test: this is the test.
#[tokio::test]
async fn shutdown_calls_graceful_stop_for_active_sessions() {
    let dir = crate::test_support::hermetic_temp_dir();
    let (mgr, fake) = make_manager(&dir).await;

    // Create a session and advance it to Active state.
    let record = mgr
        .create(
            "shutdown-integration-task".into(),
            Some(PathBuf::from("/tmp/wt-shutdown")),
            Some("shutdown-integ".into()),
            None,
            None,
            None,
        )
        .await
        .expect("create");

    // Manually advance the state to Active so shutdown() includes it.
    {
        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();

    // Call the real shutdown() — in test builds SIGTERM_GRACE_SECS=0, so no delay.
    mgr.shutdown().await;

    // Assert graceful_stop was recorded for our session.
    let gs_calls = fake.graceful_stop_calls.lock().unwrap();
    assert!(
        gs_calls.iter().any(|(n, _)| n == &tmux_name),
        "shutdown must call graceful_stop for every Active session; \
         expected {tmux_name} in graceful_stop_calls but got {gs_calls:?}"
    );

    // Assert kill_session was called as part of graceful_stop.
    let kills = fake.kill_calls.lock().unwrap();
    assert!(
        kills.iter().any(|n| n == &tmux_name),
        "shutdown must ultimately kill_session for every Active session"
    );
}

#[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)"
    );
}