trusty-mpm 1.8.2

//! Record → wire-shape conversion helpers shared across the managed-session
//! route handlers.
//!
//! Why: `managed_routes/mod.rs` was at the 500-SLOC production cap; these
//! conversion helpers are pure and self-contained, so extracting them (rather
//! than a handler) keeps `mod.rs` focused on request/response shapes and
//! routing while leaving every call site's import path unchanged (all three
//! non-JSON helpers are re-exported through `mod.rs` at their original
//! visibility).
//! What: [`record_to_json`] (MCP wire shape, crate-public), [`record_to_summary`]
//! (HTTP wire shape), [`attach_cmd_for`], and [`parse_id`].
//! Test: `serializers_include_source_id` in `super::tests`; the handler tests
//! throughout `managed_routes` exercise these indirectly.

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

use axum::http::StatusCode;

use crate::core::manifest::HarnessPlan;
use crate::core::paths::FrameworkPaths;
use crate::core::session_assets::session_plan_under;
use crate::core::update_check::{CatalogHashes, DeployedAgentHashes};
use crate::session_manager::{
    InjectionStatus, ManagedSessionId, ManagedSessionState, ManagedTmuxDriver, NumberedSlot,
    SessionRecord,
};

use super::SessionSummary;

/// Serialize a [`SessionRecord`] to the flat JSON shape the MCP tools return.
///
/// Why: the MCP tools return JSON values (not axum responses); reusing the same
/// field set as `SpawnResponse`/[`SessionSummary`] keeps the MCP and HTTP
/// payloads consistent for the driver skill. `source_id` is included so MCP
/// callers can filter or reconnect by project identity, matching what the HTTP
/// `GET /api/v1/sessions/managed` path already exposes (#1733).
/// What: maps the record to a JSON object including the derived `attach_cmd`
/// and `source_id` (null when the session has no project identity).
/// Test: `serializers_include_source_id` unit test in `super::tests`; also
/// covered by `crate::daemon::mcp_session` tests that assert echoed fields.
pub fn record_to_json(r: &SessionRecord) -> serde_json::Value {
    serde_json::json!({
        "id": r.id.to_string(),
        "name": r.tmux_name,
        "state": r.state.to_string(),
        "workspace_path": r.workspace_path.as_ref().map(|p| p.to_string_lossy().to_string()),
        "cwd": r.cwd.to_string_lossy().to_string(),
        "task": r.task,
        "repo_url": r.repo_url,
        "branch": r.branch,
        "created_at": r.created_at.to_rfc3339(),
        "last_activity_at": r.last_activity_at.map(|t| t.to_rfc3339()),
        "attach_cmd": attach_cmd_for(&r.tmux_name),
        "runtime": r.runtime.as_str(),
        "pending_decision": r.pending_decision,
        "proposed_default": r.proposed_default,
        "source_id": r.source_id,
        "deliverable_id": r.deliverable_id.map(|id| id.to_string()),
        "pane_id": r.pane_id,
        "injection_status": injection_status_wire(r.injection_status),
    })
}

/// Map [`InjectionStatus`] to its wire form, `None` for "never attempted"
/// (#2364).
///
/// Why: shared by [`record_to_json`] and [`record_to_summary`] so the MCP and
/// HTTP wire shapes cannot drift on what "no injection happened" looks like —
/// both omit the field (JSON `null`/absent) rather than emitting the literal
/// `"not_applicable"` string, keeping the field silent for the (common)
/// sessions injection never applies to.
/// What: `NotApplicable` → `None`; every other variant → `Some(<snake_case>)`
/// via [`InjectionStatus`]'s `Display`.
/// Test: `injection_status_wire_omits_not_applicable`,
/// `injection_status_wire_stringifies_other_variants` in `super::tests`.
fn injection_status_wire(status: InjectionStatus) -> Option<String> {
    match status {
        InjectionStatus::NotApplicable => None,
        other => Some(other.to_string()),
    }
}

/// Convert a [`SessionRecord`] into a wire [`SessionSummary`].
///
/// Why: the API exposes a flat, string-typed summary so clients don't depend on
/// the internal record shape.
/// What: maps every record field to its serialized form. `unresumable` always
/// starts `false` here — it requires an async filesystem probe
/// (`session_manager::resume_workdir::is_unresumable`), which this function
/// cannot perform since it is a pure/sync conversion shared by handlers that
/// have no reason to pay that cost (spawn/reactivate/decommission are never
/// mid-flight through the dead-workspace predicate). The list/get handlers
/// (#2595) overwrite the field on the returned value when they can await the
/// probe.
/// Test: covered by the list/get handler tests.
pub(crate) fn record_to_summary(r: &SessionRecord) -> SessionSummary {
    SessionSummary {
        id: r.id.to_string(),
        name: r.tmux_name.clone(),
        state: r.state.to_string(),
        // #3531: captured once, here, BEFORE any later display-only
        // reconciliation (`reconcile_live_state`) can overwrite `state` above
        // — see `SessionSummary::persisted_state`'s doc for why resume/restart
        // decisions must key off this instead.
        persisted_state: r.state.to_string(),
        workspace_path: r
            .workspace_path
            .as_ref()
            .map(|p| p.to_string_lossy().to_string()),
        repo_url: r.repo_url.clone(),
        branch: r.branch.clone(),
        created_at: r.created_at.to_rfc3339(),
        last_activity_at: r.last_activity_at.map(|t| t.to_rfc3339()),
        pending_decision: r.pending_decision.clone(),
        proposed_default: r.proposed_default.clone(),
        source_id: r.source_id.clone(),
        task: Some(r.task.clone()),
        cwd: Some(r.cwd.to_string_lossy().to_string()),
        claude_session_id: r.claude_session_id.clone(),
        deliverable_id: r.deliverable_id.map(|id| id.to_string()),
        pane_id: r.pane_id.clone(),
        injection_status: injection_status_wire(r.injection_status),
        unresumable: false,
        stale_assets: false,
        stale_assets_unchecked: false,
        // `attached` is a live-tmux reconciliation only the list handler
        // computes (see [`super::list_managed_sessions`]); every other summary
        // builder leaves it at its `false` default.
        attached: false,
        // `slot`/`deleted` (#3034) are meaningful only on the numbered `tm ls`
        // listing surface; every other caller of `record_to_summary` leaves
        // them at their "not applicable" defaults.
        slot: 0,
        deleted: false,
        // #6568: read straight off the record's `stop_cause`, so every listing
        // surface says the same thing about a parked session.
        auto_resume_parked: r.auto_resume_park_reason().map(str::to_string),
        kind: r.kind, // #8942
    }
}

/// Reconcile each summary's displayed `state`/`attached` against LIVE tmux.
///
/// Why: a record's PERSISTED `state` can read `stopped` after a daemon restart
/// (before the reconcile pass runs) even though its tmux session is alive and a
/// client is attached — the `tm ls` picker then wrongly showed EVERY session
/// `(stopped)` and offered a destructive `restart`. Real tmux is the
/// authoritative signal, so the list handler probes it once and corrects the
/// DISPLAYED state here (read-time reconciliation, mirroring how
/// `unresumable`/`stale_assets` are computed without mutating the store).
/// What: for each summary whose record is in the transient `Active`/`Stopped`
/// pair, sets `state` to `"active"` when its tmux is confirmed live (else
/// `"stopped"`), and sets `attached` when its session is in `attached`.
/// Terminal states (`Decommissioned`/`Deleted`), `Provisioning`, and `Errored`
/// are left as persisted — a live tmux name must never resurrect a
/// deleted/decommissioned record's label, and `errored`/`provisioning` carry
/// information a bare liveness probe would erase.
///
/// Pane-scoped liveness (#3714): a bare `live.contains(name)` proves only that
/// SOME tmux session currently uses this NAME — under the #3692 duplicate-name
/// condition, a stale record's own pane can be long gone while an UNRELATED
/// live session (owned by a DIFFERENT record) still happens to share its
/// name. Trusting the name alone there resurrected the stale record's
/// DISPLAYED `state` to `"active"` while its `persisted_state` correctly kept
/// reading `"stopped"` — the exact same-record `state`/`persisted_state`
/// contradiction the issue reported (two consecutive `tm` runs seconds apart
/// read as flip-flopping between "already active" and "stopped, will kill
/// pane" for the identical command). When a record has a captured `pane_id`,
/// liveness is decided by [`ManagedTmuxDriver::pane_exists_checked`] — the
/// TRI-STATE variant of `pane_exists` (#3714 review finding 2): `Some(true)`
/// (THAT SPECIFIC pane confirmed inside a session named `name`) reads
/// `"active"`; `Some(false)` (confirmed absent) reads `"stopped"` — this is
/// the case that closes the contradiction above; `None` (the query itself
/// failed — e.g. a transient tmux hiccup, NOT a confirmed absence) falls
/// back to the bare name check instead of asserting the pane is gone, so a
/// transient failure can never flip a genuinely live record to `"stopped"`
/// (the distinct failure mode `pane_exists`'s fail-closed `bool` — correct
/// for the MUTATION guard in `SessionManager::rename`, which keeps using
/// `pane_exists` unchanged — would otherwise reproduce here for DISPLAY). A
/// legacy record with no captured `pane_id` also falls back to the
/// name-only check; no stronger signal exists for it. Pane probes only run
/// for records whose name IS in `live` (short-circuited by `&&`), so a
/// fleet where most sessions are genuinely stopped pays no extra tmux
/// round-trips.
/// Test: `reconcile_live_state_flips_stopped_to_active_when_alive`,
/// `reconcile_live_state_leaves_terminal_states`,
/// `reconcile_live_state_leaves_persisted_state_untouched`,
/// `reconcile_live_state_prefers_pane_scoped_liveness_over_name_membership`,
/// `reconcile_live_state_legacy_record_without_pane_id_uses_name_only`,
/// `reconcile_live_state_pane_query_error_does_not_flip_live_record_to_stopped`
/// in `super::tests`; end-to-end coverage in the `session_lifecycle`
/// integration suite.
pub(super) fn reconcile_live_state(
    tmux: &dyn ManagedTmuxDriver,
    summaries: &mut [SessionSummary],
    records: &[SessionRecord],
    live: &std::collections::HashSet<String>,
    attached: &std::collections::HashSet<String>,
) {
    for (summary, record) in summaries.iter_mut().zip(records.iter()) {
        let name = &record.tmux_name;
        let name_live = live.contains(name);
        let is_live = match record.pane_id.as_deref() {
            Some(pane_id) if name_live => match tmux.pane_exists_checked(name, pane_id) {
                Some(alive) => alive,
                // Query failed — "could not determine" is NOT "confirmed
                // absent"; fall back to the name-only signal rather than
                // flipping a possibly-live record to "stopped" on a
                // transient tmux hiccup (#3714 review finding 2).
                None => name_live,
            },
            _ => name_live,
        };
        summary.attached = is_live && attached.contains(name);
        if matches!(
            record.state,
            ManagedSessionState::Active | ManagedSessionState::Stopped
        ) {
            summary.state = if is_live { "active" } else { "stopped" }.to_string();
        }
    }
}

/// Probe live tmux ONCE and reconcile every summary's displayed state in place.
///
/// Why: both `list_managed_sessions` (the fleet list) AND `get_managed_session`
/// (the single-record fetch that `tm session resume <id>` reads) must reconcile
/// the persisted state against real tmux — otherwise the highest-consequence
/// path (`resume`) can act on a stale `stopped` for a live session and
/// destructively recreate its pane (code-critic HIGH). Centralising the probe
/// here keeps the two call sites a single line each and, crucially, keeps the
/// FAIL-CLOSED contract in ONE place: a transient tmux enumeration error must
/// NOT collapse to "zero live sessions" (which would reconcile every Active
/// record to `stopped` and offer fleet-wide destructive restarts). On error we
/// SKIP reconciliation entirely and serve the persisted state unchanged.
/// What: calls [`ManagedTmuxDriver::list_sessions`]; on `Ok`, gathers the live +
/// attached name sets and applies [`reconcile_live_state`]; on `Err`, logs a
/// warning and leaves `summaries` at their persisted state.
/// Test: `reconcile_against_tmux_fails_closed_on_enumeration_error` in
/// `super::tests` (the fail-closed path); the success path is covered by
/// `reconcile_live_state_flips_stopped_to_active_when_alive` and end-to-end by
/// the `session_lifecycle` integration suite.
pub(super) fn reconcile_against_tmux(
    tmux: &dyn ManagedTmuxDriver,
    summaries: &mut [SessionSummary],
    records: &[SessionRecord],
) {
    match tmux.list_sessions() {
        Ok(names) => {
            let live: std::collections::HashSet<String> = names.into_iter().collect();
            let attached: std::collections::HashSet<String> =
                tmux.attached_session_names().into_iter().collect();
            reconcile_live_state(tmux, summaries, records, &live, &attached);
        }
        Err(e) => {
            tracing::warn!(
                "tmux session enumeration failed ({e}); serving persisted session \
                 state WITHOUT live reconciliation (fail-closed — never treat a \
                 probe error as 'all sessions stopped')"
            );
        }
    }
}

/// Build a tombstone `SessionSummary` for a deleted slot (issue #3034).
///
/// Why: `tm ls`'s stable numbering shows a `-- deleted --` placeholder for a
/// slot whose session record no longer exists in the store, so operators
/// never mistake a shifted neighbor for the session they meant. This is the
/// single place that shape is constructed so every numbered-listing caller
/// renders an identical, unambiguously-empty row.
/// What: every string/option field is blank/`None`; `state` is the literal
/// `"deleted"` — distinct from `"decommissioned"`, and paired with
/// `deleted: true`, so the CLI's default-view filter (`is_live_session_state`)
/// never hides this row from the default (non-`--all`) view, including via the
/// #4994 dead-record arm, which sits below the `"deleted"` one; `slot` is the
/// caller-supplied stable number and `deleted` is always `true`.
/// Test: `list_assigns_stable_slot_numbers_and_tombstones_deleted_one`
/// (integration test, `tests/session_manager_slots.rs`).
pub(super) fn tombstone_summary(slot: u32) -> SessionSummary {
    SessionSummary {
        id: String::new(),
        name: String::new(),
        state: "deleted".to_string(),
        persisted_state: String::new(),
        workspace_path: None,
        repo_url: None,
        branch: None,
        created_at: String::new(),
        last_activity_at: None,
        pending_decision: None,
        proposed_default: None,
        source_id: None,
        task: None,
        cwd: None,
        claude_session_id: None,
        deliverable_id: None,
        pane_id: None,
        injection_status: None,
        unresumable: false,
        stale_assets: false,
        stale_assets_unchecked: false,
        attached: false,
        slot,
        deleted: true,
        auto_resume_parked: None,
        kind: crate::session_manager::SessionKind::Ordinary,
    }
}

/// Whether a record's state has a deployed workspace whose asset staleness is
/// MEANINGFUL to compute at all.
///
/// Why: `Provisioning` workspaces have not deployed yet (every managed
/// artifact would spuriously read as "new"/stale) and `Decommissioned` ones
/// have no workspace left to probe — both would only add noise, not signal.
/// This is the same state set `sync_assets::syncable` gates the actual
/// redeploy on, so "we can tell you it's stale" and "we can fix it" never
/// disagree. It is the gate for the ON-DEMAND single-session path
/// ([`record_to_summary_checked`], which `tm session resume` reads); the fleet
/// LIST path applies the stricter [`probe_staleness_in_list`] on top of it.
/// What: `true` for `Active`, `Stopped`, and `Errored`.
/// Test: `checked_summaries_flags_stale_assets_only_for_relevant_states`,
/// `record_to_summary_checked_still_flags_stale_stopped_session`.
fn staleness_meaningful_for(state: &ManagedSessionState) -> bool {
    matches!(
        state,
        ManagedSessionState::Active | ManagedSessionState::Stopped | ManagedSessionState::Errored
    )
}

/// Whether a record gets the asset-staleness probe on the FLEET LIST path.
///
/// Why (issue #4322): the probe is filesystem-bound — ~95 `read_to_string`
/// calls and tens of megabytes per session, uncached — and `tm ls` paid it for
/// every `Active`/`Stopped`/`Errored` record on EVERY invocation. On a real
/// fleet the majority of those records are `Stopped` sessions idle for days
/// (20 of 32 in the reported measurement, ~56% of the total I/O) whose
/// deployed assets cannot possibly have drifted since the last listing —
/// nothing writes to a stopped session's workspace. Staleness is only
/// ACTIONABLE at resume time, and that path fetches the session individually
/// (`GET …/managed/{id}` → [`record_to_summary_checked`]), which still probes
/// under [`staleness_meaningful_for`]. Dropping `Stopped` from the LIST fan-out
/// removes that work from the interactive, per-keystroke-latency surface
/// without removing the signal from the surface that acts on it.
/// What: `true` for `Active` and `Errored` only. A `Stopped` record's summary
/// therefore carries `stale_assets: false` — which is NOT a "fresh" verdict,
/// so [`super::SessionSummary::stale_assets_unchecked`] is set instead and the
/// CLI renders an explicit `[assets ?]` marker rather than silence.
/// Test: `checked_summaries_does_not_probe_stopped_sessions`.
fn probe_staleness_in_list(state: &ManagedSessionState) -> bool {
    matches!(
        state,
        ManagedSessionState::Active | ManagedSessionState::Errored
    )
}

/// Resolve every record's plan and compute ONE [`CatalogHashes`] per distinct
/// `(agent_source, skill_source)` pair — the SHARED half of the staleness
/// comparison (issue #2444 review, MEDIUM finding: `checked_summaries`
/// originally called `session_assets_stale` — a full, independent catalog
/// recompose — once PER SESSION on every `tm sessions ls`, recomposing the
/// same ~40+ catalog agents redundantly for every session sharing the default
/// source).
///
/// Why: a staleness comparison splits into an expensive CATALOG-side half
/// (composing agents, reading skill bodies — identical for every session
/// sharing the same resolved `(agent_source, skill_source)` pair) and a
/// cheap DEPLOYED-side half (this session's own manifest + on-disk files).
/// Grouping records by their resolved source pair via
/// [`crate::core::session_assets::session_plan`] and computing
/// [`CatalogHashes::compute`] ONCE per distinct pair — which collapses to a
/// SINGLE compute for the common case where every session resolves the
/// shared default bundled/catalog source, and stays exactly correct for the
/// rare session carrying its own project-level manifest override — removes
/// the N-times recompose entirely. This half is deliberately kept SEQUENTIAL
/// and run in a single blocking task by [`stale_assets_for_many`]: fanning it
/// out would race N tasks into the same empty cache and reinstate exactly the
/// N-times recompose it exists to prevent. Both the manifest resolution and
/// the catalog compose are filesystem-bound, so this function is
/// synchronous/blocking.
/// What: returns one [`StalenessInput`] per record, in input order, each
/// carrying a shared `Arc` handle to its catalog hashes. `pub(super)` only so
/// the invariant test below can assert the SHARING structurally.
/// Test: `staleness_inputs_computes_one_catalog_per_source_pair_shared_by_arc`
/// (the anti-regression pin: same source pair ⇒ pointer-equal `Arc`, so no
/// second compute exists for a spawned task to perform),
/// `checked_summaries_stale_assets_independent_per_session_sharing_one_catalog`,
/// `checked_summaries_flags_stale_assets_only_for_relevant_states` in
/// `super::tests`.
///
/// #5040: `base` — the directory the framework install nests under — is an
/// argument rather than a `$HOME` read, for the reason
/// [`crate::core::session_assets::session_plan_under`] gives. This is the batch
/// half of that seam, so a test of the fan-out below is hermetic without
/// redirecting the process-global `$HOME`. Production passes
/// `FrameworkPaths::home_base()` from [`checked_summaries`] /
/// [`record_to_summary_checked`].
pub(super) fn staleness_inputs_under(
    records: Vec<SessionRecord>,
    base: &std::path::Path,
) -> Vec<StalenessInput> {
    let mut cache: HashMap<(PathBuf, PathBuf), Arc<CatalogHashes>> = HashMap::new();
    // #4322: the second shared half. `fw.agent_deploy_dir()` is ONE
    // machine-global directory for every session (#4409 exempted it from the
    // per-workspace rewrite), so its 42 files were being read once per session
    // for byte-identical answers. Keyed by (deployed agent dir, catalog source
    // pair) because the snapshot's stem set comes from the catalog — two
    // different catalogs would need two different snapshots even against the
    // same directory. In practice both maps collapse to a single entry.
    let mut agents: HashMap<(PathBuf, PathBuf, PathBuf), Arc<DeployedAgentHashes>> = HashMap::new();
    let mut out = Vec::with_capacity(records.len());
    for record in records {
        let (fw, plan) = session_plan_under(&record, base);
        let key = (plan.agent_source.clone(), plan.skill_source.clone());
        let catalog = cache
            .entry(key.clone())
            .or_insert_with(|| {
                Arc::new(CatalogHashes::compute(
                    &plan.agent_source,
                    &plan.skill_source,
                    &fw.skill_deploy_dir(),
                ))
            })
            .clone();
        let agents_dir = fw.agent_deploy_dir();
        let deployed_agents = agents
            .entry((agents_dir.clone(), key.0, key.1))
            .or_insert_with(|| Arc::new(DeployedAgentHashes::read(&agents_dir, &catalog)))
            .clone();
        out.push((record.id, fw, plan, catalog, deployed_agents));
    }
    out
}

/// One session's fully-resolved staleness comparison inputs: its id, its
/// workspace-scoped paths and plan, and a SHARED handle to the catalog-side
/// hashes for its resolved `(agent_source, skill_source)` pair.
///
/// Why: [`stale_assets_for_many`] hands one of these to each per-session
/// blocking task. `Arc<CatalogHashes>` (rather than a clone of the hashes) is
/// what lets the expensive catalog compose stay shared across the fan-out
/// while each task still owns everything it needs — the whole point of the
/// #2444 split, preserved under parallelism. `Arc<DeployedAgentHashes>`
/// (#4322) is the same trick applied to the DEPLOYED agent tree, which is one
/// machine-global directory every session resolves identically.
type StalenessInput = (
    ManagedSessionId,
    FrameworkPaths,
    HarnessPlan,
    Arc<CatalogHashes>,
    Arc<DeployedAgentHashes>,
);

/// Compute `stale_assets` for every record, sharing the catalog work and
/// running the per-session comparisons CONCURRENTLY on the blocking pool
/// (issue #4322).
///
/// Why: the shared-catalog design (#2444) removed the redundant per-session
/// RECOMPOSE, but left the remaining per-session work — ~95 `read_to_string`
/// calls against that session's own `.claude/{agents,skills}` tree — running
/// SERIALLY inside a single `spawn_blocking`. Those reads are independent
/// across sessions and entirely I/O-bound (the daemon burns a constant ~0.13 s
/// of CPU while `tm ls` blocks for 5–9 s cold), so serializing them makes cold
/// latency scale linearly in fleet size for no reason. This mirrors the
/// `unresumable` fan-out [`checked_summaries`] already performs — same
/// `JoinSet` pattern, `spawn_blocking` instead of `spawn` because the work is
/// synchronous filesystem I/O rather than `tokio::fs`.
/// What: resolves every session's plan and computes one [`CatalogHashes`] per
/// distinct source pair in ONE blocking task ([`staleness_inputs_under`] — the
/// sharing that must NOT be parallelized, or each task would redundantly
/// recompose the same catalog), then spawns one blocking task per session for
/// the cheap deployed-side [`session_asset_staleness_with_catalog`] half. A
/// panicking task is simply absent from the returned map, leaving that
/// summary at its `false` default rather than failing the whole listing.
/// Test: `checked_summaries_stale_assets_independent_per_session_sharing_one_catalog`,
/// `checked_summaries_flags_stale_assets_only_for_relevant_states`,
/// `stale_assets_for_many_reads_shared_agent_dir_once_for_the_whole_fleet`,
/// `stale_assets_for_many_sees_an_agent_change_on_the_very_next_call`,
/// `stale_assets_for_many_keeps_per_session_verdicts_correct_under_concurrency`,
/// `stale_assets_for_many_computes_catalog_exactly_once_per_source_pair`
/// (issue #4326 review HIGH: the prior pin only exercised [`staleness_inputs_under`]
/// directly, never this actual hot path, and stayed green when
/// `CatalogHashes::compute` was moved into the fan-out below) in
/// `super::tests`. `pub(super)` (rather than private) so that regression test
/// can call this function directly instead of the higher-level
/// `checked_summaries`, which would also exercise the unrelated `unresumable`
/// fan-out.
///
/// #5040: `base` is the framework install's base directory, supplied by the
/// caller instead of read from `$HOME`. Every test of this path used to deploy
/// its fixture into `FrameworkPaths::default()` under a `fake_home()` guard, so
/// its correctness depended on no other test in the binary rewriting `$HOME`
/// mid-run. `#[serial_test::serial]` does not provide that — it orders
/// `#[serial]` tests against each other only, and under `cargo nextest`'s
/// per-test processes it orders nothing. Passing the base in removes the
/// dependency instead of guarding it: a test hands over its own temp dir and
/// mutates no environment, so no concurrent test can reach it. Owned rather
/// than borrowed because the resolve step runs inside `spawn_blocking`.
pub(super) async fn stale_assets_for_many_under(
    records: Vec<SessionRecord>,
    base: PathBuf,
) -> HashMap<ManagedSessionId, bool> {
    let inputs = tokio::task::spawn_blocking(move || staleness_inputs_under(records, &base))
        .await
        .unwrap_or_default();
    let mut probes = tokio::task::JoinSet::new();
    for (id, fw, plan, catalog, deployed_agents) in inputs {
        // INVARIANT: a spawned task receives an ALREADY-COMPUTED
        // `Arc<CatalogHashes>` AND an already-read `Arc<DeployedAgentHashes>`,
        // and must never call `CatalogHashes::compute` or
        // `DeployedAgentHashes::read` itself. Moving either in here would give
        // every session its own copy of shared work — the catalog case is the
        // exact N-times recompose #2444's review removed, and the deployed-agent
        // case is the N-times re-read of one machine-global directory #4322
        // removed. Only what is genuinely per-session — this session's own
        // per-workspace skill manifest and skill files — belongs in the fan-out.
        probes.spawn_blocking(move || {
            (
                id,
                crate::core::session_assets::session_asset_staleness_with_shared(
                    &fw,
                    &plan,
                    &catalog,
                    &deployed_agents,
                )
                .stale,
            )
        });
    }
    let mut result = HashMap::with_capacity(probes.len());
    while let Some(res) = probes.join_next().await {
        if let Ok((id, stale)) = res {
            result.insert(id, stale);
        }
    }
    result
}

/// Run [`stale_assets_for_many_under`] for a single record — the
/// [`record_to_summary_checked`] (single-session GET) call site, which has no
/// batching benefit (N=1) but reuses the same blocking-pool handoff.
async fn probe_stale_assets(record: SessionRecord, base: PathBuf) -> bool {
    let id = record.id;
    stale_assets_for_many_under(vec![record], base)
        .await
        .get(&id)
        .copied()
        .unwrap_or(false)
}

/// [`record_to_summary`] plus the async `unresumable` probe (#2595).
///
/// Why: `list_managed_sessions`/`get_managed_session` are the two handlers
/// that can afford the filesystem probe (both already `.await` the session
/// manager). Folding "convert, then overwrite the flag" into one call here —
/// rather than repeating it inline at each handler — keeps `mod.rs`'s handler
/// bodies a single line each (`mod.rs` sits at its 500-SLOC production cap).
/// What: [`record_to_summary`], then overwrites `unresumable` via
/// `session_manager::resume_workdir::is_unresumable`.
///
/// Staleness (#4322): this SINGLE-session path keeps the full
/// [`staleness_meaningful_for`] gate — including `Stopped`, which the fleet
/// LIST path now skips. `GET …/managed/{id}` is what `tm session resume`
/// reads, and resume is precisely the moment a stopped session's asset drift
/// becomes actionable; paying ~95 file reads for ONE session on an explicit,
/// infrequent fetch is not the cost that made `tm ls` slow.
/// Test: `list_marks_dead_stopped_session_unresumable`,
/// `list_leaves_live_and_healthy_stopped_sessions_unmarked`,
/// `record_to_summary_checked_still_flags_stale_stopped_session` in
/// `super::tests`.
pub(super) async fn record_to_summary_checked(r: &SessionRecord) -> SessionSummary {
    // #5040: production resolves the framework base exactly as before.
    record_to_summary_checked_under(r, crate::core::paths::FrameworkPaths::home_base()).await
}

/// [`record_to_summary_checked`] with the framework install's base directory
/// supplied by the caller instead of read from `$HOME` (#5040).
///
/// Why: `record_to_summary_checked_still_flags_stale_stopped_session` deployed
/// its drifted fixture into `FrameworkPaths::default()` under a `fake_home()`
/// guard, which made the test a WRITER of the process-global `$HOME`. Three
/// unrelated tests in this binary read live `$HOME` concurrently and caught the
/// tempdir value, failing on roughly one local `cargo test -p trusty-mpm` run in
/// three. The same seam [`stale_assets_for_many_under`] opened for the batch path
/// removes the write instead of ordering it.
/// What: identical to [`record_to_summary_checked`], with `base` threaded to
/// [`probe_stale_assets`].
/// Test: `record_to_summary_checked_still_flags_stale_stopped_session`.
pub(super) async fn record_to_summary_checked_under(
    r: &SessionRecord,
    base: PathBuf,
) -> SessionSummary {
    let mut summary = record_to_summary(r);
    summary.unresumable = crate::session_manager::resume_workdir::is_unresumable(r).await;
    if staleness_meaningful_for(&r.state) {
        summary.stale_assets = probe_stale_assets(r.clone(), base).await;
    }
    summary
}

/// [`record_to_summary`] for MANY records, running each `unresumable` probe
/// concurrently rather than one-at-a-time (#2595 review, MEDIUM finding 4).
///
/// Why: `list_managed_sessions` originally awaited [`record_to_summary_checked`]
/// sequentially in a `for` loop — for a fleet of N sessions that serializes N
/// `tokio::fs::try_exists`-backed round trips even though they are entirely
/// independent. `is_unresumable` already short-circuits with NO I/O for any
/// state other than `Stopped`/`Errored` (the state gate), so only that subset
/// ever pays the probe cost — fan out exactly those via `tokio::task::JoinSet`
/// (unconditional `tokio` dependency; the workspace's optional `futures` crate
/// is feature-gated and not assumed available to every daemon route) rather
/// than every record.
/// What: builds every summary synchronously first via [`record_to_summary`]
/// (cheap, no I/O — preserves `records`' order), then spawns one
/// `is_unresumable` task per `Stopped`/`Errored` record (each tagged with its
/// original index so `JoinSet`'s out-of-completion-order yield can never
/// reorder the response), and patches each result back into its slot as
/// tasks complete. A panicking probe task is dropped silently — that summary
/// simply keeps its `record_to_summary` default (`unresumable: false`)
/// rather than failing the whole list response.
/// Test: `checked_summaries_preserves_input_order_and_flags_only_dead_sessions`
/// in `super::tests`; end-to-end coverage via `list_marks_dead_stopped_session_unresumable`,
/// `list_leaves_live_and_healthy_stopped_sessions_unmarked` in
/// `tests/session_manager_mvp.rs`.
pub(super) async fn checked_summaries(records: &[SessionRecord]) -> Vec<SessionSummary> {
    checked_summaries_with(records, true).await
}

/// [`checked_summaries`] with the #2444 asset-staleness probe made optional
/// (issue #4335, folds into #4322).
///
/// Why: `stale_assets` is the single most expensive thing this endpoint
/// computes — a filesystem-bound catalog compose per distinct
/// `(agent_source, skill_source)` pair plus a manifest+on-disk comparison per
/// session — and on a COLD daemon (nothing cached, dozens of sessions) it
/// pushes the whole response past the caller's timeout. Callers that never
/// READ the flag were paying for it anyway and then failing open when it took
/// too long, which is exactly how `guided::nested_session_guard` silently
/// stopped guarding. Letting such a caller opt out is strictly better than
/// raising every timeout to accommodate work nobody wants.
/// What: identical to [`checked_summaries`] except that when `probe_assets` is
/// `false` the entire second pass is skipped and every `stale_assets` stays at
/// its `record_to_summary` default of `false`. The `unresumable` probe is NOT
/// optional — it is cheap (state-gated, no I/O outside `Stopped`/`Errored`)
/// and the guard's callers do read it. A `false` here means "not computed",
/// not "computed and found current", so only a caller that ignores the field
/// may pass it.
/// Test: `checked_summaries_slim_skips_stale_assets_probe`.
pub(super) async fn checked_summaries_with(
    records: &[SessionRecord],
    probe_assets: bool,
) -> Vec<SessionSummary> {
    // #5040: production resolves the framework base exactly as before.
    checked_summaries_with_under(
        records,
        probe_assets,
        crate::core::paths::FrameworkPaths::home_base(),
    )
    .await
}

/// [`checked_summaries_with`] with the framework install's base directory
/// supplied by the caller instead of read from `$HOME` (#5040).
///
/// Why: the five `checked_summaries_*` tests each deployed a drifted fixture
/// into `FrameworkPaths::default()` under a `fake_home()` guard, making every
/// one of them a WRITER of the process-global `$HOME`. A guard that restores the
/// variable on drop is the race, not a fix for it: `session_plan_under_matches_
/// session_plan_at_home`, `managed_prune_worktrees_parity_defaults_to_a_preview`
/// and `managed_reconcile_worktrees_parity` all read live `$HOME` concurrently,
/// and an in-process `cargo test -p trusty-mpm` run caught the tempdir value in
/// roughly one run of three. Threading the base through is what lets those
/// writes be deleted outright.
/// What: identical to [`checked_summaries_with`], with `base` threaded to
/// [`stale_assets_for_many_under`]. The `unresumable` pass reads no framework
/// path at all, so it is unaffected.
/// Test: `checked_summaries_preserves_input_order_and_flags_only_dead_sessions`,
/// `checked_summaries_flags_stale_assets_only_for_relevant_states`,
/// `checked_summaries_does_not_probe_stopped_sessions`,
/// `checked_summaries_slim_skips_stale_assets_probe`,
/// `checked_summaries_stale_assets_independent_per_session_sharing_one_catalog`.
pub(super) async fn checked_summaries_with_under(
    records: &[SessionRecord],
    probe_assets: bool,
    base: PathBuf,
) -> Vec<SessionSummary> {
    let mut summaries: Vec<SessionSummary> = records.iter().map(record_to_summary).collect();

    let mut probes = tokio::task::JoinSet::new();
    for (idx, r) in records.iter().enumerate() {
        if matches!(
            r.state,
            ManagedSessionState::Stopped | ManagedSessionState::Errored
        ) {
            let r = r.clone();
            probes.spawn(async move {
                let unresumable = crate::session_manager::resume_workdir::is_unresumable(&r).await;
                (idx, unresumable)
            });
        }
    }
    while let Some(res) = probes.join_next().await {
        if let Ok((idx, unresumable)) = res {
            summaries[idx].unresumable = unresumable;
        }
    }

    // Second, independent pass for the #2444 asset-staleness probe.
    // `stale_assets_for_many` computes the shared catalog half ONCE and then
    // fans the per-session deployed-side comparisons across the blocking pool
    // (#4322). `Stopped` records are deliberately EXCLUDED here (see
    // `probe_staleness_in_list`) — their summaries carry
    // `stale_assets_unchecked` instead, so "no marker" is never silently read
    // as "assets fresh". A slim caller (#4335's `probe_assets: false`) skips
    // this ENTIRE pass — every meaningful-state record is left
    // `stale_assets_unchecked` too, for the identical reason: the probe
    // genuinely did not run, so silence must never be misread as "checked and
    // fresh" regardless of WHICH gate (state or slim) skipped it.
    for (idx, r) in records.iter().enumerate() {
        summaries[idx].stale_assets_unchecked = staleness_meaningful_for(&r.state)
            && (!probe_assets || !probe_staleness_in_list(&r.state));
    }
    let probed: Vec<SessionRecord> = records
        .iter()
        .filter(|r| probe_assets && probe_staleness_in_list(&r.state))
        .cloned()
        .collect();
    if !probed.is_empty() {
        let stale_map = stale_assets_for_many_under(probed, base).await;
        for (idx, r) in records.iter().enumerate() {
            if let Some(&stale) = stale_map.get(&r.id) {
                summaries[idx].stale_assets = stale;
            }
        }
    }

    summaries
}

/// Build the numbered `tm ls` summaries for one listing call (issue #3034).
///
/// Why: `list_managed_sessions` combines the daemon's stable slot assignment
/// with the existing async `unresumable`/`stale_assets` probes
/// ([`checked_summaries`]), the LIVE-tmux `state`/`attached` reconciliation
/// ([`reconcile_against_tmux`], #3302/#3531 — a running/attached session must
/// never read `(stopped)` just because its persisted state went stale), and
/// the `?source_id=` filter, in the order that keeps all three correct:
/// filter the numbered rows FIRST (so a session outside the filter never pays
/// the probe cost), THEN probe and reconcile only the live records that
/// remain, THEN stitch each summary back to its slot number — while a
/// tombstoned row skips both entirely and gets a blank [`tombstone_summary`]
/// instead.
/// What: takes the full [`NumberedSlot`] list (already observed against the
/// registry for every record — filtered or not, by the caller), a tmux driver
/// for the live-state reconciliation, plus an optional `source_id` filter;
/// returns one [`SessionSummary`] per surviving row, in slot order. The
/// stable `slot` field is the identity every by-number CLI surface resolves
/// against, so it is set on the row AFTER reconciliation/probing overwrite
/// `state`/`attached`/`unresumable`/`stale_assets` — sort/filter on the CLI
/// side only ever reorders or subsets this returned `Vec`, it never
/// recomputes `slot`.
/// Test: `list_assigns_stable_slot_numbers_and_tombstones_deleted_one`
/// (integration test, `tests/session_manager_slots.rs`).
pub(super) async fn numbered_summaries(
    numbered: Vec<NumberedSlot>,
    tmux: &dyn ManagedTmuxDriver,
    source_id_filter: Option<&str>,
    probe_assets: bool,
) -> Vec<SessionSummary> {
    let filtered: Vec<NumberedSlot> = numbered
        .into_iter()
        .filter(|n| source_id_filter.is_none_or(|sid| n.source_id.as_deref() == Some(sid)))
        .collect();
    let live_records: Vec<SessionRecord> =
        filtered.iter().filter_map(|n| n.record.clone()).collect();
    let mut live_summaries = checked_summaries_with(&live_records, probe_assets).await;
    reconcile_against_tmux(tmux, &mut live_summaries, &live_records);
    let mut live_summaries = live_summaries.into_iter();
    filtered
        .into_iter()
        .map(|n| match n.record {
            Some(_) => {
                let mut s = live_summaries
                    .next()
                    .expect("live_summaries has one entry per Some(record) row, same order");
                s.slot = n.slot;
                s
            }
            None => tombstone_summary(n.slot),
        })
        .collect()
}

/// Build the tmux attach command string for a session.
///
/// Why: clients need the exact attach command without hardcoding the convention.
/// What: returns `tmux attach-session -t '=<name>'` — exact and quoted for
/// zsh (#8443), via `trusty_common::tmux::shell_attach_command`.
/// Test: attach-cmd handler test.
pub(super) fn attach_cmd_for(tmux_name: &str) -> String {
    crate::core::tmux::shell_attach_command(tmux_name)
}

/// Parse a UUID path segment into a [`ManagedSessionId`].
///
/// Why: handlers receive the id as a string; an invalid UUID must produce a 400
/// rather than a 404 or panic.
/// What: parses the string into a UUID, mapping failure to a `400` tuple.
/// Test: covered by handler tests that pass an invalid id.
pub(super) fn parse_id(id_str: &str) -> Result<ManagedSessionId, (StatusCode, String)> {
    id_str
        .parse::<uuid::Uuid>()
        .map(ManagedSessionId::from)
        .map_err(|_| {
            (
                StatusCode::BAD_REQUEST,
                format!("invalid session id: {id_str}"),
            )
        })
}