80f16094f1
Closes the #3110 split — lib.rs is now just the crate doc comment and the pub mod list. journal.rs's new doc comment fixes a pre-existing bug: the old JournalPriority doc text in lib.rs was actually half Capability's doc (a leftover from an earlier reorder that moved the code but not the comment above it).
1759 lines
78 KiB
Rust
1759 lines
78 KiB
Rust
//! 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<Broker>,
|
|
pub approvals: Arc<Approvals>,
|
|
pub questions: Arc<OperatorQuestions>,
|
|
/// Scheduled-prompts queue. One sqlite connection,
|
|
/// internal mutex; the worker drains due rows and the manager
|
|
/// handlers insert / cancel through the same handle.
|
|
pub scheduled_prompts: Arc<crate::scheduled_prompts::ScheduledPrompts>,
|
|
/// Full build-log capture. `lifecycle::run` /
|
|
/// `lifecycle::prebuild_toplevel` `start()` a row per attempt,
|
|
/// pipe every stdout/stderr line into it, and `finish()` it on
|
|
/// child exit. Dashboard reads it via `list_recent_for_agent` /
|
|
/// `get_full` for the per-card chip + side-panel viewer. See
|
|
/// `build_logs.rs` for retention.
|
|
pub build_logs: Arc<crate::build_logs::BuildLogs>,
|
|
/// Audit trail of agent-initiated privileged actions (infra restart,
|
|
/// …). See `audit_log.rs`. Same dir as `build_logs`.
|
|
pub audit_log: Arc<crate::audit_log::AuditLog>,
|
|
/// URL of the hyperhive flake (no fragment). Inlined into per-agent
|
|
/// `flake.nix` files as `inputs.hyperhive.url`.
|
|
pub hyperhive_flake: String,
|
|
/// URL of the narrow `docs/` source (no fragment). Inlined into the
|
|
/// meta `flake.nix` as `inputs.hyperhive-docs.url` and threaded to
|
|
/// each agent as `hyperhive.docs.source`. Its own store path
|
|
/// so doc edits don't re-hash `hyperhive_flake`.
|
|
pub hyperhive_docs_flake: String,
|
|
/// Store-path URL of the nixpkgs to wire into the meta flake as
|
|
/// `inputs.nixpkgs.url`. Populated by `--nixpkgs-flake` (set by the
|
|
/// NixOS module to `"path:${pkgs.path}"` so the meta flake always
|
|
/// tracks the same nixpkgs the host evaluated with — which is the
|
|
/// host's nixpkgs when `inputs.hyperhive.inputs.nixpkgs.follows =
|
|
/// "nixpkgs"` is set in the host flake). Empty string = legacy
|
|
/// `follows = "hyperhive/nixpkgs"` behaviour.
|
|
pub nixpkgs_flake: String,
|
|
/// 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<String>,
|
|
/// TCP port the host's hive-c0re dashboard listens on. Inlined into
|
|
/// each per-agent flake so the agent's web UI can build the right
|
|
/// rebuild-button URL pointing back at the dashboard.
|
|
pub dashboard_port: u16,
|
|
/// Operator pronouns (free text) — `she/her` by default, set via
|
|
/// the NixOS module option `services.hive-c0re.operatorPronouns`.
|
|
/// Reaches each container as the `HIVE_OPERATOR_PRONOUNS` env var
|
|
/// (injected into systemd.services.<harness>.environment by the
|
|
/// meta flake); the harness substitutes it into the agent /
|
|
/// manager system prompt at boot.
|
|
pub operator_pronouns: String,
|
|
/// Per-model context-window sizes in tokens. Set via the host-level
|
|
/// `services.hive-c0re.contextWindowTokens` NixOS option; injected
|
|
/// into each container as `HIVE_CONTEXT_WINDOW_TOKENS_<KEY_UPPER>`
|
|
/// by the meta flake renderer. The harness uses these to derive
|
|
/// compaction / auto-reset watermarks and exposes the active value
|
|
/// on `/api/state` as `context_window_tokens`.
|
|
pub context_window_tokens: std::collections::HashMap<String, u64>,
|
|
/// Per-agent systemd `CPUQuota=` value (e.g. `"200%"`). Written into
|
|
/// the `container@h-<name>.service.d/` drop-in on every spawn/rebuild.
|
|
pub agent_cpu_quota: String,
|
|
/// Per-agent systemd `MemoryMax=` value (e.g. `"4G"`). Same drop-in.
|
|
pub agent_memory_max: String,
|
|
/// Per-agent systemd `CPUWeight=` (cgroup v2 `cpu.weight`, 1..=10000).
|
|
/// Same drop-in. A *relative share* under contention, not a cap — see
|
|
/// the `agentCpuWeight` NixOS option for the full semantics. `None`
|
|
/// (option set to `null`) omits the setting: kernel default, and the
|
|
/// drop-in is byte-identical to the one written before weights existed.
|
|
pub agent_cpu_weight: Option<u32>,
|
|
/// Per-agent systemd `IOWeight=` (cgroup v2 `io.weight`, 1..=10000).
|
|
/// Same drop-in, same relative-share and `None` semantics as
|
|
/// [`Self::agent_cpu_weight`].
|
|
pub agent_io_weight: Option<u32>,
|
|
/// Operator-tunable model→price table backing the hive-wide cost
|
|
/// estimate on the ST4TS tab. Set via `services.hyperhive.modelPrices`
|
|
/// and passed to `hive-c0re serve --model-prices <json>`. Models not
|
|
/// in the table fall back to the built-in estimate in `hive_stats`.
|
|
/// hive-c0re-local (read only by the `/api/stats-hive` handler), so —
|
|
/// unlike `context_window_tokens` — it is *not* part of `HiveEnv` and
|
|
/// is never injected into containers.
|
|
pub model_prices: crate::hive_stats::PriceTable,
|
|
agents: Mutex<HashMap<String, AgentSocket>>,
|
|
/// Agents whose lifecycle action (currently just spawn) is in flight.
|
|
/// Read by the dashboard to render a spinner; cleared when the action
|
|
/// resolves (success or failure).
|
|
/// Agents whose container is being taken down by work with **no queue node
|
|
/// behind it** (destroy, migration), so the crash watcher must not report
|
|
/// the disappearance as a crash. Not a pill — see [`CrashWatchSuppression`].
|
|
crash_suppressed: Mutex<HashSet<String>>,
|
|
/// 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**. The
|
|
/// out-of-band suppression guard, which has no node behind it, uses
|
|
/// [`NO_NODE_LABEL`].
|
|
recent_transient: Mutex<HashMap<(String, String), (bool, std::time::Instant)>>,
|
|
/// Timestamps of recent unexpected container crashes, keyed by agent.
|
|
/// Fed by `crash_watch` each time it classifies a stop as a crash (so
|
|
/// a crash-looping container — which `Restart=on-failure` flips back
|
|
/// to running between polls — accumulates one entry per down-transition,
|
|
/// not just whatever its point-in-time state happens to be). Read by
|
|
/// the dashboard's `agents_crashing` banner warning via
|
|
/// `recent_crash_counts`, which prunes entries older than its window.
|
|
recent_crashes: Mutex<HashMap<String, Vec<std::time::Instant>>>,
|
|
/// Agents with a graceful stop in progress. Set by the `GracefulStop`
|
|
/// orchestration; read by `socket_server::handle_recv`, which returns
|
|
/// `Response::GracefulStop` (instead of polling the broker) while an
|
|
/// agent is in this set — the inbound fence. Cleared when the agent
|
|
/// reports `GracefulStopComplete` or the container is stopped.
|
|
graceful_stop_pending: Mutex<HashSet<String>>,
|
|
/// Logical agent names that were running at the last broad-scope
|
|
/// `hivectl stop`. A subsequent broad-scope `hivectl start` restores
|
|
/// only this set (intersected with the requested scope) rather than
|
|
/// every configured container, so agents the operator intentionally
|
|
/// left stopped stay stopped. `None` when no broad stop has happened
|
|
/// since the last start (or since daemon boot) — start then falls
|
|
/// back to "start all". Mirrored to the broker `kv` table
|
|
/// (`last_stopped_running` key) so the snapshot also survives a
|
|
/// hive-c0re restart between the stop-all and the start: this
|
|
/// in-memory copy is the fast path, the persisted copy the backstop.
|
|
last_stopped_running: Mutex<Option<Vec<String>>>,
|
|
/// Unified wire-facing event channel feeding the dashboard SSE
|
|
/// stream. Carries broker messages (mirrored from `broker.subscribe`
|
|
/// by the forwarder task in `main.rs`) and dashboard-only mutation
|
|
/// events (approval added/resolved, question added/answered, etc.).
|
|
/// Snapshot endpoints capture `event_seq` before reading state so
|
|
/// the client can dedupe its buffered live traffic against the
|
|
/// snapshot.
|
|
dashboard_events: broadcast::Sender<DashboardEvent>,
|
|
event_seq: AtomicU64,
|
|
/// Count of dashboard-triggered `meta-update` runs currently in
|
|
/// flight. `post_meta_update` returns 200 immediately and does the
|
|
/// multi-minute `nix flake update` + agent-rebuild ripple in a
|
|
/// background task, so without this the META INPUTS panel showed
|
|
/// no sign anything was happening. Held via
|
|
/// `MetaUpdateGuard`; a count > 0 surfaces on `/api/state` as
|
|
/// `meta_update_running` and via the `MetaUpdateRunning` event.
|
|
meta_updates_active: AtomicU64,
|
|
/// Last container snapshot seen by `rescan_containers_and_emit`,
|
|
/// keyed by `ContainerView.name`. The rescan diffs a fresh
|
|
/// `container_view::build_all` against this map and emits one
|
|
/// `ContainerStateChanged` per added/changed row and one
|
|
/// `ContainerRemoved` per disappeared row. Async — guarded by a
|
|
/// tokio mutex so the rescan can `await` `lifecycle::list` /
|
|
/// `is_running` without blocking other coordinator paths.
|
|
last_containers: tokio::sync::Mutex<HashMap<String, ContainerView>>,
|
|
/// Global job-DAG queue. Every container/meta op (rebuild,
|
|
/// meta-update, first-spawn, power changes) is submitted as a DAG
|
|
/// of primitive nodes; a single scheduler drives them with
|
|
/// build-slot + per-agent-lease gating so the dashboard renders one
|
|
/// ordered view of pending + running work. See `job_queue/` for
|
|
/// the dedup rules, resource classes, and history retention.
|
|
pub job_queue: Arc<crate::job_queue::JobQueue>,
|
|
/// Durable per-agent power intent (`wanted: Up | Offline`) — the
|
|
/// spec half of desired-state reconciliation; the queue's
|
|
/// `Reconcile` nodes converge observed state to it.
|
|
pub power: Arc<crate::power::PowerStore>,
|
|
/// Shutdown signal broadcast to all background tasks. Sending
|
|
/// `true` asks every loop to exit after its current work item.
|
|
/// Use `shutdown_rx()` to subscribe; `request_shutdown()` to fire.
|
|
shutdown_tx: watch::Sender<bool>,
|
|
}
|
|
|
|
/// Hive-wide configuration that lifecycle and meta operations need.
|
|
/// Extracted from `Coordinator` so callers can pass a single struct
|
|
/// instead of repeating the same 6 arguments everywhere.
|
|
///
|
|
/// Cloned from `Coordinator` via [`Coordinator::hive_env`]. All fields
|
|
/// are cheap to clone (small strings + small map); lifecycle ops are
|
|
/// infrequent enough that the copy cost is irrelevant.
|
|
///
|
|
/// Also the container-injected subset of the `hive-c0re serve --config`
|
|
/// JSON (see [`ServeConfig`]): `#[serde(default)]` lets a partial config
|
|
/// fall back to [`Default`] (the canonical defaults that used to live in
|
|
/// the clap `default_value`s), and bare `hive-c0re serve` with no
|
|
/// `--config` boots from `Default`.
|
|
#[derive(Clone, Debug, serde::Deserialize)]
|
|
#[serde(default)]
|
|
pub struct HiveEnv {
|
|
pub hyperhive_flake: String,
|
|
/// Store-path URL of the narrow `docs/` source, wired into the meta
|
|
/// flake as `inputs.hyperhive-docs.url` and threaded to each agent as
|
|
/// `hyperhive.docs.source`. Separate from `hyperhive_flake`
|
|
/// so a doc edit only re-locks this input, not the whole source.
|
|
pub hyperhive_docs_flake: String,
|
|
pub nixpkgs_flake: String,
|
|
/// 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<String>,
|
|
pub dashboard_port: u16,
|
|
pub operator_pronouns: String,
|
|
pub context_window_tokens: std::collections::HashMap<String, u64>,
|
|
/// Per-agent systemd `CPUQuota=` value (e.g. `"200%"`).
|
|
pub agent_cpu_quota: String,
|
|
/// Per-agent systemd `MemoryMax=` value (e.g. `"4G"`).
|
|
pub agent_memory_max: String,
|
|
/// Per-agent systemd `CPUWeight=` — cgroup v2 `cpu.weight`, 1..=10000.
|
|
/// Relative share under contention, not a cap; see `agentCpuWeight`.
|
|
/// `None` = leave the setting out of the drop-in (kernel default).
|
|
pub agent_cpu_weight: Option<u32>,
|
|
/// Per-agent systemd `IOWeight=` — cgroup v2 `io.weight`, 1..=10000.
|
|
/// Same semantics as [`Self::agent_cpu_weight`]; see `agentIoWeight`.
|
|
pub agent_io_weight: Option<u32>,
|
|
}
|
|
|
|
impl Default for HiveEnv {
|
|
fn default() -> Self {
|
|
Self {
|
|
hyperhive_flake: "/etc/hyperhive".to_string(),
|
|
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 <file>` JSON: the
|
|
/// container-injected [`HiveEnv`] (flattened) plus the hive-c0re-local
|
|
/// `model_prices` table (read only by the `/api/stats-hive` handler, never
|
|
/// injected into containers — hence kept out of `HiveEnv`). The NixOS
|
|
/// module writes this so the `ExecStart` carries one `--config` flag
|
|
/// instead of every host-level setting as its own JSON-blob argument.
|
|
/// `#[serde(default)]` lets any field be omitted and fall back to its
|
|
/// canonical default.
|
|
#[derive(Clone, Debug, serde::Deserialize)]
|
|
#[serde(default)]
|
|
pub struct ServeConfig {
|
|
#[serde(flatten)]
|
|
pub env: HiveEnv,
|
|
pub model_prices: crate::hive_stats::PriceTable,
|
|
/// Number of concurrent nix-heavy job-queue nodes (prebuild /
|
|
/// profile-swap / create / meta lock). hive-c0re-local like
|
|
/// `model_prices` — never injected into containers. Set via
|
|
/// `services.hyperhive.c0re.buildSlots`.
|
|
pub build_slots: usize,
|
|
}
|
|
|
|
impl Default for ServeConfig {
|
|
fn default() -> Self {
|
|
Self {
|
|
env: HiveEnv::default(),
|
|
model_prices: crate::hive_stats::PriceTable::default(),
|
|
build_slots: 1,
|
|
}
|
|
}
|
|
}
|
|
|
|
#[cfg(test)]
|
|
mod serve_config_tests {
|
|
use super::ServeConfig;
|
|
|
|
#[test]
|
|
fn deserialises_flat_with_per_field_defaults() {
|
|
// The flattened HiveEnv fields + the local model_prices all live at
|
|
// the JSON top level. Present fields are read; omitted ones fall
|
|
// back to their canonical defaults (so a partial config is valid).
|
|
let sc: ServeConfig = serde_json::from_str(
|
|
r#"{
|
|
"hyperhive_flake": "/x",
|
|
"dashboard_port": 1234,
|
|
"model_prices": {
|
|
"opus": {"input": 5.0, "output": 25.0, "cache_read": 0.5, "cache_write": 10.0}
|
|
}
|
|
}"#,
|
|
)
|
|
.expect("flat ServeConfig should deserialise");
|
|
assert_eq!(sc.env.hyperhive_flake, "/x");
|
|
assert_eq!(sc.env.dashboard_port, 1234);
|
|
// omitted → HiveEnv::default()
|
|
assert_eq!(sc.env.operator_pronouns, "she/her");
|
|
assert_eq!(sc.env.agent_cpu_quota, "200%");
|
|
assert_eq!(sc.env.context_window_tokens.get("sonnet"), Some(&1_000_000));
|
|
assert!(sc.model_prices.contains_key("opus"));
|
|
}
|
|
|
|
#[test]
|
|
fn empty_object_is_all_defaults() {
|
|
let sc: ServeConfig = serde_json::from_str("{}").expect("empty config valid");
|
|
assert_eq!(sc.env.dashboard_port, 7000);
|
|
assert_eq!(sc.env.hyperhive_flake, "/etc/hyperhive");
|
|
assert!(sc.model_prices.is_empty());
|
|
}
|
|
}
|
|
|
|
/// Per-agent filesystem paths that lifecycle operations write to.
|
|
/// Assembled from `Coordinator`'s static path helpers so callers
|
|
/// don't repeat the same `Coordinator::agent_*_dir(name)` calls.
|
|
#[derive(Clone, Debug)]
|
|
pub struct AgentPaths {
|
|
/// Runtime socket dir (tmpfs, recreated per boot).
|
|
pub agent: 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<Item = (&'a str, bool)>,
|
|
) -> HashMap<String, bool> {
|
|
let mut out: HashMap<String, bool> = HashMap::new();
|
|
for (agent, deliberate) in entries {
|
|
*out.entry(agent.to_owned()).or_default() |= deliberate;
|
|
}
|
|
out
|
|
}
|
|
|
|
/// Tombstone label for work with **no queue node behind it** — the
|
|
/// out-of-band operations (destroy, migration) that hold a
|
|
/// [`CrashWatchGuard`] instead of appearing in the derived transient set.
|
|
///
|
|
/// [`Coordinator::recent_transient`] is keyed by `(agent, label)` so concurrent
|
|
/// pills can't overwrite each other's `deliberate_stop`; a guard has no node and
|
|
/// therefore no node label, so it needs one of its own. Angle-bracketed to keep
|
|
/// it out of the `NodeKind::as_str` namespace — no node can ever render this.
|
|
const NO_NODE_LABEL: &str = "<no-node>";
|
|
|
|
/// RAII handle returned by [`Coordinator::suppress_crash_watch`]. While held,
|
|
/// the crash watcher treats this container disappearing as **expected**.
|
|
///
|
|
/// This is *not* a dashboard pill. Transients are derived from running queue
|
|
/// nodes and nothing stores them. But destroy and migration take a container
|
|
/// down without a node behind them, so nothing in the graph says the
|
|
/// disappearance was intended — and without that, `crash_watch` fires a
|
|
/// `ContainerCrash` for every destroy and every migrated agent, and the manager
|
|
/// tries to "recover" containers that were removed on purpose.
|
|
///
|
|
/// It is held rather than stamped once because
|
|
/// [`crate::workers::crash_watch`]'s grace window is finite and these
|
|
/// operations are not: a long destroy would outlive a single tombstone. The
|
|
/// tombstone is stamped on drop, covering the poll that lands just after.
|
|
///
|
|
/// Goes away entirely once destroy + migration are real queue nodes.
|
|
#[must_use = "suppression lasts as long as the guard; bind it for the operation's duration \
|
|
(`let _guard = coord.suppress_crash_watch(...)`). An unbound call drops it \
|
|
immediately and the very next poll can report a deliberate stop as a crash."]
|
|
pub struct CrashWatchSuppression {
|
|
coord: Arc<Coordinator>,
|
|
name: String,
|
|
}
|
|
|
|
impl Drop for CrashWatchSuppression {
|
|
fn drop(&mut self) {
|
|
self.coord
|
|
.crash_suppressed
|
|
.lock()
|
|
.unwrap()
|
|
.remove(&self.name);
|
|
// Tombstone the release so the next poll — which may land in the
|
|
// window between the container going away and this guard dropping —
|
|
// still reads the stop as deliberate.
|
|
self.coord.recent_transient.lock().unwrap().insert(
|
|
(self.name.clone(), NO_NODE_LABEL.to_owned()),
|
|
(true, std::time::Instant::now()),
|
|
);
|
|
}
|
|
}
|
|
|
|
/// RAII guard for the `meta-update` in-progress flag, held for the
|
|
/// duration of a `run_meta_update` background task. Created by
|
|
/// `Coordinator::meta_update_guard`. Drop decrements the active-run
|
|
/// count; the count crossing back to 0 emits
|
|
/// `MetaUpdateRunning { running: false }`, so a concurrent pair of
|
|
/// updates only flips the dashboard flag once.
|
|
pub struct MetaUpdateGuard {
|
|
coord: Arc<Coordinator>,
|
|
}
|
|
|
|
impl Drop for MetaUpdateGuard {
|
|
fn drop(&mut self) {
|
|
if self
|
|
.coord
|
|
.meta_updates_active
|
|
.fetch_sub(1, Ordering::SeqCst)
|
|
== 1
|
|
{
|
|
self.coord
|
|
.emit_dashboard_event(DashboardEvent::MetaUpdateRunning {
|
|
seq: self.coord.next_seq(),
|
|
running: false,
|
|
});
|
|
}
|
|
}
|
|
}
|
|
|
|
/// Field-named payload for [`Coordinator::emit_approval_resolved`].
|
|
/// Mirrors the `ApprovalResolved` dashboard-event fields. `agent`
|
|
/// borrows from the caller; `approval_kind` / `status` are
|
|
/// compile-time constants.
|
|
pub struct ApprovalResolved<'a> {
|
|
pub id: i64,
|
|
pub agent: &'a str,
|
|
pub approval_kind: &'static str,
|
|
pub sha_short: Option<String>,
|
|
pub status: &'static str,
|
|
pub note: Option<String>,
|
|
pub description: Option<String>,
|
|
}
|
|
|
|
/// Field-named payload for [`Coordinator::emit_approval_added`].
|
|
/// Mirrors the `ApprovalAdded` dashboard-event fields. `agent`
|
|
/// borrows from the caller; `approval_kind` is a compile-time
|
|
/// constant. `pr_number` is set for `merge_config_pr` only.
|
|
pub struct ApprovalAdded<'a> {
|
|
pub id: i64,
|
|
pub agent: &'a str,
|
|
pub approval_kind: &'static str,
|
|
pub sha_short: Option<String>,
|
|
pub description: Option<String>,
|
|
pub pr_number: Option<u64>,
|
|
}
|
|
|
|
/// Field-named payload for [`Coordinator::emit_question_added`].
|
|
/// Mirrors the `QuestionAdded` dashboard-event fields; all references
|
|
/// share the caller's lifetime.
|
|
pub struct QuestionAdded<'a> {
|
|
pub id: i64,
|
|
pub asker: &'a str,
|
|
pub question: &'a str,
|
|
pub options: &'a [String],
|
|
pub multi: bool,
|
|
pub deadline_at: Option<i64>,
|
|
pub target: Option<&'a str>,
|
|
}
|
|
|
|
impl Coordinator {
|
|
pub fn open(
|
|
db_path: &Path,
|
|
env: HiveEnv,
|
|
model_prices: crate::hive_stats::PriceTable,
|
|
build_slots: usize,
|
|
) -> Result<Self> {
|
|
let HiveEnv {
|
|
hyperhive_flake,
|
|
hyperhive_docs_flake,
|
|
nixpkgs_flake,
|
|
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<BuildLogs>` through every public entry
|
|
// point in the lifecycle surface.
|
|
crate::build_logs::install(build_logs.clone());
|
|
// Audit log shares the same db dir; install its process-wide
|
|
// handle so privileged-action recording sites (e.g.
|
|
// `socket_server::handle_restart_infra`) write without threading an
|
|
// `Arc<AuditLog>` through the agent-request surface.
|
|
let audit_log =
|
|
Arc::new(crate::audit_log::AuditLog::open(build_logs_dir).context("open audit_log")?);
|
|
crate::audit_log::install(audit_log.clone());
|
|
let power = Arc::new(crate::power::PowerStore::open(db_path).context("open agent_power")?);
|
|
let (dashboard_events, _) = broadcast::channel(DASHBOARD_CHANNEL);
|
|
let (shutdown_tx, _) = watch::channel(false);
|
|
Ok(Self {
|
|
broker: Arc::new(broker),
|
|
approvals: Arc::new(approvals),
|
|
questions: Arc::new(questions),
|
|
scheduled_prompts: Arc::new(scheduled_prompts),
|
|
build_logs,
|
|
audit_log,
|
|
hyperhive_flake,
|
|
hyperhive_docs_flake,
|
|
nixpkgs_flake,
|
|
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()),
|
|
crash_suppressed: Mutex::new(HashSet::new()),
|
|
recent_transient: Mutex::new(HashMap::new()),
|
|
recent_crashes: Mutex::new(HashMap::new()),
|
|
graceful_stop_pending: Mutex::new(HashSet::new()),
|
|
last_stopped_running: Mutex::new(None),
|
|
dashboard_events,
|
|
event_seq: AtomicU64::new(0),
|
|
meta_updates_active: AtomicU64::new(0),
|
|
last_containers: tokio::sync::Mutex::new(HashMap::new()),
|
|
job_queue: Arc::new(crate::job_queue::JobQueue::new(build_slots)),
|
|
power,
|
|
shutdown_tx,
|
|
})
|
|
}
|
|
|
|
/// Snapshot the hive-wide configuration fields as a [`HiveEnv`].
|
|
/// Pass the result to `lifecycle::spawn` / `rebuild` / `meta::sync_agents`
|
|
/// instead of threading the individual fields separately.
|
|
#[must_use]
|
|
pub fn hive_env(&self) -> HiveEnv {
|
|
HiveEnv {
|
|
hyperhive_flake: self.hyperhive_flake.clone(),
|
|
hyperhive_docs_flake: self.hyperhive_docs_flake.clone(),
|
|
nixpkgs_flake: self.nixpkgs_flake.clone(),
|
|
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/<name>`), obtained
|
|
/// from `crate::paths::agent_runtime_dir(name)` (pure path) or from
|
|
/// `lifecycle::ensure_agent_runtime_dir(name)` when the dir must be
|
|
/// created. All other paths are derived statically from `name`.
|
|
#[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>) {
|
|
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<Self>) {
|
|
let mut schedules: Vec<hive_sh4re::schedule::WireSchedule> =
|
|
match self.scheduled_prompts.list() {
|
|
Ok(rows) => rows
|
|
.into_iter()
|
|
.map(crate::socket_server::schedule_to_wire_public)
|
|
.collect(),
|
|
Err(e) => {
|
|
tracing::warn!(error = ?e, "emit_schedules_snapshot: list failed");
|
|
return;
|
|
}
|
|
};
|
|
// Strip ghost targets (destroyed agents) so the dashboard doesn't
|
|
// render dead columns. Best-effort: if the roster cache is
|
|
// momentarily contended we emit unfiltered rather than block this
|
|
// sync path — the next snapshot / page reload corrects it.
|
|
if let Some(live) = self.live_container_names_blocking() {
|
|
crate::socket_server::filter_ghost_schedule_targets(&mut schedules, &live);
|
|
}
|
|
self.emit_dashboard_event(DashboardEvent::SchedulesChanged {
|
|
seq: self.next_seq(),
|
|
schedules,
|
|
});
|
|
}
|
|
|
|
/// Best-effort synchronous snapshot of live container names (the
|
|
/// logical agent names from the last `nixos-container list` scan).
|
|
/// Returns `None` when the roster cache lock is momentarily
|
|
/// contended, so a transient miss is treated as "roster unknown,
|
|
/// don't filter" rather than hiding live schedule targets. The async
|
|
/// `containers_snapshot` is the reliable path for request handlers;
|
|
/// this exists for the sync `emit_schedules_snapshot` SSE emit.
|
|
#[must_use]
|
|
pub fn live_container_names_blocking(&self) -> Option<HashSet<String>> {
|
|
self.last_containers
|
|
.try_lock()
|
|
.ok()
|
|
.map(|m| m.keys().cloned().collect())
|
|
}
|
|
|
|
/// Emit a `CapabilitiesChanged` snapshot event. Called from the
|
|
/// rebuild-queue worker after a `PermChange` / Capabilities entry
|
|
/// commits the JSON file, so the P3RM1SS10NS tab updates live.
|
|
pub fn emit_capabilities_snapshot(self: &Arc<Self>) {
|
|
use hive_sh4re::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<Self>) {
|
|
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<bool> {
|
|
self.shutdown_tx.subscribe()
|
|
}
|
|
|
|
/// Signal all background tasks to exit cleanly. The tasks break
|
|
/// out of their poll loop after completing their current work item.
|
|
/// Best-effort — does nothing if all receivers have already been
|
|
/// dropped (e.g. process is already mid-shutdown).
|
|
pub fn request_shutdown(&self) {
|
|
let _ = self.shutdown_tx.send(true);
|
|
}
|
|
|
|
/// Subscribe to the unified dashboard event channel. Used by the
|
|
/// `/dashboard/stream` SSE handler and by the broker-to-dashboard
|
|
/// forwarder task.
|
|
pub fn dashboard_subscribe(&self) -> broadcast::Receiver<DashboardEvent> {
|
|
self.dashboard_events.subscribe()
|
|
}
|
|
|
|
/// Stamp the next sequence number. Each emission of a
|
|
/// `DashboardEvent` should fill its `seq` with `next_seq()` so the
|
|
/// frame the wire carries is the one the client uses to dedupe.
|
|
pub fn next_seq(&self) -> u64 {
|
|
self.event_seq.fetch_add(1, Ordering::SeqCst) + 1
|
|
}
|
|
|
|
/// Current high-water seq. Snapshot endpoints read this *before*
|
|
/// gathering state so the (snapshot.seq, snapshot) pair satisfies:
|
|
/// any frame with `seq > snapshot.seq` is post-snapshot. The seq
|
|
/// captured here may grow during snapshot construction — clients
|
|
/// may double-apply such events, which renderers must tolerate.
|
|
pub fn current_seq(&self) -> u64 {
|
|
self.event_seq.load(Ordering::SeqCst)
|
|
}
|
|
|
|
/// Broadcast a freshly-built `DashboardEvent` (caller fills `seq`
|
|
/// via `next_seq()`). Returns silently when there are no
|
|
/// subscribers — the dashboard channel is best-effort presentation
|
|
/// plumbing, not a delivery guarantee.
|
|
pub fn emit_dashboard_event(&self, event: DashboardEvent) {
|
|
let _ = self.dashboard_events.send(event);
|
|
}
|
|
|
|
/// Mark a `meta-update` as in flight and return an RAII guard that
|
|
/// clears it on drop (including drop-via-panic). The first
|
|
/// concurrent run emits `MetaUpdateRunning { running: true }`; the
|
|
/// last one to finish emits `running: false`. The dashboard's META
|
|
/// INPUTS panel reads the flag to show a disabled "updating…"
|
|
/// state while the lock bump + rebuild ripple runs.
|
|
pub fn meta_update_guard(self: &Arc<Self>) -> MetaUpdateGuard {
|
|
if self.meta_updates_active.fetch_add(1, Ordering::SeqCst) == 0 {
|
|
self.emit_dashboard_event(DashboardEvent::MetaUpdateRunning {
|
|
seq: self.next_seq(),
|
|
running: true,
|
|
});
|
|
}
|
|
MetaUpdateGuard {
|
|
coord: Arc::clone(self),
|
|
}
|
|
}
|
|
|
|
/// True while at least one dashboard-triggered `meta-update` is
|
|
/// running. Surfaced on `/api/state` as `meta_update_running` so a
|
|
/// client that cold-loads mid-update sees the in-progress state.
|
|
pub fn meta_update_in_progress(&self) -> bool {
|
|
self.meta_updates_active.load(Ordering::SeqCst) > 0
|
|
}
|
|
|
|
/// Emit `AuditEntryAdded` immediately after a privileged-action row
|
|
/// is recorded, so the dashboard audit view live-appends it off
|
|
/// `/dashboard/stream`. Pass the [`AuditEntry`](crate::audit_log::AuditEntry)
|
|
/// returned by `audit_log::record` so the streamed event is the same
|
|
/// canonical row that was stored.
|
|
pub fn emit_audit_entry(&self, entry: crate::audit_log::AuditEntry) {
|
|
self.emit_dashboard_event(DashboardEvent::AuditEntryAdded {
|
|
seq: self.next_seq(),
|
|
entry,
|
|
});
|
|
}
|
|
|
|
/// Emit `ApprovalAdded` immediately after the row is inserted in
|
|
/// sqlite.
|
|
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<Self>) {
|
|
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 `"<root>"` when the
|
|
/// slot is `None`, keeping the wording consistent with the
|
|
/// `<parent>` sentinel's "root → operator" routing (see
|
|
/// `docs/conventions.md::Recipient sentinels`). The
|
|
/// notifications fire as ordinary broker messages with
|
|
/// `from = hive_sh4re::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<Self>,
|
|
moves: &[(&str, Option<&str>)],
|
|
) -> std::result::Result<(), String> {
|
|
if moves.is_empty() {
|
|
return Ok(());
|
|
}
|
|
// bulk_commit_topology applies all set_parent calls under one lock
|
|
// and returns (child, old_parent) for every move that changed.
|
|
let changed = crate::meta::bulk_commit_topology(moves).await?;
|
|
|
|
// Send per-agent notifications for each changed move.
|
|
for (child, old_parent) in &changed {
|
|
// Find the new parent from the moves slice.
|
|
let new_parent = moves
|
|
.iter()
|
|
.find(|(c, _)| *c == child)
|
|
.and_then(|(_, np)| *np);
|
|
let old_label = old_parent.as_deref().unwrap_or("<root>");
|
|
let new_label = new_parent.unwrap_or("<root>");
|
|
if let Some(op) = old_parent.as_deref() {
|
|
let _ = self.broker.send(&hive_sh4re::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<ContainerView> {
|
|
let last = self.last_containers.lock().await;
|
|
let mut out: Vec<ContainerView> = last.values().cloned().collect();
|
|
out.sort_by(|a, b| a.name.cmp(&b.name));
|
|
out
|
|
}
|
|
|
|
pub fn register_agent(self: &Arc<Self>, name: &str) -> Result<PathBuf> {
|
|
// Idempotent: drop any existing listener so re-registration (e.g. on rebuild,
|
|
// or after a hive-c0re restart cleared /run/hyperhive) gets a fresh socket.
|
|
self.unregister_agent(name);
|
|
let agent_dir = crate::paths::agent_runtime_dir(name);
|
|
std::fs::create_dir_all(&agent_dir)
|
|
.with_context(|| format!("create agent dir {}", agent_dir.display()))?;
|
|
let socket_path = Self::socket_path(name);
|
|
// Hand the full Coordinator to the per-agent socket — it
|
|
// needs broker + operator_questions to handle the agent-side
|
|
// `ask` / `answer` tools, not just the broker.
|
|
let socket = socket_server::start(name, &socket_path, self.clone())?;
|
|
self.agents.lock().unwrap().insert(name.to_owned(), socket);
|
|
Ok(agent_dir)
|
|
}
|
|
|
|
pub fn unregister_agent(&self, name: &str) {
|
|
if let Some(socket) = self.agents.lock().unwrap().remove(name) {
|
|
socket.handle.abort();
|
|
let _ = std::fs::remove_file(&socket.path);
|
|
}
|
|
}
|
|
pub fn list_agents(&self) -> Vec<String> {
|
|
self.agents.lock().unwrap().keys().cloned().collect()
|
|
}
|
|
|
|
/// 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);
|
|
}
|
|
|
|
/// Record the set of agents that were running at a broad-scope
|
|
/// `hivectl stop`, so the next broad-scope `start` restores exactly
|
|
/// this set. See the `last_stopped_running` field doc. Persists a
|
|
/// copy to the broker `kv` table (best-effort) so the snapshot
|
|
/// survives a hive-c0re restart; the in-memory copy is the fast path.
|
|
pub fn set_last_stopped_running(&self, agents: Vec<String>) {
|
|
match serde_json::to_string(&agents) {
|
|
Ok(json) => {
|
|
if let Err(e) = self.broker.kv_set(LAST_STOPPED_RUNNING_KEY, &json) {
|
|
tracing::warn!(error = ?e, "persist last_stopped_running failed (in-memory copy still set)");
|
|
}
|
|
}
|
|
Err(e) => tracing::warn!(error = ?e, "serialise last_stopped_running failed"),
|
|
}
|
|
*self.last_stopped_running.lock().unwrap() = Some(agents);
|
|
}
|
|
|
|
/// Take (and clear) the recorded broad-stop running set, if any. A
|
|
/// broad-scope `start` uses this to restore only the previously
|
|
/// running agents; `None` means "no record — start all". Falls back
|
|
/// to the persisted `kv` copy when the in-memory snapshot is empty
|
|
/// (hive-c0re restarted between the stop-all and the start). Clears
|
|
/// the persisted copy either way — the snapshot is one-shot.
|
|
pub fn take_last_stopped_running(&self) -> Option<Vec<String>> {
|
|
let result = self
|
|
.last_stopped_running
|
|
.lock()
|
|
.unwrap()
|
|
.take()
|
|
.or_else(|| {
|
|
self.broker
|
|
.kv_get(LAST_STOPPED_RUNNING_KEY)
|
|
.ok()
|
|
.flatten()
|
|
.and_then(|json| serde_json::from_str::<Vec<String>>(&json).ok())
|
|
});
|
|
if let Err(e) = self.broker.kv_delete(LAST_STOPPED_RUNNING_KEY) {
|
|
tracing::warn!(error = ?e, "clear persisted last_stopped_running failed");
|
|
}
|
|
result
|
|
}
|
|
|
|
/// Per-agent `deliberate_stop` for transients cleared within the last
|
|
/// `grace` seconds — i.e. agents the operator just acted on, whose stop the
|
|
/// crash watcher should NOT classify as a crash. Lazily reaps entries older
|
|
/// than `grace` so the map stays bounded by the active agent count.
|
|
///
|
|
/// Carries only the safety bit, not the display label: nothing downstream
|
|
/// should be able to re-derive a stop/crash decision from a pill's wording.
|
|
pub fn recent_transient_within(&self, grace: std::time::Duration) -> HashMap<String, bool> {
|
|
let now = std::time::Instant::now();
|
|
let mut map = self.recent_transient.lock().unwrap();
|
|
map.retain(|_, (_, ts)| now.duration_since(*ts) <= grace);
|
|
fold_tombstones_by_agent(map.iter().map(|((a, _), (d, _))| (a.as_str(), *d)))
|
|
}
|
|
|
|
/// Record an unexpected crash for `agent`. Called by the crash
|
|
/// watcher whenever it classifies a container stop as a crash (not an
|
|
/// operator action). Append-only here; pruning happens lazily on read
|
|
/// in `recent_crash_counts`.
|
|
pub fn record_crash(&self, agent: &str) {
|
|
self.recent_crashes
|
|
.lock()
|
|
.unwrap()
|
|
.entry(agent.to_owned())
|
|
.or_default()
|
|
.push(std::time::Instant::now());
|
|
}
|
|
|
|
/// Per-agent count of crashes within the last `window`. Lazily reaps
|
|
/// older timestamps and drops agents with none left, so the map stays
|
|
/// bounded and only lists agents actively crashing. Powers the
|
|
/// dashboard's `agents_crashing` banner warning.
|
|
pub fn recent_crash_counts(&self, window: std::time::Duration) -> HashMap<String, usize> {
|
|
let now = std::time::Instant::now();
|
|
let mut map = self.recent_crashes.lock().unwrap();
|
|
map.retain(|_, times| {
|
|
times.retain(|ts| now.duration_since(*ts) <= window);
|
|
!times.is_empty()
|
|
});
|
|
map.iter().map(|(k, v)| (k.clone(), v.len())).collect()
|
|
}
|
|
|
|
/// Tell the crash watcher that `name`'s container is going down **on
|
|
/// purpose**, for the lifetime of the returned guard. See
|
|
/// [`CrashWatchSuppression`] for why this exists at all.
|
|
///
|
|
/// Only for the operations with no queue node behind them. Anything the
|
|
/// job queue runs answers this from the node itself
|
|
/// ([`crate::job_queue::NodeKind::takes_container_down`]) and must not come
|
|
/// through here.
|
|
///
|
|
/// The guard's `Drop` runs even on task cancellation, so an aborted HTTP
|
|
/// request or a panic mid-destroy can't leave a container permanently
|
|
/// exempt from crash reporting.
|
|
pub fn suppress_crash_watch(self: &Arc<Self>, name: &str) -> CrashWatchSuppression {
|
|
self.crash_suppressed
|
|
.lock()
|
|
.unwrap()
|
|
.insert(name.to_owned());
|
|
CrashWatchSuppression {
|
|
coord: self.clone(),
|
|
name: name.to_owned(),
|
|
}
|
|
}
|
|
|
|
/// Whether a no-node operation is currently taking this container down.
|
|
#[must_use]
|
|
pub fn crash_watch_suppressed(&self, name: &str) -> bool {
|
|
self.crash_suppressed.lock().unwrap().contains(name)
|
|
}
|
|
|
|
/// 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.
|
|
///
|
|
/// Work with no queue node behind it (destroy, migration) therefore shows
|
|
/// **no pill** — there is nothing in the graph to derive one from. Its
|
|
/// crash-watch suppression is a separate, narrower thing
|
|
/// ([`Coordinator::suppress_crash_watch`]); the pill comes back for free
|
|
/// once those become real nodes.
|
|
/// ⚠️ **A `Vec` per agent, not one entry.** `running_transients` tests
|
|
/// status alone, so a lease-exempt `Prebuild` for `a` and a lease-holding
|
|
/// `StopForUpdate` for `a` are both live pills. Collapsing them to one
|
|
/// would pick arbitrarily and, for the crash watcher, silently lose a
|
|
/// `deliberate_stop = true` behind a `false` — reporting an intentional
|
|
/// stop as a crash.
|
|
#[must_use]
|
|
pub fn transient_snapshot(&self) -> HashMap<String, Vec<RunningTransient>> {
|
|
let mut out: HashMap<String, Vec<RunningTransient>> = HashMap::new();
|
|
for t in self.job_queue.running_transients() {
|
|
out.entry(t.agent.clone()).or_default().push(t);
|
|
}
|
|
out
|
|
}
|
|
|
|
/// Drop a system message into the given agent's inbox. Wakes the
|
|
/// turn loop with a "you were just (re)started" hint — operator
|
|
/// caused the transition, agent picks up where it left off
|
|
/// (notes are in the bind-mounted state dir, last turn is in
|
|
/// --continue's session). Best-effort; broker errors are logged
|
|
/// but don't propagate.
|
|
pub fn kick_agent(&self, name: &str, reason: &str) {
|
|
// Sub-agents bind their state at /agents/<name>/state. The
|
|
// manager has both /state (legacy mount) and /agents
|
|
// bind-mounted, so /agents/<name>/state resolves there too —
|
|
// use that uniformly so the wake message has one canonical
|
|
// path that works everywhere.
|
|
let body = format!(
|
|
"{reason}\n\nYou were just (re)started by the operator. \
|
|
If you were mid-task, check `/agents/{name}/state/` for \
|
|
your notes and pick up where you left off. claude's \
|
|
`--continue` session is intact, so prior context is \
|
|
still in your window."
|
|
);
|
|
if let Err(e) = self.broker.send(&hive_sh4re::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.
|
|
pub async fn push_todo(
|
|
&self,
|
|
agent: &str,
|
|
subsystem: &str,
|
|
key: Option<String>,
|
|
summary: String,
|
|
source: Option<String>,
|
|
) -> 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,
|
|
};
|
|
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.
|
|
pub async fn push_todo_submitter(
|
|
&self,
|
|
approval_id: i64,
|
|
subsystem: &str,
|
|
key: Option<String>,
|
|
summary: String,
|
|
source: Option<String>,
|
|
) -> Result<(), String> {
|
|
let target = self.submitter_or_manager(approval_id);
|
|
self.push_todo(&target, subsystem, key, summary, source)
|
|
.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<String> {
|
|
const HINT: &str =
|
|
"\n\n⚠️ _hint: this was a broadcast and may not need any action from you_";
|
|
let broadcast_body = format!("{body}{HINT}");
|
|
let mut errors = Vec::new();
|
|
for agent_name in self.list_agents() {
|
|
if agent_name == from {
|
|
continue;
|
|
}
|
|
if let Err(e) = self.broker.send(&hive_sh4re::inbox::Message {
|
|
from: hive_sh4re::manager::trusted_sender(from),
|
|
to: agent_name.clone(),
|
|
body: broadcast_body.clone(),
|
|
in_reply_to: None,
|
|
}) {
|
|
errors.push(format!("{agent_name}: {e}"));
|
|
}
|
|
}
|
|
errors
|
|
}
|
|
|
|
pub fn socket_path(name: &str) -> PathBuf {
|
|
crate::paths::agent_runtime_dir(name).join("mcp.sock")
|
|
}
|
|
|
|
/// Manager-editable proposed config repo. Bind-mounted into the manager
|
|
/// container as `/agents/<name>/config/`.
|
|
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<hive_types::Ident> {
|
|
let Ok(rd) = std::fs::read_dir(crate::paths::agents_root()) else {
|
|
return Vec::new();
|
|
};
|
|
let mut out: Vec<hive_types::Ident> = rd
|
|
.flatten()
|
|
.filter(|e| e.file_type().is_ok_and(|t| t.is_dir()))
|
|
.filter_map(|e| hive_types::Ident::parse(&e.file_name().into_string().ok()?).ok())
|
|
.collect();
|
|
out.sort();
|
|
out
|
|
}
|
|
|
|
/// 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<hive_types::Ident> {
|
|
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"
|
|
);
|
|
}
|
|
}
|