hyperhive/hive-c0re/src/meta.rs

1049 lines
42 KiB
Rust

//! Single hive-c0re-owned flake at `/var/lib/hyperhive/meta/` that
//! exports one `nixosConfiguration` per agent and drives the system-wide
//! deploy audit trail. Flow (`sync_agents`, two-phase `prepare_deploy` /
//! `finalize_deploy` / `abort_deploy`, `lock_update_hyperhive`):
//! `docs/approvals.md::Meta flake`.
use std::path::{Path, PathBuf};
use anyhow::{Context, Result, bail};
use tokio::process::Command;
use tokio::sync::Mutex;
use crate::coordinator::HiveEnv;
use crate::lifecycle;
const META_ROOT: &str = "/var/lib/hyperhive/meta";
const APPLIED_ROOT: &str = "/var/lib/hyperhive/applied";
const GIT_NAME: &str = "c0re";
const GIT_EMAIL: &str = "c0re@hyperhive.local";
/// Single-writer lock around every meta-repo operation. Git isn't
/// safe to drive from concurrent processes against the same `.git/`
/// — two simultaneous `git add` / `commit` invocations race on
/// `.git/index.lock`; if either dies before releasing, the lock
/// sticks and the next operation hits "another git process seems to
/// be running" until somebody `rm`s it manually. Holding this mutex
/// across each public function's git+nix calls makes parallel
/// rebuilds (`auto_update` + dashboard-triggered + apply-commit)
/// take turns instead of colliding.
static META_LOCK: Mutex<()> = Mutex::const_new(());
/// Where the manager sees this directory inside its container (RO bind).
#[allow(dead_code)] // wired up by set_nspawn_flags in a follow-up commit
pub const CONTAINER_MANAGER_META_MOUNT: &str = "/meta";
#[derive(Debug, Clone)]
pub struct AgentSpec {
pub name: String,
pub is_manager: bool,
pub port: u16,
}
#[must_use]
pub fn meta_dir() -> PathBuf {
PathBuf::from(META_ROOT)
}
/// Idempotently reconcile the meta repo with the current agent set.
/// First call inits the git repo, runs `nix flake lock`, and lands a
/// seed commit. Subsequent calls only touch `flake.nix` when the
/// rendered contents differ from disk; an unchanged `flake.nix` is a
/// no-op.
#[allow(dead_code, clippy::implicit_hasher)] // first caller lands in a later commit
pub async fn sync_agents(hive: &HiveEnv, agents: &[AgentSpec]) -> Result<()> {
let _guard = META_LOCK.lock().await;
let dir = meta_dir();
std::fs::create_dir_all(&dir).with_context(|| format!("create {}", dir.display()))?;
let new_flake = render_flake(
&hive.hyperhive_flake,
&hive.nixpkgs_flake,
&hive.nixpkgs_unstable_flake,
hive.dashboard_port,
&hive.operator_pronouns,
&hive.context_window_tokens,
agents,
);
let flake_path = dir.join("flake.nix");
let on_disk = std::fs::read_to_string(&flake_path).unwrap_or_default();
let initial = !dir.join(".git").exists();
if !initial && on_disk == new_flake {
return Ok(());
}
// Safety guard: refuse to write an empty agent list over a non-empty
// on-disk flake. An empty `agents` slice is never intentional — it
// means `nixos-container list` failed and the caller got an empty
// fallback. Overwriting here would drop every agent from the meta
// flake and trigger unnecessary (and potentially destructive) cascade
// rebuilds. Callers that genuinely need to clear the agent list
// (there are none today) must handle this case explicitly.
if agents.is_empty() && !initial && !on_disk.is_empty() {
tracing::warn!(
"sync_agents: refusing to overwrite non-empty meta flake with empty agent list \
(nixos-container list may have failed)"
);
return Ok(());
}
std::fs::write(&flake_path, &new_flake)
.with_context(|| format!("write {}", flake_path.display()))?;
// Reconcile topology.json against the live agent set — adds
// entries for newly-spawned agents (default: manager as parent,
// manager itself as root) and drops removed agents. Operator
// overrides via the write API are preserved because reconcile
// only fills in missing entries. Idempotent; when nothing changed
// the file isn't touched.
let agent_names: Vec<String> = agents.iter().map(|a| a.name.clone()).collect();
crate::topology::reconcile(&agent_names)
.with_context(|| format!("reconcile {}", crate::topology::topology_path().display()))?;
// Refresh /var/lib/hyperhive/agent-ports.json so the hive-gateway
// nginx sees the new agent set. The file is the single source of
// truth for which agents the gateway proxies to, since the
// gateway container lives in system config and can't be rebuilt
// from meta-flake events. Atomic write (tmp + rename) so a
// partial write never trips the gateway's read.
if let Err(e) = crate::agent_ports::write(&agent_names) {
// Best-effort: a failed write doesn't block the meta-flake
// regen + container ops that follow. The gateway falls back
// to whatever map is currently on disk (or an empty map on
// first boot, meaning no per-agent routing yet).
tracing::warn!(error = ?e, "agent_ports::write failed (non-fatal)");
}
// Refresh /var/lib/hyperhive/agent-sockets.json — sibling to the
// ports map, drives the gateway's unix-socket upstreams once
// agents opt in to `HIVE_WEB_SOCKET`. Coexists with the TCP-port
// map during the transition: the gateway picks the socket
// upstream when one exists, falls back to the TCP port otherwise.
// Same best-effort + non-fatal shape. See
// `docs/gateway.md::Per-agent unix-socket upstream`.
if let Err(e) = crate::agent_sockets::write(&agent_names) {
tracing::warn!(error = ?e, "agent_sockets::write failed (non-fatal)");
}
// Refresh /var/lib/hyperhive/gateway/agents.conf — the nginx include
// file the gateway container bind-mounts and nginx reads at runtime.
// c0re triggers a reload (or start) inside hive-gateway via hive-priv
// after writing the file. Same best-effort + non-fatal shape.
if let Err(e) = crate::gateway_nginx::write(&agent_names).await {
tracing::warn!(error = ?e, "gateway_nginx::write failed (non-fatal)");
}
if initial {
git(&dir, &["init", "--initial-branch=main"]).await?;
}
// Stage flake.nix *before* running nix flake lock. When meta is
// a git repo, nix treats it as a `git+file://` self-reference;
// its dirty-tree fetcher includes index entries (tracked +
// staged) but skips untracked files, so without the stage step
// an untracked flake.nix surfaces as "source tree does not
// contain '/flake.nix'". Lock then commit once with both
// flake.nix and flake.lock — single commit per change.
git(&dir, &["add", "flake.nix"]).await?;
// Stage topology.json on every sync (regenerated by reconcile
// above when the agent set changed). git add is a no-op when the
// file content is unchanged.
if crate::topology::topology_path().exists() {
git(&dir, &["add", "topology.json"]).await?;
}
// Stage tool-groups.json when it exists. Created on first
// `set_groups` call (operator-driven); absent = all agents on
// their role defaults, no file needed. git add is a no-op when
// the file is unchanged.
if crate::tool_groups::tool_groups_path().exists() {
git(&dir, &["add", "tool-groups.json"]).await?;
}
// Stage capabilities.json when it exists. Created on first
// `set_caps` call; absent = no agents have extra capabilities.
if crate::capabilities::capabilities_path().exists() {
git(&dir, &["add", "capabilities.json"]).await?;
}
// Stage roles.json when it exists. Written by topology::write_roles /
// reconcile_roles on first role assignment or manager default seeding.
// Without this, roles.json appears as untracked in the meta repo
// (visible in `git status`) which can confuse nix dirty-tree fetches.
if crate::topology::roles_path().exists() {
git(&dir, &["add", "roles.json"]).await?;
}
nix(&dir, &["flake", "lock"]).await?;
if std::path::Path::new(&dir).join("flake.lock").exists() {
git(&dir, &["add", "flake.lock"]).await?;
}
// Build the commit message from what's actually staged so it
// reflects reality — and skip the commit entirely when nothing
// changed (avoids "nothing to commit" errors on redundant syncs).
let staged = git_staged_names(&dir).await?;
if staged.is_empty() {
return Ok(());
}
let msg = if initial {
format!("seed meta from {} agent(s)", agents.len())
} else {
// Compose a message that names every file that actually changed,
// mapping the on-disk filename to a short human label.
let labels: Vec<&str> = staged
.iter()
.filter_map(|f| match f.as_str() {
"flake.nix" => Some("flake"),
"flake.lock" => Some("lock"),
"topology.json" => Some("topology"),
"capabilities.json" => Some("capabilities"),
"tool-groups.json" => Some("tool-groups"),
"roles.json" => Some("roles"),
_ => None,
})
.collect();
if labels.is_empty() {
"meta: update".to_owned()
} else {
format!("meta: update {}", labels.join(", "))
}
};
git_commit(&dir, &msg).await?;
Ok(())
}
/// Phase 1 of an apply-commit deploy. Updates the locked rev of
/// `agent-<name>` to whatever `applied/<name>/main` currently points
/// at and **stages** the lock so `nixos-container update --flake
/// meta#<n>` (which reads via `git+file://`) sees the new rev via
/// the index. Doesn't commit — `finalize_deploy` commits on build
/// success, `abort_deploy` drops the staged change on failure so
/// meta history only carries successful deploys.
#[allow(dead_code)] // wired up by actions::run_apply_commit in a later commit
pub async fn prepare_deploy(name: &str) -> Result<()> {
let _guard = META_LOCK.lock().await;
let dir = meta_dir();
let input = format!("agent-{name}");
nix(&dir, &["flake", "update", &input]).await?;
// Stage the new lock — git+file://'s dirty-tree fetcher reads
// index entries, so the upcoming nixos-container update sees the
// bumped rev without a commit yet.
git(&dir, &["add", "flake.lock"]).await
}
/// Phase 2-success. Commit the staged lock with the deployed tag +
/// sha as the message. No-op when the rev was already at the right
/// place (nothing staged → nothing to commit).
#[allow(dead_code)]
pub async fn finalize_deploy(name: &str, sha: &str, tag: &str) -> Result<()> {
let _guard = META_LOCK.lock().await;
let dir = meta_dir();
if !has_staged_changes(&dir).await? {
return Ok(());
}
let short = &sha[..sha.len().min(12)];
git_commit(&dir, &format!("deploy {name} {tag} {short}")).await
}
/// Phase 2-failure. Unstage + restore the lock so meta returns to
/// the previously-committed shas. The failed proposal is still
/// captured in `applied/<n>`'s annotated `failed/<id>` tag.
#[allow(dead_code)]
pub async fn abort_deploy() -> Result<()> {
let _guard = META_LOCK.lock().await;
let dir = meta_dir();
git(&dir, &["restore", "--staged", "flake.lock"]).await?;
git(&dir, &["restore", "flake.lock"]).await
}
async fn has_staged_changes(dir: &Path) -> Result<bool> {
let st = lifecycle::git_command()
.current_dir(dir)
.args(["diff", "--cached", "--quiet"])
.status()
.await
.with_context(|| format!("git diff --cached in {}", dir.display()))?;
// exit 1 = differences present, 0 = no diff, other = error
match st.code() {
Some(0) => Ok(false),
Some(1) => Ok(true),
_ => bail!("git diff --cached exited unexpectedly"),
}
}
/// One-shot used by the manual-rebuild path: relock just one
/// agent's input and commit the lock change if any. Single-phase
/// (no separate finalize) because rebuild has no failure-revert
/// semantics — it always wants the latest main.
#[allow(dead_code)] // wired up by lifecycle::rebuild in this commit
pub async fn lock_update_for_rebuild(name: &str) -> Result<()> {
let _guard = META_LOCK.lock().await;
let dir = meta_dir();
let input = format!("agent-{name}");
nix(&dir, &["flake", "update", &input]).await?;
if git_is_clean(&dir).await? {
return Ok(());
}
git(&dir, &["add", "flake.lock"]).await?;
git_commit(&dir, &format!("rebuild {name}: lock update")).await
}
/// Update one or more named inputs in the meta flake and commit
/// the resulting lock change with a single combined message.
/// Used by the dashboard's "update meta inputs" form so the
/// operator can bulk-bump `hyperhive` + selected agents in one
/// shot. Each input name is passed verbatim to
/// Run `nix flake update [inputs...]` on the meta flake and commit the
/// resulting lock changes. When `inputs` is empty, updates ALL inputs
/// (bare `nix flake update`). The caller is responsible for picking
/// real input keys (e.g. via `inputs_view()` snapshotted from the lock
/// file) when targeting specific inputs.
pub async fn lock_update(inputs: &[String]) -> Result<()> {
let _guard = META_LOCK.lock().await;
let dir = meta_dir();
let mut args: Vec<&str> = vec!["flake", "update"];
for i in inputs {
args.push(i.as_str());
}
nix(&dir, &args).await?;
if git_is_clean(&dir).await? {
return Ok(());
}
git(&dir, &["add", "flake.lock"]).await?;
let msg = if inputs.is_empty() {
"lock update: all inputs".to_string()
} else if inputs.len() == 1 {
format!("lock update: {}", inputs[0])
} else {
format!("lock update: {}", inputs.join(", "))
};
git_commit(&dir, &msg).await
}
/// One-shot used by the auto-update path: pin the latest hyperhive
/// rev, commit if the lock changed. Cheaper than `sync_agents`
/// because the per-agent inputs aren't touched.
#[allow(dead_code)]
pub async fn lock_update_hyperhive() -> Result<()> {
let _guard = META_LOCK.lock().await;
let dir = meta_dir();
nix(&dir, &["flake", "update", "hyperhive"]).await?;
if git_is_clean(&dir).await? {
return Ok(());
}
git(&dir, &["add", "flake.lock"]).await?;
git_commit(&dir, "bump hyperhive").await
}
/// Write the tool-groups file for `agent` and commit it atomically
/// under `META_LOCK`. Ensures the JSON change is staged + committed
/// before the next `prepare_deploy` or `sync_agents` runs, so the
/// working tree is never left dirty by an untimely PermChange write.
pub async fn commit_tool_groups(agent: &str, groups: &[String]) -> Result<()> {
let _guard = META_LOCK.lock().await;
crate::tool_groups::set_groups(agent, groups)?;
let dir = meta_dir();
if crate::tool_groups::tool_groups_path().exists() {
git(&dir, &["add", "tool-groups.json"]).await?;
}
if has_staged_changes(&dir).await? {
git_commit(&dir, &format!("set tool-groups for {agent}")).await?;
}
Ok(())
}
/// Write the capabilities file for `agent` and commit it atomically
/// under `META_LOCK`. Same rationale as `commit_tool_groups`.
pub async fn commit_capabilities(agent: &str, caps: &[String]) -> Result<()> {
let _guard = META_LOCK.lock().await;
crate::capabilities::set_caps(agent, caps)
.map_err(|e| anyhow::anyhow!("set capabilities for {agent}: {e}"))?;
let dir = meta_dir();
if crate::capabilities::capabilities_path().exists() {
git(&dir, &["add", "capabilities.json"]).await?;
}
if has_staged_changes(&dir).await? {
git_commit(&dir, &format!("set capabilities for {agent}")).await?;
}
Ok(())
}
/// Write the topology file and commit it atomically under `META_LOCK`.
/// Returns `Err(String)` on validation failure (unknown agent, cycle,
/// etc.) — same shape as `topology::set_parent` — so callers can
/// surface the error as a user-visible message. Git failures are
/// logged as warnings and don't propagate: the topology write already
/// succeeded, and `sync_agents` will pick up any un-committed change
/// on the next run as a safety net.
pub async fn commit_topology(
child: &str,
new_parent: Option<&str>,
) -> std::result::Result<(), String> {
let _guard = META_LOCK.lock().await;
crate::topology::set_parent(child, new_parent)?;
let dir = meta_dir();
let stage = async {
git(&dir, &["add", "topology.json"]).await?;
if has_staged_changes(&dir).await? {
git_commit(
&dir,
&format!("topology: {}{}", child, new_parent.unwrap_or("<root>")),
)
.await?;
}
Ok::<_, anyhow::Error>(())
};
if let Err(e) = stage.await {
tracing::warn!(%child, ?new_parent, error = ?e, "commit_topology: topology written but git commit failed (sync_agents will recover)");
}
Ok(())
}
/// Batch variant of [`commit_topology`]: applies every `(child, new_parent)`
/// move under a single `META_LOCK` acquisition and creates **one** git commit
/// for all of them. Moves are applied in the order given; the first
/// validation error short-circuits the whole batch. True atomic write: all
/// moves are pre-validated against a cumulative in-memory state with
/// [`crate::topology::apply_set_parent`] before anything touches disk, then
/// [`crate::topology::write`] is called exactly once. If any move fails
/// validation the topology file is never modified.
///
/// The multi-move commit message uses `moves[0].1` as the destination label.
/// This is intentional: the dashboard bulk-move UI always sends a single
/// destination for all selected agents, so the message is always accurate in
/// practice.
///
/// Returns a `Vec` of `(child, old_parent)` pairs for every move that
/// actually changed the topology (idempotent same-parent moves are skipped),
/// so the caller can send targeted notifications.
///
/// # Errors
///
/// Returns a `String` error if any move fails validation (cycle, unknown
/// agent, etc.) or if the topology file cannot be written. Git-commit failure
/// is logged as a warning and does not propagate — `sync_agents` will recover.
pub async fn bulk_commit_topology(
moves: &[(&str, Option<&str>)],
) -> std::result::Result<Vec<(String, Option<String>)>, String> {
if moves.is_empty() {
return Ok(vec![]);
}
let _guard = META_LOCK.lock().await;
// Snapshot parents before any writes so we can compute the diff.
let topo_before = crate::topology::read();
// Validate all moves against a cumulative in-memory state -- no disk
// writes yet; first error aborts with the topology file untouched.
let mut next = topo_before.clone();
for (child, new_parent) in moves {
next = crate::topology::apply_set_parent(&next, child, *new_parent)?;
}
// Only flush to disk if something actually changed.
if next != topo_before {
crate::topology::write(&next).map_err(|e| format!("{e:#}"))?;
}
// Commit the whole batch as one git operation.
let dir = meta_dir();
let commit_msg = if moves.len() == 1 {
let (child, new_parent) = moves[0];
format!("topology: {}{}", child, new_parent.unwrap_or("<root>"))
} else {
let names: Vec<&str> = moves.iter().map(|(c, _)| *c).collect();
let dest = moves[0].1.unwrap_or("<root>");
format!(
"topology: move {} agents → {} ({})",
moves.len(),
dest,
names.join(", ")
)
};
let stage = async {
git(&dir, &["add", "topology.json"]).await?;
if has_staged_changes(&dir).await? {
git_commit(&dir, &commit_msg).await?;
}
Ok::<_, anyhow::Error>(())
};
if let Err(e) = stage.await {
tracing::warn!(error = ?e, "bulk_commit_topology: topology written but git commit failed (sync_agents will recover)");
}
// Return (child, old_parent) for each move that changed state.
let changed = moves
.iter()
.filter_map(|(child, new_parent)| {
let old = topo_before.get(*child).cloned().flatten();
if old.as_deref() != *new_parent {
Some((child.to_string(), old))
} else {
None
}
})
.collect();
Ok(changed)
}
fn render_flake(
hyperhive_flake: &str,
nixpkgs_flake: &str,
nixpkgs_unstable_flake: &str,
dashboard_port: u16,
operator_pronouns: &str,
context_window_tokens: &std::collections::HashMap<String, u64>,
agents: &[AgentSpec],
) -> String {
render_flake_with_lookup(
hyperhive_flake,
nixpkgs_flake,
nixpkgs_unstable_flake,
dashboard_port,
operator_pronouns,
context_window_tokens,
agents,
agent_canonical_inputs,
)
}
/// Canonical inputs meta knows how to dedup. An agent that declares one
/// of these as a top-level input in its own `flake.nix` will get a
/// `follows = "<name>"` line emitted in meta — collapsing the
/// otherwise-separate-but-identical `nixpkgs_N` nodes into a single
/// meta-level reference.
const CANONICAL_INPUTS: &[&str] = &["nixpkgs", "nixpkgs-unstable"];
/// Env vars hive-c0re forwards from its own systemd unit env into every
/// sub-agent's harness service env. Each entry is `(env_var_name,
/// host_value)`. Empty / unset vars are filtered out so absent options
/// don't render no-op `FOO = ""` lines into the meta flake.
///
/// Returns owned `String` values so the result is `'static`-friendly +
/// trivial to stub from tests (which build their own slice instead of
/// touching process-wide env).
const FORWARDED_VARS: &[&str] = &[
"HIVE_FORGE_URL",
"HIVE_FORGE_PUBLIC_URL",
"HYPERHIVE_PEERS",
"HYPERHIVE_HIVE_DOMAIN",
"HYPERHIVE_HIVE_NAME",
"HYPERHIVE_SWARM_NAME",
];
fn forwarded_env_vars() -> Vec<(&'static str, String)> {
FORWARDED_VARS
.iter()
.filter_map(|&name| {
std::env::var(name)
.ok()
.filter(|v| !v.is_empty())
.map(|v| (name, v))
})
.collect()
}
/// Read an agent's applied `flake.lock` and return the subset of
/// `CANONICAL_INPUTS` it declares as direct (root-level) inputs.
/// Returns an empty vec when the lock is missing or unparsable —
/// safe degradation, the worst case is no dedup for that agent.
fn agent_canonical_inputs(name: &str) -> Vec<&'static str> {
let path = std::path::PathBuf::from(format!("{APPLIED_ROOT}/{name}/flake.lock"));
let Ok(raw) = std::fs::read_to_string(&path) else {
return Vec::new();
};
let Ok(json) = serde_json::from_str::<serde_json::Value>(&raw) else {
return Vec::new();
};
let Some(nodes) = json.get("nodes").and_then(|v| v.as_object()) else {
return Vec::new();
};
let Some(root_name) = json.get("root").and_then(|v| v.as_str()) else {
return Vec::new();
};
let Some(root_inputs) = nodes
.get(root_name)
.and_then(|n| n.get("inputs"))
.and_then(|v| v.as_object())
else {
return Vec::new();
};
CANONICAL_INPUTS
.iter()
.copied()
.filter(|canon| root_inputs.contains_key(*canon))
.collect()
}
/// Inner render helper accepting a lookup fn so tests can stub the
/// agent flake-lock introspection.
#[allow(
clippy::too_many_lines,
reason = "templated string-builder for the meta flake — the length is one \
contiguous fmt block, splitting it would just hide the shape"
)]
fn render_flake_with_lookup<F>(
hyperhive_flake: &str,
nixpkgs_flake: &str,
nixpkgs_unstable_flake: &str,
dashboard_port: u16,
operator_pronouns: &str,
context_window_tokens: &std::collections::HashMap<String, u64>,
agents: &[AgentSpec],
lookup: F,
) -> String
where
F: Fn(&str) -> Vec<&'static str>,
{
use std::fmt::Write as _;
let mut out = String::new();
out.push_str("{\n description = \"hyperhive deployed agents\";\n inputs = {\n");
// `nixpkgs` + `nixpkgs-unstable` are top-level meta inputs with explicit
// store-path URLs. `hyperhive` then follows them via
// `hyperhive.inputs.*.follows`. This cascades through to every agent
// because `agent-<n>.inputs.nixpkgs.follows = "nixpkgs"` resolves to
// the same top-level node.
//
// Why explicit `path:` URLs instead of
// `nixpkgs.follows = "hyperhive/nixpkgs"`:
// meta points to hyperhive's *store path* as its flake input, so nix
// reads hyperhive's own pinned lock when evaluating that input — the
// host-level `follows` the operator set never propagates. Injecting the
// evaluated `pkgs.path` / nixpkgs-unstable path directly at nix-module
// evaluation time is the only reliable way to honour the host's channel
// choice.
//
// Fallback (both flake args empty): legacy `follows` wiring — used when
// hive-c0re is not built with this option wired up.
if nixpkgs_flake.is_empty() {
// Legacy path: meta defers to hyperhive's own lock.
let _ = writeln!(out, " hyperhive.url = \"{hyperhive_flake}\";");
out.push_str(" nixpkgs.follows = \"hyperhive/nixpkgs\";\n");
out.push_str(" nixpkgs-unstable.follows = \"hyperhive/nixpkgs-unstable\";\n");
} else {
let _ = writeln!(out, " nixpkgs.url = \"{nixpkgs_flake}\";");
let _ = writeln!(
out,
" nixpkgs-unstable.url = \"{nixpkgs_unstable_flake}\";"
);
let _ = writeln!(out, " hyperhive.url = \"{hyperhive_flake}\";");
out.push_str(" hyperhive.inputs.nixpkgs.follows = \"nixpkgs\";\n");
out.push_str(" hyperhive.inputs.nixpkgs-unstable.follows = \"nixpkgs-unstable\";\n");
}
for spec in agents {
let _ = writeln!(
out,
" agent-{}.url = \"git+file://{APPLIED_ROOT}/{}\";",
spec.name, spec.name,
);
// For each canonical input the agent declares in its own
// `flake.nix` (detected by reading its applied `flake.lock`),
// emit `inputs.agent-<name>.inputs.<canon>.follows = "<canon>"`.
// Collapses otherwise-separate-but-identical nixpkgs nodes
// (root + every agent's own nixpkgs) into one. Skipped
// silently for agents that don't declare the input — emitting
// follows on a non-existent input would error at
// `nix flake lock` time.
for canon in lookup(&spec.name) {
let _ = writeln!(
out,
" agent-{}.inputs.{canon}.follows = \"{canon}\";",
spec.name,
);
}
}
out.push_str(" };\n outputs =\n { self, hyperhive, ... }@inputs:\n let\n");
// Free-text operator string — escape backslash + double-quote so a
// pronouns value like `he/him \ "rare"` round-trips into a valid
// nix string literal without breaking the flake.
let pronouns_escaped = operator_pronouns.replace('\\', "\\\\").replace('"', "\\\"");
let _ = writeln!(
out,
" dashboardPort = {dashboard_port};\n operatorPronouns = \"{pronouns_escaped}\";\n mkAgent = {{ name, isManager, port, parent ? null, toolGroups ? null, capabilities ? null }}:"
);
out.push_str(
r#" let
base = if isManager
then hyperhive.nixosConfigurations.ruth
else hyperhive.nixosConfigurations.agent-base;
input = inputs."agent-${name}";
service = "hive-ag3nt";
parentEnv = if parent == null then {} else { HIVE_PARENT = parent; };
toolGroupsEnv = if toolGroups == null then {} else { HIVE_TOOL_GROUPS = toolGroups; };
capabilitiesEnv = if capabilities == null then {} else { HIVE_CAPABILITIES = capabilities; };
in
base.extendModules {
modules = [
input.nixosModules.default
{
# The harness service inside the container runs as a
# non-root unix user named after the agent (`damocles`,
# `iris`, `root`, …). UID auto-assigned by NixOS; the
# per-agent override here is what makes
# `hyperhive.user.name` match the agent's identity
# instead of the harness-base default of `"agent"`.
hyperhive.user.name = name;
programs.git.config.user = {
name = name;
email = "${name}@hyperhive.local";
};
# Container-wide env: every service + co-process daemon can
# resolve the agent's durable state dir without hard-coding it.
# `environment.variables` only writes /etc/environment (login
# shells); `systemd.globalEnvironment` is the analogue for
# systemd units so tea-login / forge-avatar-sync /
# matrix-avatar-sync etc. can read `$HYPERHIVE_STATE_DIR`
# without each service having to redeclare it.
environment.variables = {
HIVE_LABEL = name;
HYPERHIVE_STATE_DIR = "/agents/${name}/state";
HYPERHIVE_HARNESS_DIR = "/agents/${name}/harness";
};
systemd.globalEnvironment = {
HIVE_LABEL = name;
HYPERHIVE_STATE_DIR = "/agents/${name}/state";
HYPERHIVE_HARNESS_DIR = "/agents/${name}/harness";
};
systemd.services.${service}.environment = parentEnv // toolGroupsEnv // capabilitiesEnv // {
HIVE_PORT = toString port;
HIVE_LABEL = name;
HIVE_DASHBOARD_PORT = toString dashboardPort;
HIVE_OPERATOR_PRONOUNS = operatorPronouns;"#,
);
// Per-model context-window env vars declared in the host-level
// `services.hive-c0re.contextWindowTokens` option. Use a sorted
// iterator for deterministic flake output (no spurious git diffs).
let mut sorted_tokens: Vec<(&String, &u64)> = context_window_tokens.iter().collect();
sorted_tokens.sort_by_key(|(k, _)| k.as_str());
for (key, val) in &sorted_tokens {
let upper_key = key.to_ascii_uppercase();
let _ = writeln!(
out,
" HIVE_CONTEXT_WINDOW_TOKENS_{upper_key} = \"{val}\";"
);
}
// Forwarded env vars — picked up from hive-c0re's own systemd unit
// env (`services.hyperhive.*` options flow through nix/modules/
// hive-c0re.nix into the host process). We copy whatever's set into
// each sub-agent's harness service env so the in-container surfaces
// (`identity.rs`, `forge_notify`) see a consistent view across the
// whole hive. Absent host-side env (option not set) → skip emission
// → in-container accessors fall back to None / defaults gracefully.
//
// - HIVE_FORGE_URL: agents poll this for Forgejo notifications.
// - HYPERHIVE_HIVE_DOMAIN: machine-readable hive DNS.
// - HYPERHIVE_HIVE_NAME / HYPERHIVE_SWARM_NAME: human display
// names for hive + swarm.
for (var, val) in forwarded_env_vars() {
let escaped = val.replace('\\', "\\\\").replace('"', "\\\"");
let _ = writeln!(out, " {var} = \"{escaped}\";");
}
out.push_str(
r#" HYPERHIVE_STATE_DIR = "/agents/${name}/state";
HYPERHIVE_HARNESS_DIR = "/agents/${name}/harness";
};
}
];
};
in
{
nixosConfigurations = {
"#,
);
// Pull the topology map once and look up each agent's parent. An
// empty / absent topology.json yields `parent = null` for everyone
// (every container at root). `meta::sync_agents` seeds the file
// on first run with manager as root + everyone else under manager.
let topology = crate::topology::read();
let tool_groups_map = crate::tool_groups::read();
let capabilities_map = crate::capabilities::read();
for spec in agents {
let parent_attr = topology
.get(&spec.name)
.and_then(|p| p.as_ref())
.map_or_else(|| "null".to_owned(), |p| format!("\"{p}\""));
// Emit `toolGroups = "group1,group2"` when the operator has
// explicitly configured groups for this agent. Absent entry = null
// = harness falls back to AGENT_DEFAULT (no env var emitted,
// no rebuild cascade for agents whose groups haven't changed).
let groups = tool_groups_map.get(&spec.name).cloned().unwrap_or_default();
let tool_groups_attr = if groups.is_empty() {
"null".to_owned()
} else {
let joined = groups.join(",");
format!("\"{joined}\"")
};
// Emit `capabilities = "cap1,cap2"` when the operator has
// granted capabilities to this agent. Absent entry = null = no
// capability env var injected, capability-gated tools hidden.
let caps = capabilities_map
.get(&spec.name)
.cloned()
.unwrap_or_default();
let capabilities_attr = if caps.is_empty() {
"null".to_owned()
} else {
let joined = caps.join(",");
format!("\"{joined}\"")
};
let _ = writeln!(
out,
" {} = mkAgent {{ name = \"{}\"; isManager = {}; port = {}; parent = {}; toolGroups = {}; capabilities = {}; }};",
spec.name,
spec.name,
if spec.is_manager { "true" } else { "false" },
spec.port,
parent_attr,
tool_groups_attr,
capabilities_attr,
);
}
out.push_str(" };\n };\n}\n");
out
}
async fn git_is_clean(dir: &Path) -> Result<bool> {
let out = lifecycle::git_command()
.current_dir(dir)
.args(["status", "--porcelain"])
.output()
.await
.with_context(|| format!("git status in {}", dir.display()))?;
Ok(out.stdout.iter().all(u8::is_ascii_whitespace))
}
/// Return the list of file names that are currently staged (index differs
/// from HEAD). On the initial commit (`HEAD` doesn't exist yet) falls back
/// to `git diff --cached --name-only HEAD` failing gracefully by using
/// `git status --porcelain` and collecting the `A ` / `M ` prefix lines.
async fn git_staged_names(dir: &Path) -> Result<Vec<String>> {
// `--diff-filter=ACM` skips deleted entries — we only care about
// additions and modifications for message-building purposes.
let out = lifecycle::git_command()
.current_dir(dir)
.args(["diff", "--cached", "--name-only", "--diff-filter=ACM"])
.output()
.await
.with_context(|| format!("git diff --cached in {}", dir.display()))?;
if out.status.success() {
let names = String::from_utf8_lossy(&out.stdout)
.lines()
.map(str::trim)
.filter(|l| !l.is_empty())
.map(ToOwned::to_owned)
.collect();
return Ok(names);
}
// Fallback for initial commit (no HEAD yet): parse `git status --porcelain`.
let st = lifecycle::git_command()
.current_dir(dir)
.args(["status", "--porcelain"])
.output()
.await
.with_context(|| format!("git status in {}", dir.display()))?;
let names = String::from_utf8_lossy(&st.stdout)
.lines()
.filter(|l| l.starts_with("A ") || l.starts_with("M "))
.filter_map(|l| l.get(3..))
.map(ToOwned::to_owned)
.collect();
Ok(names)
}
async fn git(dir: &Path, args: &[&str]) -> Result<()> {
let out = lifecycle::git_command()
.current_dir(dir)
.args(args)
.output()
.await
.with_context(|| format!("git {} in {}", args.join(" "), dir.display()))?;
if !out.status.success() {
bail!(
"git {} failed ({}): {}",
args.join(" "),
out.status,
String::from_utf8_lossy(&out.stderr).trim()
);
}
Ok(())
}
async fn git_commit(dir: &Path, message: &str) -> Result<()> {
git(
dir,
&[
"-c",
&format!("user.name={GIT_NAME}"),
"-c",
&format!("user.email={GIT_EMAIL}"),
"commit",
"-m",
message,
],
)
.await?;
// Best-effort mirror to the bundled forge. No-op when the forge
// isn't seeded (no core token on disk); push failures log a warn
// but don't bubble up — a missing mirror shouldn't fail an
// otherwise successful deploy.
if let Err(e) = crate::forge::push_meta(dir).await {
tracing::warn!(error = ?e, "forge: meta push after commit failed (non-fatal)");
}
Ok(())
}
async fn nix(dir: &Path, args: &[&str]) -> Result<()> {
// `--extra-experimental-features` belt-and-suspenders for hosts
// that haven't set this in nix.conf. The hyperhive module's
// deploy guide assumes flakes are already enabled, but the cost
// of being defensive is one extra argv each call.
let mut all = vec!["--extra-experimental-features", "nix-command flakes"];
all.extend(args);
let out = Command::new("nix")
.current_dir(dir)
.args(&all)
.output()
.await
.with_context(|| format!("nix {} in {}", args.join(" "), dir.display()))?;
if !out.status.success() {
bail!(
"nix {} failed ({}): {}",
args.join(" "),
out.status,
String::from_utf8_lossy(&out.stderr).trim()
);
}
Ok(())
}
#[cfg(test)]
mod tests {
use super::*;
fn sample_spec(name: &str, is_manager: bool, port: u16) -> AgentSpec {
AgentSpec {
name: name.to_owned(),
is_manager,
port,
}
}
#[test]
fn render_flake_uses_explicit_nixpkgs_url_when_provided() {
let out = render_flake(
"github:example/hyperhive",
"path:/nix/store/aaaa-nixpkgs-source",
"path:/nix/store/bbbb-nixpkgs-unstable-source",
8000,
"she/her",
&std::collections::HashMap::new(),
&[sample_spec("alice", false, 9001)],
);
// Both nixpkgs + nixpkgs-unstable are top-level inputs with
// explicit URLs; hyperhive follows them.
assert!(
out.contains("nixpkgs.url = \"path:/nix/store/aaaa-nixpkgs-source\""),
"expected explicit nixpkgs.url:\n{out}"
);
assert!(
out.contains("nixpkgs-unstable.url = \"path:/nix/store/bbbb-nixpkgs-unstable-source\""),
"expected explicit nixpkgs-unstable.url:\n{out}"
);
assert!(
out.contains("hyperhive.inputs.nixpkgs.follows = \"nixpkgs\""),
"expected hyperhive.inputs.nixpkgs.follows:\n{out}"
);
assert!(
out.contains("hyperhive.inputs.nixpkgs-unstable.follows = \"nixpkgs-unstable\""),
"expected hyperhive.inputs.nixpkgs-unstable.follows:\n{out}"
);
assert!(
!out.contains("nixpkgs.follows = \"hyperhive"),
"old-style follows must not appear when flake args are set:\n{out}"
);
}
#[test]
fn render_flake_falls_back_to_follows_when_nixpkgs_flake_empty() {
// Empty nixpkgs_flake → legacy follows behaviour (backward compat
// for any code path that can't inject pkgs.path).
let out = render_flake(
"github:example/hyperhive",
"",
"",
8000,
"she/her",
&std::collections::HashMap::new(),
&[sample_spec("alice", false, 9001)],
);
assert!(
out.contains("nixpkgs.follows = \"hyperhive/nixpkgs\""),
"expected fallback follows:\n{out}"
);
assert!(
out.contains("nixpkgs-unstable.follows = \"hyperhive/nixpkgs-unstable\""),
"expected fallback unstable follows:\n{out}"
);
assert!(
!out.contains("nixpkgs.url ="),
"no explicit url should be emitted in fallback mode:\n{out}"
);
}
#[test]
fn render_flake_emits_follows_for_agents_declaring_nixpkgs() {
// Stub lookup: pretend `bitburner` declares `nixpkgs` at its
// root, while `argus` has no canonical inputs at all.
let lookup = |name: &str| -> Vec<&'static str> {
match name {
"bitburner" => vec!["nixpkgs"],
"dmatrix" => vec!["nixpkgs", "nixpkgs-unstable"],
_ => vec![],
}
};
let out = render_flake_with_lookup(
"github:example/hyperhive",
"path:/nix/store/aaaa-nixpkgs-source",
"path:/nix/store/bbbb-nixpkgs-unstable-source",
8000,
"she/her",
&std::collections::HashMap::new(),
&[
sample_spec("argus", false, 9001),
sample_spec("bitburner", false, 9002),
sample_spec("dmatrix", false, 9003),
],
lookup,
);
// bitburner declares nixpkgs → follows emitted.
assert!(
out.contains("agent-bitburner.inputs.nixpkgs.follows = \"nixpkgs\""),
"missing bitburner nixpkgs follows:\n{out}"
);
// dmatrix declares both → both follows emitted.
assert!(out.contains("agent-dmatrix.inputs.nixpkgs.follows = \"nixpkgs\""));
assert!(
out.contains("agent-dmatrix.inputs.nixpkgs-unstable.follows = \"nixpkgs-unstable\"")
);
// argus declares neither → no follows emitted for it. Asserting
// ABSENCE is the important bit: emitting a follows on a
// non-existent input errors at `nix flake lock` time.
assert!(
!out.contains("agent-argus.inputs.nixpkgs"),
"argus shouldn't have nixpkgs follows:\n{out}"
);
}
#[test]
fn render_flake_skips_canonical_follows_when_lookup_returns_empty() {
let out = render_flake_with_lookup(
"github:example/hyperhive",
"path:/nix/store/aaaa-nixpkgs-source",
"path:/nix/store/bbbb-nixpkgs-unstable-source",
8000,
"she/her",
&std::collections::HashMap::new(),
&[sample_spec("alice", false, 9001)],
|_| Vec::new(),
);
// No agent-side follows when the lookup reports nothing
// declared — protects agents whose flake.lock can't be read
// (missing / unparsable) from being broken by a follows on a
// non-existent input.
assert!(
!out.contains("agent-alice.inputs."),
"alice shouldn't have any inputs follows:\n{out}"
);
}
}