//! 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; 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::agent_server::{self, AgentSocket}; use crate::approvals::Approvals; use crate::broker::Broker; use crate::container_view::{self, ContainerView}; use crate::dashboard_events::DashboardEvent; use crate::operator_questions::OperatorQuestions; /// 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; const AGENT_RUNTIME_ROOT: &str = "/run/hyperhive/agents"; /// Manager-editable per-agent config repos. Bind-mounted RW into the manager /// container as `/agents//`. 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, pub approvals: Arc, pub questions: Arc, /// 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, /// 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, /// URL of the hyperhive flake (no fragment). Inlined into per-agent /// `flake.nix` files as `inputs.hyperhive.url`. pub hyperhive_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, /// Store-path URL for `nixpkgs-unstable` to wire as a top-level meta /// flake input. Hyperhive's `inputs.nixpkgs-unstable` then follows it. /// Set via `--nixpkgs-unstable-flake` from `hive-c0re.nix`. Empty string /// falls back to the legacy `follows = "hyperhive/nixpkgs-unstable"`. pub nixpkgs_unstable_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..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_` /// 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, /// Per-agent systemd `CPUQuota=` value (e.g. `"200%"`). Written into /// the `container@h-.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, agents: Mutex>, /// 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>, /// 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>, /// 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, 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>, /// Global rebuild queue. Every long-running container/meta op /// (rebuild, meta-update, first-spawn) goes through this queue so /// hive-c0re runs at most one at a time and the dashboard can /// render a single ordered view of pending + running work. See /// `rebuild_queue.rs` for the dedup rules + history retention. pub rebuild_queue: Arc, /// 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, } /// 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. #[derive(Clone, Debug)] pub struct HiveEnv { pub hyperhive_flake: String, pub nixpkgs_flake: String, pub nixpkgs_unstable_flake: String, pub dashboard_port: u16, pub operator_pronouns: String, pub context_window_tokens: std::collections::HashMap, /// 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 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` so the guard is freely returnable / movable. pub struct TransientGuard { coord: Arc, 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, } 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", } } } impl Coordinator { #[allow( clippy::too_many_arguments, reason = "constructor wiring host-level config (flakes, ports, pronouns, \ context-window + resource limits) into the coordinator; bundling \ into a struct would just move the same fields one level out" )] pub fn open( db_path: &Path, hyperhive_flake: String, nixpkgs_flake: String, nixpkgs_unstable_flake: String, dashboard_port: u16, operator_pronouns: String, context_window_tokens: std::collections::HashMap, agent_cpu_quota: String, agent_memory_max: String, ) -> Result { 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` through every public entry // point in the lifecycle surface. crate::build_logs::install(build_logs.clone()); 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, hyperhive_flake, nixpkgs_flake, nixpkgs_unstable_flake, dashboard_port, operator_pronouns, context_window_tokens, agent_cpu_quota, agent_memory_max, agents: Mutex::new(HashMap::new()), transient: Mutex::new(HashMap::new()), recent_transient: Mutex::new(HashMap::new()), dashboard_events, event_seq: AtomicU64::new(0), meta_updates_active: AtomicU64::new(0), last_containers: tokio::sync::Mutex::new(HashMap::new()), rebuild_queue: Arc::new(crate::rebuild_queue::RebuildQueue::new()), 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(), nixpkgs_flake: self.nixpkgs_flake.clone(), nixpkgs_unstable_flake: self.nixpkgs_unstable_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`. The caller /// must supply `agent_dir` (from `ensure_runtime`) since that /// creates the tmpfs entry on first call. All other paths are /// derived statically from `name`. /// /// ```ignore /// let paths = Coordinator::agent_paths(name, coord.ensure_runtime(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) { let queue = self.rebuild_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) { let schedules = match self.scheduled_prompts.list() { Ok(rows) => rows .into_iter() .map(crate::manager_server::schedule_to_wire_public) .collect(), Err(e) => { tracing::warn!(error = ?e, "emit_schedules_snapshot: list failed"); return; } }; self.emit_dashboard_event(DashboardEvent::SchedulesChanged { seq: self.next_seq(), schedules, }); } /// 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) { 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) { 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(); self.emit_dashboard_event(DashboardEvent::CapabilitiesChanged { seq: self.next_seq(), caps, descriptions, assignments, }); } /// 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) { 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(); self.emit_dashboard_event(DashboardEvent::ToolGroupsChanged { seq: self.next_seq(), groups, descriptions, assignments, }); } /// Update the `step` label on a running queue entry and (if it /// actually changed) re-emit the queue snapshot so the dashboard /// renders the new phase. Returns `true` when the label was new /// and an emit fired, mostly for tracing/logging callers; safe to /// ignore. No-op when `id` is `None` (e.g. callers that aren't /// running from the queue worker) or when the row isn't `Running`. pub fn set_queue_step(self: &Arc, id: Option, step: &str) { let Some(id) = id else { return }; if self.rebuild_queue.set_step(id, step) { 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 { 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 { 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) -> 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 `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, id: i64, agent: &str, approval_kind: &'static str, sha_short: Option, diff: Option, description: Option, ) { self.emit_dashboard_event(DashboardEvent::ApprovalAdded { seq: self.next_seq(), id, agent: agent.to_owned(), approval_kind, sha_short, diff, description, }); } /// 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. #[allow(clippy::too_many_arguments)] pub fn emit_approval_resolved( &self, id: i64, agent: &str, approval_kind: &'static str, sha_short: Option, status: &'static str, note: Option, description: Option, ) { 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, 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. #[allow(clippy::too_many_arguments)] pub fn emit_question_added( &self, id: i64, asker: &str, question: &str, options: &[String], multi: bool, deadline_at: Option, target: Option<&str>, ) { 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, deadline_at, 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, 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) { 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 `""` when the /// slot is `None`, keeping the wording consistent with the /// `` 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, 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(""); let new_label = new_parent.unwrap_or(""); // 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, 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(""); let new_label = new_parent.unwrap_or(""); 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 { let last = self.last_containers.lock().await; let mut out: Vec = last.values().cloned().collect(); out.sort_by(|a, b| a.name.cmp(&b.name)); out } pub fn register_agent(self: &Arc, name: &str) -> Result { // 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 = agent_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 { self.agents.lock().unwrap().keys().cloned().collect() } /// Mark an agent as in-progress (only one state per agent for now). /// /// Prefer `transient_guard` when possible — it auto-clears on drop /// even if the surrounding future is cancelled (HTTP request /// aborted, runtime shutdown mid-rebuild, panic between set and /// clear). The bare `set_transient` / `clear_transient` pair leaks /// the transient on any of those paths and the dashboard then /// shows the agent stuck in "rebuilding…" forever. pub 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, }); } pub 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(), }); } } /// 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 { 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() } /// 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, name: &str, kind: TransientKind) -> TransientGuard { self.set_transient(name, kind); TransientGuard { coord: self.clone(), name: name.to_owned(), } } pub fn transient_snapshot(&self) -> HashMap { 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//state. The // manager has both /state (legacy mount) and /agents // bind-mounted, so /agents//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); } /// 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 { 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") } /// Runtime dir for the manager. Uses the same per-agent subdir layout as /// sub-agents — the manager is just another agent under /// `AGENT_RUNTIME_ROOT` with its own subdirectory. pub fn manager_dir() -> PathBuf { Self::agent_dir(crate::lifecycle::MANAGER_NAME) } pub fn manager_socket_path() -> PathBuf { Self::socket_path(crate::lifecycle::MANAGER_NAME) } /// Ensure a runtime dir + (for sub-agents) per-agent socket exists. For /// the manager, `manager_server::start` owns the socket — just return /// the dir. For sub-agents this is `register_agent` (creates a fresh /// listener bound to `socket_path(name)`). Source directory of the /// `/run/hive/mcp.sock` bind that ends up in `set_nspawn_flags`. pub fn ensure_runtime(self: &Arc, name: &str) -> Result { 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//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 { let Ok(rd) = std::fs::read_dir(AGENT_STATE_ROOT) else { return Vec::new(); }; let mut out: Vec = 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 } }