hyperhive/hive-c0re/src/workers/crash_watch.rs

//! Per-container crash and login-state watcher. Polls every managed
//! container on a 10s interval. A classified crash only `tracing::warn!`s
//! (`"container crash detected"`) — a swarm-level grafana alert matches
//! that line directly; see the module doc on [`emit_crash_transitions`].
//! Login-state transitions (`needs_login`/`logged_in`) publish onto
//! `swarm_notices` instead. The `RECENT_TRANSIENT_GRACE` window:
//! `docs/agent-lifecycle/config-changes.md::Helper events`.

use std::collections::HashSet;
use std::sync::Arc;
use std::time::Duration;

use crate::container_view::claude_has_session;
use crate::coordinator::Coordinator;
use crate::lifecycle::{self, AGENT_PREFIX};

const POLL_INTERVAL: Duration = Duration::from_secs(10);

/// How long an operator-initiated transient stays "recently cleared"
/// for the purpose of suppressing crash events. Three full
/// `POLL_INTERVAL`s gives the post-lifecycle path comfortable
/// breathing room — the watcher will have polled at least twice
/// inside the window even with worst-case timer skew.
const RECENT_TRANSIENT_GRACE: Duration = Duration::from_secs(30);

pub fn spawn(coord: Arc<Coordinator>) {
    let mut shutdown = coord.shutdown_rx();
    tokio::spawn(async move {
        let mut prev_running: HashSet<String> = HashSet::new();
        let mut prev_logged_in: HashSet<String> = HashSet::new();
        let mut prev_sub_agents: HashSet<String> = HashSet::new();
        let mut seeded = false;
        loop {
            let list_result = lifecycle::list().await;
            if let Err(e) = &list_result {
                tracing::warn!(
                    error = ?e,
                    "crash watcher: container list unreadable — skipping this cycle"
                );
            }
            if let CyclePlan::Process(raw) = plan_cycle(list_result) {
                let mut current_running = HashSet::new();
                let mut current_logged_in = HashSet::new();
                let mut sub_agents: Vec<String> = Vec::new();
                for c in &raw {
                    let Some(logical) = c.strip_prefix(AGENT_PREFIX) else {
                        continue;
                    };
                    let logical = logical.to_owned();
                    sub_agents.push(logical.clone());
                    let state = lifecycle::running_state(&logical).await;
                    if counts_as_running(state, prev_running.contains(&logical)) {
                        current_running.insert(logical.clone());
                    }
                    if hive_types::Ident::parse(&logical)
                        .is_ok_and(|id| claude_has_session(&Coordinator::agent_claude_dir(&id)))
                    {
                        current_logged_in.insert(logical.clone());
                    }
                }

                if seeded {
                    emit_crash_transitions(&coord, &prev_running, &current_running);
                    emit_login_transitions(
                        &prev_logged_in,
                        &current_logged_in,
                        &sub_agents,
                        &prev_sub_agents,
                    )
                    .await;
                }
                // Periodic container rescan — catches state flips that
                // happen outside our mutation surface (operator runs
                // `nixos-container stop` over ssh, agent logs in via its
                // own web UI, etc.) so the dashboard converges within one
                // POLL_INTERVAL. Idempotent + cheap when nothing changed.
                coord.rescan_containers_and_emit().await;
                prev_running = current_running;
                prev_logged_in = current_logged_in;
                prev_sub_agents = sub_agents.into_iter().collect();
                seeded = true;
            }
            // On `CyclePlan::Skip`, `prev_*` and `seeded` are left exactly as
            // the last good cycle produced them, so the next successful read
            // diffs against real prior state instead of an empty one.

            tokio::select! {
                () = tokio::time::sleep(POLL_INTERVAL) => {}
                _ = shutdown.changed() => {
                    tracing::info!("crash watcher: shutdown signal received");
                    break;
                }
            }
        }
    });
}

/// What this poll cycle does with `lifecycle::list()`'s result.
enum CyclePlan {
    /// The list read; diff `Vec` against `prev` as usual.
    Process(Vec<String>),
    /// The read failed: skip the cycle outright.
    Skip,
}

/// Turn a `lifecycle::list()` result into this poll's plan. An unreadable
/// list is not "zero containers" — treating it as such made every
/// previously-running agent land in `prev.difference(current)`, i.e. look
/// like it crashed simultaneously with every other agent. So a read failure
/// plans `Skip`: no diff against `prev`, no crash or login events this
/// cycle, and (at the call site) `prev` stays exactly what the last good
/// read left it, so the next successful read compares against real state
/// rather than an empty one.
fn plan_cycle(list_result: anyhow::Result<Vec<String>>) -> CyclePlan {
    match list_result {
        Ok(raw) => CyclePlan::Process(raw),
        Err(_) => CyclePlan::Skip,
    }
}

/// An unreadable unit state (`None`) keeps the agent's last observed state,
/// so a hung or failed `systemctl` is neither a crash nor a recovery.
fn counts_as_running(state: Option<bool>, was_running: bool) -> bool {
    state.unwrap_or(was_running)
}

fn emit_crash_transitions(coord: &Coordinator, prev: &HashSet<String>, current: &HashSet<String>) {
    let transients = coord.transient_snapshot();
    // Operator actions whose RAII guard already cleared but only just;
    // suppresses the race where `lifecycle::kill` returns + drops the
    // guard between two crash-watch polls.
    let recent = coord.recent_transient_within(RECENT_TRANSIENT_GRACE);
    for stopped in prev.difference(current) {
        // One source: a running queue node that declared it takes the container
        // down. There used to be a second — an imperative suppression guard for
        // operations with no node behind them — and destroy becoming a DAG
        // removed the last caller, so intent now has exactly one home.
        // `any`, not "the" pill: an agent can have several running nodes at
        // once (a lease-exempt build alongside a lease-holding stop), and it
        // only takes one of them expecting the container down for this to be
        // a deliberate stop rather than a crash.
        let active = transients
            .get(stopped)
            .map(|sts| sts.iter().any(|st| st.takes_container_down));
        let recently_cleared = recent.get(stopped).copied();
        if is_deliberate_stop(active, recently_cleared) {
            continue;
        }
        // The one place this fires: a swarm-level grafana alert
        // (`hh-container-crashed` in `nix/host-modules/swarm-grafana.nix`)
        // matches this exact message against the log store and pages the
        // operator directly — no agent (and no per-hive warning)
        // in the loop. See `docs/scheduler/observability.md`.
        tracing::warn!(agent = %stopped, "container crash detected");
    }
}

/// Pure classifier: did an operation take this container down on purpose, or
/// did it crash? Splits the matcher out so it has a focused unit test without
/// needing a Coordinator fixture. `active` is the currently-set transient's
/// `deliberate_stop` (if any), `recently_cleared` is one whose RAII guard
/// dropped within the grace window.
///
/// This reads the flag the transient's creator set and nothing else. It used to
/// match a `TransientKind`, which meant a pill's *display* vocabulary decided a
/// crash-alert question — so renaming or adding a label silently moved the
/// alerting boundary. Whoever starts the operation knows whether the container
/// is meant to go down; nothing downstream can re-derive it.
fn is_deliberate_stop(active: Option<bool>, recently_cleared: Option<bool>) -> bool {
    active.unwrap_or(false) || recently_cleared.unwrap_or(false)
}

async fn emit_login_transitions(
    prev: &HashSet<String>,
    current: &HashSet<String>,
    sub_agents: &[String],
    prev_sub_agents: &HashSet<String>,
) {
    for agent in current.difference(prev) {
        tracing::info!(%agent, "agent logged in");
        crate::swarm_notices::notify(
            "core",
            Some(format!("logged_in:{agent}")),
            format!("agent '{agent}' logged in"),
            None,
        )
        .await;
    }
    // Detect transitions into "needs login": an agent that was previously
    // logged-in goes unsigned (credentials deleted), OR a brand-new agent
    // appears without a session.
    //
    // prev_needs uses prev_sub_agents (the agent set from the last tick) so
    // that a newly-spawned agent — which does not appear in prev_sub_agents —
    // is absent from prev_needs even though it's not in prev_logged_in.
    // Without this, new agents land in both prev_needs and current_needs and
    // the set difference is empty, silently dropping the event.
    let prev_needs: HashSet<&str> = prev_sub_agents
        .iter()
        .map(String::as_str)
        .filter(|n| !prev.contains(*n))
        .collect();
    let current_needs: HashSet<&str> = sub_agents
        .iter()
        .map(String::as_str)
        .filter(|n| !current.contains(*n))
        .collect();
    for agent in current_needs.difference(&prev_needs) {
        tracing::info!(%agent, "agent needs login");
        crate::swarm_notices::notify(
            "core",
            Some(format!("needs_login:{agent}")),
            format!("agent '{agent}' needs login"),
            None,
        )
        .await;
    }
}

#[cfg(test)]
mod tests {
    use super::*;

    /// The regression this whole fix exists for: a read failure must not
    /// reach `emit_crash_transitions` at all, so `prev` is left untouched
    /// for the next cycle to diff against.
    #[test]
    fn list_error_skips_the_cycle() {
        let result: anyhow::Result<Vec<String>> =
            Err(anyhow::anyhow!("connect to hive-priv socket"));
        assert!(matches!(plan_cycle(result), CyclePlan::Skip));
    }

    /// Control: a real drop from running to absent is a *readable*, empty
    /// list — it must still be processed (and so still reach
    /// `emit_crash_transitions`, which warns the crash), unlike a
    /// read failure.
    #[test]
    fn readable_empty_list_still_processes() {
        let result: anyhow::Result<Vec<String>> = Ok(Vec::new());
        match plan_cycle(result) {
            CyclePlan::Process(raw) => assert!(raw.is_empty()),
            CyclePlan::Skip => panic!("a readable empty list must not be treated as unreadable"),
        }
    }

    #[test]
    fn unreadable_unit_state_keeps_the_last_observation() {
        assert!(counts_as_running(None, true), "no crash on a timeout");
        assert!(!counts_as_running(None, false), "no recovery on a timeout");
        // Control: a readable stop still drops out of the running set.
        assert!(!counts_as_running(Some(false), true));
        assert!(counts_as_running(Some(true), false));
    }

    #[test]
    fn deliberate_when_either_source_says_so() {
        // Active guard, and the race repro: a lifecycle action completes and
        // drops its guard between two polls, so only `recent` still carries it.
        assert!(is_deliberate_stop(Some(true), None));
        assert!(is_deliberate_stop(None, Some(true)));
        assert!(is_deliberate_stop(Some(true), Some(true)));
    }

    #[test]
    fn not_deliberate_with_no_transient_at_all() {
        // The real-crash case — warns "container crash detected".
        assert!(!is_deliberate_stop(None, None));
    }

    #[test]
    fn not_deliberate_when_the_operation_was_bringing_the_container_up() {
        // The case that used to be spelled `Spawning` / `Starting`: an
        // operation IS in flight, but it is not one that takes the container
        // down, so a container that vanishes under it really did crash.
        //
        // This is why `deliberate_stop` is carried rather than inferred from
        // the pill — `Create` and `Start` hold the agent's lease exactly like
        // `Stop` does, so "has a pill" cannot answer this.
        assert!(!is_deliberate_stop(Some(false), None));
        assert!(!is_deliberate_stop(None, Some(false)));
        assert!(!is_deliberate_stop(Some(false), Some(false)));
    }
}