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::approvals::Approvals;
use crate::broker::Broker;
use crate::container_view::{self, ContainerView};
use crate::dashboard_events::DashboardEvent;
use crate::operator_questions::OperatorQuestions;
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;

/// Broker `kv` key under which the broad-stop running-agents snapshot is
/// persisted, so a `hivectl start` after a hive-c0re restart can still
/// restore only the previously-running agents. See
/// `Coordinator::set_last_stopped_running`.
const LAST_STOPPED_RUNNING_KEY: &str = "last_stopped_running";

const AGENT_RUNTIME_ROOT: &str = "/run/hyperhive/agents";
/// Manager-editable per-agent config repos. Bind-mounted RW into the manager
/// container as `/agents/<name>/`. Hive-c0re only writes to these on first
/// spawn (initial commit); after that it's manager-only.
const AGENT_STATE_ROOT: &str = "/var/lib/hyperhive/agents";
/// Hive-c0re-only authoritative per-agent config repos. Containers build from
/// these. Manager has no filesystem access; the only way to update is via
/// `request_apply_commit` + user approval.
const APPLIED_STATE_ROOT: &str = "/var/lib/hyperhive/applied";

pub struct Coordinator {
    pub broker: Arc<Broker>,
    pub approvals: Arc<Approvals>,
    pub questions: Arc<OperatorQuestions>,
    /// Scheduled-prompts queue. One sqlite connection,
    /// internal mutex; the worker drains due rows and the manager
    /// handlers insert / cancel through the same handle.
    pub scheduled_prompts: Arc<crate::scheduled_prompts::ScheduledPrompts>,
    /// Full build-log capture. `lifecycle::run` /
    /// `lifecycle::prebuild_toplevel` `start()` a row per attempt,
    /// pipe every stdout/stderr line into it, and `finish()` 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>,
    /// Audit trail of agent-initiated privileged actions (infra restart,
    /// …). See `audit_log.rs`. Same dir as `build_logs`.
    pub audit_log: Arc<crate::audit_log::AuditLog>,
    /// URL of the hyperhive flake (no fragment). Inlined into per-agent
    /// `flake.nix` files as `inputs.hyperhive.url`.
    pub hyperhive_flake: String,
    /// URL of the narrow `docs/` source (no fragment). Inlined into the
    /// meta `flake.nix` as `inputs.hyperhive-docs.url` and threaded to
    /// each agent as `hyperhive.docs.source`. Its own store path
    /// so doc edits don't re-hash `hyperhive_flake`.
    pub hyperhive_docs_flake: String,
    /// Store-path URL of the nixpkgs to wire into the meta flake as
    /// `inputs.nixpkgs.url`. Populated by `--nixpkgs-flake` (set by the
    /// NixOS module to `"path:${pkgs.path}"` so the meta flake always
    /// tracks the same nixpkgs the host evaluated with — which is the
    /// host's nixpkgs when `inputs.hyperhive.inputs.nixpkgs.follows =
    /// "nixpkgs"` is set in the host flake). Empty string = legacy
    /// `follows = "hyperhive/nixpkgs"` behaviour.
    pub nixpkgs_flake: String,
    /// TCP port the host's hive-c0re dashboard listens on. Inlined into
    /// each per-agent flake so the agent's web UI can build the right
    /// rebuild-button URL pointing back at the dashboard.
    pub dashboard_port: u16,
    /// Operator pronouns (free text) — `she/her` by default, set via
    /// the NixOS module option `services.hive-c0re.operatorPronouns`.
    /// Reaches each container as the `HIVE_OPERATOR_PRONOUNS` env var
    /// (injected into systemd.services.<harness>.environment by the
    /// meta flake); the harness substitutes it into the agent /
    /// manager system prompt at boot.
    pub operator_pronouns: String,
    /// Per-model context-window sizes in tokens. Set via the host-level
    /// `services.hive-c0re.contextWindowTokens` NixOS option; injected
    /// into each container as `HIVE_CONTEXT_WINDOW_TOKENS_<KEY_UPPER>`
    /// by the meta flake renderer. The harness uses these to derive
    /// compaction / auto-reset watermarks and exposes the active value
    /// on `/api/state` as `context_window_tokens`.
    pub context_window_tokens: std::collections::HashMap<String, u64>,
    /// 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,
    /// Operator-tunable model→price table backing the hive-wide cost
    /// estimate on the ST4TS tab. Set via `services.hyperhive.modelPrices`
    /// and passed to `hive-c0re serve --model-prices <json>`. Models not
    /// in the table fall back to the built-in estimate in `hive_stats`.
    /// hive-c0re-local (read only by the `/api/stats-hive` handler), so —
    /// unlike `context_window_tokens` — it is *not* part of `HiveEnv` and
    /// is never injected into containers.
    pub model_prices: crate::hive_stats::PriceTable,
    agents: Mutex<HashMap<String, AgentSocket>>,
    /// Agents whose lifecycle action (currently just spawn) is in flight.
    /// Read by the dashboard to render a spinner; cleared when the action
    /// resolves (success or failure).
    transient: Mutex<HashMap<String, TransientState>>,
    /// 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.
    recent_transient: Mutex<HashMap<String, (TransientKind, std::time::Instant)>>,
    /// Timestamps of recent unexpected container crashes, keyed by agent.
    /// Fed by `crash_watch` each time it classifies a stop as a crash (so
    /// a crash-looping container — which `Restart=on-failure` flips back
    /// to running between polls — accumulates one entry per down-transition,
    /// not just whatever its point-in-time state happens to be). Read by
    /// the dashboard's `agents_crashing` banner warning via
    /// `recent_crash_counts`, which prunes entries older than its window.
    recent_crashes: Mutex<HashMap<String, Vec<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>>,
    /// Logical agent names that were running at the last broad-scope
    /// `hivectl stop`. A subsequent broad-scope `hivectl start` restores
    /// only this set (intersected with the requested scope) rather than
    /// every configured container, so agents the operator intentionally
    /// left stopped stay stopped. `None` when no broad stop has happened
    /// since the last start (or since daemon boot) — start then falls
    /// back to "start all". Mirrored to the broker `kv` table
    /// (`last_stopped_running` key) so the snapshot also survives a
    /// hive-c0re restart between the stop-all and the start: this
    /// in-memory copy is the fast path, the persisted copy the backstop.
    last_stopped_running: Mutex<Option<Vec<String>>>,
    /// 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 (approval added/resolved, question added/answered, 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,
    /// Count of dashboard-triggered `meta-update` runs currently in
    /// flight. `post_meta_update` returns 200 immediately and does the
    /// multi-minute `nix flake update` + agent-rebuild ripple in a
    /// background task, so without this the META INPUTS panel showed
    /// no sign anything was happening. Held via
    /// `MetaUpdateGuard`; a count > 0 surfaces on `/api/state` as
    /// `meta_update_running` and via the `MetaUpdateRunning` event.
    meta_updates_active: 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/meta op (rebuild,
    /// meta-update, 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 URL of the narrow `docs/` source, wired into the meta
    /// flake as `inputs.hyperhive-docs.url` and threaded to each agent as
    /// `hyperhive.docs.source`. Separate from `hyperhive_flake`
    /// so a doc edit only re-locks this input, not the whole source.
    pub hyperhive_docs_flake: String,
    pub nixpkgs_flake: String,
    pub dashboard_port: u16,
    pub operator_pronouns: String,
    pub context_window_tokens: std::collections::HashMap<String, u64>,
    /// 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,
}

impl Default for HiveEnv {
    fn default() -> Self {
        Self {
            hyperhive_flake: "/etc/hyperhive".to_string(),
            hyperhive_docs_flake: String::new(),
            nixpkgs_flake: String::new(),
            dashboard_port: 7000,
            operator_pronouns: "she/her".to_string(),
            context_window_tokens: std::collections::HashMap::from([
                ("haiku".to_string(), 200_000),
                ("sonnet".to_string(), 1_000_000),
                ("opus".to_string(), 1_000_000),
            ]),
            agent_cpu_quota: "200%".to_string(),
            agent_memory_max: "4G".to_string(),
        }
    }
}

/// On-disk shape of the `hive-c0re serve --config <file>` JSON: the
/// container-injected [`HiveEnv`] (flattened) plus the hive-c0re-local
/// `model_prices` table (read only by the `/api/stats-hive` handler, 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,
    pub model_prices: crate::hive_stats::PriceTable,
    /// Number of concurrent nix-heavy job-queue nodes (prebuild /
    /// profile-swap / create / meta lock). hive-c0re-local like
    /// `model_prices` — never injected into containers. Set via
    /// `services.hyperhive.c0re.buildSlots`.
    pub build_slots: usize,
}

impl Default for ServeConfig {
    fn default() -> Self {
        Self {
            env: HiveEnv::default(),
            model_prices: crate::hive_stats::PriceTable::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 model_prices 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,
                "model_prices": {
                    "opus": {"input": 5.0, "output": 25.0, "cache_read": 0.5, "cache_write": 10.0}
                }
            }"#,
        )
        .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.operator_pronouns, "she/her");
        assert_eq!(sc.env.agent_cpu_quota, "200%");
        assert_eq!(sc.env.context_window_tokens.get("sonnet"), Some(&1_000_000));
        assert!(sc.model_prices.contains_key("opus"));
    }

    #[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!(sc.model_prices.is_empty());
    }
}

/// 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_dir: PathBuf,
    /// Manager-editable proposed config repo.
    pub proposed_dir: PathBuf,
    /// Hive-c0re-authoritative applied config repo.
    pub applied_dir: PathBuf,
    /// Claude OAuth credentials (survives purge boundary).
    pub claude_dir: PathBuf,
    /// Agent durable notes + forge token (survives purge boundary).
    pub notes_dir: PathBuf,
}

/// Per-agent in-progress state that the dashboard surfaces between approve
/// click and container ready.
#[derive(Debug, Clone)]
pub struct TransientState {
    pub kind: TransientKind,
    pub since: std::time::Instant,
}

/// RAII handle returned by `Coordinator::transient_guard`. Cleared on
/// drop — including drop-via-cancellation, the path that bare
/// `set_transient` / `clear_transient` pairs leaked through. Holds an
/// `Arc<Coordinator>` so the guard is freely returnable / movable.
#[must_use = "the guard clears the transient when dropped; bind it for the operation's \
              duration (`let _guard = coord.transient_guard(...)`). An unbound call drops \
              it immediately and un-sets the transient at once — the exact footgun this guards against."]
pub struct TransientGuard {
    coord: Arc<Coordinator>,
    name: String,
}

impl Drop for TransientGuard {
    fn drop(&mut self) {
        self.coord.clear_transient(&self.name);
    }
}

/// RAII guard for the `meta-update` in-progress flag, held for the
/// duration of a `run_meta_update` background task. Created by
/// `Coordinator::meta_update_guard`. Drop decrements the active-run
/// count; the count crossing back to 0 emits
/// `MetaUpdateRunning { running: false }`, so a concurrent pair of
/// updates only flips the dashboard flag once.
pub struct MetaUpdateGuard {
    coord: Arc<Coordinator>,
}

impl Drop for MetaUpdateGuard {
    fn drop(&mut self) {
        if self
            .coord
            .meta_updates_active
            .fetch_sub(1, Ordering::SeqCst)
            == 1
        {
            self.coord
                .emit_dashboard_event(DashboardEvent::MetaUpdateRunning {
                    seq: self.coord.next_seq(),
                    running: false,
                });
        }
    }
}

#[derive(Debug, Clone, Copy)]
pub enum TransientKind {
    /// `lifecycle::spawn` is running (nixos-container create + update + start).
    Spawning,
    /// `lifecycle::start` is running.
    Starting,
    /// `lifecycle::kill` is running.
    Stopping,
    /// `lifecycle::restart` is running.
    Restarting,
    /// `lifecycle::rebuild` is running (nixos-container update).
    Rebuilding,
    /// `actions::destroy` is running.
    Destroying,
}

impl TransientKind {
    /// Wire/UI label. Matches the strings the dashboard already
    /// renders in the transient spinner.
    pub fn as_str(self) -> &'static str {
        match self {
            TransientKind::Spawning => "spawning",
            TransientKind::Starting => "starting",
            TransientKind::Stopping => "stopping",
            TransientKind::Restarting => "restarting",
            TransientKind::Rebuilding => "rebuilding",
            TransientKind::Destroying => "destroying",
        }
    }
}

/// Field-named payload for [`Coordinator::emit_approval_resolved`].
/// Mirrors the `ApprovalResolved` dashboard-event fields. `agent`
/// borrows from the caller; `approval_kind` / `status` are
/// compile-time constants.
pub struct ApprovalResolved<'a> {
    pub id: i64,
    pub agent: &'a str,
    pub approval_kind: &'static str,
    pub sha_short: Option<String>,
    pub status: &'static str,
    pub note: Option<String>,
    pub description: Option<String>,
}

/// Field-named payload for [`Coordinator::emit_approval_added`].
/// Mirrors the `ApprovalAdded` dashboard-event fields. `agent`
/// borrows from the caller; `approval_kind` is a compile-time
/// constant. `pr_number` is set for `merge_config_pr` only.
pub struct ApprovalAdded<'a> {
    pub id: i64,
    pub agent: &'a str,
    pub approval_kind: &'static str,
    pub sha_short: Option<String>,
    pub diff: Option<String>,
    pub description: Option<String>,
    pub pr_number: Option<u64>,
}

/// Field-named payload for [`Coordinator::emit_question_added`].
/// Mirrors the `QuestionAdded` dashboard-event fields; all references
/// share the caller's lifetime.
pub struct QuestionAdded<'a> {
    pub id: i64,
    pub asker: &'a str,
    pub question: &'a str,
    pub options: &'a [String],
    pub multi: bool,
    pub deadline_at: Option<i64>,
    pub target: Option<&'a str>,
}

impl Coordinator {
    pub fn open(
        db_path: &Path,
        env: HiveEnv,
        model_prices: crate::hive_stats::PriceTable,
        build_slots: usize,
    ) -> Result<Self> {
        let HiveEnv {
            hyperhive_flake,
            hyperhive_docs_flake,
            nixpkgs_flake,
            dashboard_port,
            operator_pronouns,
            context_window_tokens,
            agent_cpu_quota,
            agent_memory_max,
        } = env;
        let broker = Broker::open(db_path).context("open broker")?;
        let approvals = Approvals::open(db_path).context("open approvals")?;
        let questions = OperatorQuestions::open(db_path).context("open operator_questions")?;
        let scheduled_prompts = crate::scheduled_prompts::ScheduledPrompts::open(db_path)
            .context("open scheduled_prompts")?;
        // 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")?,
        );
        // Install the process-wide handle so `lifecycle::run` /
        // `lifecycle::prebuild_toplevel` 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());
        // Audit log shares the same db dir; install its process-wide
        // handle so privileged-action recording sites (e.g.
        // `socket_server::handle_restart_infra`) write without threading an
        // `Arc<AuditLog>` through the agent-request surface.
        let audit_log =
            Arc::new(crate::audit_log::AuditLog::open(build_logs_dir).context("open audit_log")?);
        crate::audit_log::install(audit_log.clone());
        let power = Arc::new(crate::power::PowerStore::open(db_path).context("open agent_power")?);
        let (dashboard_events, _) = broadcast::channel(DASHBOARD_CHANNEL);
        let (shutdown_tx, _) = watch::channel(false);
        Ok(Self {
            broker: Arc::new(broker),
            approvals: Arc::new(approvals),
            questions: Arc::new(questions),
            scheduled_prompts: Arc::new(scheduled_prompts),
            build_logs,
            audit_log,
            hyperhive_flake,
            hyperhive_docs_flake,
            nixpkgs_flake,
            dashboard_port,
            operator_pronouns,
            context_window_tokens,
            agent_cpu_quota,
            agent_memory_max,
            model_prices,
            agents: Mutex::new(HashMap::new()),
            transient: Mutex::new(HashMap::new()),
            recent_transient: Mutex::new(HashMap::new()),
            recent_crashes: Mutex::new(HashMap::new()),
            graceful_stop_pending: Mutex::new(HashSet::new()),
            last_stopped_running: Mutex::new(None),
            dashboard_events,
            event_seq: AtomicU64::new(0),
            meta_updates_active: 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::spawn` / `rebuild` / `meta::sync_agents`
    /// instead of threading the individual fields separately.
    #[must_use]
    pub fn hive_env(&self) -> HiveEnv {
        HiveEnv {
            hyperhive_flake: self.hyperhive_flake.clone(),
            hyperhive_docs_flake: self.hyperhive_docs_flake.clone(),
            nixpkgs_flake: self.nixpkgs_flake.clone(),
            dashboard_port: self.dashboard_port,
            operator_pronouns: self.operator_pronouns.clone(),
            context_window_tokens: self.context_window_tokens.clone(),
            agent_cpu_quota: self.agent_cpu_quota.clone(),
            agent_memory_max: self.agent_memory_max.clone(),
        }
    }

    /// Assemble the per-agent filesystem paths for `name`. `agent_dir`
    /// is the runtime directory (`/run/hyperhive/agents/<name>`), obtained
    /// from `Coordinator::agent_dir(name)` (pure path) or from
    /// `lifecycle::ensure_agent_runtime_dir` + `agent_dir` when the dir
    /// must be created. All other paths are derived statically from `name`.
    #[must_use]
    pub fn agent_paths(name: &str, agent_dir: PathBuf) -> AgentPaths {
        AgentPaths {
            agent_dir,
            proposed_dir: Self::agent_proposed_dir(name),
            applied_dir: Self::agent_applied_dir(name),
            claude_dir: Self::agent_claude_dir(name),
            notes_dir: Self::agent_notes_dir(name),
        }
    }

    /// Emit a `RebuildQueueChanged` snapshot event. 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.
    pub fn emit_rebuild_queue_snapshot(self: &Arc<Self>) {
        let queue = self.job_queue.snapshot();
        self.emit_dashboard_event(DashboardEvent::RebuildQueueChanged {
            seq: self.next_seq(),
            queue,
        });
    }

    /// Emit a `SchedulesChanged` snapshot event. Called from every
    /// schedule mutation site (operator API handlers + the worker
    /// after each tick that fires or rearms a row) so the dashboard's
    /// scheduled-prompts tab updates live without polling.
    pub fn emit_schedules_snapshot(self: &Arc<Self>) {
        let mut schedules: Vec<hive_sh4re::WireSchedule> = match self.scheduled_prompts.list() {
            Ok(rows) => rows
                .into_iter()
                .map(crate::socket_server::schedule_to_wire_public)
                .collect(),
            Err(e) => {
                tracing::warn!(error = ?e, "emit_schedules_snapshot: list failed");
                return;
            }
        };
        // Strip ghost targets (destroyed agents) so the dashboard doesn't
        // render dead columns. Best-effort: if the roster cache is
        // momentarily contended we emit unfiltered rather than block this
        // sync path — the next snapshot / page reload corrects it.
        if let Some(live) = self.live_container_names_blocking() {
            crate::socket_server::filter_ghost_schedule_targets(&mut schedules, &live);
        }
        self.emit_dashboard_event(DashboardEvent::SchedulesChanged {
            seq: self.next_seq(),
            schedules,
        });
    }

    /// Best-effort synchronous snapshot of live container names (the
    /// logical agent names from the last `nixos-container list` scan).
    /// Returns `None` when the roster cache lock is momentarily
    /// contended, so a transient miss is treated as "roster unknown,
    /// don't filter" rather than hiding live schedule targets. The async
    /// `containers_snapshot` is the reliable path for request handlers;
    /// this exists for the sync `emit_schedules_snapshot` SSE emit.
    #[must_use]
    pub fn live_container_names_blocking(&self) -> Option<HashSet<String>> {
        self.last_containers
            .try_lock()
            .ok()
            .map(|m| m.keys().cloned().collect())
    }

    /// Emit a `RemindersChanged` snapshot event. Called from every
    /// reminder mutation site (agent `remind` calls, operator cancel /
    /// retry, and the scheduler after each delivery batch) so the
    /// dashboard's pending-reminders list stays live without polling.
    pub fn emit_reminders_snapshot(self: &Arc<Self>) {
        let reminders = match self.broker.list_pending_reminders() {
            Ok(rows) => rows,
            Err(e) => {
                tracing::warn!(error = ?e, "emit_reminders_snapshot: list failed");
                return;
            }
        };
        self.emit_dashboard_event(DashboardEvent::RemindersChanged {
            seq: self.next_seq(),
            reminders,
        });
    }

    /// Emit a `CapabilitiesChanged` snapshot event. Called from the
    /// rebuild-queue worker after a `PermChange` / Capabilities entry
    /// commits the JSON file, so the P3RM1SS10NS tab updates live.
    pub fn emit_capabilities_snapshot(self: &Arc<Self>) {
        use hive_sh4re::Capability;
        let caps = Capability::ALL.iter().map(|c| c.as_str()).collect();
        let descriptions = Capability::ALL
            .iter()
            .map(|c| (c.as_str(), c.description()))
            .collect();
        let assignments = crate::capabilities::read();
        // Best-effort roster (sync path); on a contended cache miss we
        // emit explicit keys only — the HTTP refetch fills the rest in.
        let roster = self.live_container_names_blocking().unwrap_or_default();
        // Capability default is "no extra caps" — empty default slice.
        let (agents, effective) =
            crate::dashboard::permissions::roster_and_effective(roster, &assignments, &[]);
        self.emit_dashboard_event(DashboardEvent::CapabilitiesChanged {
            seq: self.next_seq(),
            caps,
            descriptions,
            assignments,
            agents,
            effective,
        });
    }

    /// Emit a `ToolGroupsChanged` snapshot event. Called from the
    /// rebuild-queue worker after a `PermChange` / `ToolGroups` entry
    /// commits the JSON file, so the P3RM1SS10NS tab updates live.
    pub fn emit_tool_groups_snapshot(self: &Arc<Self>) {
        use hive_sh4re::ToolGroup;
        let groups = ToolGroup::ALL.iter().map(|g| g.as_str()).collect();
        let descriptions = ToolGroup::ALL
            .iter()
            .map(|g| (g.as_str(), g.description()))
            .collect();
        let assignments = crate::tool_groups::read();
        let roster = self.live_container_names_blocking().unwrap_or_default();
        let (agents, effective) = crate::dashboard::permissions::roster_and_effective(
            roster,
            &assignments,
            &crate::dashboard::permissions::tool_group_default_names(),
        );
        self.emit_dashboard_event(DashboardEvent::ToolGroupsChanged {
            seq: self.next_seq(),
            groups,
            descriptions,
            assignments,
            agents,
            effective,
        });
    }

    /// Update the `step` label on the currently-running node of DAG
    /// `id` and (if it actually changed) re-emit the queue snapshot so
    /// the dashboard renders the new phase. DAG-id-only surface for
    /// the opaque approval pipeline in `actions.rs`, whose callbacks
    /// don't know node ids (its DAGs are single-node, so the lookup is
    /// exact); queue executors use the precise per-node sink in
    /// `job_queue::exec` instead. No-op when `id` is `None` (callers
    /// not running from the queue) or when nothing is `Running`.
    pub fn set_queue_step(self: &Arc<Self>, id: Option<u64>, step: &str) {
        let Some(id) = id else { return };
        if self.job_queue.set_step_running(id, step) {
            self.emit_rebuild_queue_snapshot();
        }
    }

    /// Link a `build_logs` row to the currently-running node of DAG
    /// `id` and re-emit the snapshot. Same DAG-id-only compatibility
    /// surface as [`Self::set_queue_step`].
    pub fn set_queue_build_log(self: &Arc<Self>, id: Option<u64>, log_id: i64) {
        let Some(id) = id else { return };
        if self.job_queue.set_build_log_id_running(id, log_id) {
            self.emit_rebuild_queue_snapshot();
        }
    }

    /// 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);
    }

    /// Mark a `meta-update` as in flight and return an RAII guard that
    /// clears it on drop (including drop-via-panic). The first
    /// concurrent run emits `MetaUpdateRunning { running: true }`; the
    /// last one to finish emits `running: false`. The dashboard's META
    /// INPUTS panel reads the flag to show a disabled "updating…"
    /// state while the lock bump + rebuild ripple runs.
    pub fn meta_update_guard(self: &Arc<Self>) -> MetaUpdateGuard {
        if self.meta_updates_active.fetch_add(1, Ordering::SeqCst) == 0 {
            self.emit_dashboard_event(DashboardEvent::MetaUpdateRunning {
                seq: self.next_seq(),
                running: true,
            });
        }
        MetaUpdateGuard {
            coord: Arc::clone(self),
        }
    }

    /// True while at least one dashboard-triggered `meta-update` is
    /// running. Surfaced on `/api/state` as `meta_update_running` so a
    /// client that cold-loads mid-update sees the in-progress state.
    pub fn meta_update_in_progress(&self) -> bool {
        self.meta_updates_active.load(Ordering::SeqCst) > 0
    }

    /// Emit `AuditEntryAdded` immediately after a privileged-action row
    /// is recorded, so the dashboard audit view live-appends it off
    /// `/dashboard/stream`. Pass the [`AuditEntry`](crate::audit_log::AuditEntry)
    /// returned by `audit_log::record` so the streamed event is the same
    /// canonical row that was stored.
    pub fn emit_audit_entry(&self, entry: crate::audit_log::AuditEntry) {
        self.emit_dashboard_event(DashboardEvent::AuditEntryAdded {
            seq: self.next_seq(),
            entry,
        });
    }

    /// Emit `ApprovalAdded` immediately after the row is inserted in
    /// sqlite. Caller passes the diff text it already computed (or
    /// `None` for spawn approvals which carry no diff).
    pub fn emit_approval_added(&self, ev: ApprovalAdded<'_>) {
        let ApprovalAdded {
            id,
            agent,
            approval_kind,
            sha_short,
            diff,
            description,
            pr_number,
        } = ev;
        self.emit_dashboard_event(DashboardEvent::ApprovalAdded {
            seq: self.next_seq(),
            id,
            agent: agent.to_owned(),
            approval_kind,
            sha_short,
            diff,
            description,
            pr_number,
        });
    }

    /// Emit `ApprovalResolved` after `mark_approved` / `mark_denied` /
    /// `mark_failed` lands. `resolved_at` is stamped from the system
    /// clock here so call sites don't repeat the conversion; if you
    /// already have an authoritative timestamp from the db update,
    /// the tiny skew between "row updated" and "event emitted" is
    /// presentation-only and doesn't matter to clients.
    ///
    /// Takes [`ApprovalResolved`] rather than a positional arg list so
    /// the seven fields are named at every call site.
    pub fn emit_approval_resolved(&self, ev: ApprovalResolved<'_>) {
        let ApprovalResolved {
            id,
            agent,
            approval_kind,
            sha_short,
            status,
            note,
            description,
        } = ev;
        let resolved_at = 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::ApprovalResolved {
            seq: self.next_seq(),
            id,
            agent: agent.to_owned(),
            approval_kind,
            sha_short,
            status,
            resolved_at: hive_sh4re::wire_time::from_secs(resolved_at),
            note,
            description,
        });
    }

    /// Emit `QuestionAdded` after a question is inserted. Fires for
    /// both operator-targeted (`target = None`) and peer-to-peer
    /// (`target = Some(agent)`) threads — the dashboard surfaces
    /// both, distinguishing visually + offering operator override.
    pub fn emit_question_added(&self, ev: &QuestionAdded<'_>) {
        let &QuestionAdded {
            id,
            asker,
            question,
            options,
            multi,
            deadline_at,
            target,
        } = ev;
        let asked_at = std::time::SystemTime::now()
            .duration_since(std::time::UNIX_EPOCH)
            .ok()
            .and_then(|d| i64::try_from(d.as_secs()).ok())
            .unwrap_or(0);
        let question_refs = crate::dashboard::scan_validated_paths(question);
        self.emit_dashboard_event(DashboardEvent::QuestionAdded {
            seq: self.next_seq(),
            id,
            asker: asker.to_owned(),
            question: question.to_owned(),
            options: options.to_vec(),
            multi,
            asked_at: hive_sh4re::wire_time::from_secs(asked_at),
            deadline_at: deadline_at.map(hive_sh4re::wire_time::from_secs),
            target: target.map(str::to_owned),
            question_refs,
        });
    }

    /// Emit `QuestionResolved` when a question transitions to
    /// answered (operator answer, peer answer, operator override on
    /// a peer thread, operator cancel, or ttl watchdog). Both
    /// operator-targeted and peer threads fire so the dashboard's
    /// derived store can move the row from pending to history.
    pub fn emit_question_resolved(
        &self,
        id: i64,
        answer: &str,
        answerer: &str,
        cancelled: bool,
        target: Option<&str>,
    ) {
        let answered_at = std::time::SystemTime::now()
            .duration_since(std::time::UNIX_EPOCH)
            .ok()
            .and_then(|d| i64::try_from(d.as_secs()).ok())
            .unwrap_or(0);
        let answer_refs = crate::dashboard::scan_validated_paths(answer);
        self.emit_dashboard_event(DashboardEvent::QuestionResolved {
            seq: self.next_seq(),
            id,
            answer: answer.to_owned(),
            answerer: answerer.to_owned(),
            answered_at: hive_sh4re::wire_time::from_secs(answered_at),
            cancelled,
            target: target.map(str::to_owned),
            answer_refs,
        });
    }

    /// 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_update` / `needs_login` / `deployed_sha`
    /// resolves to — lifecycle ops, destroy, approve (post-spawn),
    /// rebuild, meta-update, 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>) {
        let fresh = container_view::build_all(self).await;
        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,
            });
        }
    }

    /// Apply a topology reparent + fan the resulting notifications
    /// out to the three affected agents. On success, drops a
    /// one-line system message into the inbox of:
    ///
    /// 1. The **old parent** (if any) — `"{child} moved out of your
    ///    subtree to {new_parent_or_root}"`.
    /// 2. The **new parent** (if any) — `"{child} just moved into
    ///    your subtree (was previously under
    ///    {old_parent_or_root})"`.
    /// 3. The **moved agent** — `"your parent changed from
    ///    {old_parent_or_root} to {new_parent_or_root}"`.
    ///
    /// `_or_root` resolves to the literal string `"<root>"` when the
    /// slot is `None`, keeping the wording consistent with the
    /// `<parent>` sentinel's "root → operator" routing (see
    /// `docs/conventions.md::Recipient sentinels`). The
    /// notifications fire as ordinary broker messages with
    /// `from = hive_sh4re::SYSTEM_SENDER` so the dashboard renders
    /// them under the existing system-source styling.
    ///
    /// Idempotent: if `topology::set_parent` skipped the disk write
    /// (same-parent no-op), no messages fire. The `topology` module's
    /// validation (`apply_set_parent`: unknown agent, cycle, etc.)
    /// runs *before* any message is sent, so a refused move never
    /// notifies anyone.
    ///
    /// Also drives a `rescan_containers_and_emit` after the write so
    /// the dashboard tree re-renders without polling.
    pub async fn reparent_with_notify(
        self: &Arc<Self>,
        child: &str,
        new_parent: Option<&str>,
    ) -> std::result::Result<(), String> {
        // Snapshot the old parent BEFORE the write so the
        // notifications can describe both sides of the change.
        let topo_before = crate::topology::read();
        let old_parent = topo_before.get(child).cloned().flatten();

        // The disk write + git commit happen here under META_LOCK, so
        // the topology.json change is committed atomically and the
        // working tree is never left dirty between the write and the
        // next meta operation. Topology validation + idempotent
        // fast-path inside `set_parent` may short-circuit (same
        // parent → no-op). We mirror the same idempotent shape for
        // the notifications: if nothing changed, send nothing.
        crate::meta::commit_topology(child, new_parent).await?;

        let changed = old_parent.as_deref() != new_parent;
        if changed {
            let old_label = old_parent.as_deref().unwrap_or("<root>");
            let new_label = new_parent.unwrap_or("<root>");
            // System-source notifications. Best-effort: a failed broker
            // send is logged + ignored so a transient sqlite error
            // doesn't bubble out and unwind the topology write.
            if let Some(op) = old_parent.as_deref() {
                let _ = self.broker.send(&hive_sh4re::Message {
                    from: hive_sh4re::SYSTEM_SENDER.to_owned(),
                    to: op.to_owned(),
                    body: format!("{child} moved out of your subtree to {new_label}"),
                    in_reply_to: None,
                });
            }
            if let Some(np) = new_parent {
                let _ = self.broker.send(&hive_sh4re::Message {
                    from: hive_sh4re::SYSTEM_SENDER.to_owned(),
                    to: np.to_owned(),
                    body: format!(
                        "{child} just moved into your subtree (was previously under {old_label})"
                    ),
                    in_reply_to: None,
                });
            }
            // Coalesce: if a prior move notification is already pending in
            // the broker (agent was offline for multiple moves), update it
            // in-place so the agent sees "A to C" not "A to B" then "B to C".
            let _ = self
                .broker
                .send_coalescing_reparent(child, old_label, new_label);
        }

        // Rescan + diff-emit regardless of whether messages fired —
        // even an idempotent no-op might have happened concurrently
        // with another lifecycle event the dashboard cares about.
        self.rescan_containers_and_emit().await;
        Ok(())
    }

    /// Batch version of [`reparent_with_notify`]: applies all moves under a
    /// single `META_LOCK` acquisition (one git commit) then sends per-agent
    /// notifications for each move that actually changed topology.
    /// First validation failure aborts the whole batch with no disk writes.
    ///
    /// # Errors
    ///
    /// Propagates any error returned by [`crate::meta::bulk_commit_topology`]
    /// (validation failure or topology-file write error).
    pub async fn reparent_bulk_with_notify(
        self: &Arc<Self>,
        moves: &[(&str, Option<&str>)],
    ) -> std::result::Result<(), String> {
        if moves.is_empty() {
            return Ok(());
        }
        // bulk_commit_topology applies all set_parent calls under one lock
        // and returns (child, old_parent) for every move that changed.
        let changed = crate::meta::bulk_commit_topology(moves).await?;

        // Send per-agent notifications for each changed move.
        for (child, old_parent) in &changed {
            // Find the new parent from the moves slice.
            let new_parent = moves
                .iter()
                .find(|(c, _)| *c == child)
                .and_then(|(_, np)| *np);
            let old_label = old_parent.as_deref().unwrap_or("<root>");
            let new_label = new_parent.unwrap_or("<root>");
            if let Some(op) = old_parent.as_deref() {
                let _ = self.broker.send(&hive_sh4re::Message {
                    from: hive_sh4re::SYSTEM_SENDER.to_owned(),
                    to: op.to_owned(),
                    body: format!("{child} moved out of your subtree to {new_label}"),
                    in_reply_to: None,
                });
            }
            if let Some(np) = new_parent {
                let _ = self.broker.send(&hive_sh4re::Message {
                    from: hive_sh4re::SYSTEM_SENDER.to_owned(),
                    to: np.to_owned(),
                    body: format!(
                        "{child} just moved into your subtree (was previously under {old_label})"
                    ),
                    in_reply_to: None,
                });
            }
            let _ = self
                .broker
                .send_coalescing_reparent(child, old_label, new_label);
        }

        self.rescan_containers_and_emit().await;
        Ok(())
    }

    /// 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 = Self::agent_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 broker + operator_questions to handle the agent-side
        // `ask` / `answer` tools, not 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()
    }

    /// Mark an agent as in-progress (only one state per agent for now).
    ///
    /// Private on purpose: the RAII [`TransientGuard`] (via
    /// [`Coordinator::transient_guard`]) is the only door, so the paired
    /// `clear_transient` always runs on drop even if the surrounding future
    /// is cancelled (HTTP request aborted, runtime shutdown mid-rebuild,
    /// panic). A bare set with no guaranteed clear would leak the transient
    /// and leave the dashboard stuck in "rebuilding…" forever.
    fn set_transient(&self, name: &str, kind: TransientKind) {
        self.transient.lock().unwrap().insert(
            name.to_owned(),
            TransientState {
                kind,
                since: std::time::Instant::now(),
            },
        );
        // 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: kind.as_str(),
            since_unix,
        });
    }

    /// Clear an agent's transient state. Private: only reachable through
    /// [`TransientGuard`]'s `Drop`, which guarantees it runs (see
    /// [`Coordinator::set_transient`]).
    fn clear_transient(&self, name: &str) {
        let removed = self.transient.lock().unwrap().remove(name);
        if let Some(state) = removed {
            // Stamp the tombstone so the crash watcher can still see
            // "operator kicked this off recently" on its next 10s poll
            // — without this, the clear-then-poll race produced a
            // spurious ContainerCrash on every operator stop/restart.
            // Old entries get reaped lazily on read so the map doesn't
            // grow unbounded.
            self.recent_transient
                .lock()
                .unwrap()
                .insert(name.to_owned(), (state.kind, std::time::Instant::now()));
            self.emit_dashboard_event(DashboardEvent::TransientCleared {
                seq: self.next_seq(),
                name: name.to_owned(),
            });
        }
    }

    /// 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);
    }

    /// Record the set of agents that were running at a broad-scope
    /// `hivectl stop`, so the next broad-scope `start` restores exactly
    /// this set. See the `last_stopped_running` field doc. Persists a
    /// copy to the broker `kv` table (best-effort) so the snapshot
    /// survives a hive-c0re restart; the in-memory copy is the fast path.
    pub fn set_last_stopped_running(&self, agents: Vec<String>) {
        match serde_json::to_string(&agents) {
            Ok(json) => {
                if let Err(e) = self.broker.kv_set(LAST_STOPPED_RUNNING_KEY, &json) {
                    tracing::warn!(error = ?e, "persist last_stopped_running failed (in-memory copy still set)");
                }
            }
            Err(e) => tracing::warn!(error = ?e, "serialise last_stopped_running failed"),
        }
        *self.last_stopped_running.lock().unwrap() = Some(agents);
    }

    /// Take (and clear) the recorded broad-stop running set, if any. A
    /// broad-scope `start` uses this to restore only the previously
    /// running agents; `None` means "no record — start all". Falls back
    /// to the persisted `kv` copy when the in-memory snapshot is empty
    /// (hive-c0re restarted between the stop-all and the start). Clears
    /// the persisted copy either way — the snapshot is one-shot.
    pub fn take_last_stopped_running(&self) -> Option<Vec<String>> {
        let result = self
            .last_stopped_running
            .lock()
            .unwrap()
            .take()
            .or_else(|| {
                self.broker
                    .kv_get(LAST_STOPPED_RUNNING_KEY)
                    .ok()
                    .flatten()
                    .and_then(|json| serde_json::from_str::<Vec<String>>(&json).ok())
            });
        if let Err(e) = self.broker.kv_delete(LAST_STOPPED_RUNNING_KEY) {
            tracing::warn!(error = ?e, "clear persisted last_stopped_running failed");
        }
        result
    }

    /// Set of agents whose transient was 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.
    pub fn recent_transient_within(
        &self,
        grace: std::time::Duration,
    ) -> HashMap<String, TransientKind> {
        let now = std::time::Instant::now();
        let mut map = self.recent_transient.lock().unwrap();
        map.retain(|_, (_, ts)| now.duration_since(*ts) <= grace);
        map.iter()
            .map(|(k, (kind, _))| (k.clone(), *kind))
            .collect()
    }

    /// Record an unexpected crash for `agent`. Called by the crash
    /// watcher whenever it classifies a container stop as a crash (not an
    /// operator action). Append-only here; pruning happens lazily on read
    /// in `recent_crash_counts`.
    pub fn record_crash(&self, agent: &str) {
        self.recent_crashes
            .lock()
            .unwrap()
            .entry(agent.to_owned())
            .or_default()
            .push(std::time::Instant::now());
    }

    /// Per-agent count of crashes within the last `window`. Lazily reaps
    /// older timestamps and drops agents with none left, so the map stays
    /// bounded and only lists agents actively crashing. Powers the
    /// dashboard's `agents_crashing` banner warning.
    pub fn recent_crash_counts(&self, window: std::time::Duration) -> HashMap<String, usize> {
        let now = std::time::Instant::now();
        let mut map = self.recent_crashes.lock().unwrap();
        map.retain(|_, times| {
            times.retain(|ts| now.duration_since(*ts) <= window);
            !times.is_empty()
        });
        map.iter().map(|(k, v)| (k.clone(), v.len())).collect()
    }

    /// Set a transient state and return a guard that clears it on drop.
    /// Use this from any path where the surrounding future could be
    /// cancelled or panic between set and clear (HTTP handlers, spawned
    /// tasks). The guard's `Drop` runs even on task cancellation, so
    /// the dashboard's spinner can't get pinned forever.
    pub fn transient_guard(self: &Arc<Self>, name: &str, kind: TransientKind) -> TransientGuard {
        self.set_transient(name, kind);
        TransientGuard {
            coord: self.clone(),
            name: name.to_owned(),
        }
    }

    pub fn transient_snapshot(&self) -> HashMap<String, TransientState> {
        self.transient.lock().unwrap().clone()
    }

    /// 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) {
        // Sub-agents bind their state at /agents/<name>/state. The
        // manager has both /state (legacy mount) and /agents
        // bind-mounted, so /agents/<name>/state 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::Message {
            from: hive_sh4re::SYSTEM_SENDER.to_owned(),
            to: name.to_owned(),
            body,
            in_reply_to: None,
        }) {
            tracing::warn!(error = ?e, %name, "kick_agent: broker.send failed");
        }
    }

    /// Push a `HelperEvent` into the manager's inbox. Encoded as JSON in
    /// `Message::body`; sender = `SYSTEM_SENDER`. The manager harness
    /// recognises the sender and parses the body. Best-effort: a serde or
    /// broker error is logged but does not propagate.
    pub fn notify_manager(&self, event: &hive_sh4re::HelperEvent) {
        self.notify_agent(hive_sh4re::MANAGER_AGENT, event);
    }

    /// Route an approval-scoped helper event to the agent that submitted
    /// approval `approval_id` (the authenticated socket caller at submit
    /// time). Legacy rows with no recorded submitter — and any lookup
    /// failure — fall back to the root agent, preserving the prior
    /// always-root behaviour.
    pub fn notify_submitter(&self, approval_id: i64, event: &hive_sh4re::HelperEvent) {
        let target = self
            .approvals
            .submitter_of(approval_id)
            .unwrap_or_default()
            .unwrap_or_else(|| hive_sh4re::MANAGER_AGENT.to_owned());
        self.notify_agent(&target, event);
    }

    /// Push a `HelperEvent` into an arbitrary agent's inbox. Encoded
    /// the same way as `notify_manager` (sender = `SYSTEM_SENDER`,
    /// body = JSON-encoded event). Used to route `QuestionAnswered`
    /// events back to the agent that called `ask`, `QuestionAsked`
    /// events to the target of a peer question, etc.
    pub fn notify_agent(&self, agent: &str, event: &hive_sh4re::HelperEvent) {
        self.notify_agent_from(hive_sh4re::SYSTEM_SENDER, agent, event);
    }

    /// Same as `notify_agent` but with an explicit sender. Use this
    /// when the event originates from a known agent or the operator
    /// (e.g. `QuestionAnswered` — the answerer should be the `from`,
    /// not `system`) so the recipient's terminal shows the right name.
    pub fn notify_agent_from(&self, from: &str, agent: &str, event: &hive_sh4re::HelperEvent) {
        let body = match serde_json::to_string(event) {
            Ok(s) => s,
            Err(e) => {
                tracing::warn!(error = ?e, "failed to encode helper event");
                return;
            }
        };
        if let Err(e) = self.broker.send(&hive_sh4re::Message {
            from: from.to_owned(),
            to: agent.to_owned(),
            body,
            in_reply_to: None,
        }) {
            tracing::warn!(error = ?e, target = %agent, "failed to push helper event");
        }
    }

    /// 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::Message {
                from: from.to_owned(),
                to: agent_name.clone(),
                body: broadcast_body.clone(),
                in_reply_to: None,
            }) {
                errors.push(format!("{agent_name}: {e}"));
            }
        }
        errors
    }

    pub fn agent_dir(name: &str) -> PathBuf {
        PathBuf::from(format!("{AGENT_RUNTIME_ROOT}/{name}"))
    }

    pub fn socket_path(name: &str) -> PathBuf {
        Self::agent_dir(name).join("mcp.sock")
    }

    /// Ensure a runtime dir + (for sub-agents) per-agent socket exists.
    ///
    /// **Prefer the split form:**
    /// - dir creation → `lifecycle::ensure_agent_runtime_dir(name)`
    /// - listener registration → `register_agent(name)` (eagerly, on first
    ///   spawn) or leave it to `mcp_sockets::spawn_poll` (reconcile within
    ///   10 s, safe for the start + rebuild paths where the container
    ///   takes longer than that to boot).
    ///
    /// This method is kept as a convenience shim for any remaining callers
    /// that need the combined semantics in one call.
    pub fn ensure_runtime(self: &Arc<Self>, name: &str) -> Result<PathBuf> {
        if name == crate::lifecycle::MANAGER_NAME {
            let dir = Self::agent_dir(name);
            std::fs::create_dir_all(&dir)
                .with_context(|| format!("create manager dir {}", dir.display()))?;
            return Ok(dir);
        }
        self.register_agent(name)
    }

    /// Per-agent state root (parent of `config/`, future `prompts/`, etc.).
    pub fn agent_state_root(name: &str) -> PathBuf {
        PathBuf::from(format!("{AGENT_STATE_ROOT}/{name}"))
    }

    /// Manager-editable proposed config repo. Bind-mounted into the manager
    /// container as `/agents/<name>/config/`.
    pub fn agent_proposed_dir(name: &str) -> PathBuf {
        Self::agent_state_root(name).join("config")
    }

    /// 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: &str) -> PathBuf {
        Self::agent_state_root(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: &str) -> PathBuf {
        Self::agent_state_root(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: &str) -> PathBuf {
        Self::agent_state_root(name).join("harness")
    }

    /// Authoritative applied config repo. Hive-c0re-only.
    pub fn agent_applied_dir(name: &str) -> PathBuf {
        PathBuf::from(format!("{APPLIED_STATE_ROOT}/{name}"))
    }

    /// 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<String> {
        let Ok(rd) = std::fs::read_dir(AGENT_STATE_ROOT) else {
            return Vec::new();
        };
        let mut out: Vec<String> = rd
            .flatten()
            .filter(|e| e.file_type().is_ok_and(|t| t.is_dir()))
            .filter_map(|e| e.file_name().into_string().ok())
            .collect();
        out.sort();
        out
    }

    /// Agents that have an operator-approved proposed config repo but
    /// were never deployed — proposed `.git` exists, applied `.git` does
    /// not. Their `topology.json` parent edge (written at `InitConfig`
    /// approval) must survive `topology::reconcile` until the first
    /// apply-commit spawns the container. Distinct from tombstones,
    /// which have an applied repo from a prior deploy.
    #[must_use]
    pub fn pending_init_names() -> Vec<String> {
        Self::kept_state_names()
            .into_iter()
            .filter(|n| {
                Self::agent_proposed_dir(n).join(".git").exists()
                    && !Self::agent_applied_dir(n).join(".git").exists()
            })
            .collect()
    }
}