trusty-mpm 1.3.4

//! Auto-prune definitively-dead (`unresumable`) managed-session records at
//! listing time (owner request 2026-07-29; hardened per code-critic WARN,
//! PR #4384 review).
//!
//! Why: extracted out of `session_picker.rs` (which sits at the 500-SLOC
//! production cap, mirroring why `session_picker_render.rs`/
//! `session_picker_rename.rs` were split out before it). Before this module
//! existed, a session flagged `unresumable` (its workspace GC-pruned out from
//! under it) sat in the picker forever printing `DEAD: workspace removed; use
//! [d<N>] to remove the record` — the operator had to notice the row and type
//! the number themselves, every time.
//!
//! The critic's WARN (then BLOCK on re-review) identified this feature's
//! first cut as converting an advisory flag into an unconditional,
//! unconfirmed, uncapped, unverified destructive action. Four independent
//! hardenings close that gap:
//!   1. **Record-only removal** — [`decommission_dead_record`] calls the
//!      `/decommission` route with `record_only=true`
//!      (`SessionManager::decommission_record_only`), which never touches
//!      the filesystem. A remount between the listing-time probe and this
//!      call can never lose data, because nothing is ever deleted here.
//!   2. **Two-sightings confirmation** — [`auto_prune_dead_records_at`] only
//!      acts on a session id whose `unresumable` flag was ALSO seen on a
//!      strictly earlier call, tracked via a small persisted marker file
//!      ([`load_seen`]/[`save_seen`]). A single-shot blip (one bad listing)
//!      never triggers a prune.
//!   3. **Per-call cap** — at most [`AUTO_PRUNE_CAP`] confirmed records are
//!      decommissioned per call; the rest are reported as "pending
//!      confirmation" rather than acted on. A bad-mount day cannot
//!      mass-tombstone an entire fleet in one `tm ls`.
//!   4. **Response-body verification (a daemon/client version-skew guard)**
//!      — `record_only=true` alone is not proof of safety: a daemon process
//!      still running a build older than this fix has no
//!      `Query<DecommissionQuery>` extractor on the route at all, silently
//!      ignores the param (axum drops query params a handler doesn't
//!      declare), and falls through to the OLD unconditional full-teardown
//!      path — which ALSO returns 200. [`decommission_dead_record`] parses
//!      the response body and requires `workspace_removed == false` before
//!      counting a call as safe; anything else trips
//!      [`STALE_DAEMON_MARKER_KEY`], which disables further auto-prune
//!      attempts until the sentinel expires ([`STALE_DAEMON_TTL_SECS`], 1
//!      hour) — self-healing: one probe per hour against a genuinely stale
//!      daemon, not a permanent lockout once it's restarted. Records held
//!      back by this gate fold into [`AutoPruneOutcome::pending`], never
//!      silently disappearing from the reported count.
//!
//! What: [`auto_prune_dead_records`] is the production entry point (resolves
//! the marker file under `~/.trusty-mpm`); [`auto_prune_dead_records_at`] is
//! the testable core (explicit marker-file path); [`AutoPruneOutcome`] is the
//! three-part result (`kept`, `pruned`, `pending`) `fetch_live_sessions`
//! reports to the operator.
//!
//! Test: `auto_prune_dead_records_removes_confirmed_unresumable_records`,
//! `auto_prune_dead_records_first_sighting_is_not_pruned`,
//! `auto_prune_dead_records_keeps_workspace_present_records`,
//! `auto_prune_dead_records_is_noop_when_nothing_is_dead`,
//! `auto_prune_dead_records_honors_the_cap`,
//! `auto_prune_dead_records_stops_sweep_when_daemon_reports_workspace_removed`,
//! `auto_prune_dead_records_stale_daemon_sentinel_expires_after_ttl`
//! in `tests_behavior_d_tests.rs`; the record-only server-side guarantee is
//! pinned by `decommission_record_only_never_removes_existing_workspace`
//! (`session_manager::tests`).

use std::collections::HashMap;
use std::path::Path;

use trusty_mpm::client::ManagedSessionSummary;

/// Maximum number of CONFIRMED dead records one call may decommission (critic
/// HIGH finding #1, owner request 2026-07-29).
///
/// Why: "a bad-mount day must not mass-tombstone every session in one `tm
/// ls`" — even a session flagged dead on two consecutive listings might all
/// share one now-unmounted volume. Capping bounds the blast radius of any
/// single call regardless of how many records are confirmed-eligible.
const AUTO_PRUNE_CAP: usize = 5;

/// Basename of the two-sightings confirmation marker file, under the
/// framework root (`~/.trusty-mpm/` by default).
const SEEN_MARKER_FILENAME: &str = "auto-prune-seen.json";

/// Reserved key inside the marker file recording "a stale daemon already
/// proved it ignores `record_only`" (critic CRITICAL finding, 2026-07-30
/// re-review of PR #4384).
///
/// Why: `?record_only=true` is read by a `Query<DecommissionQuery>`
/// extractor this feature ADDS to the `/decommission` handler. Axum silently
/// drops a query param a handler doesn't declare — a daemon process still
/// running a build older than this fix has no such extractor and falls
/// through to the OLD unconditional full-teardown `decommission()` for EVERY
/// auto-prune call, regardless of the query string sent. The HTTP status
/// alone cannot distinguish that from a genuine record-only success (both
/// return 200) — only the response body's `workspace_removed` field can
/// (see [`decommission_dead_record`]). Once that's observed, auto-prune must
/// stay off for the rest of THIS invocation — the daemon does not un-stale
/// itself mid-process, and the picker's own re-fetch loop calls
/// [`auto_prune_dead_records_at`] repeatedly within one process. Storing this
/// in the SAME marker file (rather than process-global state) keeps the
/// behavior scoped to `marker_path` — exactly like every other piece of
/// state this module tracks — so tests using distinct tempdir paths can
/// never bleed into each other despite `cargo test` running many tests in
/// one process.
/// What: an unlikely-to-collide sentinel key (real session ids are UUIDs);
/// its value is the RFC 3339 timestamp of first detection — checked against
/// [`STALE_DAEMON_TTL_SECS`] by [`stale_daemon_sentinel_active`], not merely
/// its presence.
/// Test: `auto_prune_dead_records_stops_sweep_when_daemon_reports_workspace_removed`,
/// `auto_prune_dead_records_stale_daemon_sentinel_expires_after_ttl`.
const STALE_DAEMON_MARKER_KEY: &str = "__stale_daemon_detected__";

/// TTL for the `STALE_DAEMON_MARKER_KEY` sentinel: 1 hour (owner request
/// 2026-07-30 follow-up).
///
/// Why: a daemon that was stale an hour ago may have been restarted since —
/// wedging auto-prune off permanently would defeat the point of a
/// self-healing safeguard once the operator (or the daemon's own restart
/// machinery) catches up. Expiring the sentinel lets the NEXT invocation
/// retry; if the daemon is STILL stale, [`decommission_dead_record`]'s
/// body-check trips again immediately and rewrites a fresh sentinel — one
/// probe per hour against a genuinely stale daemon, never a permanent
/// lockout.
const STALE_DAEMON_TTL_SECS: i64 = 60 * 60;

/// Result of one [`auto_prune_dead_records`] / [`auto_prune_dead_records_at`]
/// call.
///
/// Why: the caller (`fetch_live_sessions`) needs three independent numbers to
/// report honestly — what it removed, what it left alone because a
/// confirmation window or the cap wasn't met, and the surviving list to
/// render — rather than collapsing them into a single count.
pub(crate) struct AutoPruneOutcome {
    /// Every session NOT successfully decommissioned this call — healthy
    /// sessions, first-sightings awaiting confirmation, confirmed sessions
    /// over the cap, and any record whose decommission attempt failed.
    pub(crate) kept: Vec<ManagedSessionSummary>,
    /// Count of records actually decommissioned this call (never more than
    /// [`AUTO_PRUNE_CAP`]).
    pub(crate) pruned: usize,
    /// Count of `unresumable` records NOT acted on this call — a first
    /// sighting (not yet confirmed), confirmed-but-over-the-cap, or held
    /// back because the stale-daemon gate is active (owner request
    /// 2026-07-30 follow-up — this must fold in here or the reported count
    /// under-represents exactly the batch that gate is protecting).
    pub(crate) pending: usize,
}

/// Production entry point: resolve the confirmation marker file under the
/// default framework root and delegate to [`auto_prune_dead_records_at`].
///
/// Why: the framework root (`~/.trusty-mpm`) is where every other piece of
/// `tm`'s per-user state already lives (`FrameworkPaths::default`);
/// resolving it here (rather than inside the testable core) keeps the core
/// injectable for tests without touching the real home directory.
/// What: `FrameworkPaths::default().root` joined with
/// [`SEEN_MARKER_FILENAME`].
/// Test: covered transitively by every `fetch_live_sessions` call site; the
/// decision logic itself is tested via [`auto_prune_dead_records_at`]
/// directly.
pub(crate) async fn auto_prune_dead_records(
    client: &reqwest::Client,
    url: &str,
    sessions: Vec<ManagedSessionSummary>,
) -> AutoPruneOutcome {
    let marker_path = trusty_mpm::core::paths::FrameworkPaths::default()
        .root
        .join(SEEN_MARKER_FILENAME);
    auto_prune_dead_records_at(client, url, sessions, &marker_path).await
}

/// Auto-prune definitively-dead (`unresumable`) session records at listing
/// time — the testable core (owner request 2026-07-29).
///
/// Why: before this, an operator had to notice the `DEAD: workspace removed`
/// row and manually type `d<N>` for every session whose workspace had been
/// GC-pruned out from under it — Bob's ask was for the picker to just clean
/// these up on its own. The critic's WARN required this to never be a
/// single-observation, uncapped, disk-mutating action — see the module doc's
/// three hardenings.
///
/// SAFETY BOUNDARY (#4344, `114de333`): a record whose worktree removal was
/// previously REFUSED because the tree was dirty keeps its `workspace_path`
/// pointed at a directory that still genuinely exists on disk. `unresumable`
/// (`session_manager::resume_workdir::is_unresumable`) probes that exact
/// path with `tokio::fs::try_exists` and returns `false` the instant ANY
/// candidate (`last_cwd`/`workspace_path`/`cwd`) is found present — so a
/// dirty-retained record can never read `unresumable == true` in the first
/// place, and never enters this function's `dead` partition at all.
///
/// What: partitions `sessions` into `(keep, dead)` by `s.unresumable`. Any
/// `keep` session that still has a confirmation-marker entry has recovered
/// (its workspace reappeared) — that entry is dropped so a later
/// re-appearance starts a fresh confirmation window. Each `dead` record is
/// then classified against the persisted marker file ([`load_seen`]): a
/// fresh id is recorded as a first sighting (NOT acted on this call); an id
/// already present (seen on a STRICTLY earlier call) is CONFIRMED-eligible.
/// At most [`AUTO_PRUNE_CAP`] confirmed records are decommissioned via
/// [`decommission_dead_record`] — the record-only route, never a raw
/// `fs::remove` or a hand-edited store entry. Everything else (healthy,
/// first-sighting, over-cap, or a failed decommission attempt) is returned
/// in `kept`. The marker file is persisted only when it actually changed.
/// Test: `auto_prune_dead_records_removes_confirmed_unresumable_records`,
/// `auto_prune_dead_records_first_sighting_is_not_pruned`,
/// `auto_prune_dead_records_keeps_workspace_present_records`,
/// `auto_prune_dead_records_is_noop_when_nothing_is_dead`,
/// `auto_prune_dead_records_honors_the_cap`,
/// `auto_prune_dead_records_stale_daemon_sentinel_expires_after_ttl`.
pub(crate) async fn auto_prune_dead_records_at(
    client: &reqwest::Client,
    url: &str,
    sessions: Vec<ManagedSessionSummary>,
    marker_path: &Path,
) -> AutoPruneOutcome {
    let (dead, mut kept): (Vec<_>, Vec<_>) = sessions.into_iter().partition(|s| s.unresumable);

    let mut seen = load_seen(marker_path);
    let mut changed = false;

    // A session that surfaced HEALTHY this call has recovered — clear any
    // stale marker so a later re-appearance of unresumable starts fresh.
    for s in &kept {
        if seen.remove(&s.id).is_some() {
            changed = true;
        }
    }

    if dead.is_empty() {
        if changed {
            save_seen(marker_path, &seen);
        }
        return AutoPruneOutcome {
            kept,
            pruned: 0,
            pending: 0,
        };
    }

    let now = chrono::Utc::now().to_rfc3339();
    let mut confirmed = Vec::new();
    let mut first_sighting = Vec::new();
    for s in dead {
        if seen.contains_key(&s.id) {
            confirmed.push(s);
        } else {
            seen.insert(s.id.clone(), now.clone());
            changed = true;
            first_sighting.push(s);
        }
    }

    let cap = AUTO_PRUNE_CAP.min(confirmed.len());
    let to_prune: Vec<_> = confirmed.drain(..cap).collect();

    // Critic CRITICAL (2026-07-30): a daemon that already proved it ignores
    // `record_only` — within its 1-hour TTL — must never be asked to
    // decommission again this call. An EXPIRED sentinel is cleared here so a
    // now-current daemon isn't wedged off forever.
    let already_stale_daemon = stale_daemon_sentinel_active(&seen);
    if !already_stale_daemon && seen.remove(STALE_DAEMON_MARKER_KEY).is_some() {
        changed = true;
    }
    let mut newly_stale_daemon = false;
    let mut pruned = 0usize;
    // Owner request 2026-07-30 follow-up: records held back by the
    // stale-daemon gate (already-active OR newly-tripped-mid-loop) must fold
    // into `pending` — otherwise "N more dead records pending confirmation"
    // under-reports exactly the batch this gate is protecting.
    let mut stale_daemon_held = 0usize;

    if already_stale_daemon {
        stale_daemon_held = to_prune.len();
        kept.extend(to_prune);
    } else {
        let mut iter = to_prune.into_iter();
        for s in iter.by_ref() {
            match decommission_dead_record(client, url, &s.id).await {
                DecommissionOutcome::Pruned => {
                    pruned += 1;
                    seen.remove(&s.id);
                    changed = true;
                }
                DecommissionOutcome::Failed => kept.push(s),
                DecommissionOutcome::StaleDaemon => {
                    newly_stale_daemon = true;
                    stale_daemon_held += 1;
                    kept.push(s);
                    break;
                }
            }
        }
        // `iter.by_ref()` leaves every un-visited item (after the `break`)
        // in `iter` — collect them back rather than silently dropping them.
        let remainder: Vec<_> = iter.collect();
        stale_daemon_held += remainder.len();
        kept.extend(remainder);
    }

    if newly_stale_daemon {
        seen.insert(STALE_DAEMON_MARKER_KEY.to_string(), now);
        changed = true;
    }
    if already_stale_daemon || newly_stale_daemon {
        eprintln!(
            "tm: auto-prune: daemon may be running an older build; skipping \
             automatic pruning until it's restarted"
        );
    }

    let pending = confirmed.len() + first_sighting.len() + stale_daemon_held;
    kept.extend(confirmed);
    kept.extend(first_sighting);

    if changed {
        save_seen(marker_path, &seen);
    }

    AutoPruneOutcome {
        kept,
        pruned,
        pending,
    }
}

/// Whether the persisted `STALE_DAEMON_MARKER_KEY` sentinel is still within
/// its [`STALE_DAEMON_TTL_SECS`] TTL (owner request 2026-07-30 follow-up).
///
/// Why: see [`STALE_DAEMON_TTL_SECS`]'s doc — a permanent lockout would
/// outlive the daemon restart that fixes it.
/// What: `true` only when the sentinel is present AND parses as an RFC 3339
/// timestamp no more than [`STALE_DAEMON_TTL_SECS`] old. A missing, corrupt,
/// or expired sentinel returns `false` (retry allowed) — the caller is
/// responsible for clearing an expired entry from `seen` so it doesn't
/// linger.
/// Test: `auto_prune_dead_records_stale_daemon_sentinel_expires_after_ttl`.
fn stale_daemon_sentinel_active(seen: &HashMap<String, String>) -> bool {
    seen.get(STALE_DAEMON_MARKER_KEY)
        .and_then(|ts| chrono::DateTime::parse_from_rfc3339(ts).ok())
        .is_some_and(|ts| {
            chrono::Utc::now().signed_duration_since(ts)
                < chrono::Duration::seconds(STALE_DAEMON_TTL_SECS)
        })
}

/// Load the persisted first-sighting map, tolerating a missing or corrupt
/// file (treated as empty — a listing must never fail merely because its
/// confirmation marker couldn't be read).
///
/// What: `{ "<session-id>": "<rfc3339 first-sighting timestamp>" }`.
fn load_seen(path: &Path) -> HashMap<String, String> {
    std::fs::read_to_string(path)
        .ok()
        .and_then(|raw| serde_json::from_str(&raw).ok())
        .unwrap_or_default()
}

/// Best-effort persist of the first-sighting map; a write failure is logged
/// to stderr and never propagated (the listing must never fail merely
/// because the marker file couldn't be written).
fn save_seen(path: &Path, seen: &HashMap<String, String>) {
    if let Some(parent) = path.parent() {
        let _ = std::fs::create_dir_all(parent);
    }
    match serde_json::to_string(seen) {
        Ok(json) => {
            if let Err(e) = std::fs::write(path, json) {
                eprintln!("tm: auto-prune: failed to persist confirmation marker: {e}");
            }
        }
        Err(e) => eprintln!("tm: auto-prune: failed to serialize confirmation marker: {e}"),
    }
}

/// Outcome of one [`decommission_dead_record`] call.
///
/// Why: an HTTP 200 alone is NOT proof that `record_only` was honored — a
/// stale daemon lacking the `Query<DecommissionQuery>` extractor also
/// returns 200 (from the OLD unconditional full-teardown path). Only the
/// response body's `workspace_removed` field can tell the two apart, so the
/// caller needs a THIRD outcome distinct from plain success/failure.
enum DecommissionOutcome {
    /// The daemon confirmed `workspace_removed == false` — genuinely
    /// record-only, safe to count as pruned.
    Pruned,
    /// The response proves (or fails to rule out) that the daemon actually
    /// removed the workspace — `workspace_removed == true`, the field was
    /// missing, or the body didn't even parse as JSON. Only reachable on a
    /// 2xx status; treated as "a stale daemon may have just deleted this,"
    /// never assumed safe.
    StaleDaemon,
    /// Non-2xx status or a transport error — the call simply failed.
    Failed,
}

/// POST the existing `/decommission` route in RECORD-ONLY mode for one
/// confirmed-dead record, best-effort (critic HIGH finding #1; response-body
/// verification added per critic CRITICAL finding, 2026-07-30).
///
/// Why: kept as the single I/O primitive behind [`auto_prune_dead_records_at`]
/// so a per-record transport/HTTP failure can never abort the rest of the
/// listing — one stuck record must not hide every other session behind it.
/// POSTing `?record_only=true` alone is NOT sufficient proof of safety: a
/// daemon process still running a build that predates this route's
/// `Query<DecommissionQuery>` extractor silently ignores the param (axum
/// drops query params a handler doesn't declare) and falls through to the
/// OLD unconditional full-teardown `decommission()` — which also returns
/// 200. This project's daemon is long-lived by design (a CLI upgrade never
/// bounces it), so that version skew is the ROUTINE case, not an edge case.
/// What: POSTs `?record_only=true`; on a 2xx response, parses the JSON body
/// and requires `workspace_removed == false` before returning
/// [`DecommissionOutcome::Pruned`] — the ONLY outcome that proves nothing was
/// deleted. `workspace_removed == true`, a missing field, or an unparseable
/// body all return [`DecommissionOutcome::StaleDaemon`]. A non-2xx status or
/// a transport error is logged to stderr and returns
/// [`DecommissionOutcome::Failed`].
/// Test: `auto_prune_dead_records_removes_confirmed_unresumable_records`
/// drives the happy path through a real loopback daemon;
/// `auto_prune_dead_records_stops_sweep_when_daemon_reports_workspace_removed`
/// drives the stale-daemon path through a stub server that ignores
/// `record_only` and always reports `workspace_removed: true`.
async fn decommission_dead_record(
    client: &reqwest::Client,
    url: &str,
    id: &str,
) -> DecommissionOutcome {
    let resp = match client
        .post(format!("{url}/api/v1/sessions/managed/{id}/decommission"))
        .query(&[("record_only", "true")])
        .send()
        .await
    {
        Ok(resp) => resp,
        Err(e) => {
            eprintln!("tm: auto-prune: failed to remove dead record {id}: {e}");
            return DecommissionOutcome::Failed;
        }
    };
    if !resp.status().is_success() {
        eprintln!(
            "tm: auto-prune: failed to remove dead record {id}: HTTP {}",
            resp.status()
        );
        return DecommissionOutcome::Failed;
    }
    match resp.json::<serde_json::Value>().await {
        Ok(body) => match body.get("workspace_removed").and_then(|v| v.as_bool()) {
            Some(false) => DecommissionOutcome::Pruned,
            _ => DecommissionOutcome::StaleDaemon,
        },
        Err(_) => DecommissionOutcome::StaleDaemon,
    }
}