hyperhive/hive-c0re/src/coordinator.rs

//! Runtime state + config shared between the host admin socket, the manager
//! socket, and the per-agent sockets: the broker, configured `agent_flake`,
//! and the map of registered agent sockets.

use std::collections::{HashMap, HashSet};
use std::path::{Path, PathBuf};
use std::sync::atomic::{AtomicU64, Ordering};
use std::sync::{Arc, Mutex};

use anyhow::{Context, Result};
use tokio::sync::{broadcast, watch};

use crate::broker::Broker;
use crate::container_view::{self, ContainerView};
use crate::dashboard_events::DashboardEvent;
use crate::job_queue::RunningTransient;
use crate::socket_server::{self, AgentSocket};

/// Capacity of the dashboard event channel. Slow browser subscribers
/// (idle tab, throttled connection) drop frames past this — that's
/// fine, the seq dedupe makes a reconnect resync safe.
const DASHBOARD_CHANNEL: usize = 256;

pub struct Coordinator {
    pub broker: Arc<Broker>,
    /// Full build-log capture. `lifecycle::run` `start()`s a row per attempt,
    /// pipes every stdout/stderr line into it, and `finish()`es it on
    /// child exit. Dashboard reads it via `list_recent_for_agent` /
    /// `get_full` for the per-card chip + side-panel viewer. See
    /// `build_logs.rs` for retention.
    pub build_logs: Arc<crate::build_logs::BuildLogs>,
    /// Store path of the hyperhive flake source (no fragment). Used to
    /// identify the running revision (`current_flake_rev`).
    pub hyperhive_flake: String,
    /// Store path of the `claude-code` build every agent runs, handed to
    /// each container at start as `HYPERHIVE_CLAUDE_CODE_PATH`. Set by
    /// the NixOS module option
    /// `services.hyperhive.deploy.hive-controller.claudeCodePackage`
    /// (which resolves the package and hands us its path). `None` = every
    /// agent keeps the `claude-code` from its own nixpkgs.
    ///
    /// A path rather than a flake input because containers share the
    /// host's `/nix/store`: the binary is already reachable inside them,
    /// closure and all. The host module is what keeps it from being
    /// garbage-collected — see that option.
    pub claude_code_path: Option<String>,
    /// TCP port the host's hive-c0re dashboard listens on. Handed to each
    /// container at start as `HIVE_DASHBOARD_PORT`, so the agent's web UI
    /// can build the rebuild-button URL pointing back at the dashboard.
    pub dashboard_port: u16,
    /// Per-agent systemd `CPUQuota=` value (e.g. `"200%"`). Written into
    /// the `container@h-<name>.service.d/` drop-in on every spawn/rebuild.
    pub agent_cpu_quota: String,
    /// Per-agent systemd `MemoryMax=` value (e.g. `"4G"`). Same drop-in.
    pub agent_memory_max: String,
    /// Per-agent systemd `CPUWeight=` (cgroup v2 `cpu.weight`, 1..=10000).
    /// Same drop-in. A *relative share* under contention, not a cap — see
    /// the `agentCpuWeight` NixOS option for the full semantics. `None`
    /// (option set to `null`) omits the setting: kernel default, and the
    /// drop-in is byte-identical to the one written before weights existed.
    pub agent_cpu_weight: Option<u32>,
    /// Per-agent systemd `IOWeight=` (cgroup v2 `io.weight`, 1..=10000).
    /// Same drop-in, same relative-share and `None` semantics as
    /// [`Self::agent_cpu_weight`].
    pub agent_io_weight: Option<u32>,
    agents: Mutex<HashMap<String, AgentSocket>>,
    /// Tombstone for transients that have JUST been cleared. The
    /// crash watcher polls every 10s and would race the
    /// drop-clears-immediately path of `TransientGuard`: an operator
    /// kill / restart sets `Stopping` → runs `nixos-container stop` →
    /// drop clears the transient → poll fires next tick and sees the
    /// container missing-from-running with no active transient →
    /// spurious "container stopped without an operator action"
    /// message.
    ///
    /// `clear_transient` stamps the cleared kind here with an
    /// `Instant`; `recent_transient_within(grace)` returns the set of
    /// agents whose tombstone is still inside the grace window. Crash
    /// watcher consults both this and the active map before declaring
    /// a stop deliberate.
    ///
    /// 🚨 **Keyed by `(agent, label)`, not by agent.** An agent can have several
    /// pills clearing in the same window — the transient set tests status alone,
    /// so a lease-exempt `Prebuild` and a lease-holding `StopForUpdate` are both
    /// live and both clear. Keyed by agent, the last clear *overwrites* the
    /// others: a `Prebuild` (`deliberate_stop = false`) landing after a
    /// `StopForUpdate` (`true`) leaves the tombstone reading `false`, and the
    /// crash watcher then reports an intentional stop as a **crash**.
    recent_transient: Mutex<HashMap<(String, String), (bool, std::time::Instant)>>,
    /// Agents with a graceful stop in progress. Set by the `GracefulStop`
    /// orchestration; read by `socket_server::handle_recv`, which returns
    /// `Response::GracefulStop` (instead of polling the broker) while an
    /// agent is in this set — the inbound fence. Cleared when the agent
    /// reports `GracefulStopComplete` or the container is stopped.
    graceful_stop_pending: Mutex<HashSet<String>>,
    /// Agents with a pause acknowledgement in progress — set by the pause
    /// DAG's signal node, cleared when the agent reports
    /// `PauseAcknowledged` (or the drain node's timeout fallback fires).
    /// Same shape as `graceful_stop_pending`, one set per concern rather
    /// than reusing it: a pause and a graceful stop are independent
    /// orchestrations that can be in flight for different agents
    /// simultaneously.
    pause_pending: Mutex<HashSet<String>>,
    /// Agents an update has taken down and not yet brought back: set when
    /// `StopForUpdate` stops a running container, cleared when the tail
    /// `Reconcile` / `Start` settles the agent. Published in the agent's
    /// status snapshot so the swarm reads the gap as an update and not as a
    /// stopped or lost agent. In memory only: a hive restart mid-update loses
    /// the flag, and the status sweep then reads the agent as stopped until it
    /// is back, as it did before the flag existed.
    updating: Mutex<HashSet<String>>,
    /// Woken when [`Self::set_updating`] changes the set, so the status
    /// publisher sends the change at once instead of at its next sweep.
    updating_changed: tokio::sync::Notify,
    /// Unified wire-facing event channel feeding the dashboard SSE
    /// stream. Carries broker messages (mirrored from `broker.subscribe`
    /// by the forwarder task in `main.rs`) and dashboard-only mutation
    /// events (container state, rebuild queue, etc.).
    /// Snapshot endpoints capture `event_seq` before reading state so
    /// the client can dedupe its buffered live traffic against the
    /// snapshot.
    dashboard_events: broadcast::Sender<DashboardEvent>,
    event_seq: AtomicU64,

    /// Last container snapshot seen by `rescan_containers_and_emit`,
    /// keyed by `ContainerView.name`. The rescan diffs a fresh
    /// `container_view::build_all` against this map and emits one
    /// `ContainerStateChanged` per added/changed row and one
    /// `ContainerRemoved` per disappeared row. Async — guarded by a
    /// tokio mutex so the rescan can `await` `lifecycle::list` /
    /// `is_running` without blocking other coordinator paths.
    last_containers: tokio::sync::Mutex<HashMap<String, ContainerView>>,
    /// Global job-DAG queue. Every container op (rebuild,
    /// first-spawn, power changes) is submitted as a DAG
    /// of primitive nodes; a single scheduler drives them with
    /// build-slot + per-agent-lease gating so the dashboard renders one
    /// ordered view of pending + running work. See `job_queue/` for
    /// the dedup rules, resource classes, and history retention.
    pub job_queue: Arc<crate::job_queue::JobQueue>,
    /// Durable per-agent power intent (`wanted: Up | Offline`) — the
    /// spec half of desired-state reconciliation; the queue's
    /// `Reconcile` nodes converge observed state to it.
    pub power: Arc<crate::power::PowerStore>,
    /// Shutdown signal broadcast to all background tasks. Sending
    /// `true` asks every loop to exit after its current work item.
    /// Use `shutdown_rx()` to subscribe; `request_shutdown()` to fire.
    shutdown_tx: watch::Sender<bool>,
}

/// Hive-wide configuration that lifecycle and meta operations need.
/// Extracted from `Coordinator` so callers can pass a single struct
/// instead of repeating the same 6 arguments everywhere.
///
/// Cloned from `Coordinator` via [`Coordinator::hive_env`]. All fields
/// are cheap to clone (small strings + small map); lifecycle ops are
/// infrequent enough that the copy cost is irrelevant.
///
/// Also the container-injected subset of the `hive-c0re serve --config`
/// JSON (see [`ServeConfig`]): `#[serde(default)]` lets a partial config
/// fall back to [`Default`] (the canonical defaults that used to live in
/// the clap `default_value`s), and bare `hive-c0re serve` with no
/// `--config` boots from `Default`.
#[derive(Clone, Debug, serde::Deserialize)]
#[serde(default)]
pub struct HiveEnv {
    pub hyperhive_flake: String,
    /// Store path of the `claude-code` agents run, or `None` for "each
    /// agent keeps the one out of its own nixpkgs". Travels into the
    /// container as `HYPERHIVE_CLAUDE_CODE_PATH` — a plain string, kept
    /// alive host-side by the module that resolved it.
    pub claude_code_path: Option<String>,
    pub dashboard_port: u16,
    /// Per-agent systemd `CPUQuota=` value (e.g. `"200%"`).
    pub agent_cpu_quota: String,
    /// Per-agent systemd `MemoryMax=` value (e.g. `"4G"`).
    pub agent_memory_max: String,
    /// Per-agent systemd `CPUWeight=` — cgroup v2 `cpu.weight`, 1..=10000.
    /// Relative share under contention, not a cap; see `agentCpuWeight`.
    /// `None` = leave the setting out of the drop-in (kernel default).
    pub agent_cpu_weight: Option<u32>,
    /// Per-agent systemd `IOWeight=` — cgroup v2 `io.weight`, 1..=10000.
    /// Same semantics as [`Self::agent_cpu_weight`]; see `agentIoWeight`.
    pub agent_io_weight: Option<u32>,
}

impl Default for HiveEnv {
    fn default() -> Self {
        Self {
            hyperhive_flake: "/etc/hyperhive".to_string(),
            claude_code_path: None,
            dashboard_port: 7000,
            agent_cpu_quota: "200%".to_string(),
            agent_memory_max: "4G".to_string(),
            // Slightly below the kernel default of 100, so agent
            // containers yield to host services + the infra containers
            // (which carry no drop-in and stay at 100) under contention.
            agent_cpu_weight: Some(80),
            agent_io_weight: Some(80),
        }
    }
}

/// On-disk shape of the `hive-c0re serve --config <file>` JSON: the
/// container-injected [`HiveEnv`] (flattened) plus the hive-c0re-local
/// settings that are never injected into containers — hence kept out of
/// `HiveEnv`. The NixOS module writes this so the `ExecStart` carries one
/// `--config` flag
/// instead of every host-level setting as its own JSON-blob argument.
/// `#[serde(default)]` lets any field be omitted and fall back to its
/// canonical default.
#[derive(Clone, Debug, serde::Deserialize)]
#[serde(default)]
pub struct ServeConfig {
    #[serde(flatten)]
    pub env: HiveEnv,
    /// Number of concurrent closure-fetching job-queue nodes (prebuild /
    /// profile-swap / create / meta lock). hive-c0re-local — never
    /// injected into containers. Set via
    /// `services.hyperhive.deploy.hive-controller.buildSlots`.
    pub build_slots: usize,
}

impl Default for ServeConfig {
    fn default() -> Self {
        Self {
            env: HiveEnv::default(),
            build_slots: 1,
        }
    }
}

#[cfg(test)]
mod serve_config_tests {
    use super::ServeConfig;

    #[test]
    fn deserialises_flat_with_per_field_defaults() {
        // The flattened HiveEnv fields + the local build_slots all live at
        // the JSON top level. Present fields are read; omitted ones fall
        // back to their canonical defaults (so a partial config is valid).
        let sc: ServeConfig = serde_json::from_str(
            r#"{
                "hyperhive_flake": "/x",
                "dashboard_port": 1234,
                "build_slots": 3
            }"#,
        )
        .expect("flat ServeConfig should deserialise");
        assert_eq!(sc.env.hyperhive_flake, "/x");
        assert_eq!(sc.env.dashboard_port, 1234);
        // omitted → HiveEnv::default()
        assert_eq!(sc.env.agent_cpu_quota, "200%");
        assert_eq!(sc.build_slots, 3);
    }

    #[test]
    fn a_leftover_model_prices_key_is_ignored() {
        // Configs written before the hive stats page went carry the key.
        let sc: ServeConfig = serde_json::from_str(r#"{"model_prices": {"opus": {"input": 5.0}}}"#)
            .expect("an unknown key must not break startup");
        assert_eq!(sc.build_slots, 1);
    }

    #[test]
    fn empty_object_is_all_defaults() {
        let sc: ServeConfig = serde_json::from_str("{}").expect("empty config valid");
        assert_eq!(sc.env.dashboard_port, 7000);
        assert_eq!(sc.env.hyperhive_flake, "/etc/hyperhive");
        assert_eq!(sc.build_slots, 1);
    }
}

/// Per-agent filesystem paths that lifecycle operations write to.
/// Assembled from `Coordinator`'s static path helpers so callers
/// don't repeat the same `Coordinator::agent_*_dir(name)` calls.
#[derive(Clone, Debug)]
pub struct AgentPaths {
    /// Runtime socket dir (tmpfs, recreated per boot).
    pub agent: PathBuf,
    /// Claude OAuth credentials (survives purge boundary).
    pub claude: PathBuf,
    /// Agent durable notes + forge token (survives purge boundary).
    pub notes: PathBuf,
}

/// Collapse per-pill tombstones to one answer per agent: **was any recently
/// cleared pill for this agent a deliberate stop?**
///
/// `OR`, not last-write-wins — that is the whole fix. An agent can clear several
/// pills inside one grace window (the transient set tests status alone, so a
/// lease-exempt `Prebuild` and a lease-holding `StopForUpdate` are both live),
/// and taking the last one means an incidental `false` erases a real `true`,
/// which the crash watcher then reads as a container **crash**.
///
/// Asks the same question of the cleared set that `crash_watch` asks of the
/// active set with `.any(…)`, so the two agree by construction. A free function
/// so it is testable without a `Coordinator` fixture — same reason
/// `crash_watch::is_deliberate_stop` is one.
fn fold_tombstones_by_agent<'a>(
    entries: impl Iterator<Item = (&'a str, bool)>,
) -> HashMap<String, bool> {
    let mut out: HashMap<String, bool> = HashMap::new();
    for (agent, deliberate) in entries {
        *out.entry(agent.to_owned()).or_default() |= deliberate;
    }
    out
}

/// The two ways [`Coordinator::set_paused_by_name`] can fail. See its own
/// doc comment for why this is a distinct type rather than one
/// `anyhow::Error` (`Write` already wraps one, `BadName` never needs to).
#[derive(Debug)]
pub enum SetPausedByNameError {
    /// `name` did not parse as a [`hive_types::Ident`] — the parse error's
    /// own message.
    BadName(&'static str),
    Write(anyhow::Error),
}

impl std::fmt::Display for SetPausedByNameError {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        match self {
            SetPausedByNameError::BadName(msg) => write!(f, "bad agent name: {msg}"),
            SetPausedByNameError::Write(e) => write!(f, "{e}"),
        }
    }
}

impl Coordinator {
    pub fn open(db_path: &Path, env: HiveEnv, build_slots: usize) -> Result<Self> {
        let HiveEnv {
            hyperhive_flake,
            claude_code_path,
            dashboard_port,
            agent_cpu_quota,
            agent_memory_max,
            agent_cpu_weight,
            agent_io_weight,
        } = env;
        let broker = Broker::open(db_path).context("open broker")?;
        // BuildLogs wants a directory (it picks its own `build_logs.sqlite`
        // file under it); every other opener here takes the sibling
        // sqlite-file path itself. Derive the dir from `db_path`'s
        // parent so the two shapes line up.
        let build_logs_dir = db_path.parent().unwrap_or_else(|| Path::new("."));
        let build_logs = Arc::new(
            crate::build_logs::BuildLogs::open(build_logs_dir).context("open build_logs")?,
        );
        // This process is the database's only writer and has started no
        // build yet, so a row still in-flight belongs to a dead process.
        let reaped = build_logs.reap_orphans()?;
        if reaped > 0 {
            tracing::warn!(
                reaped,
                "build_logs: closed rows left in-flight by a previous run"
            );
        }
        // Install the process-wide handle so `lifecycle::run` can write
        // without us having to thread an `Arc<BuildLogs>` through every public entry
        // point in the lifecycle surface.
        crate::build_logs::install(build_logs.clone());
        let power = Arc::new(crate::power::PowerStore::default());
        let (dashboard_events, _) = broadcast::channel(DASHBOARD_CHANNEL);
        let (shutdown_tx, _) = watch::channel(false);
        Ok(Self {
            broker: Arc::new(broker),
            build_logs,
            hyperhive_flake,
            claude_code_path,
            dashboard_port,
            agent_cpu_quota,
            agent_memory_max,
            agent_cpu_weight,
            agent_io_weight,
            agents: Mutex::new(HashMap::new()),
            recent_transient: Mutex::new(HashMap::new()),
            graceful_stop_pending: Mutex::new(HashSet::new()),
            pause_pending: Mutex::new(HashSet::new()),
            updating: Mutex::new(HashSet::new()),
            updating_changed: tokio::sync::Notify::new(),
            dashboard_events,
            event_seq: AtomicU64::new(0),

            last_containers: tokio::sync::Mutex::new(HashMap::new()),
            job_queue: Arc::new(crate::job_queue::JobQueue::new(build_slots)),
            power,
            shutdown_tx,
        })
    }

    /// Snapshot the hive-wide configuration fields as a [`HiveEnv`].
    /// Pass the result to `lifecycle::swap_update` / `write_dropins`
    /// instead of threading the individual fields separately.
    #[must_use]
    pub fn hive_env(&self) -> HiveEnv {
        HiveEnv {
            hyperhive_flake: self.hyperhive_flake.clone(),
            claude_code_path: self.claude_code_path.clone(),
            dashboard_port: self.dashboard_port,
            agent_cpu_quota: self.agent_cpu_quota.clone(),
            agent_memory_max: self.agent_memory_max.clone(),
            agent_cpu_weight: self.agent_cpu_weight,
            agent_io_weight: self.agent_io_weight,
        }
    }

    /// Assemble the per-agent filesystem paths for `name`. `agent_dir`
    /// is the runtime directory (`/run/hyperhive/agents/<name>`), obtained
    /// from `crate::paths::agent_runtime_dir(name)` (pure path) or from
    /// `lifecycle::ensure_agent_runtime_dir(name)` when the dir must be
    /// created. All other paths are derived statically from `name`.
    ///
    /// # Errors
    ///
    /// `name` is not a valid [`hive_types::Ident`]. Job payloads carry agent
    /// names as plain strings, one source being the swarm's published wanted
    /// state, so this is an error for the caller to report, not an invariant.
    pub fn agent_paths(name: &str, agent_dir: PathBuf) -> Result<AgentPaths> {
        let name = hive_types::Ident::parse(name)
            .map_err(|e| anyhow::anyhow!("invalid agent name {name:?}: {e}"))?;
        Ok(AgentPaths {
            agent: agent_dir,
            claude: Self::agent_claude_dir(&name),
            notes: Self::agent_notes_dir(&name),
        })
    }

    /// Emit a `RebuildQueueChanged` tick. Called from the queue mutation
    /// helpers (`enqueue` / `finish` / `cancel`-adjacent wrappers below) and
    /// the worker so every state transition surfaces on the dashboard without
    /// extra plumbing.
    ///
    /// Carries no queue payload — clients refetch `/api/jobq/graph`. The name
    /// keeps `snapshot` because *that* is still what a client ends up with;
    /// what changed is who serves it.
    pub fn emit_rebuild_queue_snapshot(self: &Arc<Self>) {
        self.emit_dashboard_event(DashboardEvent::RebuildQueueChanged {
            seq: self.next_seq(),
        });
    }

    /// Subscribe to the shutdown watch channel. Background tasks call
    /// this at spawn time and break their loop when the receiver
    /// transitions to `true` (via `Coordinator::request_shutdown`).
    /// A closed channel (i.e. the Coordinator was dropped) also
    /// signals tasks to exit.
    pub fn shutdown_rx(&self) -> watch::Receiver<bool> {
        self.shutdown_tx.subscribe()
    }

    /// Signal all background tasks to exit cleanly. The tasks break
    /// out of their poll loop after completing their current work item.
    /// Best-effort — does nothing if all receivers have already been
    /// dropped (e.g. process is already mid-shutdown).
    pub fn request_shutdown(&self) {
        let _ = self.shutdown_tx.send(true);
    }

    /// Subscribe to the unified dashboard event channel. Used by the
    /// `/dashboard/stream` SSE handler and by the broker-to-dashboard
    /// forwarder task.
    pub fn dashboard_subscribe(&self) -> broadcast::Receiver<DashboardEvent> {
        self.dashboard_events.subscribe()
    }

    /// Stamp the next sequence number. Each emission of a
    /// `DashboardEvent` should fill its `seq` with `next_seq()` so the
    /// frame the wire carries is the one the client uses to dedupe.
    pub fn next_seq(&self) -> u64 {
        self.event_seq.fetch_add(1, Ordering::SeqCst) + 1
    }

    /// Current high-water seq. Snapshot endpoints read this *before*
    /// gathering state so the (snapshot.seq, snapshot) pair satisfies:
    /// any frame with `seq > snapshot.seq` is post-snapshot. The seq
    /// captured here may grow during snapshot construction — clients
    /// may double-apply such events, which renderers must tolerate.
    pub fn current_seq(&self) -> u64 {
        self.event_seq.load(Ordering::SeqCst)
    }

    /// Broadcast a freshly-built `DashboardEvent` (caller fills `seq`
    /// via `next_seq()`). Returns silently when there are no
    /// subscribers — the dashboard channel is best-effort presentation
    /// plumbing, not a delivery guarantee.
    pub fn emit_dashboard_event(&self, event: DashboardEvent) {
        let _ = self.dashboard_events.send(event);
    }

    /// Rebuild the per-container snapshot, diff it against the last
    /// one cached on `self`, and emit one
    /// `DashboardEvent::ContainerStateChanged` per added/changed row
    /// and one `DashboardEvent::ContainerRemoved` per disappeared row.
    /// Call after any mutation that could affect what
    /// `nixos-container list` returns or what a row's
    /// `running` / `needs_login`
    /// resolves to — lifecycle ops, destroy,
    /// rebuild, and the crash-watcher's periodic poll.
    /// Cheap when nothing changed (one `nixos-container list` + a
    /// `HashMap` diff + zero emits).
    pub async fn rescan_containers_and_emit(self: &Arc<Self>) {
        // An unreadable list says nothing about which containers exist, so
        // the last snapshot stands rather than every row being reported gone.
        let fresh = match container_view::build_all().await {
            Ok(fresh) => fresh,
            Err(e) => {
                tracing::warn!(error = ?e, "container rescan: list failed; keeping last snapshot");
                return;
            }
        };
        crate::host_stats::refresh_pending_logins(&fresh);
        crate::host_stats::record_login_states(&fresh);
        let mut last = self.last_containers.lock().await;
        let mut changed_or_new = Vec::new();
        let mut removed = Vec::new();
        // Diff into change vs. add.
        for view in &fresh {
            match last.get(&view.name) {
                Some(prev) if prev == view => {} // unchanged
                _ => changed_or_new.push(view.clone()),
            }
        }
        // Anything in `last` but not in `fresh` is gone.
        let fresh_names: std::collections::HashSet<&str> =
            fresh.iter().map(|c| c.name.as_str()).collect();
        for name in last.keys() {
            if !fresh_names.contains(name.as_str()) {
                removed.push(name.clone());
            }
        }
        // Rebuild the cache from the fresh snapshot.
        last.clear();
        for c in fresh {
            last.insert(c.name.clone(), c);
        }
        drop(last);
        for c in changed_or_new {
            self.emit_dashboard_event(DashboardEvent::ContainerStateChanged {
                seq: self.next_seq(),
                container: c,
            });
        }
        for name in removed {
            self.emit_dashboard_event(DashboardEvent::ContainerRemoved {
                seq: self.next_seq(),
                name,
            });
        }
    }

    /// Read-only snapshot of the last cached container view. Used by
    /// `/api/state` to cold-load page-open clients without re-running
    /// `nixos-container list` themselves; the
    /// `rescan_containers_and_emit` calls keep this fresh.
    pub async fn containers_snapshot(&self) -> Vec<ContainerView> {
        let last = self.last_containers.lock().await;
        let mut out: Vec<ContainerView> = last.values().cloned().collect();
        out.sort_by(|a, b| a.name.cmp(&b.name));
        out
    }

    pub fn register_agent(self: &Arc<Self>, name: &str) -> Result<PathBuf> {
        // Idempotent: drop any existing listener so re-registration (e.g. on rebuild,
        // or after a hive-c0re restart cleared /run/hyperhive) gets a fresh socket.
        self.unregister_agent(name);
        let agent_dir = crate::paths::agent_runtime_dir(name);
        std::fs::create_dir_all(&agent_dir)
            .with_context(|| format!("create agent dir {}", agent_dir.display()))?;
        let socket_path = Self::socket_path(name);
        // Hand the full Coordinator to the per-agent socket — it needs
        // more than just the broker.
        let socket = socket_server::start(name, &socket_path, self.clone())?;
        self.agents.lock().unwrap().insert(name.to_owned(), socket);
        Ok(agent_dir)
    }

    pub fn unregister_agent(&self, name: &str) {
        if let Some(socket) = self.agents.lock().unwrap().remove(name) {
            socket.handle.abort();
            let _ = std::fs::remove_file(&socket.path);
        }
    }
    pub fn list_agents(&self) -> Vec<String> {
        self.agents.lock().unwrap().keys().cloned().collect()
    }

    /// Emit the "a pill appeared" edge, for the job-queue scheduler publishing
    /// the transitions of its derived set.
    pub(crate) fn emit_transient_set(&self, name: &str, label: String) {
        // Live-update dashboards. `since_unix` is wall-clock so the
        // browser can tick "Ns spawning…" without polling. The
        // intra-process map keeps using `Instant` for monotonicity.
        let since_unix = std::time::SystemTime::now()
            .duration_since(std::time::UNIX_EPOCH)
            .ok()
            .and_then(|d| i64::try_from(d.as_secs()).ok())
            .unwrap_or(0);
        self.emit_dashboard_event(DashboardEvent::TransientSet {
            seq: self.next_seq(),
            name: name.to_owned(),
            transient_kind: label,
            since_unix,
        });
    }

    /// Emit the "a pill went away" edge **and stamp the tombstone the crash
    /// watcher reads**, for the job-queue scheduler.
    ///
    /// 🚨 The stamp is not bookkeeping. Without it the clear-then-poll race
    /// produced a spurious crash warning on **every** operator stop/restart:
    /// the transient is gone by the time the 10s poll looks, so a deliberate
    /// stop is indistinguishable from a crash. `recent_transient_within` is what
    /// closes that window, which is why the clear has to be an *event* — a
    /// derived read of current state cannot answer "was one here a moment ago?".
    ///
    /// Old entries are reaped lazily on read, so the map stays bounded.
    /// `label` identifies *which* pill cleared. An agent can hold several at
    /// once, so both the tombstone key and the wire event need it — without it
    /// a client that received two `TransientSet`s cannot tell which one this
    /// clears, and the tombstone silently overwrites a concurrent pill's
    /// `deliberate_stop`.
    pub(crate) fn emit_transient_cleared(&self, name: &str, label: &str, deliberate_stop: bool) {
        self.recent_transient.lock().unwrap().insert(
            (name.to_owned(), label.to_owned()),
            (deliberate_stop, std::time::Instant::now()),
        );
        self.emit_dashboard_event(DashboardEvent::TransientCleared {
            seq: self.next_seq(),
            name: name.to_owned(),
            transient_kind: label.to_owned(),
        });
    }

    /// Mark `name` as being updated (`true`) or settled (`false`), and wake the
    /// status publisher when that changed something.
    pub fn set_updating(&self, name: &str, updating: bool) {
        let changed = {
            let mut set = self.updating.lock().unwrap();
            if updating {
                set.insert(name.to_owned())
            } else {
                set.remove(name)
            }
        };
        if changed {
            self.updating_changed.notify_one();
        }
    }

    /// Whether an update has `name` down and not yet back up.
    #[must_use]
    pub fn is_updating(&self, name: &str) -> bool {
        self.updating.lock().unwrap().contains(name)
    }

    /// Resolves when the set of updating agents changed since it was last
    /// awaited, for the status publisher.
    pub async fn updating_changed(&self) {
        self.updating_changed.notified().await;
    }

    /// Mark `name` as having a graceful stop in progress. While set,
    /// `socket_server::handle_recv` returns `Response::GracefulStop` for
    /// this agent instead of polling the broker (the inbound fence).
    pub fn mark_graceful_stop(&self, name: &str) {
        self.graceful_stop_pending
            .lock()
            .unwrap()
            .insert(name.to_owned());
    }

    /// Whether a graceful stop is pending for `name`.
    #[must_use]
    pub fn is_graceful_stop_pending(&self, name: &str) -> bool {
        self.graceful_stop_pending.lock().unwrap().contains(name)
    }

    /// Clear the graceful-stop flag for `name` (agent reported
    /// `GracefulStopComplete`, or the stop finished / was abandoned).
    pub fn clear_graceful_stop(&self, name: &str) {
        self.graceful_stop_pending.lock().unwrap().remove(name);
    }

    /// Mark `name` as having a pause acknowledgement in progress. Set by
    /// the pause DAG's signal node; the drain node polls
    /// `is_pause_pending` until the agent reports back or the timeout
    /// fallback fires.
    pub fn mark_pause_pending(&self, name: &str) {
        self.pause_pending.lock().unwrap().insert(name.to_owned());
    }

    /// Whether a pause acknowledgement is pending for `name`.
    #[must_use]
    pub fn is_pause_pending(&self, name: &str) -> bool {
        self.pause_pending.lock().unwrap().contains(name)
    }

    /// Clear the pause-pending flag for `name` (agent reported
    /// `PauseAcknowledged`, or the drain node's wait finished / timed out).
    pub fn clear_pause_pending(&self, name: &str) {
        self.pause_pending.lock().unwrap().remove(name);
    }

    /// Per-agent `deliberate_stop` for transients cleared within the last
    /// `grace` seconds — i.e. agents the operator just acted on, whose stop the
    /// crash watcher should NOT classify as a crash. Lazily reaps entries older
    /// than `grace` so the map stays bounded by the active agent count.
    ///
    /// Carries only the safety bit, not the display label: nothing downstream
    /// should be able to re-derive a stop/crash decision from a pill's wording.
    pub fn recent_transient_within(&self, grace: std::time::Duration) -> HashMap<String, bool> {
        let now = std::time::Instant::now();
        let mut map = self.recent_transient.lock().unwrap();
        map.retain(|_, (_, ts)| now.duration_since(*ts) <= grace);
        fold_tombstones_by_agent(map.iter().map(|((a, _), (d, _))| (a.as_str(), *d)))
    }

    /// Every live transient, keyed by agent.
    ///
    /// **Derived on read, stored nowhere.** Straight off the running graph, so
    /// there is no cached copy to go stale, leak, or disagree with what is
    /// actually running.
    ///
    /// Migration is the last operation with no queue node behind it, so it
    /// shows **no pill** — there is nothing in the graph to derive one from.
    /// It gets one for free once it becomes real nodes, the way destroy did.
    ///
    /// ⚠️ **A `Vec` per agent, not one entry.** `running_transients` tests
    /// status alone, so a lease-exempt `Prebuild` for `a` and a lease-holding
    /// `StopForUpdate` for `a` are both live pills. Collapsing them to one
    /// would pick arbitrarily and, for the crash watcher, silently lose a
    /// `deliberate_stop = true` behind a `false` — reporting an intentional
    /// stop as a crash.
    #[must_use]
    pub fn transient_snapshot(&self) -> HashMap<String, Vec<RunningTransient>> {
        let mut out: HashMap<String, Vec<RunningTransient>> = HashMap::new();
        for t in self.job_queue.running_transients() {
            out.entry(t.agent.clone()).or_default().push(t);
        }
        out
    }

    /// Drop a system message into the given agent's inbox. Wakes the
    /// turn loop with a "you were just (re)started" hint — operator
    /// caused the transition, agent picks up where it left off
    /// (notes are in the bind-mounted state dir, last turn is in
    /// --continue's session). Best-effort; broker errors are logged
    /// but don't propagate.
    pub fn kick_agent(&self, name: &str, reason: &str) {
        // Every agent binds its state at /agents/<name>/state, and a
        // `manage_root_agent` holder has all of /agents bind-mounted, so
        // the path resolves there too — use that uniformly so the wake
        // message has one canonical path that works everywhere.
        let body = format!(
            "{reason}\n\nYou were just (re)started by the operator. \
             If you were mid-task, check `/agents/{name}/state/` for \
             your notes and pick up where you left off. claude's \
             `--continue` session is intact, so prior context is \
             still in your window."
        );
        if let Err(e) = self.broker.send(&hive_sh4re::inbox::Message {
            from: hive_sh4re::manager::trusted_sender(hive_sh4re::manager::SYSTEM_SENDER),
            to: name.to_owned(),
            body,
            in_reply_to: None,
        }) {
            tracing::warn!(error = ?e, %name, "kick_agent: broker.send failed");
        }
    }

    /// Deliver `body` to every currently-registered agent except the sender,
    /// appending the standard broadcast hint. Returns a list of per-agent
    /// error strings for any that failed (empty = all ok).
    pub fn broadcast_send(&self, from: &str, body: &str) -> Vec<String> {
        const HINT: &str =
            "\n\n⚠️ _hint: this was a broadcast and may not need any action from you_";
        let broadcast_body = format!("{body}{HINT}");
        let mut errors = Vec::new();
        for agent_name in self.list_agents() {
            if agent_name == from {
                continue;
            }
            if let Err(e) = self.broker.send(&hive_sh4re::inbox::Message {
                from: hive_sh4re::manager::trusted_sender(from),
                to: agent_name.clone(),
                body: broadcast_body.clone(),
                in_reply_to: None,
            }) {
                errors.push(format!("{agent_name}: {e}"));
            }
        }
        errors
    }

    pub fn socket_path(name: &str) -> PathBuf {
        crate::paths::agent_runtime_dir(name).join("mcp.sock")
    }

    /// Per-agent Claude credentials dir. Bind-mounted RW into the agent
    /// container at `/root/.claude` so OAuth state survives container
    /// destroy/recreate. Each agent owns its own token lineage — sharing
    /// would break on the first refresh-token rotation.
    pub fn agent_claude_dir(name: &hive_types::Ident) -> PathBuf {
        crate::paths::agent_state_dir(name).join("claude")
    }

    /// Per-agent durable knowledge dir. Bind-mounted RW into the agent
    /// container at `/agents/{name}/state`. Survives destroy/recreate.
    /// Agent-visible — claude is told to write long-lived notes here.
    pub fn agent_notes_dir(name: &hive_types::Ident) -> PathBuf {
        crate::paths::agent_state_dir(name).join("state")
    }

    /// Per-agent harness-internal state dir. Bind-mounted RW into the
    /// agent container at `/agents/{name}/harness`. Holds sqlite dbs
    /// and config files owned by the harness (`hyperhive-events.sqlite`,
    /// `hyperhive-turn-stats.sqlite`, `hyperhive-model`) — kept separate
    /// from the agent-visible `state/` so claude's "my notes" view is
    /// uncluttered and the host vacuum has a clean sweep root.
    pub fn agent_harness_dir(name: &hive_types::Ident) -> PathBuf {
        crate::paths::agent_state_dir(name).join("harness")
    }

    /// Host-side path of the pause marker — the same file the harness
    /// resolves in-container via `hive_agent::paths::paused_marker`,
    /// reached through the harness bind-mount. Its presence means the
    /// agent's turn loop is parked: the harness still serves its web UI
    /// and MCP daemons, but drives no turns, so inbox messages queue up
    /// unacked until the marker is removed.
    ///
    /// Both sides only ever *stat* or create/remove this file, so there
    /// is no protocol between them and pause survives a container
    /// restart (and can be set on a stopped container).
    pub fn agent_paused_marker(name: &hive_types::Ident) -> PathBuf {
        Self::agent_harness_dir(name).join(hive_sh4re::paths::PAUSED_MARKER_FILE)
    }

    /// Whether `name` is currently paused. A stat error (missing agent
    /// dir, permissions) reads as "not paused" — the pause indicator is
    /// advisory on the host side, and the harness is the component that
    /// actually enforces it.
    #[must_use]
    pub fn is_paused(name: &hive_types::Ident) -> bool {
        Self::agent_paused_marker(name).exists()
    }

    /// Create or remove the pause marker, **via hive-priv**. Idempotent in
    /// both directions: pausing an already-paused agent (or resuming a
    /// running one) is a no-op rather than an error, so the dashboard
    /// toggle and `hivectl pause|resume` don't have to read-then-write.
    ///
    /// The write cannot happen in-process. The harness dir is chowned to
    /// the agent user on the container's first boot (mode 0755), and this
    /// daemon runs as the unprivileged `hive-core` user — so the read side
    /// ([`Self::is_paused`], a stat) works while a direct `fs::write` here
    /// fails with `EACCES` on every agent that has ever booted. The root
    /// helper owns both directions instead.
    ///
    /// # Errors
    ///
    /// Returns the hive-priv error when the marker can't be created (agent
    /// dir missing and un-creatable, disk full) or can't be removed, and
    /// the transport error when the helper is unreachable. A `NotFound` on
    /// removal is *not* an error — that's the idempotent resume case.
    pub async fn set_paused(name: &hive_types::Ident, paused: bool) -> anyhow::Result<()> {
        crate::priv_client::set_agent_paused(name.as_str(), paused).await
    }

    /// [`Self::set_paused`] from a plain agent-name string, parsing it
    /// first — the exact shape both `dashboard::lifecycle_ops`'s
    /// `/api/pause`/`/api/resume` handlers and `workers::wanted`'s
    /// reconcile loop need, previously duplicated between them (mara, PR
    /// review: "could this be a bit less duplicated code?"). A distinct
    /// error variant per failure mode, not a single
    /// `anyhow::Error`, because the two callers report the two cases
    /// differently: a bad name is the caller's mistake (400 there, a
    /// warned skip in the reconcile loop), a write failure is this
    /// host's (500 there, also a warned skip here).
    ///
    /// # Errors
    ///
    /// `BadName` when `name` is not a legal [`hive_types::Ident`];
    /// `Write` for everything [`Self::set_paused`] itself can fail with.
    pub async fn set_paused_by_name(name: &str, paused: bool) -> Result<(), SetPausedByNameError> {
        let id = hive_types::Ident::parse(name).map_err(SetPausedByNameError::BadName)?;
        Self::set_paused(&id, paused)
            .await
            .map_err(SetPausedByNameError::Write)
    }

    /// Enumerate names that have a persistent state dir under
    /// `/var/lib/hyperhive/agents/` (i.e. config / claude creds /
    /// notes survive). Includes both currently-existing containers and
    /// destroyed-but-kept tombstones; callers filter the latter by
    /// subtracting `lifecycle::list()`.
    #[must_use]
    pub fn kept_state_names() -> Vec<hive_types::Ident> {
        let Ok(rd) = std::fs::read_dir(crate::paths::agents_root()) else {
            return Vec::new();
        };
        let mut out: Vec<hive_types::Ident> = rd
            .flatten()
            .filter(|e| e.file_type().is_ok_and(|t| t.is_dir()))
            .filter_map(|e| hive_types::Ident::parse(&e.file_name().into_string().ok()?).ok())
            .collect();
        out.sort();
        out
    }
}

/// `push_todo` summary text for a rebuild outcome, for
/// `job_queue::exec::run_emit_rebuilt`. Pure + independently testable; the
/// `sha`/`tag` fields are part of the summary text, since a todo carries one
/// string, not a structured payload.
#[must_use]
pub fn rebuilt_todo_summary(
    agent: &str,
    ok: bool,
    note: Option<&str>,
    sha: Option<&str>,
    tag: Option<&str>,
) -> String {
    use std::fmt::Write as _;

    if !ok {
        return format!(
            "agent '{agent}' rebuild FAILED: {}",
            note.unwrap_or("unknown error")
        );
    }

    let mut summary = format!("agent '{agent}' rebuilt");
    if let Some(sha) = sha {
        let _ = write!(summary, " @ {sha}");
    }
    if let Some(tag) = tag {
        let _ = write!(summary, " ({tag})");
    }
    summary
}

#[cfg(test)]
mod tombstone_fold_tests {
    use super::fold_tombstones_by_agent;

    #[test]
    fn a_deliberate_stop_survives_an_incidental_clear_on_the_same_agent() {
        // The regression this fold exists for. `a` clears two pills in one
        // grace window: a `StopForUpdate` that really does take the container
        // down, and a lease-exempt `Prebuild` that doesn't. Keyed by agent
        // alone the second overwrote the first and the crash watcher reported
        // a deliberate stop as a crash — order must not matter.
        let out = fold_tombstones_by_agent([("a", true), ("a", false)].into_iter());
        assert_eq!(out.get("a"), Some(&true));

        let reversed = fold_tombstones_by_agent([("a", false), ("a", true)].into_iter());
        assert_eq!(reversed.get("a"), Some(&true));
    }

    #[test]
    fn all_incidental_clears_stay_false() {
        // The other direction has to keep working: nothing deliberate cleared,
        // so a container going down now really is a crash.
        let out = fold_tombstones_by_agent([("a", false), ("a", false)].into_iter());
        assert_eq!(out.get("a"), Some(&false));
    }

    #[test]
    fn agents_do_not_bleed_into_each_other() {
        let out = fold_tombstones_by_agent([("a", true), ("b", false)].into_iter());
        assert_eq!(out.get("a"), Some(&true));
        assert_eq!(out.get("b"), Some(&false));
        assert_eq!(out.get("c"), None);
    }
}

#[cfg(test)]
mod rebuilt_todo_summary_tests {
    use super::rebuilt_todo_summary;

    #[test]
    fn failure_reports_the_note() {
        assert_eq!(
            rebuilt_todo_summary("iris", false, Some("build failed"), None, None),
            "agent 'iris' rebuild FAILED: build failed"
        );
    }

    #[test]
    fn failure_without_a_note_says_unknown() {
        assert_eq!(
            rebuilt_todo_summary("iris", false, None, None, None),
            "agent 'iris' rebuild FAILED: unknown error"
        );
    }

    #[test]
    fn success_with_no_sha_or_tag_is_bare() {
        assert_eq!(
            rebuilt_todo_summary("iris", true, None, None, None),
            "agent 'iris' rebuilt"
        );
    }

    #[test]
    fn success_with_sha_and_tag_carries_both() {
        assert_eq!(
            rebuilt_todo_summary("iris", true, None, Some("abc123"), Some("deployed/42")),
            "agent 'iris' rebuilt @ abc123 (deployed/42)"
        );
    }

    #[test]
    fn success_with_sha_only() {
        assert_eq!(
            rebuilt_todo_summary("iris", true, None, Some("abc123"), None),
            "agent 'iris' rebuilt @ abc123"
        );
    }
}

#[cfg(test)]
mod agent_paths_tests {
    use super::Coordinator;

    /// A name from a job payload that is not an `Ident` fails the caller's
    /// node instead of panicking the executor task.
    #[test]
    fn a_malformed_name_is_an_error_not_a_panic() {
        let err = Coordinator::agent_paths("../etc", std::path::PathBuf::from("/nonexistent"))
            .expect_err("a malformed agent name must be rejected");
        assert!(format!("{err:#}").contains("invalid agent name"), "{err:#}");
    }

    #[test]
    fn a_valid_name_builds_its_paths() {
        let paths = Coordinator::agent_paths("iris", std::path::PathBuf::from("/run/x"))
            .expect("a valid agent name");
        assert_eq!(paths.agent, std::path::PathBuf::from("/run/x"));
    }
}

#[cfg(test)]
mod open_tests {
    use super::{Coordinator, HiveEnv};

    fn open(db: &std::path::Path) -> Coordinator {
        Coordinator::open(db, HiveEnv::default(), 1).expect("open coordinator")
    }

    /// A build log row the previous process left open is closed as failed
    /// when the next one opens the coordinator.
    #[tokio::test]
    async fn open_closes_build_logs_left_in_flight() {
        let dir = tempfile::tempdir().expect("tempdir");
        let db = dir.path().join("broker.sqlite");
        let id = open(&db)
            .build_logs
            .start("alice", "run", "cmd", None)
            .expect("start");

        let coord = open(&db);
        let row = coord.build_logs.get_full(id).expect("get").expect("Some");
        assert_eq!(row.header.status.as_deref(), Some("fail"));
        assert!(row.header.finished_at.is_some());
    }
}