hyperhive/hive-c0re/src/container_view.rs

594 lines
26 KiB
Rust

//! `ContainerView` + the snapshot builder that turns
//! `nixos-container list` (plus per-agent state on disk) into the row
//! shape the dashboard renders. Extracted from `dashboard.rs` so the
//! coordinator's rescan-and-emit helper can build the same view and
//! diff against the last snapshot to fire
//! `ContainerStateChanged` / `ContainerRemoved` events.
use std::collections::HashMap;
use std::path::Path;
use chrono::{DateTime, Utc};
use serde::Serialize;
use crate::coordinator::Coordinator;
use crate::lifecycle::{self, AGENT_PREFIX, UnitState};
// Independent per-agent flags, each its own badge on the dashboard card
// and each diffed separately by `rescan_containers_and_emit`. Grouping
// them into nested structs would need `serde(flatten)` to keep the JSON
// the frontend already consumes, for no readability gain — same
// rationale as `LifecycleScope` in hive-host-sock.
#[allow(
clippy::struct_excessive_bools,
reason = "flat wire projection of independent per-agent flags"
)]
#[derive(Serialize, Clone, PartialEq, Eq, Debug)]
pub struct ContainerView {
/// Logical agent name (no `h-` prefix). Used in action URLs.
pub name: String,
/// Container name as nixos-container sees it (`h-foo`). Internal only;
/// not serialized to API responses since the dashboard no longer displays it.
#[serde(skip)]
pub container: String,
pub port: u16,
pub running: bool,
/// The container's unit is in systemd's `failed` state — it exhausted
/// its bounded restarts and gave up, rather than being stopped
/// deliberately.
///
/// Orthogonal to `running` rather than a variant of it: a failed unit
/// is not running, but a not-running unit is usually just *off*. That
/// distinction is the whole point — without it a container that gave
/// up is indistinguishable from one an operator stopped on purpose.
#[serde(default)]
pub failed: bool,
pub needs_update: bool,
pub needs_login: bool,
/// First 12 chars of the sha the meta flake currently has locked
/// for this agent's input.
#[serde(skip_serializing_if = "Option::is_none")]
pub deployed_sha: Option<String>,
/// Name of this agent's parent in the agent hierarchy. `None`
/// marks the agent as root-level; the dashboard renders it without
/// indentation. Sourced from `meta/topology.json` (single source of
/// truth, hive-c0re-owned) — NOT from per-agent agent.nix, because
/// an agent shouldn't be able to unilaterally declare its own place
/// in the tree. See `docs/agent-lifecycle/agent-hierarchy.md::Where the tree lives`.
#[serde(default, skip_serializing_if = "Option::is_none")]
pub parent: Option<String>,
/// The Claude model the agent's harness is currently using, read from
/// `state/hyperhive-harness.json["active_model"]`. `None` when the
/// agent has never started a turn or the field is absent. Only
/// meaningful when `running` is true; the dashboard skips the badge
/// for stopped containers.
#[serde(default, skip_serializing_if = "Option::is_none")]
pub active_model: Option<String>,
/// The agent's current free-text status (`set_status`). `None` when
/// unset, or when the container isn't running — see
/// [`read_agent_status_live`] for why a stopped agent's on-disk
/// status is treated as stale rather than surfaced.
#[serde(default, skip_serializing_if = "Option::is_none")]
pub status_text: Option<String>,
/// When `status_text` was last set, RFC 3339 UTC on the wire (see
/// `hive_sh4re::wire_time`). `None` exactly when `status_text` is
/// `None`.
#[serde(default, skip_serializing_if = "Option::is_none")]
pub status_set_at: Option<DateTime<Utc>>,
/// The agent's turn loop is parked: the container is up and still
/// serving its web UI / MCP daemons, but it drives no turns and its
/// inbox messages queue unacked until it resumes. Sourced from the
/// pause marker in the harness dir (`Coordinator::is_paused`), so it
/// stays true across a container restart — and, unlike `needs_login`,
/// is reported for stopped containers too: pausing a stopped agent is
/// a legitimate way to keep it idle when it next boots.
#[serde(default)]
pub paused: bool,
/// Effective systemd `CPUQuota=` for this container (e.g. `"400%"`) —
/// the per-agent override from `meta/resource-limits.json` when set,
/// otherwise the hive-wide `agentCpuQuota`. Always populated: there
/// is no "unset" state to render, only "same as everyone else".
/// Reflects what the *drop-in says*, which is what the next start
/// will enforce — not a live cgroup reading.
pub cpu_quota: String,
/// Effective systemd `MemoryMax=` for this container (e.g. `"8G"`).
/// Same resolution + caveat as [`ContainerView::cpu_quota`].
pub memory_max: String,
}
impl From<ContainerView> for hive_sh4re::container::AgentStatusRow {
/// The one place a `ContainerView` becomes the wire row `host.sock`
/// serves it as — `handle_agent_status`'s poll path and the
/// `SubscribeAgentStatus` push path both go through this, so the two
/// can never quietly diverge on which fields make the cut.
///
/// `pending_reminders` has no source here: reminders are agent-local
/// now, and c0re has no cross-agent visibility into pending counts
/// anymore. Stubbed to `0` rather than dropping the wire field
/// outright — leaves `hivectl status`/the dashboard column intact
/// syntactically, just always empty, until iris's frontend follow-up
/// decides whether to drop the column entirely. `port`, `cpu_quota`,
/// `memory_max` have no equivalent on the row at all and are simply
/// not carried over. `url` has no source on `ContainerView` at all —
/// see [`agent_url`].
fn from(v: ContainerView) -> Self {
Self {
url: agent_url(&v.name),
name: v.name,
running: v.running,
failed: v.failed,
needs_update: v.needs_update,
needs_login: v.needs_login,
deployed_sha: v.deployed_sha,
pending_reminders: 0,
parent: v.parent,
paused: v.paused,
active_model: v.active_model,
status_text: v.status_text,
status_set_at: v.status_set_at,
}
}
}
/// This agent's own web UI, behind the gateway. `None` when the hive's
/// domain isn't configured — same env var, same "empty string counts as
/// unset" filter `server.rs::hive_urls()` uses for `HiveUrls::home`, kept
/// in sync with it by reading the one env var directly rather than
/// threading a resolved domain through `HiveEnv` for a value nothing else
/// needs pre-resolved.
fn agent_url(name: &str) -> Option<String> {
let domain = std::env::var("HYPERHIVE_HIVE_DOMAIN")
.ok()
.filter(|v| !v.is_empty())?;
Some(agent_url_for_domain(&domain, name))
}
/// The pure half of [`agent_url`], split out so the path-scheme formatting
/// is testable without an env var in the loop. `/agent/<name>/` is the
/// gateway's real path scheme
/// (`nix/host-modules/hive-gateway/vhosts.nix`'s per-agent `agentLocations`
/// block) — computed once here so a client (trollshell's sidebar, the
/// dashboard) never has to derive it and silently drift if that scheme
/// ever changes.
fn agent_url_for_domain(domain: &str, name: &str) -> String {
format!("https://{domain}/agent/{name}/")
}
/// Build the full container list. Wraps `lifecycle::list()` and
/// resolves every per-agent attribute the dashboard surfaces.
///
/// Takes `hive` because the effective resource limits are a per-field
/// fallback onto the hive-wide `agent_cpu_quota` / `agent_memory_max`,
/// and those live on [`crate::coordinator::HiveEnv`], not on disk. Both callers already
/// hold a `Coordinator`, so this is a parameter rather than a global.
pub async fn build_all(hive: &crate::coordinator::HiveEnv) -> Vec<ContainerView> {
let raw = lifecycle::list().await.unwrap_or_default();
let locked = read_meta_locked_revs();
// Pull the topology map once and look up each agent's parent below.
// Empty / absent topology.json → every agent root-level (safe
// degradation for fresh installs that haven't run sync_agents yet).
let topology = crate::topology::read();
// Same once-per-scan treatment as the topology map: the override file
// is read here and resolved per agent below, rather than re-read for
// every container on every SSE scan.
let limits = crate::resource_limits::read();
let mut out = Vec::new();
for c in &raw {
let Some(logical) = c.strip_prefix(AGENT_PREFIX) else {
continue;
};
// Parse the nspawn machine suffix into an Ident once at this
// enumeration origin; a suffix that isn't a valid ident isn't one
// of our agents, so skip it.
let Ok(logical) = hive_types::Ident::parse(logical) else {
continue;
};
let deployed_full = locked
.get(&format!("agent-{logical}"))
.map(std::string::String::as_str);
let needs_update =
crate::auto_update::agent_config_pending(logical.as_str(), deployed_full).await;
let deployed_sha = deployed_full.map(|s| s[..s.len().min(12)].to_owned());
let parent = topology.get(logical.as_str()).cloned().flatten();
// One `systemctl` call for both facts: `unit_state` is the same
// shell-out `is_running` makes, minus `--quiet`. Asking twice would
// double the per-agent subprocess count on every SSE scan.
let state = lifecycle::unit_state(logical.as_str()).await;
let running = state == UnitState::Active;
let failed = state == UnitState::Failed;
// needs_login fires when EITHER the claude session dir is missing
// (boot-time / fresh container) OR the harness wrote the auth-failed
// sentinel because a turn hit 401. Cleared for stopped containers —
// stale sentinel state is not meaningful when the harness isn't up.
//
// The first half doesn't apply to an api-key agent
// (`hyperhive.useApiKey`, stamped as `api_key_mode` in the
// consolidated state file by `hive_agent::harness_state::write_api_key_mode`):
// its `~/.claude/` is empty by design (no OAuth flow to complete),
// so an empty dir there means nothing — only the auth-failed
// sentinel (a real 401, meaning the configured key itself is bad)
// still counts.
//
// One `read_harness_flags` call for both fields, not a wrapper per
// field: this is the only call site that needs more than one, and
// a wrapper each would read the file twice per agent per sweep for
// no reason.
let (_, needs_login_sentinel, is_api_key) = read_harness_flags(&logical);
let needs_login = running
&& ((!is_api_key && !claude_has_session(&Coordinator::agent_claude_dir(&logical)))
|| needs_login_sentinel);
// Read the active model from the harness state file. Only surfaced
// when the container is running — stale model info from a stopped
// agent is misleading (the model may change on next boot).
let active_model = if running {
read_active_model(&logical)
} else {
None
};
// Same running-gate as `active_model` just above, and the same
// rule `read_agent_status_live` applies for its other callers —
// duplicated inline rather than calling that helper because it
// would re-derive `running` with its own `lifecycle::is_running`
// shell-out when this loop already has it for free.
let (status_text, status_set_at) = if running {
let (text, set_at) = read_agent_status(&logical);
(text, set_at.map(hive_sh4re::wire_time::from_secs))
} else {
(None, None)
};
let paused = Coordinator::is_paused(&logical);
let (cpu_quota, memory_max) = crate::resource_limits::effective_from(
&limits,
logical.as_str(),
&hive.agent_cpu_quota,
&hive.agent_memory_max,
);
out.push(ContainerView {
port: lifecycle::agent_web_port(logical.as_str()),
running,
failed,
container: c.clone(),
name: logical.into_string(),
needs_update,
needs_login,
deployed_sha,
parent,
active_model,
status_text,
status_set_at,
paused,
cpu_quota,
memory_max,
});
}
out
}
/// Host-side mirror of `hive_agent::login::has_session`. Returns true
/// if the agent's bound `~/.claude/` dir on disk contains any regular
/// file. Reads each `build_all()` so a login driven from the agent's
/// own web UI reflects on the next snapshot.
pub fn claude_has_session(dir: &Path) -> bool {
let Ok(entries) = std::fs::read_dir(dir) else {
return false;
};
entries
.flatten()
.any(|e| e.file_type().is_ok_and(|t| t.is_file()))
}
/// Read `rate_limited` + `needs_login` (auth-failed sentinel) +
/// `api_key_mode` from the consolidated `hyperhive-harness.json`. Falls
/// back to the legacy individual sentinel files written by older harness
/// builds so in-place upgrades don't lose state during the transition
/// window — `api_key_mode` has no legacy equivalent (it postdates the
/// consolidated file), so that fallback just says `false`, the correct
/// answer for any harness build old enough to have never written it.
fn read_harness_flags(name: &hive_types::Ident) -> (bool, bool, bool) {
let dir = Coordinator::agent_notes_dir(name);
if let Ok(raw) = std::fs::read_to_string(dir.join("hyperhive-harness.json"))
&& let Ok(v) = serde_json::from_str::<serde_json::Value>(&raw)
{
let rl = v
.get("rate_limited")
.and_then(serde_json::Value::as_bool)
.unwrap_or(false);
let nl = v
.get("needs_login")
.and_then(serde_json::Value::as_bool)
.unwrap_or(false);
let akm = v
.get("api_key_mode")
.and_then(serde_json::Value::as_bool)
.unwrap_or(false);
return (rl, nl, akm);
}
// Legacy fallback: presence of individual sentinel files.
let rate_limited = dir.join("hyperhive-rate-limited").exists();
let needs_login = dir.join("hyperhive-needs-login").exists();
(rate_limited, needs_login, false)
}
/// Read the agent's free-text status and the Unix timestamp when it was last set
/// (derived from the file's mtime). Returns `(None, None)` when the file is absent
/// or empty.
///
/// NB: a caller building `AgentMeta` for a *stopped* container must clear
/// the result — the on-disk status is a stale snapshot from before the
/// stop. [`read_agent_status_live`] applies that rule, and is what the
/// socket server and the swarm status reader actually call.
fn read_agent_status(name: &hive_types::Ident) -> (Option<String>, Option<i64>) {
let path = Coordinator::agent_notes_dir(name).join("hyperhive-status");
let meta = std::fs::metadata(&path).ok();
// Read at most STATUS_MAX_CHARS * 4 + 2 bytes: 4 is the max UTF-8 byte
// width per char, +2 for the trailing newline. Guards against a
// pathologically large file written outside the harness validation path.
let s = {
use std::io::Read as _;
std::fs::File::open(&path).ok().and_then(|f| {
let cap = (crate::limits::STATUS_MAX_CHARS * 4 + 2) as u64;
let mut buf = String::new();
f.take(cap).read_to_string(&mut buf).ok().map(|_| buf)
})
};
let text = s
.as_deref()
.map(str::trim)
.filter(|t| !t.is_empty())
.map(str::to_owned);
let mtime = meta.and_then(|m| {
m.modified().ok().and_then(|t| {
t.duration_since(std::time::UNIX_EPOCH)
.ok()
.and_then(|d| i64::try_from(d.as_secs()).ok())
})
});
if text.is_none() {
(None, None)
} else {
(text, mtime)
}
}
/// Wraps `read_agent_status` with the same "stopped containers have
/// stale state" gate `build_all` uses. Returns `(None, None, false)`
/// when the container isn't running so callers don't have to know
/// about the sentinel rules — they just hand back what we give them.
///
/// Returned tuple is `(status_text, status_set_at, running)`.
/// `name` is the logical agent name (same as the broker recipient).
pub async fn read_agent_status_live(
name: &hive_types::Ident,
) -> (Option<String>, Option<i64>, bool) {
if !lifecycle::is_running(name.as_str()).await {
return (None, None, false);
}
let (text, set_at) = read_agent_status(name);
(text, set_at, true)
}
/// Read the active Claude model from `hyperhive-harness.json` (the
/// consolidated harness state file in the agent's state dir). Written by
/// the harness on startup and on every `set_model` / `emit_status` call,
/// so it always reflects the resolved priority (nix config > runtime
/// override > default). Returns `None` when the field is absent or the
/// harness has not yet started a turn.
fn read_active_model(name: &hive_types::Ident) -> Option<String> {
let path = Coordinator::agent_notes_dir(name).join("hyperhive-harness.json");
let raw = std::fs::read_to_string(path).ok()?;
let v: serde_json::Value = serde_json::from_str(&raw).ok()?;
v.get("active_model")
.and_then(serde_json::Value::as_str)
.filter(|s| !s.is_empty())
.map(str::to_owned)
}
/// Host-side hive + swarm display names, read from the c0re service's
/// own process env. The hyperhive NixOS module sets these from
/// `services.hyperhive.hiveName` + `services.hyperhive.swarm.name`
/// (the hive names itself; the swarm it joins is named one level out).
/// The agent-side
/// `hive-agent::identity::{hive_name, swarm_name}` accessors read the
/// same env vars after they're forwarded into each sub-agent's
/// harness service environment by `meta::render_flake`; surfacing
/// them here from c0re's own env keeps the manager + agent
/// `GetAgentMeta` paths consistent without a round-trip to the
/// target container.
///
/// Returns `(hive_name, swarm_name)`. Each is `None` when the
/// corresponding env var is unset or empty.
#[must_use]
pub fn hive_swarm_names() -> (Option<String>, Option<String>) {
let read = |var: &str| -> Option<String> { std::env::var(var).ok().filter(|s| !s.is_empty()) };
(read("HYPERHIVE_HIVE_NAME"), read("HYPERHIVE_SWARM_NAME"))
}
/// Map of `agent-<n>` → locked sha from meta's flake.lock. Used to
/// render the `deployed:<sha12>` chip per container row.
fn read_meta_locked_revs() -> HashMap<String, String> {
let Ok(raw) = std::fs::read_to_string(crate::paths::meta_flake_lock()) else {
return HashMap::new();
};
parse_locked_revs(&raw)
}
/// The parsing half of [`read_meta_locked_revs`], split out so it can be
/// exercised without a `flake.lock` on disk.
///
/// Every failure is the same empty map: the chip is decoration, and a
/// malformed lock must not take the dashboard down.
fn parse_locked_revs(raw: &str) -> HashMap<String, String> {
let mut out = HashMap::new();
let Ok(json) = serde_json::from_str::<serde_json::Value>(raw) else {
return out;
};
let Some(nodes) = json.get("nodes").and_then(|v| v.as_object()) else {
return out;
};
let Some(root_name) = json.get("root").and_then(|v| v.as_str()) else {
return out;
};
let Some(root_inputs) = nodes
.get(root_name)
.and_then(|n| n.get("inputs"))
.and_then(|v| v.as_object())
else {
return out;
};
for alias in root_inputs.keys() {
let target_name = match root_inputs.get(alias) {
Some(serde_json::Value::String(s)) => s.clone(),
_ => continue,
};
if let Some(rev) = nodes
.get(&target_name)
.and_then(|n| n.get("locked"))
.and_then(|v| v.get("rev"))
.and_then(|v| v.as_str())
{
out.insert(alias.clone(), rev.to_owned());
}
}
out
}
#[cfg(test)]
mod tests {
use super::parse_locked_revs;
/// Shape of a real `meta/flake.lock`: `root` names a node whose
/// `inputs` map alias → node name, and each node carries `locked.rev`.
fn lock(inputs: &str, nodes: &str) -> String {
format!(r#"{{"root":"root","nodes":{{"root":{{"inputs":{{{inputs}}}}},{nodes}}}}}"#)
}
/// Keyed by the **alias** (what the chip renders), not by the node
/// name it points at. Those differ whenever nix dedupes a node —
/// `agent-bob` → `agent-bob_2` — so the fixture makes them differ,
/// otherwise the test cannot tell the two keyings apart.
#[test]
fn maps_each_alias_to_its_locked_rev() {
let raw = lock(
r#""agent-alice":"agent-alice","agent-bob":"agent-bob_2""#,
r#""agent-alice":{"locked":{"rev":"aaa111"}},"agent-bob_2":{"locked":{"rev":"bbb222"}}"#,
);
let got = parse_locked_revs(&raw);
assert_eq!(got.get("agent-alice").map(String::as_str), Some("aaa111"));
assert_eq!(
got.get("agent-bob").map(String::as_str),
Some("bbb222"),
"keyed by alias, not by the deduped node name"
);
assert!(!got.contains_key("agent-bob_2"));
assert_eq!(got.len(), 2);
}
/// A `follows` input is stored as an array of path segments, not a
/// node name. Skipping it is why the match arm is a `continue`.
#[test]
fn a_follows_input_is_skipped_without_losing_its_siblings() {
let raw = lock(
r#""agent-alice":"agent-alice","nixpkgs":["agent-alice","nixpkgs"]"#,
r#""agent-alice":{"locked":{"rev":"aaa111"}}"#,
);
let got = parse_locked_revs(&raw);
assert_eq!(got.len(), 1, "the sibling still resolves");
assert!(!got.contains_key("nixpkgs"));
}
#[test]
fn an_input_whose_node_has_no_rev_is_skipped() {
let raw = lock(
r#""agent-alice":"agent-alice","agent-bob":"agent-bob""#,
r#""agent-alice":{"locked":{"rev":"aaa111"}},"agent-bob":{"locked":{}}"#,
);
let got = parse_locked_revs(&raw);
assert_eq!(got.len(), 1);
assert!(got.contains_key("agent-alice"));
}
/// Every malformed shape yields an empty map rather than a panic —
/// the chip is decoration and must not take the dashboard down.
#[test]
fn malformed_input_yields_an_empty_map() {
for raw in [
"not json at all",
"{}",
r#"{"root":"root"}"#,
r#"{"nodes":{"root":{"inputs":{}}}}"#,
r#"{"root":"missing","nodes":{"root":{"inputs":{"a":"a"}}}}"#,
] {
assert!(parse_locked_revs(raw).is_empty(), "for input: {raw}");
}
// Control: the well-formed shape these are degraded from does
// resolve, so the emptiness above is the guard and not the parser
// being inert.
let ok = lock(r#""a":"a""#, r#""a":{"locked":{"rev":"abc"}}"#);
assert_eq!(parse_locked_revs(&ok).len(), 1);
}
/// The `From` impl both `handle_agent_status`'s poll path and
/// `stream_agent_status`'s push path go through — pins the field
/// mapping so the two can't quietly diverge, and that `port` /
/// `cpu_quota` / `memory_max` (no equivalent on the row) are dropped
/// on purpose rather than by omission.
#[test]
fn agent_status_row_carries_every_field_the_row_has_and_drops_the_rest() {
use super::ContainerView;
use chrono::{TimeZone, Utc};
use hive_sh4re::container::AgentStatusRow;
let view = ContainerView {
name: "alice".to_owned(),
container: "h-alice".to_owned(),
port: 7000,
running: true,
failed: false,
needs_update: true,
needs_login: false,
deployed_sha: Some("abc123def456".to_owned()),
parent: Some("bob".to_owned()),
active_model: Some("claude-opus".to_owned()),
status_text: Some("shipping".to_owned()),
status_set_at: Some(Utc.timestamp_opt(1_700_000_000, 0).unwrap()),
paused: true,
cpu_quota: "400%".to_owned(),
memory_max: "8G".to_owned(),
};
let row = AgentStatusRow::from(view);
assert_eq!(row.name, "alice");
assert!(row.running);
assert!(!row.failed);
assert!(row.needs_update);
assert!(!row.needs_login);
assert_eq!(row.deployed_sha.as_deref(), Some("abc123def456"));
assert_eq!(row.parent.as_deref(), Some("bob"));
assert!(row.paused);
assert_eq!(row.active_model.as_deref(), Some("claude-opus"));
assert_eq!(row.status_text.as_deref(), Some("shipping"));
assert_eq!(
row.status_set_at,
Some(Utc.timestamp_opt(1_700_000_000, 0).unwrap())
);
// No source for reminders on this side (see the `From` impl's doc
// comment) — always the stub, never left uninitialised.
assert_eq!(row.pending_reminders, 0);
// `url` depends on process-global env (`HYPERHIVE_HIVE_DOMAIN`), so
// it isn't asserted here — mutating env vars in a test race with
// every other test in this binary. `agent_url_for_domain` below
// covers the actual path-formatting logic without touching env.
}
/// The pure half of the per-agent URL: given a domain, is the path
/// exactly `/agent/<name>/`? (`agent_url` itself, the thin env-reading
/// wrapper, is intentionally untested — see the note on the test
/// above.)
#[test]
fn agent_url_for_domain_uses_the_gateways_real_path_scheme() {
assert_eq!(
super::agent_url_for_domain("hive.example.com", "alice"),
"https://hive.example.com/agent/alice/"
);
}
}