d94bc2188d
`topology.json` was a map of `name -> parent | null`, and that value fed the whole agent hierarchy: `<parent>` / `<children>` recipient sentinels, the reparenting API (CLI verb, wire verb, dashboard endpoints, DAG node), the dashboard tree, the rebuild depth sort, and an unconditional bind-mount grant giving every agent RW on its direct children's state. Per the operator's ruling the field goes, and with it all of the above. The file survives as what remains once the value is gone: the roster of agent names, which is the set `ManageRootAgent` grants mounts over. It is now a JSON array; `read` still accepts the old map shape and keeps its keys, so a hive that upgrades across this does not blank its roster (and so no capability holder loses its mounts for the length of that window). Two sites kept their behaviour under a different recipient rather than losing it. Both addressed `<parent>`, which the broker already resolved to `operator` for a root agent, and every agent is now what that fallback called a root: - the harness's turn-failure / plugin-failure notification (`Surface::send_to_parent` -> `send_to_operator`), and - the send allow-list's always-permitted escape hatch, so an agent with a restrictive allow-list still has a way to say it is stuck. What is NOT preserved, deliberately: an agent with no capability no longer sees any other agent's dirs. `ManageRootAgent`'s own grant is unchanged -- still every agent in the roster, still state RW + config RO, still no `harness`. The dashboard's reparenting control (the M0V3 picker) is deleted with its CSS. The tree rendering that reads `ContainerView.parent` is left for the frontend owner -- it degrades to a flat list with the field gone.
1568 lines
70 KiB
Rust
1568 lines
70 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::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>,
|
|
/// 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>,
|
|
/// 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 `services.hyperhive.agent.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 `services.hyperhive.agent.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>>,
|
|
/// Tombstone for transients that have JUST been cleared. The
|
|
/// crash watcher polls every 10s and would race the
|
|
/// drop-clears-immediately path of `TransientGuard`: an operator
|
|
/// kill / restart sets `Stopping` → runs `nixos-container stop` →
|
|
/// drop clears the transient → poll fires next tick and sees the
|
|
/// container missing-from-running with no active transient →
|
|
/// spurious "container stopped without an operator action"
|
|
/// message.
|
|
///
|
|
/// `clear_transient` stamps the cleared kind here with an
|
|
/// `Instant`; `recent_transient_within(grace)` returns the set of
|
|
/// agents whose tombstone is still inside the grace window. Crash
|
|
/// watcher consults both this and the active map before declaring
|
|
/// a stop deliberate.
|
|
///
|
|
/// 🚨 **Keyed by `(agent, label)`, not by agent.** An agent can have several
|
|
/// pills clearing in the same window — the transient set tests status alone,
|
|
/// so a lease-exempt `Prebuild` and a lease-holding `StopForUpdate` are both
|
|
/// live and both clear. Keyed by agent, the last clear *overwrites* the
|
|
/// others: a `Prebuild` (`deliberate_stop = false`) landing after a
|
|
/// `StopForUpdate` (`true`) leaves the tombstone reading `false`, and the
|
|
/// crash watcher then reports an intentional stop as a **crash**.
|
|
recent_transient: Mutex<HashMap<(String, String), (bool, std::time::Instant)>>,
|
|
/// 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>>,
|
|
/// Agents with a pause acknowledgement in progress — set by the pause
|
|
/// DAG's signal node, cleared when the agent reports
|
|
/// `PauseAcknowledged` (or the drain node's timeout fallback fires).
|
|
/// Same shape as `graceful_stop_pending`, one set per concern rather
|
|
/// than reusing it: a pause and a graceful stop are independent
|
|
/// orchestrations that can be in flight for different agents
|
|
/// simultaneously.
|
|
pause_pending: Mutex<HashSet<String>>,
|
|
/// 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, 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
|
|
/// `services.hyperhive.agent.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 `services.hyperhive.agent.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
|
|
}
|
|
|
|
/// 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>,
|
|
}
|
|
|
|
/// The two ways [`Coordinator::set_paused_by_name`] can fail. See its own
|
|
/// doc comment for why this is a distinct type rather than one
|
|
/// `anyhow::Error` (`Write` already wraps one, `BadName` never needs to).
|
|
#[derive(Debug)]
|
|
pub enum SetPausedByNameError {
|
|
/// `name` did not parse as a [`hive_types::Ident`] — the parse error's
|
|
/// own message.
|
|
BadName(&'static str),
|
|
Write(anyhow::Error),
|
|
}
|
|
|
|
impl std::fmt::Display for SetPausedByNameError {
|
|
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
|
|
match self {
|
|
SetPausedByNameError::BadName(msg) => write!(f, "bad agent name: {msg}"),
|
|
SetPausedByNameError::Write(e) => write!(f, "{e}"),
|
|
}
|
|
}
|
|
}
|
|
|
|
impl Coordinator {
|
|
pub fn open(
|
|
db_path: &Path,
|
|
env: HiveEnv,
|
|
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 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());
|
|
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),
|
|
scheduled_prompts: Arc::new(scheduled_prompts),
|
|
build_logs,
|
|
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/<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| <&str>::from(*c)).collect();
|
|
let descriptions = Capability::ALL
|
|
.iter()
|
|
.map(|c| (<&str>::from(*c), 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| <&str>::from(*g)).collect();
|
|
let descriptions = ToolGroup::ALL
|
|
.iter()
|
|
.map(|g| (<&str>::from(*g), 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 `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,
|
|
});
|
|
}
|
|
|
|
/// 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,
|
|
});
|
|
}
|
|
}
|
|
|
|
/// Read-only snapshot of the last cached container view. Used by
|
|
/// `/api/state` to cold-load page-open clients without re-running
|
|
/// `nixos-container list` themselves; the
|
|
/// `rescan_containers_and_emit` calls keep this fresh.
|
|
pub async fn containers_snapshot(&self) -> Vec<ContainerView> {
|
|
let last = self.last_containers.lock().await;
|
|
let mut out: Vec<ContainerView> = last.values().cloned().collect();
|
|
out.sort_by(|a, b| a.name.cmp(&b.name));
|
|
out
|
|
}
|
|
|
|
pub fn register_agent(self: &Arc<Self>, name: &str) -> Result<PathBuf> {
|
|
// Idempotent: drop any existing listener so re-registration (e.g. on rebuild,
|
|
// or after a hive-c0re restart cleared /run/hyperhive) gets a fresh socket.
|
|
self.unregister_agent(name);
|
|
let agent_dir = crate::paths::agent_runtime_dir(name);
|
|
std::fs::create_dir_all(&agent_dir)
|
|
.with_context(|| format!("create agent dir {}", agent_dir.display()))?;
|
|
let socket_path = Self::socket_path(name);
|
|
// Hand the full Coordinator to the per-agent socket — it needs
|
|
// more than just the broker (approvals, scheduled_prompts, ...).
|
|
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);
|
|
}
|
|
|
|
/// 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<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()
|
|
}
|
|
|
|
/// Every live transient, keyed by agent.
|
|
///
|
|
/// **Derived on read, stored nowhere.** Straight off the running graph, so
|
|
/// there is no cached copy to go stale, leak, or disagree with what is
|
|
/// actually running.
|
|
///
|
|
/// Migration is the last operation with no queue node behind it, so it
|
|
/// shows **no pill** — there is nothing in the graph to derive one from.
|
|
/// It gets one for free once it becomes real nodes, the way destroy did.
|
|
///
|
|
/// ⚠️ **A `Vec` per agent, not one entry.** `running_transients` tests
|
|
/// status alone, so a lease-exempt `Prebuild` for `a` and a lease-holding
|
|
/// `StopForUpdate` for `a` are both live pills. Collapsing them to one
|
|
/// would pick arbitrarily and, for the crash watcher, silently lose a
|
|
/// `deliberate_stop = true` behind a `false` — reporting an intentional
|
|
/// stop as a crash.
|
|
#[must_use]
|
|
pub fn transient_snapshot(&self) -> HashMap<String, Vec<RunningTransient>> {
|
|
let mut out: HashMap<String, Vec<RunningTransient>> = HashMap::new();
|
|
for t in self.job_queue.running_transients() {
|
|
out.entry(t.agent.clone()).or_default().push(t);
|
|
}
|
|
out
|
|
}
|
|
|
|
/// Drop a system message into the given agent's inbox. Wakes the
|
|
/// turn loop with a "you were just (re)started" hint — operator
|
|
/// caused the transition, agent picks up where it left off
|
|
/// (notes are in the bind-mounted state dir, last turn is in
|
|
/// --continue's session). Best-effort; broker errors are logged
|
|
/// but don't propagate.
|
|
pub fn kick_agent(&self, name: &str, reason: &str) {
|
|
// 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.
|
|
/// `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<String>,
|
|
summary: String,
|
|
source: Option<String>,
|
|
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<String>,
|
|
summary: String,
|
|
source: Option<String>,
|
|
) -> 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) — e.g. `ContainerCrash`, `NeedsUpdate`.
|
|
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
|
|
/// rather than the system itself, so the recipient's terminal
|
|
/// shows the right name instead of `system`.
|
|
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")
|
|
}
|
|
|
|
/// 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/<name>/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
|
|
}
|
|
|
|
/// [`Self::set_paused`] from a plain agent-name string, parsing it
|
|
/// first — the exact shape both `dashboard::lifecycle_ops`'s
|
|
/// `/api/pause`/`/api/resume` handlers and `workers::wanted`'s
|
|
/// reconcile loop need, previously duplicated between them (mara, PR
|
|
/// review: "could this be a bit less duplicated code?"). A distinct
|
|
/// error variant per failure mode, not a single
|
|
/// `anyhow::Error`, because the two callers report the two cases
|
|
/// differently: a bad name is the caller's mistake (400 there, a
|
|
/// warned skip in the reconcile loop), a write failure is this
|
|
/// host's (500 there, also a warned skip here).
|
|
///
|
|
/// # Errors
|
|
///
|
|
/// `BadName` when `name` is not a legal [`hive_types::Ident`];
|
|
/// `Write` for everything [`Self::set_paused`] itself can fail with.
|
|
pub async fn set_paused_by_name(name: &str, paused: bool) -> Result<(), SetPausedByNameError> {
|
|
let id = hive_types::Ident::parse(name).map_err(SetPausedByNameError::BadName)?;
|
|
Self::set_paused(&id, paused)
|
|
.await
|
|
.map_err(SetPausedByNameError::Write)
|
|
}
|
|
|
|
/// Enumerate names that have a persistent state dir under
|
|
/// `/var/lib/hyperhive/agents/` (i.e. config / claude creds /
|
|
/// notes survive). Includes both currently-existing containers and
|
|
/// destroyed-but-kept tombstones; callers filter the latter by
|
|
/// subtracting `lifecycle::list()`.
|
|
#[must_use]
|
|
pub fn kept_state_names() -> Vec<hive_types::Ident> {
|
|
let Ok(rd) = std::fs::read_dir(crate::paths::agents_root()) else {
|
|
return Vec::new();
|
|
};
|
|
let mut out: Vec<hive_types::Ident> = rd
|
|
.flatten()
|
|
.filter(|e| e.file_type().is_ok_and(|t| t.is_dir()))
|
|
.filter_map(|e| hive_types::Ident::parse(&e.file_name().into_string().ok()?).ok())
|
|
.collect();
|
|
out.sort();
|
|
out
|
|
}
|
|
|
|
/// 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 before the first
|
|
/// spawn) 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"
|
|
);
|
|
}
|
|
}
|