trusty-mpm 1.8.2

//! trusty-mpm daemon library.
//!
//! Why: the daemon's HTTP API and shared state are useful beyond the `trusty-mpmd`
//! binary — sibling crates (e.g. the Telegram bot's test suite) reuse the real
//! `api::router` and `DaemonState` to drive in-process integration tests without
//! a live daemon. Exposing the modules as a library makes that possible.
//! What: re-exports the daemon's modules as `pub` so both `main.rs` and external
//! consumers can build against them.
//! Test: the modules carry their own `#[cfg(test)]` suites; `cargo test
//! -p trusty-mpm-daemon` exercises them.

pub mod api;
pub mod audit;
pub mod bug_report;
/// The build-lease admission decision log (#8261).
pub mod build_lease_routes;
/// The machine-wide builder-slot claim route and its read-only census (#6892).
pub mod builder_slot_routes;
/// The peer message bus (DOC-60 §5.3) — envelope, pub/sub, instance registry.
pub mod bus;
/// Response-size bound for the `session_context_catchup` digest (#5557).
pub mod catchup_bounds;
/// The `session_context_catchup` response body and its `resolved_note` (#8408).
pub mod catchup_payload;
/// Whether the repo has moved past the snapshot a resume would act on (#7501).
pub mod catchup_superseded;
pub mod claude_config;
pub mod coordinator;
/// Delegation query routes merged into the main router (#4480).
pub mod delegation_routes;
pub mod discovery;
pub mod doctor;
// #7685: declared here for the same reason #7424's row below is — `doctor.rs`
// sits AT the 500-SLOC production cap. `pub` because `tm doctor --fix` calls
// `repair_auto_memory` from the `tm` binary.
pub mod doctor_auto_memory;
// #8236: declared here for the same reason — a credential in a user-readable
// LaunchAgent plist. `pub` because `tm doctor --fix` calls
// `repair_launchd_plist_secrets` from the `tm` binary.
pub mod doctor_launchd_secrets;
// #8236: the repair half, split from the row so neither file approaches the
// 500-SLOC cap. `pub` for the same reason.
pub mod doctor_launchd_secrets_repair;
// #8236: `tm doctor --fix-launchd-secrets` — that repair alone, plus the chmod.
pub mod doctor_launchd_secrets_scoped;
// #8236 item 8: can the DAEMON's own resolver reach each credential it needs?
// Read-only, key-only, and bounded so it can never hang on a Keychain dialog.
pub mod doctor_credential_reach;
// #7424: declared here rather than inside `doctor.rs` — that file sits AT the
// 500-SLOC production cap, so its `mod` + `use` pair would not fit. The check
// is reached as `super::doctor_startup_context::…` from `doctor::run_doctor`.
mod doctor_startup_context;
// #8415: the launchd `ProcessType` of the tm jobs that start tmux servers.
// Declared here for the same cap reason as the rows above.
pub mod doctor_launchd_process_type;
// #8415: the observed priority of the RUNNING tmux server, which a plist fix
// does not lift until the server restarts.
mod doctor_tmux_priority;
// #8926: trusty-* processes listening on TCP, against the ADR-0032 allowlist.
mod doctor_tcp_listeners;
// #8378 PR-C: the instructional-content source and the installed pin (ADR-0064).
mod doctor_content;
pub mod error;
pub mod idle_nudge;
pub mod idle_reaper;
// #8942, #9101: the shutdown legacy-registry pass, which kills nothing.
mod legacy_reap;
pub mod llm_overseer;
pub mod lock;
/// The cloud log-drain scheduler (#6535, Phase 3 of #6533).
pub mod log_drain;
pub mod managed_routes;
pub mod manager;
pub mod mcp_backend;
pub mod mcp_bugreport;
pub mod mcp_console;
// #6927: the Disk dashboard survey backing the `disk_survey` MCP tool.
pub mod mcp_context;
pub mod mcp_disk;
// #8985: the last complete `disk_survey` pass, refreshed in the background.
pub(crate) mod disk_survey_cache;
pub mod mcp_project;
// #7357: the backfill both registration transports share.
pub mod mcp_proxy;
pub mod mcp_session;
pub mod project_adoption;
// #6431: record-only bulk deletion for the console's unknown-state bucket.
pub mod mcp_session_bulk;
pub mod openapi;
pub mod optimizer;
pub mod orphan_gc;
pub mod overseer_compose;
/// The `/pair/*` route registrations, extracted from `api.rs` (#4480).
pub mod pairing_routes;
pub mod pairing_store;
pub mod provisioning;
// #6288: the method families served on the Unix socket.
pub mod rpc;
pub mod runtime_reap;
// #6285: the one trusty-search client in this crate. The `discover` module that
// read `~/.trusty-search/http_addr` retired with the daemon's HTTP listener.
pub mod search_rpc;
pub mod services;
pub(crate) mod session_record_kind;
pub mod sm_stdio;
/// The hardened Unix-socket listener served alongside HTTP (#6288, ADR-0032).
pub mod socket;
pub(crate) mod spawn_command;
pub mod state;
pub mod tmux;
pub mod watcher;

use std::net::SocketAddr;
use std::sync::Arc;

use tracing::info;

pub use state::DaemonState;

/// Run the resident HTTP daemon: API, hook relay, dashboard feed.
///
/// Why: the HTTP boot sequence is shared by both the standalone `trusty-mpmd`
/// shim and the unified `trusty-mpm daemon` subcommand; living in the library
/// keeps a single source of truth.
/// What: binds both listeners — the TCP port and the RPC socket (#6288) — then
/// hands them to [`serve_http`].
///
/// # Errors
///
/// When either listener cannot be bound, or as [`serve_http`].
///
/// Test: `cargo run -p trusty-mpm-cli -- daemon` logs "trusty-mpm daemon
/// starting" and `curl localhost:7880/health` returns `ok`; the dual-serve and
/// drain behaviour is covered by `tests/daemon_dual_serve.rs`.
pub async fn run_http(state: Arc<DaemonState>, addr: SocketAddr) -> anyhow::Result<()> {
    info!("trusty-mpm daemon starting on {addr}");
    let listener = tokio::net::TcpListener::bind(addr).await?;
    // #6288: both binds complete before anything is served, for the same reason
    // `run_daemon` publishes nothing until both succeed — a half-bound daemon
    // must not exist.
    let rpc = socket::bind(&socket::socket_path()?).await?;
    serve_http(state, listener, rpc).await
}

/// Run the daemon's background tasks, HTTP API, and RPC socket on already-bound
/// listeners.
///
/// Why both listeners arrive pre-bound: the CLI performs auto port selection
/// (binding an ephemeral port when the configured one is busy) and needs the
/// daemon to serve on that exact listener. The socket is pre-bound for a
/// stronger reason (#6288) — `run_daemon` writes a lock file naming this
/// daemon, and a lock must never name a daemon that then fails to start. Both
/// binds happen in the caller, before anything is published, so a UDS bind
/// failure aborts before the lock is written rather than leaving a record with
/// no cleanup path.
///
/// What: discovers the trusty sidecars, spawns the file watcher and the
/// dead-session reaper, then serves the axum router on `listener` AND the RPC
/// router on `rpc` until both drain on SIGTERM/SIGINT.
///
/// # Errors
///
/// When the HTTP server exits with one.
///
/// Test: `serve_http_drains_both_listeners_on_shutdown` in
/// `tests/daemon_dual_serve.rs` drives this body through
/// [`serve_with_shutdown`]; the socket half in isolation is covered by
/// `socket_tests.rs`.
pub async fn serve_http(
    state: Arc<DaemonState>,
    listener: tokio::net::TcpListener,
    rpc: socket::BoundSocket,
) -> anyhow::Result<()> {
    serve_with_shutdown(state, listener, rpc, trusty_common::shutdown_signal()).await
}

/// This daemon's host-state refusal, read against the root it actually runs on.
///
/// Why: every background path in this module holds an `Arc<DaemonState>` and so
/// holds the daemon's framework root, but each one asked
/// [`crate::core::host_state_gate::host_state_access`], which reads `$HOME`
/// alone. A daemon isolated by its root under an untouched `$HOME` therefore
/// passed, and `reap_loop`, `orphan_gc_loop`, and the shutdown reap all shelled
/// `tmux list-sessions` / `list-panes` at the operator's real server every tick
/// (#6348 review round 2). Routing all three through one named helper is what
/// keeps them from drifting apart again — a new loop that forgets the root now
/// has to reach past this function to do it.
/// What: defers to
/// [`crate::core::host_state_gate::host_state_access_for_root`] with
/// [`DaemonState::framework_root`], returning the operator-facing sentence when
/// the environment blocks and `None` when it allows.
/// Test: `host_state_refusal_blocks_a_scratch_framework_root`,
/// `reap_loop_leaves_the_registry_alone_on_a_scratch_framework_root`;
/// `scratch_root_daemon_does_not_spawn_tmux` proves the whole quadrant
/// end-to-end with a real spawn instrument.
fn host_state_refusal(state: &DaemonState) -> Option<String> {
    crate::core::host_state_gate::host_state_access_for_root(state.framework_root()).skip_reason()
}

/// [`serve_http`]'s body, with the shutdown future supplied by the caller.
///
/// Why: [`serve_http`] waits on SIGTERM/SIGINT, which a test cannot deliver to
/// its own process without affecting the whole test binary. Taking the future
/// as a parameter is what lets `serve_http_drains_both_listeners_on_shutdown`
/// drive the REAL drain — the same fan-out, the same unlink — rather than
/// re-implementing the loop, which would pass whether or not the fan-out below
/// exists.
///
/// # Errors
///
/// As [`serve_http`].
///
/// Test: `serve_http_drains_both_listeners_on_shutdown`.
pub async fn serve_with_shutdown(
    state: Arc<DaemonState>,
    listener: tokio::net::TcpListener,
    rpc: socket::BoundSocket,
    shutdown: impl std::future::Future<Output = ()> + Send + 'static,
) -> anyhow::Result<()> {
    // #5784: report a gated environment as gated. Without this branch the
    // refusal reads as "tmux not found", which sends an operator hunting for a
    // missing binary instead of showing them the reassigned `$HOME` and the
    // variable that lifts the gate.
    match crate::core::host_state_gate::host_state_access().skip_reason() {
        Some(reason) => tracing::warn!(
            "tmux and host-process discovery are DISABLED for this daemon — {reason}"
        ),
        None if tmux::TmuxDriver::is_available() => info!("tmux control model available"),
        None => info!("tmux not found — sessions will need the PTY or SDK control model"),
    }

    // #6285: neither sidecar has an address to discover any more — trusty-memory
    // (#6286) and trusty-search both serve a socket derived from the data dir,
    // so the startup banner names the path rather than a port read off disk.
    match search_rpc::search_socket() {
        Ok(socket) => info!("trusty-search at {}", socket.display()),
        Err(e) => tracing::warn!("cannot resolve the trusty-search socket: {e:#}"),
    }

    // Auto-discover existing Claude Code sessions — both tmux panes and native
    // Terminal.app processes — so they appear in the dashboard and the Telegram
    // bot without a manual `/adopt`.
    let discovered = discovery::discover_all(&state).await;
    if discovered.adopted > 0 {
        info!(
            "auto-discovered {} Claude Code session(s)",
            discovered.adopted
        );
    }

    // Create a cancellation token so background loops exit cleanly on shutdown.
    // Each spawned loop receives a child token; the parent is cancelled inside
    // `with_graceful_shutdown` AFTER axum drains in-flight requests.
    let cancel = tokio_util::sync::CancellationToken::new();

    // Spawn the multi-session file watcher as a background task.
    // Pass a child cancel token so the watcher exits cleanly on daemon shutdown.
    let fw = watcher::FileWatcher::new(Arc::clone(&state));
    tokio::spawn(fw.spawn(cancel.child_token()));

    // #8233 (owner ruling 2026-09-18): give the session manager the runtime
    // relaunch its AUTOMATIC resume paths need, before anything can sweep. Until
    // this is installed `resume_auto` prepares a pane and marks the record
    // `Active` with nothing in it — the fleet state a daemon restart produced.
    // The boot reconcile inside `session_manager()` defers its own auto-resume
    // to the supervisor for exactly that reason; see `session_manager::reconcile`.
    managed_routes::auto_relaunch::install_auto_relauncher(&state).await;
    // #8233 review round 2 (finding 4): sweep launch specs abandoned by an
    // earlier process. Each carries `GH_TOKEN` and `CLAUDE_CODE_OAUTH_TOKEN` in
    // cleartext, and the per-launch sweep only runs when a NEXT launch happens.
    crate::runtime::launch_spec::reap_orphans();

    // Spawn the periodic dead-session reaper with a cancel token.
    tokio::spawn(reap_loop(Arc::clone(&state), cancel.child_token()));

    // Spawn the idle auto-stop reaper if enabled (#1816).
    // Gated entirely behind `idle_auto_stop.enabled`; no-op when disabled.
    spawn_idle_reaper_if_enabled(Arc::clone(&state), cancel.child_token()).await;

    // Orphan-GC: clean accumulated leaked managed sessions. Runs ONE sweep now
    // (boot cleanup of orphans inherited across restarts) and then periodically.
    // Default ON; set TRUSTY_MPM_ORPHAN_GC=0 to disable entirely.
    if orphan_gc_enabled() {
        tokio::spawn(orphan_gc_loop(Arc::clone(&state), cancel.child_token()));
    } else {
        info!("orphan-GC disabled via TRUSTY_MPM_ORPHAN_GC");
    }

    // #7504: automatic post-merge worktree reclaim. Disk management is a daemon
    // function (epic #7505), so this is ON by default — a reclaim the PM must
    // remember to run is the manual step #2919 already shipped. The env gate, the
    // cadence and the spawn all live in the service module, which keeps this file
    // under its SLOC cap and keeps the policy next to the loop it governs.
    services::merged_pr_reclaim::spawn_if_enabled(Arc::clone(&state), cancel.child_token());
    // #8451: evict idle builder-slot dirs once their volume passes its budget.
    services::slot_pool_evict::spawn_if_enabled(Arc::clone(&state), cancel.child_token());
    project_adoption::spawn_startup_marker_migration(&state); // #8511

    // Cloud log drain (#6535): OFF unless `log_drain.enabled` is true, so the
    // default host spawns nothing. A malformed section is reported here and
    // recorded for the `log_drain` doctor row rather than silently skipped —
    // an operator who configured a drain must not have to guess why no bytes
    // moved. The loop never crashes the daemon: a failed pass warns and retries
    // on the next tick.
    {
        let home = dirs::home_dir().unwrap_or_else(|| std::path::PathBuf::from("."));
        let framework_root = crate::core::paths::FrameworkPaths::default().root;
        let tt_config = crate::core::trusty_tools_config::TrustyToolsConfig::load();
        match log_drain::should_spawn(&tt_config, &home) {
            Ok(true) => {
                tokio::spawn(log_drain::log_drain_loop(
                    framework_root,
                    home,
                    cancel.child_token(),
                ));
            }
            Ok(false) => info!("log drain disabled (no `log_drain:` section, or enabled: false)"),
            Err(reason) => {
                tracing::error!("log drain NOT started — invalid `log_drain:` config: {reason}");
                // Persist the verdict so `tm doctor` reports it even though no
                // pass ever ran.
                let dir = log_drain::state_dir(&framework_root);
                log_drain::save_status(&dir, &log_drain::config_error_status(&reason));
            }
        }
    }

    // Run inproject-hygiene for all managed base clones (#1709): fetch,
    // safety-gated `git merge --ff-only` (#2177) to the default branch, prune
    // stale worktrees. It never runs `git reset --hard`. #7965 moved the env
    // gate, the #5784 host-state refusal and the maintenance lane into the
    // service module, so this file carries one line per background loop and the
    // bound lives next to the sweep it governs.
    managed_routes::inproject_hygiene_sweep::spawn_if_enabled(Arc::clone(&state));

    // #3304 + ADR-0011 loopback-only doctrine (#3330): the daemon binds
    // loopback ONLY — there is no secondary (Tailscale) listener anymore, so
    // the router-wide guard always trusts just the loopback default.
    // Non-loopback ingress is exclusively trusty-console's job.
    let app = api::origin_guard::guard_router(
        api::router(Arc::clone(&state)),
        trusty_common::server::SelfOrigins::default(),
    );
    // Issue #2332: the startup banner carries the build version + PID so a
    // `~/Library/Logs/trusty-mpm/stderr.log` timeline can be correlated
    // against `cargo install` / restart events without guessing which
    // binary a long-lived daemon is actually running.
    info!(
        "daemon listening (trusty-mpm v{} pid {}); press Ctrl-C to stop",
        env!("CARGO_PKG_VERSION"),
        std::process::id()
    );
    // `into_make_service_with_connect_info` makes the peer `SocketAddr` available
    // to handlers via `ConnectInfo` — the loopback-only `POST /rpc` gate (#1221)
    // depends on it. Without connect-info the `ConnectInfo` extractor would 500.
    //
    // `with_graceful_shutdown` lets the daemon reap every live tmux session it
    // owns before the process exits (#1455). Without it the `reap_loop` /
    // `orphan_gc_loop` tasks are simply abandoned on SIGTERM and the sessions
    // they track leak. The shutdown future: waits for signal, cancels all
    // background actor loops (so they exit cleanly rather than being dropped
    // mid-sweep), then walks BOTH registries and kills each live session,
    // fail-open per session. Also calls `manager.shutdown()` to graceful-stop
    // all live managed sessions via SIGTERM-before-kill.
    //
    // #6288: `shutdown` is awaited ONCE, by the task below, and fanned out to
    // both listeners through a token. Awaiting a signal separately in each would
    // work on SIGTERM and fail on Ctrl-C, where `tokio::signal` delivers to
    // whichever waiter is registered — one listener would drain and the other
    // would keep accepting.
    let drain = tokio_util::sync::CancellationToken::new();
    let signal_watcher = drain.clone();
    tokio::spawn(async move {
        shutdown.await;
        signal_watcher.cancel();
    });

    let rpc_drain = drain.clone();
    // #6288 slice 2: the socket serves the SAME state the axum router was built
    // with, so the two transports are two doors onto one daemon, not two.
    let rpc_state = Arc::clone(&state);
    let rpc_task = tokio::spawn(async move {
        socket::serve_until_shutdown(rpc, rpc_state, rpc_drain.cancelled_owned()).await;
    });

    let http_drain = drain.clone();
    let reaper_state = Arc::clone(&state);
    let http_result = axum::serve(
        listener,
        app.into_make_service_with_connect_info::<std::net::SocketAddr>(),
    )
    .with_graceful_shutdown(async move {
        http_drain.cancelled_owned().await;
        // Signal all background loops to stop before we walk the session list.
        cancel.cancel();
        // Graceful-stop live SessionManager sessions (SIGTERM → 2s wait → kill).
        // This is the single authority for SM-tracked sessions, and #9101 it
        // stops only the ones it proves it owns. The legacy-registry pass
        // that follows (`reap_all_live_sessions`) kills nothing: a legacy
        // entry is a bare tmux name, which proves no ownership.
        let mgr = reaper_state.session_manager().await;
        mgr.shutdown().await;
        // Report the legacy-registry sessions, left running.
        reap_all_live_sessions(reaper_state).await;
    })
    .await;

    // #6288: cancel unconditionally, so the socket is unlinked even when axum
    // exits for a reason that was never a shutdown signal. Cancelling a token
    // the signal watcher already cancelled is a no-op, so the ordinary drain is
    // unaffected. The `?` waits until AFTER the join for the same reason —
    // returning early would leave a socket file no successor bound.
    drain.cancel();
    if let Err(e) = rpc_task.await {
        tracing::warn!(error = %e, "the rpc serve task did not exit cleanly");
    }
    http_result?;
    Ok(())
}

// #6288: the private `shutdown_signal` that used to live here was a
// line-for-line copy of `trusty_common::shutdown_signal` — same SIGTERM stream,
// same Ctrl-C race, same fallback when the handler fails to register. The
// signal is now awaited once, through the shared entry point, and fanned out to
// both listeners; see `serve_http`'s drain token.

/// Report the LEGACY DaemonState registry's sessions on shutdown, killing none.
///
/// Why: on graceful shutdown no session may be left fire-and-forget (#1452,
/// #1455), and `SessionManager::shutdown`, run first, stops every managed
/// session it proves it owns. A legacy registry entry records only a tmux
/// name, which proves nothing: boot discovery adds ANY claude pane, and a
/// restarted tmux server hands the name to an unrelated session. #9101: the
/// daemon kills only sessions it proves it owns, so this pass kills none.
/// What: collects the `tmux_name` of every legacy registry entry (NOT the
/// `SessionManager` store) and logs each as left running through
/// [`legacy_reap::skip_legacy_names`]. It never runs tmux, so it needs no
/// host-state gate (#6348) and never mutates the registry.
/// Test: `reap_all_live_sessions_is_safe_when_empty`;
/// `reap_all_live_sessions_refuses_on_a_scratch_framework_root`;
/// `the_shutdown_legacy_pass_kills_no_legacy_session` (#9101).
async fn reap_all_live_sessions(state: Arc<DaemonState>) {
    let names: std::collections::HashSet<String> = state
        .list_sessions()
        .into_iter()
        .map(|s| s.tmux_name)
        .collect();
    let skipped = legacy_reap::skip_legacy_names(names);
    info!("graceful shutdown: left {skipped} legacy session(s) running (#9101)");
}

/// Spawn the idle auto-stop background loop if `idle_auto_stop.enabled = true`.
///
/// Why: the feature is opt-in and must have ZERO overhead when disabled — this
/// function is a thin gate that reads config and early-returns with a debug log
/// when the feature flag is off. When enabled it builds a verdict provider from
/// the daemon's `ActivityMonitor` and hands off to `idle_reaper_loop`.
/// What: reads `MpmConfig::load_default().idle_auto_stop`; if disabled, returns
/// immediately. If enabled, acquires the session manager and spawns
/// `idle_reaper::idle_reaper_loop` as a background task with the provided
/// `cancel` token.
/// Test: the loop itself is tested in `idle_reaper::tests`; this wrapper is
/// covered by the integration smoke-test (daemon starts without the feature).
async fn spawn_idle_reaper_if_enabled(
    state: Arc<DaemonState>,
    cancel: tokio_util::sync::CancellationToken,
) {
    let cfg = crate::core::config::MpmConfig::load_default().idle_auto_stop;
    if !cfg.enabled {
        tracing::debug!("idle-reaper: disabled (idle_auto_stop.enabled = false); skipping");
        return;
    }
    info!(
        dry_run = cfg.dry_run,
        poll_secs = cfg.poll_interval_secs,
        idle_threshold = cfg.idle_consecutive_threshold,
        done_threshold = cfg.done_consecutive_threshold,
        "idle-reaper: enabled; spawning background loop{}",
        if cfg.dry_run {
            " (report-only — set idle_auto_stop.dry_run=false to enact teardown)"
        } else {
            ""
        }
    );
    let mgr = state.session_manager().await;
    // Cache mgr in the provider so verdict() does not re-acquire the lock on every poll.
    let provider = Arc::new(DaemonVerdictProvider {
        mgr: Arc::clone(&mgr),
        state,
    });
    tokio::spawn(idle_reaper::idle_reaper_loop(mgr, cfg, provider, cancel));
}

/// Production `IdleVerdictProvider` backed by the daemon's `ActivityMonitor`.
///
/// Why: the idle reaper trait is injectable so unit tests can avoid LLM calls;
/// the production implementation simply calls `ActivityMonitor::check` with the
/// pane content captured from `DaemonState`.
/// What: holds a cached `Arc<SessionManager>` (avoiding a lock acquisition per
/// poll) plus `Arc<DaemonState>` for the shared `ActivityMonitor`. Converts
/// `ActivityState` to a stable string via enum match — never via `{:?}` Debug
/// repr which is not a stable API contract.
/// Test: the trait is tested with `MockVerdictProvider` in `idle_reaper::tests`.
struct DaemonVerdictProvider {
    /// Cached session manager handle — acquired once at construction.
    mgr: Arc<crate::session_manager::SessionManager>,
    state: Arc<DaemonState>,
}

impl idle_reaper::IdleVerdictProvider for DaemonVerdictProvider {
    async fn verdict(&self, session_name: &str) -> Option<String> {
        use crate::activity::ActivityState;
        // Capture the most recent output of the session's RECORDED harness pane
        // (300 lines is enough for the LLM classifier). Routing through
        // `capture_pane_by_tmux_name` rather than the raw session-scoped
        // `tmux_driver().capture()` (#2515) ensures we classify the pane this
        // record actually owns, not whichever sibling window an operator left
        // active — otherwise a busy session can be auto-stopped as idle (or a
        // quiet one kept alive). `None` when no live record matches the name, the
        // recorded pane is confirmed gone, or the capture is empty: in every case
        // we skip this poll rather than act on ambiguous content.
        let pane_text = self
            .mgr
            .capture_pane_by_tmux_name(session_name, 300)
            .await?;
        // Use the shared ActivityMonitor (hash-cached — no LLM call if content unchanged).
        let monitor = self.state.activity_monitor();
        match monitor.check(session_name, &pane_text).await {
            Ok(result) => {
                // Match on the enum directly — Debug repr is not a stable contract.
                let s = match result.verdict.state {
                    ActivityState::Idle => "idle",
                    ActivityState::Done => "done",
                    ActivityState::Working => "working",
                    ActivityState::BlockedOnPermission => "blocked-on-permission",
                    ActivityState::Errored => "errored",
                    ActivityState::Unknown => "unknown",
                };
                Some(s.to_string())
            }
            Err(e) => {
                tracing::debug!(name = %session_name, "idle-reaper: activity check failed: {e}");
                None
            }
        }
    }
}

/// Interval between dead-session reap sweeps. #9010: also how recent a hook
/// event must be to hold a settled session whose `claude` is gone.
pub(crate) const REAP_INTERVAL_SECS: u64 = 60;

/// Periodically prune registry entries whose tmux session has exited.
///
/// Why: without housekeeping, dead sessions accumulate in `DaemonState`
/// forever; a slow background sweep keeps the registry honest.
/// What: every [`REAP_INTERVAL_SECS`] seconds, discovers tmux and calls
/// [`DaemonState::reap_dead_sessions`]; exits cleanly when `cancel` fires
/// rather than being dropped mid-sweep on SIGTERM. Also runs
/// [`DaemonState::sweep_delegations`] (#2864) — unconditionally, since bounding
/// delegation liveness and map growth has nothing to do with tmux being
/// discoverable, and it is here rather than on the `PreToolUse` hot path
/// precisely because it is an O(N) pass. #6556: and
/// [`services::stop_spool_drain::drain_unposted_stops`] first, so a stop a hook
/// parked while this daemon was down resolves its delegation instead of the
/// sweep aging the record out six hours later.
/// Test: the reaping rule is unit-tested via `DaemonState::reap_against`;
/// `cancel_token_stops_reap_loop_cleanly` asserts the loop exits on cancel.
async fn reap_loop(state: Arc<DaemonState>, cancel: tokio_util::sync::CancellationToken) {
    let mut tick = tokio::time::interval(std::time::Duration::from_secs(REAP_INTERVAL_SECS));
    loop {
        tokio::select! {
            _ = cancel.cancelled() => {
                info!("reap_loop: cancel signal received; exiting");
                break;
            }
            _ = tick.tick() => {
                // #6556: BEFORE the sweep. A stop a hook parked on disk while
                // this daemon was down is the truth about a delegation the
                // sweep would otherwise leave Running for six hours; replaying
                // it first means the sweep sees the recovered record, not the
                // stale one it would have had to age out.
                crate::daemon::services::stop_spool_drain::drain_unposted_stops(&state).await;
                let sweep = state.sweep_delegations();
                if sweep.staled > 0 || sweep.evicted > 0 {
                    info!(
                        "delegations: {} marked stale (no terminal signal), {} evicted (#2864)",
                        sweep.staled, sweep.evicted
                    );
                }
                // #5784: a gated environment used to fall through this
                // `if let` silently, every tick, forever. Name it once per
                // sweep so the operator sees the reason and the hatch.
                // #6348: root-aware, not $HOME-only — this loop holds the
                // daemon's framework root, so it asks about the root it has.
                if let Some(reason) = host_state_refusal(&state) {
                    tracing::warn!("reap_loop: skipped — {reason}");
                } else if let Ok(driver) = tmux::TmuxDriver::discover() {
                    let result = state.reap_dead_sessions(&driver);
                    if result.reaped > 0 {
                        info!("reaped {} dead session(s)", result.reaped);
                    }
                    if result.stopped > 0 {
                        info!(
                            "marked {} session(s) stopped (claude process exited)",
                            result.stopped
                        );
                    }
                    // #1744: also reap managed sessions whose tmux session has gone.
                    state.reap_dead_managed_sessions(&driver).await;
                    // #1814: transition managed sessions to Stopped when their tmux
                    // pane is still alive but the inner `claude` process has exited
                    // (the pane fell back to a bare shell). The #1744 reap above only
                    // covers a vanished tmux *session*; the orphan-GC deliberately
                    // KEEPS tracked sessions — so without this a runtime-exited
                    // managed session would stay `Active` forever.
                    let mgr = state.session_manager().await;
                    let stopped = runtime_reap::reap_runtime_exited_managed(
                        &mgr,
                        &driver,
                        &orphan_gc::ProcessTreeProbe,
                    )
                    .await;
                    if stopped > 0 {
                        info!(
                            "marked {stopped} managed session(s) stopped (runtime process exited, #1814)"
                        );
                    }
                }
            }
        }
    }
}

/// Default interval between orphan-GC sweeps, in seconds.
const ORPHAN_GC_INTERVAL_SECS: u64 = 60;

/// Whether the orphan-GC is enabled (default ON).
///
/// Why: operators need an escape hatch to disable the reaper entirely (e.g. on a
/// host where session tracking is known-incomplete and false positives would be
/// catastrophic). Defaulting ON keeps the self-healing behaviour without opt-in.
/// What: reads `TRUSTY_MPM_ORPHAN_GC` and delegates to the pure
/// [`parse_orphan_gc_enabled`]; a thin wrapper so the parsing is testable
/// without mutating process-global env.
/// Test: parsing covered by `parse_orphan_gc_enabled_*` below.
fn orphan_gc_enabled() -> bool {
    parse_orphan_gc_enabled(std::env::var("TRUSTY_MPM_ORPHAN_GC").ok().as_deref())
}

/// Pure parse of the `TRUSTY_MPM_ORPHAN_GC` raw value into an on/off decision.
///
/// Why: env vars are process-global, so testing the policy by mutating them is
/// racy under the multi-threaded test harness. Taking the raw value as a
/// parameter makes the decision a pure function that needs no env mutation.
/// What: returns `false` only for an explicit `0`/`false`/`off`/`no`
/// (case-insensitive, trimmed); `true` for any other value, including `None`
/// (unset) — i.e. default ON.
/// Test: `parse_orphan_gc_enabled_default_and_overrides`.
fn parse_orphan_gc_enabled(raw: Option<&str>) -> bool {
    match raw {
        Some(v) => {
            let v = v.trim().to_ascii_lowercase();
            !matches!(v.as_str(), "0" | "false" | "off" | "no")
        }
        None => true,
    }
}

/// Resolve the orphan-GC sweep interval, honouring an env override.
///
/// Why: long-running hosts may want a slower (or faster) cadence than the
/// 60-second default; the debounce is expressed in *passes*, so the interval
/// also sets the effective grace window (a fresh orphan survives ≥ one interval).
/// What: reads `TRUSTY_MPM_ORPHAN_GC_INTERVAL_SECS` and delegates to the pure
/// [`parse_orphan_gc_interval`]; a thin wrapper so the parsing is testable
/// without mutating process-global env.
/// Test: parsing covered by `parse_orphan_gc_interval_*` below.
fn orphan_gc_interval_secs() -> u64 {
    parse_orphan_gc_interval(
        std::env::var("TRUSTY_MPM_ORPHAN_GC_INTERVAL_SECS")
            .ok()
            .as_deref(),
    )
}

/// Pure parse of the `TRUSTY_MPM_ORPHAN_GC_INTERVAL_SECS` raw value to seconds.
///
/// Why: as with [`parse_orphan_gc_enabled`], keeping the parsing pure avoids
/// racy process-global env mutation in tests.
/// What: parses `raw` as a positive `u64`, falling back to
/// [`ORPHAN_GC_INTERVAL_SECS`] when `None`, unparsable, or non-positive (a zero
/// interval would busy-loop).
/// Test: `parse_orphan_gc_interval_override`.
fn parse_orphan_gc_interval(raw: Option<&str>) -> u64 {
    raw.and_then(|v| v.trim().parse::<u64>().ok())
        .filter(|n| *n > 0)
        .unwrap_or(ORPHAN_GC_INTERVAL_SECS)
}

/// Periodically reconcile live tmux against the registries and reap orphans.
///
/// Why: leaked managed (`tm-*`/`tmpm-*`) tmux sessions (from crashes or pre-RAII builds) must be
/// cleaned up self-healingly. One sweep runs immediately at boot to clear
/// accumulated orphans, then the loop runs on the configured interval.
/// What: owns a single [`orphan_gc::OrphanGc`] so the two-pass debounce persists
/// across sweeps; exits cleanly when `cancel` fires rather than being dropped
/// mid-sweep on SIGTERM. Each tick gathers the live managed panes (via
/// [`tmux::TmuxDriver::list_managed_panes`]) and both registries' tracked names
/// (via [`DaemonState::gather_tracked_names`]), then calls
/// [`orphan_gc::run_sweep`]. A tmux-listing failure skips the pass (reaps
/// nothing) rather than risking a wrong kill; likewise, if
/// [`DaemonState::gather_tracked_names`] returns a *degraded* snapshot (a
/// registry read failed, so the protected set is incomplete), `run_sweep`
/// skips its reap phase for that tick — both paths fail CLOSED.
/// Test: the reconciliation rule and debounce are unit-tested in
/// [`orphan_gc`]; the end-to-end sweep is covered by `tests/orphan_gc_sweep.rs`.
async fn orphan_gc_loop(state: Arc<DaemonState>, cancel: tokio_util::sync::CancellationToken) {
    let interval_secs = orphan_gc_interval_secs();
    info!(
        interval_secs,
        "orphan-GC enabled; running startup sweep then periodic reconciliation"
    );
    // #6118: the untracked-active skip log repeats once every N sweeps per
    // session rather than every sweep; N is overridable for live debugging.
    let mut gc = orphan_gc::OrphanGc::from_env();
    // Cross-restart debounce for the PID-file sweep (#1918): owned here (not
    // reconstructed per-tick) so its two-pass candidate history persists across
    // sweeps, exactly like `gc` above.
    let mut pid_gc = crate::core::pid_registry::PidOrphanGc::new();
    // Same two-observation discipline as `gc` and `pid_gc`, and owned here for
    // the same reason: a record must be evictable on two consecutive sweeps
    // before retention deletes it.
    let mut retention_gc = crate::session_manager::RetentionDebounce::new();
    let probe = orphan_gc::ProcessTreeProbe;
    let mut tick = tokio::time::interval(std::time::Duration::from_secs(interval_secs));
    loop {
        tokio::select! {
            _ = cancel.cancelled() => {
                info!("orphan_gc_loop: cancel signal received; exiting");
                break;
            }
            _ = tick.tick() => {
                // Record retention runs FIRST, ahead of the tmux probe below,
                // which `continue`s the whole tick when tmux is not
                // discoverable. A store-only sweep needs neither tmux nor live
                // pane data, so running it after that probe would tie it to an
                // unrelated dependency. This does NOT make it unconditional:
                // `TRUSTY_MPM_ORPHAN_GC=0` disables this entire loop, retention
                // included.
                {
                    let mgr = state.session_manager().await;
                    match mgr.sweep_default_retention(&mut retention_gc).await {
                        Ok(o) if !o.is_empty() => info!(
                            stamped = o.stamped,
                            evicted = o.evicted.len(),
                            "record retention sweep"
                        ),
                        Ok(_) => {}
                        Err(e) => tracing::warn!("record retention sweep failed: {e}"),
                    }
                }
                // #5784: same silent-`continue` trap as `reap_loop` above.
                // #6348: and root-aware for the same reason it is there.
                if let Some(reason) = host_state_refusal(&state) {
                    tracing::warn!("orphan-GC: skipped — {reason}");
                    continue;
                }
                // #4323: reclaim empty `<framework root>/sessions/<id>/` holders.
                // Scoped to `state.framework_root()` — the daemon's OWN state
                // dir — never `dirs::home_dir()`. A scratch-rooted daemon and
                // the operator's real `$HOME` are different directories, and a
                // sweep that resolved home would delete the operator's session
                // dirs from a test process (#6348's quadrant).
                //
                // Below `host_state_refusal` for the same reason every other
                // sweep in this loop is, and above `TmuxDriver::discover()`
                // because it needs no tmux: tying a state-dir sweep to tmux
                // discoverability would be an unrelated dependency. Note this
                // is NOT the retention sweep's placement — that one runs above
                // the gate because it goes through `state.session_manager()`,
                // which is root-scoped by construction; this one is not
                // analogous and must not be moved up beside it.
                //
                // Logs ONE count per sweep — the per-path shape is what put
                // 208K lines/day in the operator's log. Fail-closed and
                // non-recursive; see `core::session_store`.
                let sessions_dir =
                    crate::core::session_store::sessions_root_under(state.framework_root());
                match crate::core::session_store::reap_empty_session_dirs_at(&sessions_dir) {
                    Ok(sweep) if !sweep.is_empty() => info!(
                        removed = sweep.removed,
                        skipped = sweep.skipped,
                        failed = sweep.failed,
                        "empty session-dir sweep"
                    ),
                    Ok(_) => {}
                    Err(e) => tracing::warn!("empty session-dir sweep failed: {e}"),
                }
                let Ok(driver) = tmux::TmuxDriver::discover() else {
                    continue;
                };
                let panes = match driver.list_managed_panes() {
                    Ok(p) => p,
                    Err(e) => {
                        tracing::warn!("orphan-GC: list_managed_panes failed: {e}");
                        continue;
                    }
                };
                // NOTE: `gather_tracked_names()` and `session_manager()` are acquired in
                // separate awaits, so this snapshot is intentionally NON-atomic — a
                // session could register in the gap between the two. That race is safe:
                // the two-pass debounce in `OrphanGc` means a just-registered session
                // missed by this sweep is only ever a reap *candidate* this pass and
                // must be seen orphaned again next sweep before it can be reaped, by
                // which point `gather_tracked_names()` will include it. No lock spanning
                // both calls is needed.
                let tracked = state.gather_tracked_names().await;
                let mgr = state.session_manager().await;
                let mtmux = mgr.tmux_driver();
                // #6118: `run_sweep_fs` supplies the filesystem cwd probe, which
                // is what lets the sweep reap a pane whose worktree was deleted
                // underneath it — see `orphan_gc::CwdProbe`.
                let reaped =
                    orphan_gc::run_sweep_fs(&mut gc, &panes, &tracked, &probe, mtmux.as_ref());
                if reaped > 0 {
                    info!("orphan-GC reaped {reaped} orphaned managed session(s)");
                }

                // PID-file orphan-GC (§10.3): reap daemon-owned `claude` child
                // processes whose tmux pane is already gone — invisible to the
                // tmux-name sweep above. Live session-ids come from BOTH registries;
                // any recorded PID whose session-id is absent is a candidate, and
                // the registry SIGTERMs it only after verifying it is still a live
                // `claude` process (the §13.5 PID-reuse mitigation) AND the
                // candidate has survived the #1918 restart-safety debounce (`pid_gc`)
                // — a session-id must be an untracked-alive-claude candidate on TWO
                // consecutive sweeps before it is actually signalled, so a legacy
                // session that is alive across a daemon restart (when `self.sessions`
                // is always empty) gets one full GC interval to be re-learned before
                // any SIGTERM is sent.
                let live_ids = state.gather_live_session_ids().await;
                let pid_outcome = state.pid_registry().sweep_orphans(
                    &live_ids,
                    &crate::core::pid_registry::OsProcessProbe,
                    &mut pid_gc,
                );
                if pid_outcome.terminated > 0
                    || pid_outcome.removed_stale > 0
                    || pid_outcome.deferred > 0
                {
                    info!(
                        terminated = pid_outcome.terminated,
                        removed_stale = pid_outcome.removed_stale,
                        deferred = pid_outcome.deferred,
                        "orphan-GC PID-file sweep cleaned up orphaned claude process(es)"
                    );
                }
                // #1508: auto-reap STALE EPHEMERAL sessions as a backstop.
                let max_age = chrono::Duration::hours(crate::session_manager::MAX_EPHEMERAL_AGE_HOURS);
                match mgr.reap_aged_ephemeral(max_age).await {
                    Ok(n) if n > 0 => info!("ephemeral auto-reap decommissioned {n} stale session(s)"),
                    Ok(_) => {}
                    Err(e) => tracing::warn!("ephemeral auto-reap failed: {e}"),
                }

                // #1838: reap orphaned managed-clone worktree dirs whose session
                // records are gone. Same self-healing rationale as the tmux orphan
                // sweep above: the `.worktrees/<id>` tree under each managed base
                // clone otherwise grows without bound (94 dirs observed for one
                // project) because decommission-time cleanup (#1806/#1840) only
                // covers sessions torn down AFTER that fix landed and the manual
                // `tm session prune-worktrees` sweep is rarely run.
                //
                // What actually bounds this call (#4207/#4224 — the previous
                // wording here, "only leaf worktree dirs ... the base clone are
                // never touched", described the removed five-shape walk and is
                // no longer true: a registered worktree at ANY depth inside a
                // managed project, e.g. `<repo>/agents/scratch/wt-1`, is a
                // candidate). A directory must clear ALL of: git registers it as
                // a worktree that is not main/bare/prunable/locked — this is
                // what excludes the BASE CLONE, which is a separate
                // `git clone --bare` absent from `<repo>`'s registry entirely
                // and bare+main in its own; it is a strict descendant of the
                // managed project whose registry named it, which is what puts
                // the operator's own checkouts parked beside a project out of
                // reach by POSITION rather than by name — note this gate does
                // NOT exclude `.base`, which IS a strict descendant of `<repo>`,
                // so do not re-attribute the base clone to it; it is in neither
                // the initial nor the
                // pre-deletion active set; it carries a #3649 ownership sentinel
                // naming a provably-ownerless owner past the grace window; and
                // the #4091/#4118 dirty gate finds no uncommitted or unpushed
                // work. See `SessionManager::reap_orphaned_worktrees` for the
                // authoritative list. Inherits the orphan-GC env gate
                // (`TRUSTY_MPM_ORPHAN_GC`) since it runs inside this loop.
                let repos_root = managed_routes::inproject::repos_root();
                // #7357: this loop is the entry point for the unattended sweep,
                // so it resolves the adopted anchors under the DAEMON's own
                // framework root — the same root `project_register` writes them
                // to.
                let adopted = crate::project::adopted_anchors_under(state.framework_root());
                match mgr.reap_orphaned_worktrees(&repos_root, &adopted).await {
                    Ok(outcome) => {
                        if !outcome.removed.is_empty() {
                            info!(
                                "orphan-GC reaped {} orphaned worktree dir(s)",
                                outcome.removed.len()
                            );
                        }
                        // #3649: owner-unknown worktrees are NEVER auto-deleted —
                        // log the count so operators see them accumulating and
                        // know to run `tm doctor` / `tm session prune-worktrees
                        // --dry-run` for manual review, instead of silently
                        // wondering why disk usage keeps growing.
                        if !outcome.owner_unknown.is_empty() {
                            info!(
                                "orphan-GC left {} owner-unknown worktree dir(s) untouched \
                                 (run `tm doctor` to review, #3649)",
                                outcome.owner_unknown.len()
                            );
                        }
                        // #4311: attributed, so NOT the owner-unknown backlog
                        // above — these have an owner and a reaper, and the
                        // count is logged separately so an operator can tell
                        // "nothing may touch this" from "the agent's exit
                        // will".
                        if !outcome.agent_owned.is_empty() {
                            info!(
                                "orphan-GC left {} agent-owned worktree dir(s) to their agents' \
                                 exits (#4311)",
                                outcome.agent_owned.len()
                            );
                        }
                        // #4091: a worktree holding uncommitted or unpushed
                        // work is NEVER auto-deleted. Warn (not info) with the
                        // per-path reason — this is unsaved work sitting in a
                        // directory the sweep would otherwise have reclaimed,
                        // and the operator needs to see it, not discover it
                        // missing later.
                        for dirt in &outcome.skipped_dirty {
                            tracing::warn!(
                                path = %dirt.path.display(),
                                reason = %dirt.reason,
                                "orphan-GC left a worktree with unsaved work untouched (#4091)"
                            );
                        }
                    }
                    Err(e) => tracing::warn!("orphan-GC worktree sweep failed: {e}"),
                }

                // #2033: sweep orphaned / 0-chunk trusty-search indexes left
                // behind by managed worktrees — sessions decommissioned before
                // the decommission-time index-removal fix, or whose worktree
                // was never populated before teardown. Best-effort: a
                // discoverable-but-erroring daemon logs and is retried next tick.
                match mgr.reap_orphaned_search_indexes().await {
                    Ok(removed) if !removed.is_empty() => {
                        info!(
                            "orphan-GC removed {} orphaned trusty-search index(es)",
                            removed.len()
                        );
                    }
                    Ok(_) => {}
                    Err(e) => tracing::warn!("orphan-GC search-index sweep failed: {e}"),
                }
            }
        }
    }
}

/// Run the MCP server over stdio so a Claude Code session can call the
/// orchestration tools (`session_list`, `agent_delegate`, ...).
///
/// Why: shared by `trusty-mpmd mcp` and `trusty-mpm daemon --mcp`.
/// What: wraps [`DaemonState`] in a [`mcp_backend::StateBackend`] and pumps the
/// trusty-mcp-core stdio JSON-RPC loop.
/// Test: pipe a JSON-RPC `initialize` request to the process and observe a
/// well-formed response on stdout.
pub async fn run_mcp(state: Arc<DaemonState>) -> anyhow::Result<()> {
    info!("trusty-mpm MCP server starting on stdio");
    let backend = mcp_backend::StateBackend::new(state);
    trusty_mcp::run_stdio_loop(move |req| {
        let backend = backend.clone();
        async move { crate::mcp::dispatch(&backend, req).await }
    })
    .await
}

#[cfg(test)]
mod shutdown_reaper_tests {
    use super::{DaemonState, host_state_refusal, reap_all_live_sessions, reap_loop};
    use std::sync::Arc;

    /// A `DaemonState` rooted at a throwaway directory, plus the directory.
    ///
    /// The `$HOME` is left exactly as the test binary inherited it — that is the
    /// #6348 quadrant, and reassigning it would silence the root arm behind the
    /// `$HOME` arm that already worked.
    fn scratch_root_state() -> (Arc<DaemonState>, tempfile::TempDir) {
        let dir = tempfile::tempdir().expect("scratch framework root");
        let state = Arc::new(DaemonState::with_root(dir.path().to_path_buf()));
        (state, dir)
    }

    /// RAII guard restoring `$HOME` on drop, panic included (#6663).
    ///
    /// A copy of `session_manager::naming_tests`'s guard rather than a shared
    /// one: module-private visibility does not cross sibling module trees, and
    /// several other test modules in this crate carry the same copy for the
    /// same reason. Callers MUST be `#[serial_test::serial]` — the guard makes
    /// the mutation reversible, not atomic, and ~20 tests in this binary write
    /// `$HOME`.
    /// Test: used by
    /// `orphan_gc_loop_leaves_the_home_session_dirs_alone_on_a_scratch_root`.
    struct HomeGuard(Option<String>);
    impl Drop for HomeGuard {
        fn drop(&mut self) {
            // SAFETY: paired with `#[serial_test::serial]` — no other thread
            // reads or 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` until the returned guard drops. 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)
    }

    /// A tmux-origin session whose tmux name is certainly not live.
    fn dead_session() -> crate::core::session::Session {
        use crate::core::session::{ControlModel, Session, SessionId, SessionStatus};
        let mut s = Session::new(SessionId::new(), "/tmp/p", ControlModel::Tmux, None);
        s.tmux_name = format!("tmpm-6348-absent-{}", std::process::id());
        s.status = SessionStatus::Active;
        s
    }

    /// A scratch framework root is refused, whatever `$HOME` says.
    ///
    /// Why: this is the single gate the three background tmux paths in this
    /// module now share, so its refusal is what stops all three (#6348 review
    /// round 2).
    /// What: builds a state on a temp root and asserts the refusal exists and
    /// names the hatch. When the ambient environment is the operator's real one
    /// — the exact quadrant #6348 is about — it additionally asserts the
    /// sentence names the scratch ROOT rather than a `$HOME` mismatch, so the
    /// test cannot pass on the pre-existing `$HOME` arm alone.
    /// Test: this is the test.
    #[test]
    fn host_state_refusal_blocks_a_scratch_framework_root() {
        let (state, dir) = scratch_root_state();
        let reason = host_state_refusal(&state).expect("a scratch framework root must be refused");
        assert!(
            reason.contains(crate::core::host_state_gate::ALLOW_HOST_STATE_ENV),
            "a refusal an operator cannot lift is a dead end; got: {reason}"
        );
        if crate::core::host_state_gate::host_state_access().is_allowed() {
            assert!(
                reason.contains(&dir.path().display().to_string()),
                "under a real $HOME the refusal must be the DATA-ROOT arm, naming {}; got: {reason}",
                dir.path().display()
            );
        }
    }

    /// `reap_loop` reaps nothing on a scratch framework root.
    ///
    /// Why: the loop ticks immediately on spawn and then every
    /// [`super::REAP_INTERVAL_SECS`] seconds for the life of the daemon. Before
    /// this fix it passed the `$HOME`-only gate and swept a scratch daemon's
    /// registry against the operator's REAL tmux server, deleting every entry
    /// that server did not host.
    /// What: registers one tmux-origin session whose tmux name cannot be live,
    /// lets exactly one tick run, cancels, and asserts the entry survived. On a
    /// host with tmux installed, removing the gate fails this assertion.
    /// Test: this is the test.
    #[tokio::test]
    async fn reap_loop_leaves_the_registry_alone_on_a_scratch_framework_root() {
        let (state, _dir) = scratch_root_state();
        let session = dead_session();
        let id = session.id;
        state.register_session(session);

        let cancel = tokio_util::sync::CancellationToken::new();
        let handle = tokio::spawn(reap_loop(Arc::clone(&state), cancel.child_token()));
        // `tokio::time::interval` fires its first tick immediately, so one sweep
        // has run by the time this yields back.
        tokio::time::sleep(std::time::Duration::from_millis(250)).await;
        cancel.cancel();
        handle.await.expect("reap_loop must not panic");

        assert!(
            state.session(id).is_some(),
            "a refused sweep must reap nothing — the registry entry was removed"
        );
    }

    /// The shutdown reap refuses on a scratch framework root.
    ///
    /// Why: this path had no host-state gate at all and issued `kill_session`
    /// against the real tmux server for every name in its own registry, so a
    /// scratch daemon's shutdown could kill a live operator session that merely
    /// shared a name.
    /// What: registers one session on a scratch-rooted state and runs the reap,
    /// asserting it returns without panicking and leaves the registry intact.
    /// #9101: the pass now kills nothing on any root.
    /// Test: this is the test.
    #[tokio::test]
    async fn reap_all_live_sessions_refuses_on_a_scratch_framework_root() {
        let (state, _dir) = scratch_root_state();
        let session = dead_session();
        let id = session.id;
        state.register_session(session);

        reap_all_live_sessions(Arc::clone(&state)).await;

        assert!(
            state.session(id).is_some(),
            "the shutdown reap must not mutate the registry"
        );
        assert!(
            host_state_refusal(&state).is_some(),
            "the fixture must actually be in the refused quadrant"
        );
    }

    /// `orphan_gc_loop` completes a refused tick on a scratch framework root.
    ///
    /// Why: the loop is on by default and its sweep asks the real tmux server
    /// for every managed pane, then kills the ones no registry claims. A scratch
    /// daemon has an empty registry, so every real managed pane the operator was
    /// running looked orphaned to it.
    /// What: drives one tick on a scratch-rooted state and asserts the loop
    /// exits cleanly and the registry is untouched. It is a liveness pin, NOT a
    /// discriminating one: the sweep's kills already route through the session
    /// manager's driver, which a scratch root makes a no-op, so this test also
    /// passes with the gate removed. What actually pins the refusal is
    /// [`super::host_state_refusal`], covered by
    /// `host_state_refusal_blocks_a_scratch_framework_root`, and the absence of
    /// a tmux spawn, covered end-to-end by
    /// `scratch_root_daemon_does_not_spawn_tmux`.
    /// Test: this is the test.
    #[tokio::test]
    async fn orphan_gc_loop_ticks_cleanly_on_a_scratch_framework_root() {
        let (state, _dir) = scratch_root_state();
        let session = dead_session();
        let id = session.id;
        state.register_session(session);

        let cancel = tokio_util::sync::CancellationToken::new();
        let handle = tokio::spawn(super::orphan_gc_loop(
            Arc::clone(&state),
            cancel.child_token(),
        ));
        tokio::time::sleep(std::time::Duration::from_millis(250)).await;
        cancel.cancel();
        handle.await.expect("orphan_gc_loop must not panic");

        assert!(
            state.session(id).is_some(),
            "a refused sweep must leave the registry alone"
        );
    }

    /// The empty-session-dir sweep never touches `$HOME` (#4323, #6348).
    ///
    /// Why: the sweep DELETES directories. Its first revision called a
    /// home-resolving wrapper and ran above `host_state_refusal`, so a
    /// scratch-rooted daemon — which `scratch_root_state` deliberately leaves
    /// with the inherited `$HOME`, the #6348 quadrant — swept the operator's
    /// real `~/.trusty-mpm/sessions/`. A canary planted there was deleted by one
    /// tick of this loop. Unlike
    /// `orphan_gc_loop_ticks_cleanly_on_a_scratch_framework_root`, this test is
    /// DISCRIMINATING: it fails against that revision.
    /// What: plants an empty UUID-named directory under the sessions dir of a
    /// `$HOME` this test owns, drives the loop on a scratch-rooted state, and
    /// asserts the canary survived. Two independent things must hold for that:
    /// the sweep resolves `state.framework_root()` rather than the home
    /// directory, and it sits below the host-state gate.
    ///
    /// #6663: the canary used to be planted under `dirs::home_dir()`, i.e. the
    /// ambient `$HOME`. About twenty tests in this same lib binary redirect
    /// `$HOME` to a `TempDir`, so a canary planted while one of those overrides
    /// was live landed inside that fixture's directory and was deleted by its
    /// `Drop` — the loop under test never touched it. Owning `$HOME` here makes
    /// the canary's parent a directory this test holds for its whole duration,
    /// so no sibling's cleanup can reach it. A sibling that clobbers `$HOME`
    /// mid-test can now only send a BUGGY sweep somewhere else, which loses
    /// discrimination rather than failing; under `cargo nextest` (one process
    /// per test, the CI shards) even that cannot happen.
    ///
    /// The redirect is reversible and serialized — [`set_home`] plus
    /// `#[serial_test::serial]`. Leaving `$HOME` pointing at this test's
    /// `TempDir` after it returns would hand every later test in the binary a
    /// deleted home, which is a worse fault than the one being fixed.
    ///
    /// This is the ONE test here that leaves the inherited-`$HOME` quadrant
    /// `scratch_root_state` documents, and it stays discriminating anyway: the
    /// revision this guards against swept ABOVE the host-state gate, so which
    /// arm of that gate refuses does not decide the outcome — whether the sweep
    /// resolved the home directory does.
    /// Test: this is the test.
    #[tokio::test]
    #[serial_test::serial]
    async fn orphan_gc_loop_leaves_the_home_session_dirs_alone_on_a_scratch_root() {
        // #6663: a `$HOME` this test owns, not the operator's. Declared BEFORE
        // the guard so the guard drops FIRST — `$HOME` is restored while this
        // directory still exists, never left naming a removed path.
        let home_dir = tempfile::tempdir().expect("scratch home");
        let home = home_dir.path().to_path_buf();
        let _home = set_home(&home);

        let (state, dir) = scratch_root_state();
        assert!(
            host_state_refusal(&state).is_some(),
            "the fixture must be in the refused quadrant for this to discriminate"
        );
        assert_ne!(
            dir.path(),
            home.join(".trusty-mpm"),
            "the scratch root must not BE the home root, or the test proves nothing"
        );

        // A canary the sweep would find reclaimable: empty, hyphenated-UUID
        // named, directly under the home sessions dir.
        let canary = crate::core::session_store::sessions_root_in(&home)
            .join(uuid::Uuid::new_v4().to_string());
        std::fs::create_dir_all(&canary).expect("plant the canary");

        let cancel = tokio_util::sync::CancellationToken::new();
        let handle = tokio::spawn(super::orphan_gc_loop(
            Arc::clone(&state),
            cancel.child_token(),
        ));
        // #6663: poll for the failure instead of sleeping a fixed 250 ms and
        // looking once. `tokio::time::interval` fires its first tick
        // immediately, so a correct loop has already refused long before the
        // budget expires, while a loaded machine gets the whole budget to get
        // round to a tick that a fixed wait would have missed.
        let deadline = std::time::Instant::now() + std::time::Duration::from_secs(1);
        let mut survived = true;
        while std::time::Instant::now() < deadline {
            if !canary.is_dir() {
                survived = false;
                break;
            }
            tokio::time::sleep(std::time::Duration::from_millis(25)).await;
        }
        cancel.cancel();
        handle.await.expect("orphan_gc_loop must not panic");

        // Clean up before asserting, so a failure does not also leave litter.
        let _ = std::fs::remove_dir(&canary);
        assert!(
            survived,
            "a scratch-rooted daemon deleted {} out of the home session tree",
            canary.display()
        );
    }

    /// The shutdown reaper is a safe no-op on an empty daemon.
    ///
    /// Why: graceful shutdown must never panic, whether or not tmux is present
    /// and whether or not any session is tracked (#1455). With no registered
    /// sessions there is nothing to kill, so the sweep must complete cleanly —
    /// returning early when tmux is unavailable, or reaping zero when it is.
    /// What: builds a fresh in-memory `DaemonState` (no sessions) and runs the
    /// reaper, asserting only that it returns without panicking.
    /// Test: this is the test.
    #[tokio::test]
    async fn reap_all_live_sessions_is_safe_when_empty() {
        let state: Arc<DaemonState> = DaemonState::shared();
        // Must not panic regardless of tmux availability on the host.
        reap_all_live_sessions(state).await;
    }

    /// Cancelling the token causes `reap_loop` to exit without panic.
    ///
    /// Why: the WI-4 PR A requirement is that background loops respond to the
    /// cancellation token rather than being dropped mid-sweep on SIGTERM. This
    /// test asserts that the loop task completes cleanly (join returns `Ok`)
    /// after the token is cancelled, and never panics.
    /// What: spawns `reap_loop` with a fresh token, immediately cancels it,
    /// then awaits the task handle and asserts the join succeeded.
    /// Test: this is the test.
    #[tokio::test]
    async fn cancel_token_stops_reap_loop_cleanly() {
        let state: Arc<DaemonState> = DaemonState::shared();
        let cancel = tokio_util::sync::CancellationToken::new();
        let token = cancel.clone();
        let handle = tokio::spawn(reap_loop(state, token));
        // Cancel immediately — the loop should exit on the next select! poll.
        cancel.cancel();
        // The task must complete without panicking (join returns Ok).
        handle
            .await
            .expect("reap_loop task must not panic after cancellation");
    }
}

#[cfg(test)]
mod orphan_gc_config_tests {
    use super::{ORPHAN_GC_INTERVAL_SECS, parse_orphan_gc_enabled, parse_orphan_gc_interval};

    /// `TRUSTY_MPM_ORPHAN_GC` defaults ON and only explicit falsey values disable.
    ///
    /// Tests the PURE parser with raw values — no process-global env mutation,
    /// so it is safe to run in parallel with any sibling test (the #1458 review
    /// fix for env-var test flakiness).
    #[test]
    fn parse_orphan_gc_enabled_default_and_overrides() {
        assert!(parse_orphan_gc_enabled(None), "unset must default ON");

        for off in ["0", "false", "off", "no", "OFF", " false "] {
            assert!(!parse_orphan_gc_enabled(Some(off)), "{off:?} must disable");
        }
        for on in ["1", "true", "yes", "anything", "  "] {
            assert!(
                parse_orphan_gc_enabled(Some(on)),
                "{on:?} must keep enabled"
            );
        }
    }

    /// The interval override parses positive seconds and falls back otherwise.
    ///
    /// Tests the PURE parser with raw values — no env mutation (see #1458).
    #[test]
    fn parse_orphan_gc_interval_override() {
        assert_eq!(parse_orphan_gc_interval(None), ORPHAN_GC_INTERVAL_SECS);
        assert_eq!(parse_orphan_gc_interval(Some("30")), 30);
        assert_eq!(parse_orphan_gc_interval(Some("  45 ")), 45);

        // Zero and garbage fall back to the default (a zero interval would busy-loop).
        for bad in ["0", "nonsense", "-5", ""] {
            assert_eq!(
                parse_orphan_gc_interval(Some(bad)),
                ORPHAN_GC_INTERVAL_SECS,
                "{bad:?}"
            );
        }
    }
}