//! 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::job_queue::RunningTransient; 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"; 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, /// Audit trail of agent-initiated privileged actions (infra restart, /// …). See `audit_log.rs`. Same dir as `build_logs`. pub audit_log: Arc, /// 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, /// Store path of the `claude-code` build every agent runs, written /// into each per-agent flake as `hyperhive.claudeCodePath`. Set by /// the NixOS module option `services.hyperhive.c0re.claudeCodePackage` /// (which resolves the package and hands us its path). `None` = every /// agent keeps the `claude-code` from its own nixpkgs. /// /// A path rather than a flake input because containers share the /// host's `/nix/store`: the binary is already reachable inside them, /// closure and all. The host module is what keeps it from being /// garbage-collected — see that option. pub claude_code_path: Option, /// 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, /// Per-agent systemd `CPUWeight=` (cgroup v2 `cpu.weight`, 1..=10000). /// Same drop-in. A *relative share* under contention, not a cap — see /// the `agentCpuWeight` NixOS option for the full semantics. `None` /// (option set to `null`) omits the setting: kernel default, and the /// drop-in is byte-identical to the one written before weights existed. pub agent_cpu_weight: Option, /// Per-agent systemd `IOWeight=` (cgroup v2 `io.weight`, 1..=10000). /// Same drop-in, same relative-share and `None` semantics as /// [`Self::agent_cpu_weight`]. pub agent_io_weight: Option, /// 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 `. 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>, /// Tombstone for transients that have JUST been cleared. The /// crash watcher polls every 10s and would race the /// drop-clears-immediately path of `TransientGuard`: an operator /// kill / restart sets `Stopping` → runs `nixos-container stop` → /// drop clears the transient → poll fires next tick and sees the /// container missing-from-running with no active transient → /// spurious "container stopped without an operator action" /// message. /// /// `clear_transient` stamps the cleared kind here with an /// `Instant`; `recent_transient_within(grace)` returns the set of /// agents whose tombstone is still inside the grace window. Crash /// watcher consults both this and the active map before declaring /// a stop deliberate. /// /// 🚨 **Keyed by `(agent, label)`, not by agent.** An agent can have several /// pills clearing in the same window — the transient set tests status alone, /// so a lease-exempt `Prebuild` and a lease-holding `StopForUpdate` are both /// live and both clear. Keyed by agent, the last clear *overwrites* the /// others: a `Prebuild` (`deliberate_stop = false`) landing after a /// `StopForUpdate` (`true`) leaves the tombstone reading `false`, and the /// crash watcher then reports an intentional stop as a **crash**. recent_transient: Mutex>, /// 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>>, /// 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>, /// Agents with a pause acknowledgement in progress — set by the pause /// DAG's signal node, cleared when the agent reports /// `PauseAcknowledged` (or the drain node's timeout fallback fires). /// Same shape as `graceful_stop_pending`, one set per concern rather /// than reusing it: a pause and a graceful stop are independent /// orchestrations that can be in flight for different agents /// simultaneously. pause_pending: Mutex>, /// 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>>, /// 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 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, /// 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, /// 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. /// /// 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, /// Store path of the `claude-code` agents run, or `None` for "each /// agent keeps the one out of its own nixpkgs". Travels into the /// container as `hyperhive.claudeCodePath` — a plain string, kept /// alive host-side by the module that resolved it. pub claude_code_path: Option, 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 systemd `CPUWeight=` — cgroup v2 `cpu.weight`, 1..=10000. /// Relative share under contention, not a cap; see `agentCpuWeight`. /// `None` = leave the setting out of the drop-in (kernel default). pub agent_cpu_weight: Option, /// Per-agent systemd `IOWeight=` — cgroup v2 `io.weight`, 1..=10000. /// Same semantics as [`Self::agent_cpu_weight`]; see `agentIoWeight`. pub agent_io_weight: Option, } impl Default for HiveEnv { fn default() -> Self { Self { hyperhive_flake: "/etc/hyperhive".to_string(), hyperhive_docs_flake: String::new(), nixpkgs_flake: String::new(), claude_code_path: None, 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(), // Slightly below the kernel default of 100, so agent // containers yield to host services + the infra containers // (which carry no drop-in and stay at 100) under contention. agent_cpu_weight: Some(80), agent_io_weight: Some(80), } } } /// On-disk shape of the `hive-c0re serve --config ` 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: PathBuf, /// Manager-editable proposed config repo. pub proposed: PathBuf, /// Hive-c0re-authoritative applied config repo. pub applied: PathBuf, /// Claude OAuth credentials (survives purge boundary). pub claude: PathBuf, /// Agent durable notes + forge token (survives purge boundary). pub notes: PathBuf, } /// Collapse per-pill tombstones to one answer per agent: **was any recently /// cleared pill for this agent a deliberate stop?** /// /// `OR`, not last-write-wins — that is the whole fix. An agent can clear several /// pills inside one grace window (the transient set tests status alone, so a /// lease-exempt `Prebuild` and a lease-holding `StopForUpdate` are both live), /// and taking the last one means an incidental `false` erases a real `true`, /// which the crash watcher then reads as a container **crash**. /// /// Asks the same question of the cleared set that `crash_watch` asks of the /// active set with `.any(…)`, so the two agree by construction. A free function /// so it is testable without a `Coordinator` fixture — same reason /// `crash_watch::is_deliberate_stop` is one. fn fold_tombstones_by_agent<'a>( entries: impl Iterator, ) -> HashMap { let mut out: HashMap = HashMap::new(); for (agent, deliberate) in entries { *out.entry(agent.to_owned()).or_default() |= deliberate; } out } /// 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, }); } } } /// 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, pub status: &'static str, pub note: Option, pub description: Option, } /// 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, pub description: Option, pub pr_number: Option, } /// 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, 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 { let HiveEnv { hyperhive_flake, hyperhive_docs_flake, nixpkgs_flake, claude_code_path, dashboard_port, operator_pronouns, context_window_tokens, agent_cpu_quota, agent_memory_max, agent_cpu_weight, agent_io_weight, } = 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` 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` 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, claude_code_path, dashboard_port, operator_pronouns, context_window_tokens, agent_cpu_quota, agent_memory_max, agent_cpu_weight, agent_io_weight, model_prices, agents: Mutex::new(HashMap::new()), recent_transient: Mutex::new(HashMap::new()), recent_crashes: Mutex::new(HashMap::new()), graceful_stop_pending: Mutex::new(HashSet::new()), pause_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(), claude_code_path: self.claude_code_path.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(), agent_cpu_weight: self.agent_cpu_weight, agent_io_weight: self.agent_io_weight, } } /// Assemble the per-agent filesystem paths for `name`. `agent_dir` /// is the runtime directory (`/run/hyperhive/agents/`), obtained /// from `crate::paths::agent_runtime_dir(name)` (pure path) or from /// `lifecycle::ensure_agent_runtime_dir(name)` when the dir must be /// created. All other paths are derived statically from `name`. #[must_use] pub fn agent_paths(name: &str, agent_dir: PathBuf) -> AgentPaths { // `name` is validated upstream (spawn-approval / enqueue gate / the // MANAGER_NAME const), so an invalid ident here is a construction // bug. This is the step-3 boundary between the Ident-threaded path // builders and the job_queue layer (threaded post hive-jobq cutover). let name = hive_types::Ident::parse(name) .expect("agent_paths: name must be a valid ident (validated at spawn/enqueue)"); AgentPaths { agent: agent_dir, proposed: Self::agent_proposed_dir(&name), applied: crate::paths::applied_dir(name.as_str()), claude: Self::agent_claude_dir(&name), notes: Self::agent_notes_dir(&name), } } /// Emit a `RebuildQueueChanged` tick. Called from the queue mutation /// helpers (`enqueue` / `finish` / `cancel`-adjacent wrappers below) and /// the worker so every state transition surfaces on the dashboard without /// extra plumbing. /// /// Carries no queue payload — clients refetch `/api/jobq/graph`. The name /// keeps `snapshot` because *that* is still what a client ends up with; /// what changed is who serves it. pub fn emit_rebuild_queue_snapshot(self: &Arc) { self.emit_dashboard_event(DashboardEvent::RebuildQueueChanged { seq: self.next_seq(), }); } /// 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 mut schedules: Vec = 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> { self.last_containers .try_lock() .ok() .map(|m| m.keys().cloned().collect()) } /// 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::permissions::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) { use hive_sh4re::permissions::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, }); } /// 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 `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. pub fn emit_approval_added(&self, ev: ApprovalAdded<'_>) { let ApprovalAdded { id, agent, approval_kind, sha_short, description, pr_number, } = ev; self.emit_dashboard_event(DashboardEvent::ApprovalAdded { seq: self.next_seq(), id, agent: agent.to_owned(), approval_kind, sha_short, 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) { let fresh = container_view::build_all(&self.hive_env()).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 topology reparent(s) + fan the resulting notifications out to /// the affected agents. Applies all moves under a single `META_LOCK` /// acquisition (one git commit — a single-move call is just a /// one-element slice) then, for each move that actually changed /// topology, 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::manager::SYSTEM_SENDER` so the dashboard renders /// them under the existing system-source styling. /// /// 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::inbox::Message { from: hive_sh4re::manager::trusted_sender(hive_sh4re::manager::SYSTEM_SENDER), 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::inbox::Message { from: hive_sh4re::manager::trusted_sender(hive_sh4re::manager::SYSTEM_SENDER), 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 = crate::paths::agent_runtime_dir(name); std::fs::create_dir_all(&agent_dir) .with_context(|| format!("create agent dir {}", agent_dir.display()))?; let socket_path = Self::socket_path(name); // Hand the full Coordinator to the per-agent socket — it // needs 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 { self.agents.lock().unwrap().keys().cloned().collect() } /// Emit the "a pill appeared" edge, for the job-queue scheduler publishing /// the transitions of its derived set. pub(crate) fn emit_transient_set(&self, name: &str, label: String) { // Live-update dashboards. `since_unix` is wall-clock so the // browser can tick "Ns spawning…" without polling. The // intra-process map keeps using `Instant` for monotonicity. let since_unix = std::time::SystemTime::now() .duration_since(std::time::UNIX_EPOCH) .ok() .and_then(|d| i64::try_from(d.as_secs()).ok()) .unwrap_or(0); self.emit_dashboard_event(DashboardEvent::TransientSet { seq: self.next_seq(), name: name.to_owned(), transient_kind: label, since_unix, }); } /// Emit the "a pill went away" edge **and stamp the tombstone the crash /// watcher reads**, for the job-queue scheduler. /// /// 🚨 The stamp is not bookkeeping. Without it the clear-then-poll race /// produced a spurious `ContainerCrash` on **every** operator stop/restart: /// the transient is gone by the time the 10s poll looks, so a deliberate /// stop is indistinguishable from a crash. `recent_transient_within` is what /// closes that window, which is why the clear has to be an *event* — a /// derived read of current state cannot answer "was one here a moment ago?". /// /// Old entries are reaped lazily on read, so the map stays bounded. /// `label` identifies *which* pill cleared. An agent can hold several at /// once, so both the tombstone key and the wire event need it — without it /// a client that received two `TransientSet`s cannot tell which one this /// clears, and the tombstone silently overwrites a concurrent pill's /// `deliberate_stop`. pub(crate) fn emit_transient_cleared(&self, name: &str, label: &str, deliberate_stop: bool) { self.recent_transient.lock().unwrap().insert( (name.to_owned(), label.to_owned()), (deliberate_stop, std::time::Instant::now()), ); self.emit_dashboard_event(DashboardEvent::TransientCleared { seq: self.next_seq(), name: name.to_owned(), transient_kind: label.to_owned(), }); } /// Mark `name` as having a graceful stop in progress. While set, /// `socket_server::handle_recv` returns `Response::GracefulStop` for /// this agent instead of polling the broker (the inbound fence). pub fn mark_graceful_stop(&self, name: &str) { self.graceful_stop_pending .lock() .unwrap() .insert(name.to_owned()); } /// Whether a graceful stop is pending for `name`. #[must_use] pub fn is_graceful_stop_pending(&self, name: &str) -> bool { self.graceful_stop_pending.lock().unwrap().contains(name) } /// Clear the graceful-stop flag for `name` (agent reported /// `GracefulStopComplete`, or the stop finished / was abandoned). pub fn clear_graceful_stop(&self, name: &str) { self.graceful_stop_pending.lock().unwrap().remove(name); } /// Mark `name` as having a pause acknowledgement in progress. Set by /// the pause DAG's signal node; the drain node polls /// `is_pause_pending` until the agent reports back or the timeout /// fallback fires. pub fn mark_pause_pending(&self, name: &str) { self.pause_pending.lock().unwrap().insert(name.to_owned()); } /// Whether a pause acknowledgement is pending for `name`. #[must_use] pub fn is_pause_pending(&self, name: &str) -> bool { self.pause_pending.lock().unwrap().contains(name) } /// Clear the pause-pending flag for `name` (agent reported /// `PauseAcknowledged`, or the drain node's wait finished / timed out). pub fn clear_pause_pending(&self, name: &str) { self.pause_pending.lock().unwrap().remove(name); } /// 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) { 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> { 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::>(&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 } /// Per-agent `deliberate_stop` for transients cleared within the last /// `grace` seconds — i.e. agents the operator just acted on, whose stop the /// crash watcher should NOT classify as a crash. Lazily reaps entries older /// than `grace` so the map stays bounded by the active agent count. /// /// Carries only the safety bit, not the display label: nothing downstream /// should be able to re-derive a stop/crash decision from a pill's wording. pub fn recent_transient_within(&self, grace: std::time::Duration) -> HashMap { let now = std::time::Instant::now(); let mut map = self.recent_transient.lock().unwrap(); map.retain(|_, (_, ts)| now.duration_since(*ts) <= grace); fold_tombstones_by_agent(map.iter().map(|((a, _), (d, _))| (a.as_str(), *d))) } /// 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 { 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() } /// Every live transient, keyed by agent. /// /// **Derived on read, stored nowhere.** Straight off the running graph, so /// there is no cached copy to go stale, leak, or disagree with what is /// actually running. /// /// Migration is the last operation with no queue node behind it, so it /// shows **no pill** — there is nothing in the graph to derive one from. /// It gets one for free once it becomes real nodes, the way destroy did. /// /// ⚠️ **A `Vec` per agent, not one entry.** `running_transients` tests /// status alone, so a lease-exempt `Prebuild` for `a` and a lease-holding /// `StopForUpdate` for `a` are both live pills. Collapsing them to one /// would pick arbitrarily and, for the crash watcher, silently lose a /// `deliberate_stop = true` behind a `false` — reporting an intentional /// stop as a crash. #[must_use] pub fn transient_snapshot(&self) -> HashMap> { let mut out: HashMap> = HashMap::new(); for t in self.job_queue.running_transients() { out.entry(t.agent.clone()).or_default().push(t); } out } /// Drop a system message into the given agent's inbox. Wakes the /// turn loop with a "you were just (re)started" hint — operator /// caused the transition, agent picks up where it left off /// (notes are in the bind-mounted state dir, last turn is in /// --continue's session). Best-effort; broker errors are logged /// but don't propagate. pub fn kick_agent(&self, name: &str, reason: &str) { // 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::inbox::Message { from: hive_sh4re::manager::trusted_sender(hive_sh4re::manager::SYSTEM_SENDER), to: name.to_owned(), body, in_reply_to: None, }) { tracing::warn!(error = ?e, %name, "kick_agent: broker.send failed"); } } /// 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::manager::HelperEvent) { self.notify_agent(hive_sh4re::manager::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::manager::HelperEvent) { let target = self.submitter_or_manager(approval_id); self.notify_agent(&target, event); } /// Shared resolution for "which agent should hear about approval /// `approval_id`" — the authenticated socket caller at submit time, /// falling back to the manager for legacy rows with no recorded /// submitter (or any lookup failure). Used by both `notify_submitter` /// and `push_todo_submitter` so the fallback rule lives in one place. fn submitter_or_manager(&self, approval_id: i64) -> String { self.approvals .submitter_of(approval_id) .unwrap_or_default() .unwrap_or_else(|| hive_sh4re::manager::MANAGER_AGENT.to_owned()) } /// Push a todo directly into `agent`'s in-container todo store — a /// best-effort *live* dial of its `hive-agent-sock` socket /// (`hive_host_sock::agent_todo_socket`), same `UpsertTodo` request /// shape the built-in in-container producers (matrix/bash/forge-notify) /// already send — `subsystem` isn't restricted to that set, any /// producer can use the same shape. This is the migration target for /// `HelperEvent` variants that are pure "FYI, check when convenient" notices: the /// event stops being a broker `Message` (which always drives an /// immediate turn) and becomes a todo instead, with the same /// dedup-by-key semantics as any other producer. /// /// Deliberately push, not queue: if the agent's container is down /// (socket file absent) or the dial otherwise fails, this is a /// silent no-op (logged at `debug`/`warn`) — no retry, no fallback /// delivery. An agent that's down doesn't need a todo about /// something it'll never see appear this way; whatever mechanism /// resurfaces its state on the next boot is unrelated to this path. /// /// Returns `Ok(())` on a successful push and `Err(reason)` — a /// short human-readable string, already logged at `debug`/`warn` /// by this function — on any of the silent-no-op cases below. /// Most callers are pure fire-and-forget notices and ignore it /// (`let _ = coord.push_todo(...).await;`); callers that track a /// per-target delivery outcome (e.g. the scheduled-prompts worker's /// `last_result` column) use it instead of assuming success. /// `reopen_if_acked` forwards to `UpsertTodo` (see its doc comment) — /// `false` for every caller here except the scheduled-prompts worker. pub async fn push_todo( &self, agent: &str, subsystem: &str, key: Option, summary: String, source: Option, reopen_if_acked: bool, ) -> Result<(), String> { let Ok(ident) = hive_types::Ident::parse(agent) else { tracing::warn!(%agent, "push_todo: not a valid agent ident, skipping"); return Err(format!("'{agent}' is not a valid agent ident")); }; let path = hive_host_sock::agent_todo_socket(&ident); if !path.exists() { tracing::debug!(%agent, path = %path.display(), "push_todo: agent socket not present (offline?), skipping"); return Err(format!("agent '{agent}' socket not present (offline?)")); } let req = hive_agent_sock::Request::UpsertTodo { subsystem: subsystem.to_owned(), key, summary, source, reopen_if_acked, }; match hive_sock_client::request::<_, hive_agent_sock::Response>( &path, &req, hive_sock_client::Retry::None, ) .await { Ok(hive_agent_sock::Response::Err { message }) => { tracing::warn!(%agent, %message, "push_todo: agent rejected the todo"); Err(format!("agent '{agent}' rejected the todo: {message}")) } Err(e) => { tracing::warn!(%agent, error = ?e, "push_todo: dial failed"); Err(format!("push_todo dial to '{agent}' failed: {e:#}")) } Ok(_) => Ok(()), } } /// `push_todo` to whichever agent submitted approval `approval_id` — /// same resolution `notify_submitter` uses. Every caller here is a /// one-shot approval-resolution notice, so `reopen_if_acked` is /// unconditionally `false` — there's no reconciler/event distinction /// to make for an event that only ever fires once. pub async fn push_todo_submitter( &self, approval_id: i64, subsystem: &str, key: Option, summary: String, source: Option, ) -> Result<(), String> { let target = self.submitter_or_manager(approval_id); self.push_todo(&target, subsystem, key, summary, source, false) .await } /// 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::manager::HelperEvent) { self.notify_agent_from(hive_sh4re::manager::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::manager::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::inbox::Message { from: hive_sh4re::manager::trusted_sender(from), 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::inbox::Message { from: hive_sh4re::manager::trusted_sender(from), to: agent_name.clone(), body: broadcast_body.clone(), in_reply_to: None, }) { errors.push(format!("{agent_name}: {e}")); } } errors } pub fn socket_path(name: &str) -> PathBuf { crate::paths::agent_runtime_dir(name).join("mcp.sock") } /// The *proposed* config repo: where a config change lands before it is /// applied, and what an approved deploy promotes into `applied_dir`. /// /// **Not bind-mounted into any container.** An agent that edits a config /// clones it from the forge itself; `/agents//config` shows the /// applied (deployed) tree instead — see `config_bind_source` in /// `lifecycle/host_config.rs`. pub fn agent_proposed_dir(name: &hive_types::Ident) -> PathBuf { crate::paths::agent_state_dir(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: &hive_types::Ident) -> PathBuf { crate::paths::agent_state_dir(name).join("claude") } /// Per-agent durable knowledge dir. Bind-mounted RW into the agent /// container at `/agents/{name}/state`. Survives destroy/recreate. /// Agent-visible — claude is told to write long-lived notes here. pub fn agent_notes_dir(name: &hive_types::Ident) -> PathBuf { crate::paths::agent_state_dir(name).join("state") } /// Per-agent harness-internal state dir. Bind-mounted RW into the /// agent container at `/agents/{name}/harness`. Holds sqlite dbs /// and config files owned by the harness (`hyperhive-events.sqlite`, /// `hyperhive-turn-stats.sqlite`, `hyperhive-model`) — kept separate /// from the agent-visible `state/` so claude's "my notes" view is /// uncluttered and the host vacuum has a clean sweep root. pub fn agent_harness_dir(name: &hive_types::Ident) -> PathBuf { crate::paths::agent_state_dir(name).join("harness") } /// Host-side path of the pause marker — the same file the harness /// resolves in-container via `hive_agent::paths::paused_marker`, /// reached through the harness bind-mount. Its presence means the /// agent's turn loop is parked: the harness still serves its web UI /// and MCP daemons, but drives no turns, so inbox messages queue up /// unacked until the marker is removed. /// /// Both sides only ever *stat* or create/remove this file, so there /// is no protocol between them and pause survives a container /// restart (and can be set on a stopped container). pub fn agent_paused_marker(name: &hive_types::Ident) -> PathBuf { Self::agent_harness_dir(name).join(hive_sh4re::paths::PAUSED_MARKER_FILE) } /// Whether `name` is currently paused. A stat error (missing agent /// dir, permissions) reads as "not paused" — the pause indicator is /// advisory on the host side, and the harness is the component that /// actually enforces it. #[must_use] pub fn is_paused(name: &hive_types::Ident) -> bool { Self::agent_paused_marker(name).exists() } /// Create or remove the pause marker, **via hive-priv**. Idempotent in /// both directions: pausing an already-paused agent (or resuming a /// running one) is a no-op rather than an error, so the dashboard /// toggle and `hivectl pause|resume` don't have to read-then-write. /// /// The write cannot happen in-process. The harness dir is chowned to /// the agent user on the container's first boot (mode 0755), and this /// daemon runs as the unprivileged `hive-core` user — so the read side /// ([`Self::is_paused`], a stat) works while a direct `fs::write` here /// fails with `EACCES` on every agent that has ever booted. The root /// helper owns both directions instead. /// /// # Errors /// /// Returns the hive-priv error when the marker can't be created (agent /// dir missing and un-creatable, disk full) or can't be removed, and /// the transport error when the helper is unreachable. A `NotFound` on /// removal is *not* an error — that's the idempotent resume case. pub async fn set_paused(name: &hive_types::Ident, paused: bool) -> anyhow::Result<()> { crate::priv_client::set_agent_paused(name.as_str(), paused).await } /// 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(crate::paths::agents_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| hive_types::Ident::parse(&e.file_name().into_string().ok()?).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 { Self::kept_state_names() .into_iter() .filter(|n| { Self::agent_proposed_dir(n).join(".git").exists() && !crate::paths::applied_dir(n.as_str()).join(".git").exists() }) .collect() } } /// `push_todo`/`push_todo_submitter` summary text for a rebuild outcome — /// shared by the two `Rebuilt` call sites (`actions::finish_approval`'s /// `MergeConfigPr` arm, `job_queue::exec::run_emit_rebuilt`) so the wording /// stays identical regardless of which path fired. Pure + independently /// testable, unlike the old `HelperEvent::Rebuilt`'s separate `sha`/`tag` /// fields — those become part of the summary text itself now, since a todo /// carries one string, not a structured payload. #[must_use] pub fn rebuilt_todo_summary( agent: &str, ok: bool, note: Option<&str>, sha: Option<&str>, tag: Option<&str>, ) -> String { use std::fmt::Write as _; if !ok { return format!( "agent '{agent}' rebuild FAILED: {}", note.unwrap_or("unknown error") ); } let mut summary = format!("agent '{agent}' rebuilt"); if let Some(sha) = sha { let _ = write!(summary, " @ {sha}"); } if let Some(tag) = tag { let _ = write!(summary, " ({tag})"); } summary } #[cfg(test)] mod tombstone_fold_tests { use super::fold_tombstones_by_agent; #[test] fn a_deliberate_stop_survives_an_incidental_clear_on_the_same_agent() { // The regression this fold exists for. `a` clears two pills in one // grace window: a `StopForUpdate` that really does take the container // down, and a lease-exempt `Prebuild` that doesn't. Keyed by agent // alone the second overwrote the first and the crash watcher reported // a deliberate stop as a crash — order must not matter. let out = fold_tombstones_by_agent([("a", true), ("a", false)].into_iter()); assert_eq!(out.get("a"), Some(&true)); let reversed = fold_tombstones_by_agent([("a", false), ("a", true)].into_iter()); assert_eq!(reversed.get("a"), Some(&true)); } #[test] fn all_incidental_clears_stay_false() { // The other direction has to keep working: nothing deliberate cleared, // so a container going down now really is a crash. let out = fold_tombstones_by_agent([("a", false), ("a", false)].into_iter()); assert_eq!(out.get("a"), Some(&false)); } #[test] fn agents_do_not_bleed_into_each_other() { let out = fold_tombstones_by_agent([("a", true), ("b", false)].into_iter()); assert_eq!(out.get("a"), Some(&true)); assert_eq!(out.get("b"), Some(&false)); assert_eq!(out.get("c"), None); } } #[cfg(test)] mod rebuilt_todo_summary_tests { use super::rebuilt_todo_summary; #[test] fn failure_reports_the_note() { assert_eq!( rebuilt_todo_summary("iris", false, Some("build failed"), None, None), "agent 'iris' rebuild FAILED: build failed" ); } #[test] fn failure_without_a_note_says_unknown() { assert_eq!( rebuilt_todo_summary("iris", false, None, None, None), "agent 'iris' rebuild FAILED: unknown error" ); } #[test] fn success_with_no_sha_or_tag_is_bare() { assert_eq!( rebuilt_todo_summary("iris", true, None, None, None), "agent 'iris' rebuilt" ); } #[test] fn success_with_sha_and_tag_carries_both() { assert_eq!( rebuilt_todo_summary("iris", true, None, Some("abc123"), Some("deployed/42")), "agent 'iris' rebuilt @ abc123 (deployed/42)" ); } #[test] fn success_with_sha_only() { assert_eq!( rebuilt_todo_summary("iris", true, None, Some("abc123"), None), "agent 'iris' rebuilt @ abc123" ); } }