trusty-mpm 1.8.2

//! Replay the `SubagentStop` records a hook could not deliver (#6556).
//!
//! Why: the hook is the only process that ever learns a subagent stopped, and it
//! learns it once. When this daemon could not answer, every attempt that hook
//! made failed and it parked the stop in [`crate::core::stop_spool`] instead of
//! losing it. Without a reader that park is a write-only file: the delegation
//! still sits `Running` until the six-hour `RUNNING_STALE_AFTER_SECS` sweep,
//! holding a builder slot and a checkout. This is that reader, and the reap loop
//! is where it runs — the same loop that owns the staleness sweep this fix
//! exists to beat. The recoverable case is a WEDGED-BUT-ALIVE daemon; see the
//! [`crate::core::stop_spool`] header for why a restart is not one.
//!
//! What: [`drain_unposted_stops`] replays each parked body through the same
//! [`crate::daemon::rpc::sessions_legacy_ops::ingest_hook`] a live POST would
//! have reached, so a recovered stop terminalizes its delegation by exactly the
//! path the hook's own POST does. Every record is discarded after its replay,
//! including one the daemon refuses — a body this daemon cannot accept will not
//! become acceptable on the next tick, and leaving it would wedge the directory
//! against [`crate::core::stop_spool::MAX_UNPOSTED_STOPS`].
//!
//! Test: the module's `#[cfg(test)]` suite.

use std::sync::Arc;

use crate::core::session::SessionId;
use crate::core::stop_spool;
use crate::daemon::api::HookPost;
use crate::daemon::state::DaemonState;

/// Replay and discard every parked stop under this daemon's framework root.
///
/// Why: see the module header.
/// What: reads `<framework-root>/unposted-stops/` oldest first, deserializes
/// each body as a [`HookPost`], and ingests it. Returns how many were replayed
/// successfully. Unparsable and refused records are discarded too, and counted
/// separately only in the log — the caller's number is "stops recovered".
///
/// **A replay only ever terminalizes an EXACT `agent_id` match** (critic HIGH on
/// PR #8052). Reached through `ingest_hook`, a stop whose id no record carries
/// falls into
/// [`reconcile_by_agent_type`](crate::daemon::services::delegation_tracker), which
/// stales the single live record of the same `agent_type` that never learned an
/// id. That inference is sound for a stop arriving live; it is NOT sound for a
/// replay, because the delegation map is in-memory and the spool outlives the
/// process that wrote it — so after a restart EVERY replayed stop takes that
/// fallback, and a resumed session that re-dispatched the same agent type inside
/// the ≤60 s before the first reap tick has a LIVE agent's record terminalized
/// (#5661's shape). [`replay_target`] is therefore checked first and a record
/// with no live, exactly-matching `agent_id` is discarded unreplayed.
/// Test: `a_parked_stop_terminalizes_its_delegation`,
/// `a_replay_whose_agent_id_matches_nothing_never_stales_a_sibling`,
/// `a_parked_non_stop_event_is_discarded_unreplayed`,
/// `a_replay_cannot_reassert_an_idle_parking_nudge`,
/// `a_corrupt_record_is_discarded_rather_than_wedging_the_drain`,
/// `an_empty_spool_drains_nothing`, `the_drain_sweeps_temp_litter_before_replaying`.
pub async fn drain_unposted_stops(state: &Arc<DaemonState>) -> usize {
    let root = state.framework_root().to_path_buf();
    // Before the read, and before the empty-spool return: a tree holding only
    // litter still needs it swept (critic LOW on PR #8052).
    let swept = stop_spool::sweep_stale_temp_files(&root);
    if swept > 0 {
        tracing::warn!(
            "stop-spool: swept {swept} temp file(s) a killed hook left mid-write, older than the \
             drain window (#6556)"
        );
    }
    let parked = stop_spool::read_unposted_stops(&root);
    if parked.is_empty() {
        return 0;
    }
    let mut replayed = 0usize;
    let mut discarded = 0usize;
    for (path, body) in parked {
        match body
            .and_then(|v| serde_json::from_value::<HookPost>(v).ok())
            .and_then(|post| replay_target(state, post))
        {
            Some(post) => {
                match crate::daemon::rpc::sessions_legacy_ops::ingest_hook(state, post).await {
                    Ok(_) => replayed += 1,
                    Err(e) => {
                        discarded += 1;
                        tracing::warn!(
                            record = %path.display(),
                            "stop-spool: the daemon refused a parked SubagentStop ({e}); discarding \
                             it rather than retrying a body that cannot become acceptable (#6556)"
                        );
                    }
                }
            }
            None => {
                discarded += 1;
                tracing::warn!(
                    record = %path.display(),
                    "stop-spool: a parked record is not a replayable SubagentStop — unreadable, \
                     not a stop, or naming an agent_id no live record carries; discarding it \
                     rather than letting it stale a sibling by agent_type (#6556)"
                );
            }
        }
        stop_spool::discard_unposted_stop(&path);
    }
    tracing::info!(
        "stop-spool: recovered {replayed} SubagentStop record(s) the hook could not deliver, \
         discarded {discarded} (#6556)"
    );
    replayed
}

/// Accept `post` for replay, or refuse it.
///
/// Why: the three ways a parked body must not reach `ingest_hook`. The
/// `agent_id` rule is the one that matters — see [`drain_unposted_stops`].
/// What: `Some(post)` only when the event is `SubagentStop`, the payload names
/// an `agent_id`, and THIS daemon holds a live delegation in `post.session_id`
/// carrying exactly that id. The returned post has `payload.idle_parking`
/// stripped: that flag spawns the #2610 auto-nudge, and a nudge minutes or a
/// restart late is aimed at a turn that has already ended.
/// Test: `a_replay_whose_agent_id_matches_nothing_never_stales_a_sibling`,
/// `a_parked_non_stop_event_is_discarded_unreplayed`,
/// `a_replay_cannot_reassert_an_idle_parking_nudge`.
fn replay_target(state: &Arc<DaemonState>, mut post: HookPost) -> Option<HookPost> {
    if post.event != crate::core::hook::HookEvent::SubagentStop {
        return None;
    }
    let agent_id = post.payload.get("agent_id")?.as_str()?.to_string();
    let session = uuid::Uuid::parse_str(&post.session_id)
        .ok()
        .map(SessionId)?;
    state.find_delegation(session, |d| {
        d.agent_id.as_deref() == Some(agent_id.as_str()) && !d.status.is_terminal()
    })?;
    if let Some(obj) = post.payload.as_object_mut() {
        obj.remove("idle_parking");
    }
    Some(post)
}

#[cfg(test)]
mod stop_spool_drain_tests {
    use super::*;
    use crate::core::agent::{Delegation, DelegationStatus};
    use crate::core::paths::FrameworkPaths;

    /// A hermetic daemon whose framework root is a temp dir, plus one session.
    fn hermetic() -> (Arc<DaemonState>, tempfile::TempDir, SessionId) {
        let dir = tempfile::tempdir().expect("temp dir");
        let paths = FrameworkPaths::under(dir.path());
        let state = Arc::new(DaemonState::with_paths(&paths));
        (state, dir, SessionId(uuid::Uuid::new_v4()))
    }

    /// One running delegation the parked stop will terminalize.
    fn running(state: &DaemonState, session: SessionId, agent_id: &str) {
        let mut d = Delegation::observed(session, "rust-engineer", "task", Some("toolu_1".into()));
        d.status = DelegationStatus::Running;
        d.agent_id = Some(agent_id.to_string());
        state.upsert_delegation(d);
    }

    /// The body `tm hook` parks: the exact `POST /hooks` envelope.
    fn parked_body(session: SessionId, agent_id: &str) -> serde_json::Value {
        serde_json::json!({
            "session_id": session.0.to_string(),
            "event": "SubagentStop",
            "payload": {"agent_id": agent_id, "tool": "Agent"},
        })
    }

    /// 🔴 REGRESSION (#6556): the park is only worth writing if something reads
    /// it. A daemon that was down when the agent stopped recovers the stop on its
    /// next reap tick, instead of the record sitting `Running` for six hours.
    #[tokio::test]
    async fn a_parked_stop_terminalizes_its_delegation() {
        let (state, dir, session) = hermetic();
        running(&state, session, "agent-1");
        let root = FrameworkPaths::under(dir.path()).root;
        stop_spool::record_unposted_stop(&root, &parked_body(session, "agent-1")).expect("parked");

        assert_eq!(drain_unposted_stops(&state).await, 1);

        let records = state.delegations_for(session);
        assert_eq!(records.len(), 1);
        assert!(
            !records[0].status.is_live(),
            "the recovered stop ended the record, status {:?}",
            records[0].status
        );
        assert!(
            stop_spool::read_unposted_stops(&root).is_empty(),
            "a replayed record must not replay again"
        );
    }

    /// 🔴 REGRESSION (critic HIGH on PR #8052): a replay whose `agent_id` no
    /// live record carries must terminalize NOTHING.
    ///
    /// Why: delegations are in-memory, so the restart the spool exists across
    /// wipes every record. Replayed through `ingest_hook`, such a stop falls into
    /// `reconcile_by_agent_type`, which stales the one live record of the same
    /// `agent_type` with no id — and a resumed session re-dispatching that agent
    /// type inside the ≤60 s before the first reap tick has a LIVE agent
    /// terminalized under it (#5661's shape).
    /// Fails at cf471c0c9: the sibling reads `Stale` and the drain reports 1.
    #[tokio::test]
    async fn a_replay_whose_agent_id_matches_nothing_never_stales_a_sibling() {
        let (state, dir, session) = hermetic();
        // The restart-equivalent state: a live record of the SAME agent_type
        // that never learned an id — exactly what the fallback matches on.
        let mut sibling =
            Delegation::observed(session, "rust-engineer", "task", Some("toolu_new".into()));
        sibling.status = DelegationStatus::Running;
        state.upsert_delegation(sibling);
        let root = FrameworkPaths::under(dir.path()).root;
        let mut body = parked_body(session, "agent-gone");
        body["payload"]["agent_type"] = serde_json::json!("rust-engineer");
        stop_spool::record_unposted_stop(&root, &body).expect("parked");

        assert_eq!(
            drain_unposted_stops(&state).await,
            0,
            "a stop no live record answers is not a recovery"
        );

        let records = state.delegations_for(session);
        assert_eq!(records.len(), 1);
        assert_eq!(
            records[0].status,
            DelegationStatus::Running,
            "the newly dispatched sibling must be untouched"
        );
        assert!(stop_spool::read_unposted_stops(&root).is_empty());
    }

    /// The spool carries stops. Anything else reaching `ingest_hook` from disk
    /// would re-run a whole hook pipeline — auto-registration, the overseer —
    /// against a payload from another process lifetime.
    #[tokio::test]
    async fn a_parked_non_stop_event_is_discarded_unreplayed() {
        let (state, dir, session) = hermetic();
        running(&state, session, "agent-1");
        let root = FrameworkPaths::under(dir.path()).root;
        let mut body = parked_body(session, "agent-1");
        body["event"] = serde_json::json!("PreToolUse");
        stop_spool::record_unposted_stop(&root, &body).expect("parked");

        assert_eq!(drain_unposted_stops(&state).await, 0);
        assert!(
            state.delegations_for(session)[0].status.is_live(),
            "a non-stop event changes nothing"
        );
    }

    /// The #2610 auto-nudge fires on `payload.idle_parking`. A nudge aimed at a
    /// turn that ended a restart ago is noise, so the flag is stripped on replay.
    #[tokio::test]
    async fn a_replay_cannot_reassert_an_idle_parking_nudge() {
        let (state, _dir, session) = hermetic();
        running(&state, session, "agent-1");
        let mut body = parked_body(session, "agent-1");
        body["payload"]["idle_parking"] = serde_json::json!(true);
        let post: HookPost = serde_json::from_value(body).expect("a hook body");

        let accepted = replay_target(&state, post).expect("an exact id match replays");
        assert!(
            accepted.payload.get("idle_parking").is_none(),
            "the nudge flag does not survive the park: {:?}",
            accepted.payload
        );
    }

    /// A record this daemon cannot read is discarded, not retried forever —
    /// otherwise one corrupt file fills the spool to its cap and every later
    /// stop is dropped at the writer.
    #[tokio::test]
    async fn a_corrupt_record_is_discarded_rather_than_wedging_the_drain() {
        let (state, dir, _session) = hermetic();
        let root = FrameworkPaths::under(dir.path()).root;
        let spool = stop_spool::unposted_stops_dir(&root);
        std::fs::create_dir_all(&spool).expect("dir");
        std::fs::write(spool.join("0000-0-0.json"), "{ not json").expect("write");
        // Valid JSON, but not a hook body — the other way a record is unusable.
        std::fs::write(spool.join("0001-0-0.json"), r#"{"event":"SubagentStop"}"#).expect("write");

        assert_eq!(drain_unposted_stops(&state).await, 0);
        assert!(stop_spool::read_unposted_stops(&root).is_empty());
    }

    /// The common case: nothing parked, no work, no log noise.
    #[tokio::test]
    async fn an_empty_spool_drains_nothing() {
        let (state, _dir, _session) = hermetic();
        assert_eq!(drain_unposted_stops(&state).await, 0);
    }

    /// A spool holding ONLY a killed writer's temp file still gets swept — the
    /// sweep runs before the "nothing parked" return, or litter in an otherwise
    /// empty spool would never be reached (critic LOW on PR #8052).
    #[tokio::test]
    async fn the_drain_sweeps_temp_litter_before_replaying() {
        let (state, dir, _session) = hermetic();
        let root = FrameworkPaths::under(dir.path()).root;
        let spool = stop_spool::unposted_stops_dir(&root);
        std::fs::create_dir_all(&spool).expect("dir");
        let litter = spool.join(".tmpKilled");
        std::fs::write(&litter, "half a record").expect("write");
        let aged = std::time::SystemTime::now()
            - std::time::Duration::from_secs(stop_spool::SPOOL_DRAIN_GRACE_SECS + 5);
        std::fs::File::options()
            .write(true)
            .open(&litter)
            .expect("open")
            .set_times(std::fs::FileTimes::new().set_modified(aged))
            .expect("backdate");

        assert_eq!(drain_unposted_stops(&state).await, 0);
        assert!(!litter.exists(), "the drain swept the litter");
    }
}