vinzenz/hyperhive
vinzenz/hyperhive
Watch
0
0
Fork
You've already forked hyperhive
 0
Code Activity

build agents only from internal/meta; delete the hive's local meta flake

Browse source 
The hive no longer renders, locks or commits a meta flake. A rebuild's
MetaSync node picks the internal/meta commit and records it with the
hyperhive and config pins it holds; Prebuild, Swap and Create build from it,
and a head that does not pin the agent's current config fails the build
naming the agent. hive-priv's meta_rev is required.

Removed: meta.rs sync/lock/render, the manual-rebuild relock, the boot
sweep's hyperhive bump (MetaLock), the update-meta-inputs form and its
panel, MetaWindow, the hive's hyperhive_rev setting.

Kept: agent_set::refresh holds what sync_agents also did (topology.json, the
gateway's agent-socket map and nginx include), called from spawn, destroy,
rebuild and the migration.

The boot reconcile and the stale-start upgrade compare the agent's marker
with the hyperhive pin of internal/meta; the dashboard's deployed chip reads
the config marker a successful build writes.
This commit is contained in:
atlas 2026-10-25 13:33:15 +01:00
commit 4aa11b3a94
64 changed files with 1042 additions and 3793 deletions
Download patch file Download diff file Expand all files Collapse all files

6
hive-c0re/src/actions.rs
View file

@ -33,12 +33,12 @@ pub(crate) enum RevAdvance {
/// The rev is an ancestor of `applied/main`, so a later commit already
/// carries it and there is nothing to deploy. Deploy requests are not
/// ordered: two merges' requests can arrive in either order, and a
/// meta-update cascade fast-forwards `main` to the forge's on its own.
/// forge fetch fast-forwards `main` to the forge's on its own.
Superseded,
}
/// Fast-forward `agent`'s `applied/main` to `rev`, a commit the swarm asks
/// this hive to deploy, so the next relocking rebuild builds it.
/// this hive to deploy, so the dashboard's update chip compares against it.
///
/// No eval-verify on this path: a config that does not evaluate fails the
/// rebuild instead, with `applied/main` already at `rev`.
@ -61,7 +61,7 @@ pub(crate) async fn advance_applied_to_rev(agent: &str, rev: &str) -> Result<Rev
/// testable against a local repo.
///
/// `applied/main` is compared to `rev` twice: before the fetch, so an applied
/// rev costs no forge round-trip, and after it, because a meta-update cascade
/// rev costs no forge round-trip, and after it, because a forge fetch
/// fast-forwards `main` to the forge's and may do so meanwhile.
async fn advance_applied_main<T>(
applied_dir: &std::path::Path,

15
hive-c0re/src/agent_config/archetypes.rs
View file

@ -1,17 +1,16 @@
//! The archetype each of this hive's archetype agents is built from, as last
//! read from the swarm's `agent-archetype` bucket
//! ([`swarm_queue_client::agent_archetype`]). Stored at
//! `/var/lib/hyperhive/swarm-agent-archetypes.json`, outside the meta repo:
//! the meta render runs at boot, possibly before the queue answers, so it
//! reads this file rather than the bucket.
//! `/var/lib/hyperhive/swarm-agent-archetypes.json`: a rebuild at boot may run
//! before the queue answers, so it reads this file rather than the bucket.
//!
//! Format: agent name → [`ArchetypeRef`]. An agent with an entry has no config
//! repo: the meta flake builds it from the archetype's module, and the paths
//! that seed, relock or advance `applied/<name>` skip it. An agent with no
//! entry is a custom agent.
//! repo: `internal/meta` builds it from the archetype's module, and the paths
//! that seed or advance `applied/<name>` skip it. An agent with no entry is a
//! custom agent.
//!
//! [`refresh`] reads one agent's key; provisioning and every rebuild's meta
//! sync call it before they render the meta flake.
//! sync call it before they choose the `internal/meta` commit to build from.
use std::collections::BTreeMap;
use std::path::{Path, PathBuf};
@ -74,7 +73,7 @@ fn store_read(path: &Path, agent: &str, read: Read) -> Result<()> {
}
Read::Current(value) => value,
};
// Both are rendered into the meta flake as nix strings.
// Both are later compared with the archetype `internal/meta` lists.
if let Some(ArchetypeRef { source, name }) = &value {
for part in [source, name] {
hive_types::Ident::parse(part)

2
hive-c0re/src/agent_config/topology.rs
View file

@ -44,7 +44,7 @@ enum OnDisk {
}
/// A missing file is the empty roster (a fresh install that hasn't run
/// through `meta::sync_agents` yet).
/// through `agent_set::refresh` yet).
impl Default for OnDisk {
fn default() -> Self {
Self::Roster(BTreeSet::new())

164
hive-c0re/src/agent_set.rs Normal file
View file

@ -0,0 +1,164 @@
//! The host-side files that follow the set of agents on this hive.
//!
//! [`refresh`] runs wherever the set changes (spawn, destroy, rebuild, the
//! startup migration): it brings `topology.json`, the gateway's agent-socket
//! map and its nginx include in line with the live agents.
use std::path::Path;
use anyhow::{Context, Result, bail};
use tokio::sync::Mutex;
/// Serialises [`refresh`]: `topology.json` is read, changed and written back,
/// so two concurrent refreshes would lose one of the changes.
static REFRESH: Mutex<()> = Mutex::const_new(());
/// Bring the host-side files in line with `agents`, the live agent set.
///
/// An empty `agents` is never intentional: it means `nixos-container list`
/// failed and the caller fell back to an empty list, and writing it would drop
/// every agent from the topology and the gateway. It is refused.
///
/// # Errors
/// `HIVE_FORGE_URL` is unset, or `topology.json` could not be reconciled. The
/// gateway files are best-effort: a failure there is logged, because the
/// 10-second poll in `workers::agent_sockets` writes them again.
pub async fn refresh(agents: &[hive_meta_flake::AgentSpec]) -> Result<()> {
let _guard = REFRESH.lock().await;
require_forge_url(std::env::var("HIVE_FORGE_URL").ok().as_deref())?;
let dir = crate::paths::meta_root();
std::fs::create_dir_all(&dir).with_context(|| format!("create {}", dir.display()))?;
let Some(agent_names) = live_names(agents) else {
tracing::warn!(
"agent_set: refusing to refresh with an empty agent list \
(nixos-container list may have failed)"
);
return Ok(());
};
remove_stale_ca_files(&dir);
let pending: Vec<String> = crate::coordinator::Coordinator::pending_init_names()
.into_iter()
.map(hive_types::Ident::into_string)
.collect();
crate::topology::reconcile(&agent_names, &pending)
.with_context(|| format!("reconcile {}", crate::topology::topology_path().display()))?;
// Every agent binds a unix socket (`HIVE_WEB_SOCKET`); the gateway routes
// via this map, falling back to a computed TCP loopback port
// (`lifecycle::agent_web_port`) only while an agent's socket marker is
// absent. See `docs/networking/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)");
}
// c0re is unprivileged, so the nginx reload goes through hive-priv.
if let Err(e) = crate::gateway_nginx::write(&agent_names).await {
tracing::warn!(error = ?e, "gateway_nginx::write failed (non-fatal)");
}
Ok(())
}
/// The names of `agents`, or `None` for an empty list, which [`refresh`]
/// refuses.
fn live_names(agents: &[hive_meta_flake::AgentSpec]) -> Option<Vec<String>> {
(!agents.is_empty()).then(|| agents.iter().map(|a| a.name.clone()).collect())
}
/// The forge is not optional on a real hive: the hyperhive NixOS module sets
/// `HIVE_FORGE_URL` unconditionally on the hive-c0re unit, so its absence
/// means this daemon was started outside it. The agent option is nullable
/// (`null` legitimately means "no forge" when the modules are evaluated on
/// their own), so nothing downstream would object: every agent would come up
/// with no forge login and no hint that was unintended.
fn require_forge_url(value: Option<&str>) -> Result<()> {
if value.is_none_or(str::is_empty) {
bail!(
"HIVE_FORGE_URL is unset — the hyperhive NixOS module sets it unconditionally on \
the hive-c0re unit, so this process was started outside that unit. Refusing to \
register agents that would every one of them have no forge configured."
);
}
Ok(())
}
/// Delete the CA files earlier renders left next to the meta flake
/// (`hive-ca.pem`, and the per-peer `peer-ca-<N>.pem` before it). Agents fetch
/// the swarm root from the store at boot, so nothing reads one.
fn remove_stale_ca_files(dir: &Path) {
let Ok(entries) = std::fs::read_dir(dir) else {
return;
};
for e in entries.flatten() {
let fname = e.file_name();
let Some(name) = fname.to_str() else { continue };
if is_stale_ca_name(name) {
let _ = std::fs::remove_file(dir.join(name));
}
}
}
/// True for the filenames [`remove_stale_ca_files`] owns.
fn is_stale_ca_name(name: &str) -> bool {
name == "hive-ca.pem"
|| (name.starts_with("peer-ca-")
&& Path::new(name)
.extension()
.is_some_and(|ext| ext.eq_ignore_ascii_case("pem")))
}
#[cfg(test)]
mod tests {
use super::*;
#[test]
fn an_empty_agent_list_is_refused_and_a_live_one_is_named() {
let spec = |name: &str| hive_meta_flake::AgentSpec {
name: name.to_owned(),
port: hive_meta_flake::agent_web_port(name),
archetype: None,
pin: None,
};
assert_eq!(live_names(&[]), None);
assert_eq!(
live_names(&[spec("alice"), spec("bob")]),
Some(vec!["alice".to_owned(), "bob".to_owned()])
);
}
#[test]
fn a_forge_url_is_required() {
require_forge_url(Some("http://forge.example.test"))
.expect("a forge URL satisfies the requirement");
for missing in [None, Some("")] {
let err = require_forge_url(missing).expect_err("no forge URL must stop the refresh");
assert!(
err.to_string().contains("HIVE_FORGE_URL is unset"),
"the error must name the missing variable: {err}"
);
}
}
#[test]
fn stale_ca_files_are_removed_and_nothing_else() {
let dir = tempfile::tempdir().expect("tempdir");
for name in [
"hive-ca.pem",
"peer-ca-0.pem",
"peer-ca-1.PEM",
"topology.json",
] {
std::fs::write(dir.path().join(name), "x").expect("write");
}
// Not CA files this machinery owns, though they look similar.
for name in ["peer-ca-0.txt", "other-ca.pem"] {
std::fs::write(dir.path().join(name), "x").expect("write");
}
remove_stale_ca_files(dir.path());
for gone in ["hive-ca.pem", "peer-ca-0.pem", "peer-ca-1.PEM"] {
assert!(!dir.path().join(gone).exists(), "{gone} must go");
}
for kept in ["topology.json", "peer-ca-0.txt", "other-ca.pem"] {
assert!(dir.path().join(kept).exists(), "{kept} must survive");
}
}
}

143
hive-c0re/src/container_view.rs
View file

@ -5,7 +5,6 @@
//! diff against the last snapshot to fire
//! `ContainerStateChanged` / `ContainerRemoved` events.
use std::collections::HashMap;
use std::path::Path;
use chrono::{DateTime, Utc};
@ -45,8 +44,8 @@ pub struct ContainerView {
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.
/// First 12 chars of the config commit the agent's last successful build
/// pinned (`paths::applied_config_marker`).
#[serde(skip_serializing_if = "Option::is_none")]
pub deployed_sha: Option<String>,
/// The Claude model the agent's harness is currently using, read from
@ -142,7 +141,7 @@ fn agent_url_for_domain(domain: &str, name: &str) -> String {
/// The container list could not be read.
pub async fn build_all() -> anyhow::Result<Vec<ContainerView>> {
let raw = lifecycle::list().await?;
let locked = read_meta_locked_revs();
let mut out = Vec::new();
for c in &raw {
let Some(logical) = c.strip_prefix(AGENT_PREFIX) else {
@ -154,12 +153,12 @@ pub async fn build_all() -> anyhow::Result<Vec<ContainerView>> {
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 deployed_full =
std::fs::read_to_string(crate::paths::applied_config_marker(logical.as_str())).ok();
let needs_update =
crate::auto_update::agent_config_pending(logical.as_str(), deployed_full).await;
let deployed_sha = deployed_full.map(short_rev);
crate::auto_update::agent_config_pending(logical.as_str(), deployed_full.as_deref())
.await;
let deployed_sha = deployed_full.as_deref().map(short_rev);
// 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.
@ -380,64 +379,15 @@ pub fn hive_swarm_names() -> (Option<String>, Option<String>) {
(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
}
/// First 12 characters of a rev. Counts chars, not bytes: the value comes
/// from `flake.lock` JSON, and a byte slice at 12 can split a multi-byte char.
/// First 12 characters of a rev. Counts chars, not bytes, so a byte slice at 12
/// cannot split a multi-byte char.
fn short_rev(rev: &str) -> String {
rev.chars().take(12).collect()
}
#[cfg(test)]
mod tests {
use super::{parse_locked_revs, short_rev};
use super::short_rev;
#[test]
fn short_rev_truncates_and_keeps_short() {
@ -452,77 +402,6 @@ mod tests {
assert_eq!(short_rev(rev), "0123456789aé");
}
/// 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` (no

83
hive-c0re/src/coordinator.rs
View file

@ -37,12 +37,8 @@ pub struct Coordinator {
/// `build_logs.rs` for retention.
pub build_logs: Arc<crate::build_logs::BuildLogs>,
/// Store path of the hyperhive flake source (no fragment). Used to
/// identify the running revision (`current_flake_rev`); the meta flake
/// pins `hyperhive` by forge URL and [`Self::hyperhive_rev`] instead.
/// identify the running revision (`current_flake_rev`).
pub hyperhive_flake: String,
/// Git rev of the hyperhive checkout this host runs. Pins the meta
/// flake's `hyperhive` input; `None` makes the meta render fail.
pub hyperhive_rev: Option<String>,
/// Store path of the `claude-code` build every agent runs, handed to
/// each container at start as `HYPERHIVE_CLAUDE_CODE_PATH`. Set by
/// the NixOS module option
@ -139,14 +135,7 @@ pub struct Coordinator {
/// 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
@ -155,8 +144,8 @@ pub struct Coordinator {
/// 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
/// Global job-DAG queue. Every container op (rebuild,
/// 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
@ -189,8 +178,6 @@ pub struct Coordinator {
#[serde(default)]
pub struct HiveEnv {
pub hyperhive_flake: String,
/// Git rev of the hyperhive checkout this host runs; see [`Coordinator::hyperhive_rev`].
pub hyperhive_rev: Option<String>,
/// Store path of the `claude-code` agents run, or `None` for "each
/// agent keeps the one out of its own nixpkgs". Travels into the
/// container as `HYPERHIVE_CLAUDE_CODE_PATH` — a plain string, kept
@ -214,7 +201,6 @@ impl Default for HiveEnv {
fn default() -> Self {
Self {
hyperhive_flake: "/etc/hyperhive".to_string(),
hyperhive_rev: None,
claude_code_path: None,
dashboard_port: 7000,
agent_cpu_quota: "200%".to_string(),
@ -334,33 +320,6 @@ fn fold_tombstones_by_agent<'a>(
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,
});
}
}
}
/// 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).
@ -390,7 +349,6 @@ impl Coordinator {
) -> Result<Self> {
let HiveEnv {
hyperhive_flake,
hyperhive_rev,
claude_code_path,
dashboard_port,
agent_cpu_quota,
@ -428,7 +386,6 @@ impl Coordinator {
broker: Arc::new(broker),
build_logs,
hyperhive_flake,
hyperhive_rev,
claude_code_path,
dashboard_port,
agent_cpu_quota,
@ -443,7 +400,7 @@ impl Coordinator {
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,
@ -452,13 +409,12 @@ impl Coordinator {
}
/// Snapshot the hive-wide configuration fields as a [`HiveEnv`].
/// Pass the result to `lifecycle::provision_container` / `rebuild` / `meta::sync_agents`
/// Pass the result to `lifecycle::swap_update` / `write_dropins`
/// instead of threading the individual fields separately.
#[must_use]
pub fn hive_env(&self) -> HiveEnv {
HiveEnv {
hyperhive_flake: self.hyperhive_flake.clone(),
hyperhive_rev: self.hyperhive_rev.clone(),
claude_code_path: self.claude_code_path.clone(),
dashboard_port: self.dashboard_port,
agent_cpu_quota: self.agent_cpu_quota.clone(),
@ -553,31 +509,6 @@ impl Coordinator {
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
}
/// Rebuild the per-container snapshot, diff it against the last
/// one cached on `self`, and emit one
/// `DashboardEvent::ContainerStateChanged` per added/changed row
@ -586,7 +517,7 @@ impl Coordinator {
/// `nixos-container list` returns or what a row's
/// `running` / `needs_update` / `needs_login` / `deployed_sha`
/// resolves to — lifecycle ops, destroy,
/// rebuild, meta-update, and the crash-watcher's periodic poll.
/// rebuild, 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>) {

4
hive-c0re/src/dashboard/lifecycle_ops.rs
View file

@ -51,7 +51,7 @@ pub(super) async fn post_rebuild(
return reject;
}
if let Err(e) = state.coord.job_queue.insert_job(|b| {
crate::job_queue::templates::rebuild(b, &logical, true);
crate::job_queue::templates::rebuild(b, &logical);
Vec::new()
}) {
tracing::error!(agent = %logical, error = ?e, "rebuild: insert failed");
@ -329,7 +329,7 @@ pub(super) async fn post_update_all(State(state): State<AppState>) -> Response {
};
for logical in agents {
if let Err(e) = state.coord.job_queue.insert_job(|b| {
crate::job_queue::templates::rebuild(b, &logical, true);
crate::job_queue::templates::rebuild(b, &logical);
Vec::new()
}) {
tracing::error!(agent = %logical, error = ?e, "update-all: insert failed");

293
hive-c0re/src/dashboard/meta_inputs.rs
View file

@ -1,293 +0,0 @@
//! META INPUTS panel backend: walks `meta/flake.lock` into
//! `MetaInputView` rows for the snapshot, emits the `MetaInputsChanged`
//! event after lock bumps, and handles `POST /api/meta-update` (bulk
//! flake-input update + rebuild ripple via the job queue).
use axum::{
extract::{Form, State},
http::StatusCode,
response::{IntoResponse, Response},
};
use serde::{Deserialize, Serialize};
use utoipa::ToSchema;
use crate::coordinator::Coordinator;
use super::{AppState, error_response};
#[derive(Serialize, Clone, Debug)]
pub struct MetaInputView {
/// Input key in meta's `flake.nix` — `hyperhive`, `agent-<n>`, etc.
pub name: String,
/// Full locked sha. Not displayed verbatim; the dashboard
/// truncates to the first 12 chars for the chip.
pub rev: String,
/// Unix seconds — `locked.lastModified`. Drives the relative
/// "2h ago" timestamp on each input row.
pub last_modified: i64,
/// `original.url` if available, for the tooltip / row meta text.
#[serde(skip_serializing_if = "Option::is_none")]
pub url: Option<String>,
}
/// Walk `flake.lock`'s `nodes` graph from `root` and emit one
/// `MetaInputView` per fetched input, at **every** depth. That
/// surfaces the direct meta inputs (`hyperhive`, `agent-<n>`), the
/// agent flakes' own inputs (`agent-dmatrix/mcp-matrix`,
/// `hyperhive/nixpkgs`), and any deeper transitive inputs — so the
/// operator can bump any of them individually. Names are
/// slash-separated paths from root, the syntax `nix flake update`
/// accepts for transitive inputs.
///
/// Filtering:
/// - Inputs that resolve via a `follows` chain (lock value is an
/// array) are skipped — they alias another node, not their own
/// fetched derivation, so updating them does nothing.
/// - A node is emitted only when it carries a `locked.rev`.
/// - Each fetched node is walked exactly once (a `visited` set):
/// the lock graph shares nodes (many flakes reference one
/// nixpkgs), so without this a shared subtree re-walks per parent
/// and a cycle would recurse forever. The result is a spanning
/// tree — every input shown once, at its shallowest path.
pub(super) fn read_meta_inputs() -> Vec<MetaInputView> {
let mut out = Vec::new();
let Ok(raw) = std::fs::read_to_string(crate::paths::meta_flake_lock()) else {
return out;
};
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 mut visited = std::collections::HashSet::new();
visited.insert(root_name.to_owned());
walk_meta_inputs(nodes, root_name, "", &mut visited, &mut out);
// hyperhive first, then alphabetical. String-sorting the
// slash-paths puts every node directly above its own children
// (`agent-foo`, `agent-foo/bar`, `agent-foo/bar/baz`), so the
// result is a pre-order traversal the tree renderer can consume.
out.sort_by(|a, b| match (a.name.as_str(), b.name.as_str()) {
("hyperhive", _) => std::cmp::Ordering::Less,
(_, "hyperhive") => std::cmp::Ordering::Greater,
_ => a.name.cmp(&b.name),
});
out
}
fn walk_meta_inputs(
nodes: &serde_json::Map<String, serde_json::Value>,
node_name: &str,
prefix: &str,
visited: &mut std::collections::HashSet<String>,
out: &mut Vec<MetaInputView>,
) {
let Some(node) = nodes.get(node_name) else {
return;
};
let Some(inputs_map) = node.get("inputs").and_then(|v| v.as_object()) else {
return;
};
// Two passes: claim (and emit) every direct input of this node
// before descending into any of them. A shallow input that a
// deeper flake also references then keeps its shallow path
// rather than being captured first by the deep walk.
let mut to_recurse: Vec<(String, String)> = Vec::new();
for (alias, target) in inputs_map {
// Inputs map value is either a string (node name) or an
// array (a `follows` chain). The latter just aliases another
// node — we can't `nix flake update` it directly, so skip.
let serde_json::Value::String(target_name) = target else {
continue;
};
// Walk each fetched node once — guards shared subtrees and
// cycles, and keeps the panel free of duplicate rows.
if !visited.insert(target_name.clone()) {
continue;
}
let Some(target_node) = nodes.get(target_name) else {
continue;
};
let path = if prefix.is_empty() {
alias.clone()
} else {
format!("{prefix}/{alias}")
};
if let Some(rev) = target_node
.get("locked")
.and_then(|v| v.get("rev"))
.and_then(|v| v.as_str())
{
let last_modified = target_node
.get("locked")
.and_then(|v| v.get("lastModified"))
.and_then(serde_json::Value::as_i64)
.unwrap_or(0);
let url = target_node
.get("original")
.and_then(|v| v.get("url"))
.and_then(|v| v.as_str())
.map(str::to_owned);
out.push(MetaInputView {
name: path.clone(),
rev: rev.to_owned(),
last_modified,
url,
});
}
to_recurse.push((target_name.clone(), path));
}
// Recurse hyperhive's subtree before any agent's — without this,
// when meta's top-level `nixpkgs` is a `follows` alias the
// `String` check above skips it, and the alphabetical BTreeMap
// iteration descends into `agent-*` first. The agent walk then
// claims `nixpkgs` at `agent-X/nixpkgs` instead of
// `hyperhive/nixpkgs`, which is where the operator expects it.
// Sort by the same "hyperhive first, then alpha"
// priority `read_meta_inputs` uses for the final output.
to_recurse.sort_by(|(a, _), (b, _)| match (a.as_str(), b.as_str()) {
("hyperhive", _) => std::cmp::Ordering::Less,
(_, "hyperhive") => std::cmp::Ordering::Greater,
_ => a.cmp(b),
});
for (target_name, path) in to_recurse {
walk_meta_inputs(nodes, &target_name, &path, visited, out);
}
}
/// Snapshot meta/flake.lock's root inputs + emit
/// `MetaInputsChanged`. Call after any mutation that bumps a lock
/// (`run_meta_update`, `auto_update::rebuild_agent`).
pub(crate) fn emit_meta_inputs_snapshot(coord: &Coordinator) {
let inputs = read_meta_inputs();
coord.emit_dashboard_event(crate::dashboard_events::DashboardEvent::MetaInputsChanged {
seq: coord.next_seq(),
inputs,
});
}
/// Form for `POST /meta-update`. Inputs ride in as a comma-separated
/// list under the `inputs` field — the JS submitter joins the
/// checked boxes since axum's `Form` extractor doesn't natively
/// decode repeated keys without a helper.
#[derive(Deserialize, ToSchema)]
pub(super) struct MetaUpdateForm {
inputs: String,
}
/// Bulk-update selected meta flake inputs, then rebuild the affected
/// agents in the background.
///
/// Idempotent w.r.t. selection — choosing an input that's already at
/// the latest sha is a no-op (no commit, no rebuild ripple). Returns
/// immediately after queueing the work; dashboard polls for progress
/// via container `pending` spinners + the meta-inputs row sha update.
#[utoipa::path(
post,
path = "/api/meta-update",
request_body(content = MetaUpdateForm, content_type = "application/x-www-form-urlencoded"),
responses(
(status = 200, description = "update + rebuild ripple queued", body = String),
(status = 500, description = "no inputs selected"),
),
tag = "meta_inputs"
)]
pub(super) async fn post_meta_update(
State(state): State<AppState>,
Form(form): Form<MetaUpdateForm>,
) -> Response {
let inputs: Vec<String> = form
.inputs
.split(',')
.map(|s| s.trim().to_owned())
.filter(|s| !s.is_empty())
.collect();
if inputs.is_empty() {
return error_response("meta-update: no inputs selected");
}
// Cascade rebuild children fan out from the MetaLock node when the
// lock bump lands — appended by the scheduler so they build against
// the post-bump lock, and a failed bump simply fans out nothing.
state
.coord
.job_queue
.insert_job(|b| {
crate::job_queue::templates::meta_update(b, inputs);
Vec::new()
})
.expect("template-declared shapes are acyclic");
state.coord.emit_rebuild_queue_snapshot();
(StatusCode::OK, "ok").into_response()
}
#[cfg(test)]
mod tests {
use super::*;
#[test]
fn walk_meta_inputs_keeps_nixpkgs_under_hyperhive_post_follows_refactor() {
// Reproduce the shape where meta has
// `nixpkgs.follows = "hyperhive/nixpkgs"` at the top level
// (rendered as an array — `["hyperhive" "nixpkgs"]` — which
// walk_meta_inputs skips because we can't `nix flake update`
// a follows alias). The remaining top-level inputs are
// `hyperhive` (string) and `agent-z` (string). Without the
// hyperhive-first recursion sort, the BTreeMap alphabetical
// order descends into `agent-z` first and claims
// `nixpkgs` at `agent-z/nixpkgs`.
let raw = r#"{
"root": "root",
"version": 7,
"nodes": {
"root": {
"inputs": {
"hyperhive": "hyperhive",
"nixpkgs": ["hyperhive", "nixpkgs"],
"agent-z": "agent-z"
}
},
"hyperhive": {
"inputs": { "nixpkgs": "nixpkgs" },
"locked": {"rev": "hhrev", "lastModified": 1},
"original": {"url": "git+file:///tmp/hyperhive"}
},
"agent-z": {
"inputs": { "nixpkgs": "nixpkgs" },
"locked": {"rev": "azrev", "lastModified": 2},
"original": {"url": "git+file:///tmp/agent-z"}
},
"nixpkgs": {
"locked": {"rev": "npkrev", "lastModified": 3},
"original": {"url": "github:NixOS/nixpkgs/nixos-26.05"}
}
}
}"#;
let json: serde_json::Value = serde_json::from_str(raw).unwrap();
let nodes = json.get("nodes").unwrap().as_object().unwrap();
let root_name = json.get("root").unwrap().as_str().unwrap();
let mut visited = std::collections::HashSet::new();
visited.insert(root_name.to_owned());
let mut out = Vec::new();
walk_meta_inputs(nodes, root_name, "", &mut visited, &mut out);
let nixpkgs = out
.iter()
.find(|v| v.rev == "npkrev")
.expect("nixpkgs node should be emitted exactly once");
assert_eq!(
nixpkgs.name, "hyperhive/nixpkgs",
"nixpkgs should be claimed under hyperhive, not under agent-z. \
got: {:?}",
nixpkgs.name
);
// And the agent-z path should NOT also carry a nixpkgs entry —
// the spanning-tree visited set guarantees it's claimed once.
assert!(
!out.iter().any(|v| v.name == "agent-z/nixpkgs"),
"agent-z/nixpkgs should not be emitted (already claimed under hyperhive)"
);
}
}

11
hive-c0re/src/dashboard/mod.rs
View file

@ -37,7 +37,7 @@ use crate::lifecycle;
(name = "health", description = "hive-wide liveness/readiness probes"),
(name = "build_logs", description = "build log headers, full rows, and raw text downloads"),
(name = "lifecycle_ops", description = "agent container lifecycle: rebuild/restart/start/stop/pause/limits"),
(name = "meta_inputs", description = "bulk flake-input update for the meta flake"),
(name = "misc_api", description = "mark-all-read, compose, spawn-request, hive stats"),
(name = "rebuild_queue", description = "rebuild-queue cancel"),
(name = "state_snapshot", description = "cold-load dashboard snapshot"),
@ -54,17 +54,12 @@ mod build_logs;
pub(crate) use hive_types::Ident;
mod health;
mod lifecycle_ops;
mod meta_inputs;
mod misc_api;
mod rebuild_queue;
mod state_snapshot;
mod tombstones;
// Run after lock bumps by the job queue (`job_queue/exec.rs`); the view
// type feeds `DashboardEvent::MetaInputsChanged` (`dashboard_events.rs`).
// Re-exported to preserve the `crate::dashboard::*` paths across the split.
pub use meta_inputs::MetaInputView;
pub(crate) use meta_inputs::emit_meta_inputs_snapshot;
// Called after destroy/purge/spawn finalisation (`actions.rs`); the view
// type feeds `DashboardEvent::TombstonesChanged` (`dashboard_events.rs`).
// Re-exported to preserve the `crate::dashboard::*` paths across the split.
@ -126,7 +121,6 @@ fn app(coord: Arc<Coordinator>) -> axum::Router {
.routes(routes!(lifecycle_ops::post_resume))
.routes(routes!(lifecycle_ops::post_update_all))
.routes(routes!(tombstones::post_purge_tombstone))
.routes(routes!(meta_inputs::post_meta_update))
.routes(routes!(build_logs::get_build_log_stream))
.routes(routes!(state_snapshot::dashboard_stream))
.routes(routes!(state_snapshot::jobq_graph))
@ -377,7 +371,6 @@ mod router_build_probe {
.routes(routes!(lifecycle_ops::post_resume))
.routes(routes!(lifecycle_ops::post_update_all))
.routes(routes!(tombstones::post_purge_tombstone))
.routes(routes!(meta_inputs::post_meta_update))
.routes(routes!(build_logs::get_build_log_stream))
.routes(routes!(state_snapshot::dashboard_stream))
.routes(routes!(state_snapshot::jobq_graph))

17
hive-c0re/src/dashboard/state_snapshot.rs
View file

@ -18,7 +18,7 @@ use utoipa::IntoParams;
use crate::container_view::ContainerView;
use super::AppState;
use super::meta_inputs::{MetaInputView, read_meta_inputs};
use super::tombstones::{TombstoneView, build_tombstone_views};
#[allow(clippy::struct_excessive_bools)]
@ -51,15 +51,7 @@ pub(super) struct StateSnapshot {
/// one other. Operator resolves by renaming. The dashboard renders
/// a banner at the top listing each cluster.
port_conflicts: Vec<PortConflict>,
/// Inputs in `meta/flake.lock` the operator can selectively
/// `nix flake update`. Hyperhive first, then `agent-<n>` rows.
meta_inputs: Vec<MetaInputView>,
/// True while a dashboard-triggered `meta-update` (flake lock bump +
/// agent rebuild ripple) is running in the background. Lets a
/// client that cold-loads mid-update render the META INPUTS panel's
/// disabled "updating…" state; live transitions arrive via the
/// `MetaUpdateRunning` event.
meta_update_running: bool,
/// Whether the hive-forge container is up. When true the dashboard
/// links each container's config into the forge's `agent-configs`
/// repos.
@ -124,7 +116,7 @@ struct TransientView {
/// Cold-load snapshot of the whole dashboard.
///
/// Includes the roster, tombstones, job queue, meta inputs, and more. Live
/// Includes the roster, tombstones, job queue, and more. Live
/// clients then follow `/api/dashboard/stream` (SSE) for incremental updates
/// keyed off `seq`.
// `StateSnapshot` is a large tree of nested view types (`ContainerView`,
@ -197,8 +189,7 @@ pub(super) async fn api_state(
any_stale,
containers,
transients,
meta_inputs: read_meta_inputs(),
meta_update_running: state.coord.meta_update_in_progress(),
tombstones,
port_conflicts,
forge_present: crate::forge::is_present().await,

34
hive-c0re/src/dashboard_events.rs
View file

@ -7,7 +7,7 @@
use serde::Serialize;
use crate::container_view::ContainerView;
use crate::dashboard::{MetaInputView, TombstoneView};
use crate::dashboard::TombstoneView;
use chrono::{DateTime, Utc};
#[derive(Debug, Clone, Serialize)]
@ -96,26 +96,7 @@ pub enum DashboardEvent {
seq: u64,
tombstones: Vec<TombstoneView>,
},
/// Full snapshot of `meta/flake.lock`'s root inputs. Emitted
/// after every operation that bumps a lock: `meta-update`,
/// `rebuild_agent` (lock bumps via two-phase staging),
/// `update-all`. Same snapshot-shape rationale as
/// `TombstonesChanged` — the list is small (one row per agent
/// plus their fetched inputs).
MetaInputsChanged {
seq: u64,
inputs: Vec<MetaInputView>,
},
/// A dashboard-triggered `meta-update` started (`running: true`) or
/// finished (`running: false`). `post_meta_update` returns 200
/// immediately and runs the `nix flake update` + agent-rebuild
/// ripple in a background task — this event lets the META INPUTS
/// panel show a disabled "updating…" state for that whole window
/// instead of looking idle. Emitted by
/// `Coordinator::meta_update_guard` / `MetaUpdateGuard::drop` only
/// when the active-run count crosses 0, so concurrent updates flip
/// the flag exactly once.
MetaUpdateRunning { seq: u64, running: bool },
/// The job queue changed — **a bare trigger, no payload.**
///
/// It used to carry a full typed snapshot of the queue. Nothing reads
@ -153,8 +134,7 @@ impl DashboardEvent {
DashboardEvent::ContainerStateChanged { .. } => "container_state_changed",
DashboardEvent::ContainerRemoved { .. } => "container_removed",
DashboardEvent::TombstonesChanged { .. } => "tombstones_changed",
DashboardEvent::MetaInputsChanged { .. } => "meta_inputs_changed",
DashboardEvent::MetaUpdateRunning { .. } => "meta_update_running",
DashboardEvent::RebuildQueueChanged { .. } => "rebuild_queue_changed",
}
}
@ -219,14 +199,6 @@ mod tests {
seq: 1,
tombstones: Vec::new(),
},
DashboardEvent::MetaInputsChanged {
seq: 1,
inputs: Vec::new(),
},
DashboardEvent::MetaUpdateRunning {
seq: 1,
running: false,
},
DashboardEvent::RebuildQueueChanged { seq: 1 },
];
for ev in samples {

5
hive-c0re/src/forge/mod.rs
View file

@ -13,11 +13,10 @@ mod users;
pub(crate) use internal_meta::internal_meta_source;
pub use reader::reader_token;
pub(crate) use reconcile::FORGE_MAIN_REF;
pub use reconcile::{fetch_forge_main, reconcile_config_apply, reconcile_config_status};
pub use repos::{
clone_config_into_proposed, ensure_meta_remote, ensure_repo, fast_forward_applied_main,
fetch_config_main_into_applied, meta_read_access, push_meta,
clone_config_into_proposed, ensure_meta_remote, ensure_repo, fetch_config_main_into_applied,
meta_read_access, push_meta,
};
pub use users::core_token;

57
hive-c0re/src/forge/repos.rs
View file

@ -273,63 +273,6 @@ pub async fn fetch_config_main_into_applied(name: &str) -> bool {
true
}
/// Fast-forward `applied/<name>/main` to the config repo's current live
/// main. Unlike [`fetch_config_main_into_applied`] (reseed-only — bails
/// if `applied` already has a `.git`), this is for an `applied` repo that
/// already exists and just needs to catch up: force-sets the local ref to
/// match live main rather than requiring a fast-forward or any review
/// gate — there is no PR to review on this path, so nothing to check
/// ancestry against. Contrast the merged-PR deploy flow's CAS'd,
/// ancestry-gated advance for a reviewed PR merge, which this is not.
///
/// Best-effort, same shape as [`fetch_config_main_into_applied`]: returns
/// `false` (not an error) when there is no [`reader_token`], `applied` has
/// no `.git` yet (nothing to fast-forward — reseed instead), or any git step
/// fails.
pub async fn fast_forward_applied_main(name: &str) -> bool {
let Some(reader) = reader_token().await else {
return false;
};
let applied = crate::paths::applied_dir(name);
if !applied.join(".git").exists() {
return false;
}
let url = forge_git_url(&format!("{CONFIG_ORG}/{name}"));
let auth = reader.auth_header();
let out = crate::lifecycle::git_command_authed(&auth)
.current_dir(&applied)
.args(["fetch", "--no-tags", &url, "refs/heads/main"])
.output()
.await;
match out {
Ok(o) if o.status.success() => {}
Ok(o) => {
tracing::warn!(
%name,
stderr = %String::from_utf8_lossy(&o.stderr).trim(),
"forge: applied fast-forward fetch failed"
);
return false;
}
Err(e) => {
tracing::warn!(%name, error = ?e, "forge: applied fast-forward fetch failed");
return false;
}
}
if let Err(e) =
crate::lifecycle::git_update_ref(&applied, "refs/heads/main", "FETCH_HEAD").await
{
tracing::warn!(%name, error = ?e, "forge: applied fast-forward update-ref failed");
return false;
}
if let Err(e) = crate::lifecycle::git_read_tree_reset(&applied, "refs/heads/main").await {
tracing::warn!(%name, error = ?e, "forge: applied fast-forward read-tree failed");
return false;
}
tracing::info!(%name, "forge: fast-forwarded applied/main to agent-configs main");
true
}
/// Seed an agent's `proposed` repo from `agent-configs/<name>` on the forge.
///
/// When the swarm creates an agent it writes that agent's config to the forge

434
hive-c0re/src/job_queue/exec.rs
View file

@ -1,9 +1,8 @@
//! Node executors — one async fn per [`NodeKind`], each a thin wrapper
//! over existing `lifecycle.rs` / `meta.rs` / `actions.rs` code. Node
//! over existing `lifecycle.rs` / `meta_rev.rs` / `actions.rs` code. Node
//! executors keep their own internal error handling where it exists
//! today (cold-start fallback inside `Reconcile`, non-fatal boot-time
//! lock bump inside the sweep `MetaLock`, warn-only forge sync in the
//! `Swap` tail); DAG-level failure handling is cancel-downstream in
//! today (cold-start fallback inside `Reconcile`, warn-only forge sync in
//! the `Swap` tail); DAG-level failure handling is cancel-downstream in
//! the queue.
use std::sync::{Arc, Mutex, OnceLock};
@ -61,18 +60,18 @@ pub(super) async fn run_node(
kind: &NodeKind,
) -> (super::JobBuilder, Result<()>) {
// The agent this node targets rides the payload — empty for the agentless
// kinds (`MetaLock`), which never read it.
// kinds (the sweeps), which never read it.
let agent = kind.agent();
// Every arm is `Result<()>`; the three that grow work declare into `builder`
// *synchronously*, after their own awaits have finished. Borrowing `&builder`
// Every arm is `Result<()>`; the one that grows work declares into `builder`
// *synchronously*, after its own awaits have finished. Borrowing `&builder`
// inside an `.await` would make this future non-`Send` (see above), so the
// growth executors return what to grow rather than taking the builder.
let result = match kind {
NodeKind::MetaSync { relock, .. } => run_meta_sync(coord, agent, *relock).await,
NodeKind::MetaSync { .. } => run_meta_sync(agent).await,
NodeKind::Prebuild { .. } => run_prebuild(agent, id).await,
NodeKind::Swap { .. } => run_swap(coord, agent, id).await,
NodeKind::RebuildBookkeeping { .. } => run_rebuild_bookkeeping(coord, agent).await,
NodeKind::Provision { .. } => run_provision(coord, agent).await,
NodeKind::Provision { .. } => run_provision(agent).await,
NodeKind::Create { .. } => run_create(agent).await,
NodeKind::DestroyContainer { .. } => run_destroy_container(coord, agent).await,
NodeKind::PurgeState { .. } => {
@ -83,24 +82,7 @@ pub(super) async fn run_node(
run_destroy_bookkeeping(coord, agent).await;
Ok(())
}
NodeKind::MetaLock {
sweep,
fanout,
inputs,
} => run_meta_lock(coord, *sweep, fanout.clone(), inputs)
.await
.map(|agents| {
// `sweep` is the whole difference: it relocks per-agent like a
// manual rebuild, and it drains agents that were mid-turn when
// the host came up. A cascade does neither. Decided here rather
// than returned, since `run_meta_lock` would only be deriving
// it from the `sweep` this call site already holds.
if *sweep {
super::templates::grown_graceful_rebuilds(&builder, &agents, true);
} else {
super::templates::grown_rebuilds(&builder, &agents, false);
}
}),
NodeKind::Reconcile { .. } => run_reconcile(coord, agent).await.map(|sub| {
if let Some(kind) = sub {
super::templates::fanned_out_mechanical(&builder, kind);
@ -258,12 +240,12 @@ async fn run_wanted_pull(coord: &Arc<Coordinator>) -> Result<()> {
crate::workers::wanted::pull(coord).await
}
/// Rerender the meta flake from whatever containers still exist on disk.
/// Idempotent — a no-op when nothing changed. Lives here because the destroy
/// tail is its only caller; it moved with `destroy` when that became a DAG.
async fn sync_meta_after_lifecycle(coord: &Coordinator) -> Result<()> {
/// Refresh the host-side agent-set files from whatever containers still exist
/// on disk. Idempotent. Lives here because the destroy tail is its only
/// caller; it moved with `destroy` when that became a DAG.
async fn refresh_agent_set_after_lifecycle() -> Result<()> {
let agents = crate::lifecycle::agents_for_meta_listing().await?;
crate::meta::sync_agents(&coord.hive_env(), &agents).await
crate::agent_set::refresh(&agents).await
}
/// `nixos-container destroy`, then drop the agent from the roster and clear
@ -319,15 +301,13 @@ async fn run_purge_state(agent: &str) {
/// Post-destroy bookkeeping. Infallible by construction: every step is
/// warn-and-continue, because the destroy it follows has already succeeded and
/// none of this is undoable — a failed meta sync or power-store write is a
/// none of this is undoable — a failed agent-set refresh or power-store write is a
/// bookkeeping drift to log, not a reason to red a DAG whose container is
/// already gone.
async fn run_destroy_bookkeeping(coord: &Arc<Coordinator>, agent: &str) {
// Meta flake: drop the agent's input + nixosConfiguration so a future spawn
// under the same name re-seeds cleanly, and so the meta lock doesn't
// reference a vanished applied repo.
if let Err(e) = sync_meta_after_lifecycle(coord).await {
tracing::warn!(error = ?e, %agent, "meta sync after destroy failed");
// Drop the agent from the topology and the gateway files.
if let Err(e) = refresh_agent_set_after_lifecycle().await {
tracing::warn!(error = ?e, %agent, "agent_set refresh after destroy failed");
}
// Drop the durable power intent — a future agent of the same name seeds
// fresh from its observed state.
@ -380,21 +360,16 @@ fn run_set_wanted(coord: &Arc<Coordinator>, agent: &str, up: bool) -> Result<()>
Ok(())
}
/// The rebuild's meta preamble: runtime-dir prep, an idempotent meta
/// `sync_agents`, and the optional per-agent relock. Runs under the deploy
/// window (`Resource::MetaWindow`, held by the scheduler for this
/// node) so its commits can never land inside another node's staged
/// prepare→finalize window.
/// The rebuild's preamble: runtime-dir prep, the agent-set refresh, and the
/// `internal/meta` commit `Prebuild` and `Swap` build from.
///
/// Deliberately a separate node from the [`run_prebuild`] it feeds: that
/// build takes minutes and only *reads* the store, so keeping the global
/// window off it is what lets rebuilds of different agents overlap.
async fn run_meta_sync(coord: &Arc<Coordinator>, name: &str, relock: bool) -> Result<()> {
/// Fails when `internal/meta` does not pin the agent's current config yet:
/// there is no other flake to build it from.
async fn run_meta_sync(name: &str) -> Result<()> {
// Runs while the agent is still up — the runtime dir and MCP listener
// already exist. Use the pure path accessor; no need to re-register the
// listener (event-driven: registered at start/create).
let agent_dir = crate::paths::agent_runtime_dir(name);
let hive = coord.hive_env();
let paths = Coordinator::agent_paths(name, agent_dir)?;
// A failed read keeps the archetype last stored, so the agent is built as
// it was.
@ -402,28 +377,18 @@ async fn run_meta_sync(coord: &Arc<Coordinator>, name: &str, relock: bool) -> Re
tracing::warn!(%name, error = ?e, "rebuild: reading the swarm's archetype failed; keeping the stored one");
}
crate::lifecycle::prepare_rebuild_dirs(name, &paths).await?;
// Idempotent meta sync so a manual rebuild can also recover from a
// divergent meta repo; then bump just this agent's input. `relock =
// false` only for meta-update cascade children, where re-locking
// would revert the bump the cascade just committed.
let agents = crate::lifecycle::agents_for_meta_listing().await?;
crate::meta::sync_agents(&hive, &agents).await?;
// An archetype agent has no `agent-<name>` input to relock.
if relock && crate::archetypes::of(name)?.is_none() {
crate::meta::lock_update_for_rebuild(name).await?;
}
// The commit of the swarm's `internal/meta` that `Prebuild` and `Swap`
// build this agent from, resolved once here so both build the same flake;
// none leaves them on the hive's own flake rendered above.
let meta_rev = crate::meta_rev::resolve(name).await;
crate::meta_rev::record(name, meta_rev.as_deref())?;
crate::agent_set::refresh(&agents).await?;
// Resolved once here so `Prebuild` and `Swap` build the same flake.
let build = crate::meta_rev::resolve(name).await?;
crate::meta_rev::record(name, &build)?;
Ok(())
}
/// Out-of-band toplevel build while the container keeps serving: warm
/// `system.build.toplevel` so the later `Swap` hits cache and skips
/// straight to the profile-swap, against a meta repo the upstream
/// `MetaSync` node has already synced. The warm build is skipped when the
/// straight to the profile-swap, from the `internal/meta` commit the upstream
/// `MetaSync` node recorded. The warm build is skipped when the
/// container is already down: its only purpose is to shrink the swap's
/// downtime window, so a stopped agent (no uptime to preserve) doesn't
/// pay the double eval — `Swap` builds inline instead.
@ -433,10 +398,8 @@ async fn run_prebuild(name: &str, id: NodeId) -> Result<()> {
// stopped agent has no uptime to preserve, so skip the (expensive)
// eval and let the downstream `Swap` build inline.
if crate::lifecycle::is_running(name).await {
let flake_ref = match crate::meta_rev::recorded(name) {
Some(rev) => crate::meta_rev::flake_ref(name, &rev),
None => format!("{}#{name}", crate::paths::meta_root().display()),
};
let build = crate::meta_rev::recorded_or_err(name)?;
let flake_ref = crate::meta_rev::flake_ref(name, &build.rev);
crate::lifecycle::prebuild_toplevel(name, &flake_ref, Some(id.get())).await?;
}
Ok(())
@ -453,10 +416,9 @@ async fn run_swap(coord: &Arc<Coordinator>, name: &str, id: NodeId) -> Result<()
let agent_dir = crate::paths::agent_runtime_dir(name);
let hive = coord.hive_env();
let paths = Coordinator::agent_paths(name, agent_dir)?;
let meta_rev = crate::meta_rev::recorded(name);
let build = crate::meta_rev::recorded_or_err(name)?;
let result =
crate::lifecycle::swap_update(name, &hive, &paths, meta_rev.as_deref(), Some(id.get()))
.await;
crate::lifecycle::swap_update(name, &hive, &paths, &build.rev, Some(id.get())).await;
// On success the Ok-only bookkeeping tail (rev marker, forge/matrix
// sync, kick, rescan, snapshot) runs in the sibling `RebuildBookkeeping` node,
// which deps `AfterOk(Swap)`. On failure `RebuildBookkeeping` is cancel-cascaded
@ -476,10 +438,10 @@ async fn run_swap(coord: &Arc<Coordinator>, name: &str, id: NodeId) -> Result<()
/// build (build-slot-exempt); the agent lease taken at `Swap` is still held
/// (the whole chain up to `Reconcile` is one agent's subgraph).
async fn run_rebuild_bookkeeping(coord: &Arc<Coordinator>, name: &str) -> Result<()> {
if let Some(rev) = crate::auto_update::current_flake_rev(&coord.hyperhive_flake)
&& let Err(e) = std::fs::write(crate::paths::applied_rev_marker(name), rev)
if let Some(build) = crate::meta_rev::recorded(name)
&& let Err(e) = crate::meta_rev::mark_deployed(name, &build)
{
tracing::warn!(%name, error = ?e, "write rev marker failed");
tracing::warn!(%name, error = ?e, "write rev markers failed");
}
// The `Rebuilt` manager event is emitted exactly once per agent by the DAG's
// `EmitRebuilt` tail — emitting ok here and letting a failed tail `Reconcile`
@ -497,138 +459,33 @@ async fn run_rebuild_bookkeeping(coord: &Arc<Coordinator>, name: &str) -> Result
// alongside `run_start`'s own kick for the same boot. `hive-agent` now
// self-announces once on its own startup instead — one message per
// actual boot, not one per DAG node that happens to run during it.
// Rescan so dashboards drop the "needs update" chip; lock bump →
// meta-inputs re-render.
// Rescan so dashboards drop the "needs update" chip.
coord.rescan_containers_and_emit().await;
crate::dashboard::emit_meta_inputs_snapshot(coord);
Ok(())
}
/// First-spawn pre-create provisioning: proposed/applied repos, state
/// subvolume, and the meta `sync_agents` registration. Runs under the
/// deploy window (it declares `Resource::MetaWindow`) so its commit can't
/// land inside another node's staged deploy window.
async fn run_provision(coord: &Arc<Coordinator>, name: &str) -> Result<()> {
/// subvolume, and the agent-set refresh.
async fn run_provision(name: &str) -> Result<()> {
let agent_dir = crate::paths::agent_runtime_dir(name);
let hive = coord.hive_env();
let paths = Coordinator::agent_paths(name, agent_dir)?;
crate::lifecycle::provision_container(name, &hive, &paths).await?;
crate::lifecycle::provision_container(name, &paths).await?;
Ok(())
}
/// `nixos-container create` proper — the upstream `Provision` node
/// already registered the agent in meta, so this only reads the store
/// (no deploy-window gate needed, mirroring `Prebuild`'s build). Runtime
/// already registered the agent, so this only reads the store. Runtime
/// dir creation and MCP listener registration are deferred to the tail
/// `Reconcile` (`converge_start_preamble` + `register_agent`) so this
/// node stays purely "create", not "create + start".
/// node stays purely "create", not "create + start". Marks the agent as
/// running the pins it was created from.
async fn run_create(name: &str) -> Result<()> {
let meta_rev = crate::meta_rev::resolve(name).await;
crate::lifecycle::create_only(name, meta_rev.as_deref()).await?;
let build = crate::meta_rev::resolve(name).await?;
crate::lifecycle::create_only(name, &build.rev).await?;
crate::meta_rev::mark_deployed(name, &build)?;
Ok(())
}
/// Meta flake lock bump. Boot-sweep flavour is non-fatal (a failed
/// bump must not cancel the fan-out rebuilds — they proceed against
/// the current lock, exactly like today's sweep); the meta-update
/// flavour propagates errors, and a failed bump fans out nothing.
/// Returns the agents whose rebuild subgraphs the caller should grow into this
/// node, rather than declaring them here — the declaration has to happen outside
/// any `.await` (see [`run_node`]).
///
/// Only the agent list: *which* rebuild flavour to grow is a pure function of
/// `sweep`, which the caller passed in, so returning it too would be a round
/// trip rather than a decision.
async fn run_meta_lock(
coord: &Arc<Coordinator>,
sweep: bool,
fanout: Option<Vec<String>>,
inputs: &[String],
) -> Result<Vec<String>> {
if sweep {
if let Err(e) = crate::meta::lock_update_hyperhive().await {
tracing::warn!(error = ?e, "startup sweep: meta lock_update_hyperhive failed");
}
// Grow one rebuild subgraph per stale agent into *this* boot DAG
// (rooted on this `MetaLock`, so they build against the post-bump
// lock), rather than fanning out child DAGs. The caller grows them
// with the graceful flavour.
return Ok(fanout.unwrap_or_default());
}
let _progress = coord.meta_update_guard();
// Resolved before the lock bump so a failed container list aborts
// the update before anything is committed.
let cascade = match fanout {
Some(list) => validate_agent_names(list),
None => meta_update_cascade_agents(inputs).await?,
};
crate::meta::lock_update(inputs).await?;
// Lock file changed — meta-inputs panel re-renders.
crate::dashboard::emit_meta_inputs_snapshot(coord);
// Pull each cascade agent's own input too — an agent's config-repo main
// is trusted, so a meta-input bump is a reasonable place to also catch
// it up. Without this, a meta-input bump rebuilds every affected agent
// against whatever `applied/<name>` already happened to be locked to —
// normally current, but silently stale forever if a past deploy failed
// and nothing since retried it.
// An archetype agent has no config repo and no `agent-<name>` input, so
// it is left out here; it is still rebuilt with the cascade.
let archetypes = crate::archetypes::read_at(&crate::archetypes::path())?;
let custom: Vec<String> = cascade
.iter()
.filter(|name| !archetypes.contains_key(*name))
.cloned()
.collect();
catch_up_cascade_agents(
&custom,
crate::forge::fast_forward_applied_main,
|agent_inputs| async move { crate::meta::lock_update(&agent_inputs).await },
)
.await?;
// Grow one rebuild subgraph per affected agent into *this* meta-update
// DAG (rooted on this `MetaLock`, so they build against the post-bump
// lock), rather than fanning out child DAGs. `relock = false` — the
// cascade children must NOT re-lock, which would revert the bump this
// node just committed (the property the old `fanout_specs` meta-update
// branch encoded).
Ok(cascade)
}
/// Fast-forward each cascade agent's `applied/<name>/main` to its forge
/// config main, then relock every `agent-<name>` input in one call.
///
/// The declared `agent-<name>` input is the `applied/<name>` mirror itself,
/// and the cascade's rebuilds (`relock = false`) build whatever this relock
/// pins, so the mirror has to move first. The fast-forward is best-effort
/// per agent: a failed one leaves that agent relocked against whatever its
/// mirror already has. The relock is one combined call, so one broken input
/// fails it for the whole cascade.
///
/// The fast-forward and the relock are passed in so the order is testable
/// without nix or a forge.
async fn catch_up_cascade_agents<'a, FF, Relock>(
cascade: &'a [String],
mut fast_forward: impl FnMut(&'a str) -> FF,
relock: impl FnOnce(Vec<String>) -> Relock,
) -> Result<()>
where
FF: std::future::Future<Output = bool>,
Relock: std::future::Future<Output = Result<()>>,
{
if cascade.is_empty() {
return Ok(());
}
for name in cascade {
if !fast_forward(name).await {
tracing::warn!(
%name,
"meta-update cascade: applied/main fast-forward skipped or failed"
);
}
}
relock(cascade.iter().map(|name| format!("agent-{name}")).collect()).await
}
/// Idempotent power-converge *planner*: compare `wanted` (durable
/// intent) against observed state and, when they diverge, fan the
/// mechanical `Start` / `Stop` out as a first-class node appended to
@ -814,202 +671,3 @@ async fn run_write_dropin(coord: &Arc<Coordinator>, name: &str) -> Result<()> {
crate::lifecycle::write_dropins(name, &hive, &paths).await?;
Ok(())
}
/// Compute which agents a `nix flake update <inputs>` on the meta
/// flake affects — the fan-out set for `MetaUpdate` DAGs. Empty
/// `inputs` or any input under `hyperhive` → every container;
/// otherwise just the agents named by `agent-<name>` inputs.
/// Sorted by name — the parent field the old depth sort keyed on is
/// gone, and with every agent a root that sort already reduced to
/// this.
///
/// `inputs` is the caller-supplied flake-input-name list (the dashboard's
/// `POST /api/meta-update` form field, operator-supplied but not
/// otherwise validated) — the `agent-<name>` branch parses agent names
/// straight out of it, so each is validated through [`hive_types::Ident`]
/// before it ever reaches a filesystem path or a forge URL built from a
/// cascade agent's name. The `touched_hyperhive` branch's names need no
/// such filter: they come from `lifecycle::list()`, already real
/// container names by construction, not parsed out of caller input.
///
/// # Errors
///
/// The every-container branch could not list the containers.
pub async fn meta_update_cascade_agents(inputs: &[String]) -> Result<Vec<String>> {
let touched_hyperhive = inputs
.iter()
.any(|i| i == "hyperhive" || i.starts_with("hyperhive/"));
let touched_agents = parse_agent_input_names(inputs);
let mut names = if touched_hyperhive || inputs.is_empty() {
crate::lifecycle::agent_names(crate::lifecycle::list().await)
.context("listing containers for the meta-update cascade")?
} else {
touched_agents
};
names.sort();
Ok(names)
}
/// Parse `agent-<name>` input strings into validated agent names. Pure and
/// synchronous (no `lifecycle::list()` call) so the validation — the part
/// that actually matters for safety, since these names end up in
/// filesystem paths and forge URLs — is unit-testable without touching
/// live container state. Delegates the actual filtering to
/// [`validate_agent_names`] after stripping the `agent-` prefix.
fn parse_agent_input_names(inputs: &[String]) -> Vec<String> {
let candidates = inputs
.iter()
.filter_map(|i| i.strip_prefix("agent-"))
.map(|rest| rest.split('/').next().unwrap_or(rest).to_owned());
validate_agent_names(candidates)
}
/// Drop any name that isn't a well-formed [`hive_types::Ident`] — the same
/// validating parser every other agent-name wire field in this codebase
/// goes through, so a name that reaches
/// [`fast_forward_applied_main`](crate::forge::fast_forward_applied_main) or
/// an `agent-configs/<name>` forge URL built off it has already
/// passed the same bar as one that arrived over a socket. Not just a
/// helper for [`parse_agent_input_names`]: `run_meta_lock`'s `fanout`
/// parameter carries pre-computed agent names too, and while today's only
/// caller of that branch (the boot sweep, `sweep = true`) returns before
/// ever reaching the filesystem/git code below, nothing in the type
/// enforces that pairing — so this same filter also guards the
/// `Some(list)` arm, closing the gap for good rather than relying on it
/// staying incidental.
fn validate_agent_names(names: impl IntoIterator<Item = String>) -> Vec<String> {
names
.into_iter()
.filter(|name| {
let valid = hive_types::Ident::parse(name).is_ok();
if !valid {
tracing::warn!(%name, "meta-update cascade: dropping malformed agent name");
}
valid
})
.collect()
}
#[cfg(test)]
mod cascade_agent_tests {
use super::{catch_up_cascade_agents, parse_agent_input_names, validate_agent_names};
use std::cell::RefCell;
use std::collections::BTreeMap;
/// Runs [`catch_up_cascade_agents`] over `cascade` against a model of
/// the applied mirrors: every mirror starts at `"old"`, a fast-forward
/// moves it to `"new"` unless the agent is in `failing`. Returns the
/// mirror head the relock saw for each input it was given.
async fn relocked_heads(cascade: &[&str], failing: &[&str]) -> Vec<(String, &'static str)> {
let cascade: Vec<String> = cascade.iter().map(|&n| n.to_owned()).collect();
let heads = RefCell::new(
cascade
.iter()
.map(|n| (n.clone(), "old"))
.collect::<BTreeMap<_, _>>(),
);
let relocked = RefCell::new(Vec::new());
let fast_forward = |name: &str| {
let ok = !failing.contains(&name);
if ok {
heads.borrow_mut().insert(name.to_owned(), "new");
}
async move { ok }
};
let relock = |inputs: Vec<String>| {
for input in inputs {
let name = input.strip_prefix("agent-").expect("agent input");
let head = heads.borrow()[name];
relocked.borrow_mut().push((input, head));
}
async { anyhow::Ok(()) }
};
catch_up_cascade_agents(&cascade, fast_forward, relock)
.await
.expect("catch up");
relocked.into_inner()
}
#[tokio::test]
async fn the_cascade_relock_pins_the_fast_forwarded_head() {
assert_eq!(
relocked_heads(&["iris", "atlas"], &[]).await,
vec![
("agent-iris".to_owned(), "new"),
("agent-atlas".to_owned(), "new"),
]
);
}
#[tokio::test]
async fn a_failed_fast_forward_still_relocks_that_agent() {
assert_eq!(
relocked_heads(&["iris", "atlas"], &["iris"]).await,
vec![
("agent-iris".to_owned(), "old"),
("agent-atlas".to_owned(), "new"),
]
);
}
/// A relock with no inputs is a bare `nix flake update`, which bumps
/// every meta input.
#[tokio::test]
async fn an_empty_cascade_moves_nothing_and_relocks_nothing() {
let fast_forwarded = std::cell::Cell::new(false);
let relocked = std::cell::Cell::new(false);
let outcome = catch_up_cascade_agents(
&[],
|_| {
fast_forwarded.set(true);
async { true }
},
|_| {
relocked.set(true);
async { anyhow::Ok(()) }
},
)
.await;
outcome.expect("an empty cascade is not an error");
assert!(!fast_forwarded.get(), "nothing to fast-forward");
assert!(!relocked.get(), "an empty relock would bump every input");
}
#[test]
fn well_formed_agent_inputs_pass_through() {
let inputs = vec!["agent-iris".to_owned(), "agent-atlas".to_owned()];
assert_eq!(parse_agent_input_names(&inputs), vec!["iris", "atlas"]);
}
#[test]
fn path_traversal_in_an_agent_input_is_dropped() {
let inputs = vec!["agent-../../etc".to_owned(), "agent-iris".to_owned()];
assert_eq!(parse_agent_input_names(&inputs), vec!["iris"]);
}
#[test]
fn non_agent_inputs_are_ignored_not_misparsed() {
let inputs = vec!["hyperhive".to_owned(), "nixpkgs".to_owned()];
assert!(parse_agent_input_names(&inputs).is_empty());
}
#[test]
fn oversized_or_uppercase_names_are_dropped() {
let too_long = format!("agent-{}", "a".repeat(64));
let inputs = vec![too_long, "agent-Iris".to_owned(), "agent-iris".to_owned()];
assert_eq!(parse_agent_input_names(&inputs), vec!["iris"]);
}
#[test]
fn fanout_supplied_names_are_validated_too() {
// `run_meta_lock`'s `Some(list)` arm feeds `fanout` straight into
// this filter — pin that a bogus name in that pre-computed list
// gets dropped exactly like a parsed `agent-<name>` input would.
let names = vec![
"iris".to_owned(),
"../../etc".to_owned(),
"atlas".to_owned(),
];
assert_eq!(validate_agent_names(names), vec!["iris", "atlas"]);
}
}

4
hive-c0re/src/job_queue/mod.rs
View file

@ -1,7 +1,7 @@
//! Generic job-DAG queue + desired-state reconciliation — the host-side
//! wrapper over the domain-agnostic [`hive_jobq`] scheduler. Jobs are nodes in
//! per-request DAGs (see [`templates`]); the special cases (graceful-stop
//! watcher, deferred-start follow-up, meta-update cascade) collapse into DAG
//! watcher, deferred-start follow-up) collapse into DAG
//! *shapes* over the shared node primitives ([`model::NodeKind`]).
//!
//! [`hive_jobq`] owns the graph, the two-class resource pool, and the roll-up
@ -327,7 +327,7 @@ impl JobQueue {
// without declaring its lease, and they are work on that agent
// that the operator wants to see.
//
// Empty means an agentless container kind (`MetaLock`, `Dag`),
// Empty means an agentless kind (the sweeps and pulls),
// which targets no agent and lights nothing.
let agent = n.payload.agent();
if agent.is_empty() {

41
hive-c0re/src/job_queue/model.rs
View file

@ -16,14 +16,10 @@ pub use hive_host_sock::jobs::State;
use serde::Serialize;
/// The primitive operations — each kind maps to one executor fn in
/// `exec.rs`, a thin wrapper over existing `lifecycle.rs` / `meta.rs`
/// code. Concurrency is gated by two resource classes (see
/// `exec.rs`, a thin wrapper over existing `lifecycle.rs` / `meta_rev.rs`
/// code. Concurrency is gated by resource classes (see
/// [`Resource`](super::resource::Resource), declared per node where the node
/// is constructed rather than derived from its kind); the
/// meta *repo* is serialized by `meta::META_LOCK` inside the wrapped
/// functions themselves, which is why there is no `GitCommit` node —
/// a standalone commit node would open a dirty-working-tree window
/// between nodes that the fused `meta.rs` ops deliberately close.
/// is constructed rather than derived from its kind).
///
/// What each kind does, the resource/lease rules, and the DAG shapes they
/// compose into live in the node-inventory table and surrounding sections
@ -33,10 +29,9 @@ use serde::Serialize;
#[serde(tag = "kind", rename_all = "snake_case")]
#[strum(serialize_all = "snake_case")]
pub enum NodeKind {
/// The rebuild's meta-repo preamble. `relock = false` only for
/// meta-update cascade rebuilds (re-locking would revert the bump the
/// cascade just committed).
MetaSync { agent: String, relock: bool },
/// The rebuild's preamble: runtime-dir prep and the `internal/meta`
/// commit the build reads.
MetaSync { agent: String },
/// Out-of-band toplevel build while the container keeps serving.
Prebuild { agent: String },
/// `nixos-container update` profile-swap.
@ -46,7 +41,7 @@ pub enum NodeKind {
/// First-spawn pre-create provisioning.
Provision { agent: String },
/// First-spawn `nixos-container create` proper. Assumes the
/// upstream `Provision` node already registered the agent in meta.
/// upstream `Provision` node already registered the agent.
Create { agent: String },
/// `nixos-container destroy` plus the un-registration that follows it.
///
@ -60,15 +55,7 @@ pub enum NodeKind {
PurgeState { agent: String },
/// The post-destroy bookkeeping tail.
DestroyBookkeeping { agent: String },
/// Meta flake lock bump. `sweep = false`: `meta::lock_update` with this
/// node's own `inputs`; `sweep = true`: `meta::lock_update_hyperhive`
/// (boot sweep, `inputs` empty). See _Notable collapses_ for how the
/// scheduler appends the fan-out `Rebuild` DAGs on completion.
MetaLock {
sweep: bool,
fanout: Option<Vec<String>>,
inputs: Vec<String>,
},
/// Idempotent power converge *planner*: fans a child
/// [`NodeKind::Start`] / [`NodeKind::Stop`] out at runtime rather than
/// doing the mechanical work itself.
@ -140,11 +127,7 @@ impl hive_jobq_wire::WireNode for NodeKind {
if !agent.is_empty() {
data.insert("agent".to_owned(), agent.into());
}
if let NodeKind::MetaLock { inputs, .. } = self
&& !inputs.is_empty()
{
data.insert("inputs".to_owned(), inputs.clone().into());
}
// Not in the payload at all — the build log is keyed on node identity
// in a side table, which is why `data` is handed the id.
if let Some(log) = crate::build_logs::global().and_then(|h| h.id_for_node(id)) {
@ -160,8 +143,7 @@ impl hive_jobq_wire::WireNode for NodeKind {
impl NodeKind {
/// The agent this node targets, or `""` for agentless kinds
/// ([`NodeKind::MetaLock`] on the `hyperhive` pseudo-agent, and the
/// swarm-wide sweeps and pulls).
/// (the swarm-wide sweeps and pulls).
#[must_use]
pub fn agent(&self) -> &str {
match self {
@ -186,8 +168,7 @@ impl NodeKind {
| NodeKind::AgentWindow { agent }
| NodeKind::EmitRebuilt { agent, .. }
| NodeKind::SetWanted { agent, .. } => agent,
NodeKind::MetaLock { .. }
| NodeKind::ForgeSweep
NodeKind::ForgeSweep
| NodeKind::MatrixSweep
| NodeKind::KnowledgePull
| NodeKind::WantedPull => "",

8
hive-c0re/src/job_queue/power.rs
View file

@ -103,7 +103,7 @@ fn start_chain(
// Rebuild subtree chained behind the `SetWanted` head. `MetaSync`, the
// `AgentWindow` brace and `Reconcile` are their own group roots
// (top-level, per `rebuild_nodes`) — hence all four names.
let roots = rebuild_nodes(builder, agent, true, Some(wanted));
let roots = rebuild_nodes(builder, agent, Some(wanted));
vec![
wanted.guid(),
roots.meta_sync.guid(),
@ -287,14 +287,12 @@ pub async fn start_many(
coord: &Arc<Coordinator>,
agents: &[String],
) -> anyhow::Result<Vec<NodeId>> {
let current = crate::auto_update::current_flake_rev(&coord.hyperhive_flake);
let current = crate::meta_rev::current_hyperhive_pin().await;
let mut targets = Vec::with_capacity(agents.len());
for agent in agents {
let running = lifecycle::is_running(agent).await;
let stored = read_rev_marker(&crate::paths::applied_rev_marker(agent));
let stale = current
.as_ref()
.is_some_and(|rev| stored.as_deref() != Some(rev.as_str()));
let stale = !crate::meta_rev::marker_is_fresh(current.as_deref(), stored.as_deref());
if !running && stale {
tracing::info!(%agent, "start: rev stale + agent down — rebuild-then-start");
}

16
hive-c0re/src/job_queue/resource.rs
View file

@ -11,15 +11,13 @@
//!
//! All of a node's resource edges are acquired **atomically**
//! (`try_acquire_all`) — a node never holds one resource while waiting on
//! another, so the multi-resource nodes (a `MetaLock` wants a build slot *and*
//! the meta window) cannot deadlock against each other.
//! another, so the multi-resource nodes (the rebuild brace wants a build slot
//! *and* the agent lease) cannot deadlock against each other.
/// The two resource classes the queue gates concurrency on, as the crate's
/// The resource classes the queue gates concurrency on, as the crate's
/// generic resource type `R`. What each holds, who declares it, and the
/// exemptions: see `docs/scheduler/coordinator.md`'s _Scheduler semantics_
/// (build slots, the per-agent lease) and _Two further layers protect the
/// meta repo_ (the deploy window) sections — this enum stays the one-line
/// summary, not a second copy.
/// section — this enum stays the one-line summary, not a second copy.
#[derive(Debug, Clone, PartialEq, Eq, Hash)]
pub enum Resource {
/// One of the `buildSlots` permits, held by a nix-heavy node for its
@ -28,9 +26,7 @@ pub enum Resource {
/// The per-agent lifecycle lease — globally exclusive per agent across
/// all DAGs.
Agent(String),
/// The meta-repo mutation window — a global singleton held by any node
/// that mutates the meta repo, so two meta mutations never interleave.
MetaWindow,
/// The `/knowledge` working tree, held by every `KnowledgePull`. Two
/// overlapping `reset` / `clean` / `pull` passes there fail on
/// `.git/index.lock` and the remote-tracking ref lock.
@ -53,7 +49,7 @@ impl hive_jobq_wire::WireResource for Resource {
match self {
Resource::BuildSlot => "build-slot".to_owned(),
Resource::Agent(agent) => format!("agent:{agent}"),
Resource::MetaWindow => "meta-window".to_owned(),
Resource::KnowledgeTree => "knowledge-tree".to_owned(),
Resource::MatrixSweep => "matrix-sweep".to_owned(),
}

2
hive-c0re/src/job_queue/scheduler.rs
View file

@ -14,7 +14,7 @@
//! focused terminal node (`EmitRebuilt`), dispatched through `exec::run_node`
//! like any other node once the DAG settles.
//!
//! In-DAG growth (a `MetaLock` growing rebuild subgraphs, a `Reconcile` fanning
//! In-DAG growth (a `Reconcile` fanning
//! its `Start`/`Stop`) is declared onto the builder each node is handed, and
//! inserted as part of completing that node. Completion itself is not this
//! module's job any more: it happens *inside* the future

91
hive-c0re/src/job_queue/templates.rs
View file

@ -55,31 +55,6 @@ fn emit_rebuilt_tails(builder: &JobBuilder, agent: &str, roots: &[Handle<'_>]) {
);
}
/// Declare one rebuild subgraph per agent onto the builder a running
/// [`NodeKind::MetaLock`] was handed.
///
/// **Into the emitter's own builder, not as new DAGs.** Growing in-DAG is what
/// roots each subgraph on the `MetaLock`, so the whole sweep (or meta-update
/// cascade) stays one unit of work the operator can watch and cancel, and every
/// rebuild builds against the lock the emitter just bumped.
///
/// Same reason as [`fanned_out_mechanical`] for living here: this was the
/// second construction site declaring nodes inline in an executor.
pub(crate) fn grown_rebuilds(builder: &JobBuilder, agents: &[String], relock: bool) {
for agent in agents {
rebuild_nodes(builder, agent, relock, None);
}
}
/// As [`grown_rebuilds`], but each agent gets its `Signal` → `Drain` window
/// before being stopped. The boot sweep's flavour: it stops agents that were
/// mid-turn when the host came up, so they drain rather than being cut off.
pub(crate) fn grown_graceful_rebuilds(builder: &JobBuilder, agents: &[String], relock: bool) {
for agent in agents {
graceful_rebuild_nodes(builder, agent, relock, None);
}
}
/// Declare the mechanical node a [`NodeKind::Reconcile`] planner fans out
/// (`Start` / `Stop`) onto the builder it was handed while running.
///
@ -162,7 +137,7 @@ pub(crate) fn pause_quiesce<'a>(builder: &'a JobBuilder, agent: &str) -> Handle<
/// cover every node in the subgraph without the caller knowing its shape.
#[derive(Debug, Clone, Copy)]
pub(crate) struct RebuildRoots<'a> {
/// The meta-repo preamble.
/// The preamble.
pub meta_sync: Handle<'a>,
/// The brace holding the agent lease and the build slot — its roll-up
/// carries the whole mechanical subtree (`Prebuild`, the quiesce chain,
@ -182,13 +157,9 @@ impl<'a> RebuildRoots<'a> {
/// The rebuild node subtree — three group roots (`MetaSync`, the `AgentWindow`
/// brace, `Reconcile`). `after`, when given, is the node it chains behind. The
/// shape itself is in `docs/scheduler/coordinator.md`; the code below is the source of
/// truth for it, so only the three choices a reader would otherwise undo are
/// truth for it, so only the two choices a reader would otherwise undo are
/// called out here:
///
/// - **`MetaSync` is a sibling root, not the brace's parent** — it owns the
/// *global* `MetaWindow`, and a resource is held across the holder's whole
/// subtree, so parenting the rebuild under it would serialise every agent's
/// nix build behind one hive-wide window.
/// - **`StopForUpdate` waits on `Prebuild` as well as `Drain`** — running the
/// drain early is the win; taking the container *down* early is pure downtime.
/// - **`Reconcile` is a top-level root, not a child** — so it survives the
@ -198,15 +169,12 @@ impl<'a> RebuildRoots<'a> {
fn rebuild_subtree<'a>(
builder: &'a JobBuilder,
agent: &str,
relock: bool,
graceful: bool,
after: Option<Handle<'a>>,
) -> RebuildRoots<'a> {
let a = || agent.to_owned();
let mut meta_sync = builder
.node(NodeKind::MetaSync { agent: a(), relock })
.needs(Resource::MetaWindow);
let mut meta_sync = builder.node(NodeKind::MetaSync { agent: a() });
if let Some(after) = after {
meta_sync = meta_sync.after_ok(after);
}
@ -263,17 +231,15 @@ fn rebuild_subtree<'a>(
}
}
/// The rebuild subtree, stopping the agent outright — the shape five of the six
/// call sites want. `relock` re-locks the meta flake inside `MetaSync`; `after`,
/// when given, is the node this subgraph chains behind. See
/// [`rebuild_subtree`] for the structure.
/// The rebuild subtree, stopping the agent outright — the shape most call
/// sites want. `after`, when given, is the node this subgraph chains behind.
/// See [`rebuild_subtree`] for the structure.
pub(crate) fn rebuild_nodes<'a>(
builder: &'a JobBuilder,
agent: &str,
relock: bool,
after: Option<Handle<'a>>,
) -> RebuildRoots<'a> {
rebuild_subtree(builder, agent, relock, false, after)
rebuild_subtree(builder, agent, false, after)
}
/// As [`rebuild_nodes`], but the agent gets a `Signal` → `Drain` window to
@ -286,17 +252,15 @@ pub(crate) fn rebuild_nodes<'a>(
pub(crate) fn graceful_rebuild_nodes<'a>(
builder: &'a JobBuilder,
agent: &str,
relock: bool,
after: Option<Handle<'a>>,
) -> RebuildRoots<'a> {
rebuild_subtree(builder, agent, relock, true, after)
rebuild_subtree(builder, agent, true, after)
}
/// One uniform rebuild shape — no `was_running` branch. `StopForUpdate`
/// noops when already down; the tail `Reconcile` auto-noops the start
/// when `wanted = Offline` (a rebuild of a deliberately-stopped agent
/// leaves it stopped). `relock = false` only for meta-update cascade
/// children.
/// leaves it stopped).
///
/// Closed by an [`NodeKind::EmitRebuilt`] tail edged onto all three group-roots
/// (`MetaSync`, the `AgentWindow` brace, `Reconcile`) — the brace's roll-up
@ -307,8 +271,8 @@ pub(crate) fn graceful_rebuild_nodes<'a>(
///
/// Returns those three roots, so a caller that needs to wait on the rebuild can
/// name them.
pub fn rebuild(builder: &JobBuilder, agent: &str, relock: bool) -> Vec<hive_jobq::NodeGuid> {
let roots = rebuild_nodes(builder, agent, relock, None);
pub fn rebuild(builder: &JobBuilder, agent: &str) -> Vec<hive_jobq::NodeGuid> {
let roots = rebuild_nodes(builder, agent, None);
emit_rebuilt_tails(builder, agent, &roots.all());
vec![
roots.meta_sync.guid(),
@ -318,7 +282,7 @@ pub fn rebuild(builder: &JobBuilder, agent: &str, relock: bool) -> Vec<hive_jobq
}
/// First deploy of an agent this hive has never seen, asked for by the swarm:
/// `Provision` (proposed/applied repos, state subvolume, meta registration)
/// `Provision` (proposed/applied repos, state subvolume, agent-set registration)
/// then `Create` (`nixos-container create`), drop-in write, then `Reconcile`
/// starts the container (`wanted = Up`, seeded by
/// `swarm_status::queue_first_deploy`). All-or-nothing: `Provision`
@ -330,9 +294,7 @@ pub fn rebuild(builder: &JobBuilder, agent: &str, relock: bool) -> Vec<hive_jobq
/// Returns the group root so a caller can wait on the whole subtree.
pub fn first_deploy(builder: &JobBuilder, agent: &str) -> Vec<hive_jobq::NodeGuid> {
let a = || agent.to_owned();
let provision = builder
.node(NodeKind::Provision { agent: a() })
.needs(Resource::MetaWindow);
let provision = builder.node(NodeKind::Provision { agent: a() });
let create = builder
.node(NodeKind::Create { agent: a() })
.needs(Resource::BuildSlot)
@ -403,31 +365,10 @@ pub fn destroy(builder: &JobBuilder, agent: &str, purge: bool) {
.after_ok(last);
}
/// Meta-input lock bump. The `MetaLock` executor grows one rebuild subgraph
/// per affected agent into *this same* DAG on completion (declared onto the
/// builder it was handed) — appended *after* the bump lands so their prebuilds
/// run against the post-bump lock, and a failed bump appends nothing
/// (replacing the old fan-out-child-DAGs dance).
/// `transient = Rebuilding` because those appended subgraphs are rebuilds:
/// it's applied per-agent at claim time (the `MetaLock` head needs no lease,
/// so the "hyperhive" pseudo-agent gets no pill), giving each cascade agent
/// crash-watch suppression during its `Swap` — the property the old child
/// `Rebuild` DAGs carried via their own transient.
pub fn meta_update(builder: &JobBuilder, inputs: Vec<String>) {
let _lock = builder
.node(NodeKind::MetaLock {
sweep: false,
fanout: None,
inputs,
})
.needs(Resource::BuildSlot)
.needs(Resource::MetaWindow);
}
// The boot is assembled inline in `workers/auto_update.rs::submit_boot_tree`
// as ONE `Boot` DAG (a sweep `MetaLock` root that grows rebuild subgraphs
// in-DAG, plus a `Reconcile` root per drifted agent) — no anchor node and no
// per-agent child DAGs.
// as ONE `Boot` DAG (a graceful rebuild subgraph per stale agent, plus a
// `Reconcile` root per drifted agent) — no anchor node and no per-agent child
// DAGs.
/// One matrix user/space sweep. Holding [`Resource::MatrixSweep`] is what keeps
/// two passes from overlapping, whichever caller submitted each.

82
hive-c0re/src/job_queue/tests.rs
View file

@ -51,7 +51,7 @@ fn insert_named(
}
fn rebuild(builder: &JobBuilder, agent: &str) -> Vec<hive_jobq::NodeGuid> {
templates::rebuild(builder, agent, true)
templates::rebuild(builder, agent)
}
/// Restart shape with every agent treated as **running** — the online
@ -468,7 +468,7 @@ fn rebuild_chain_is_declared_serial() {
fn graceful_rebuild_chain_drains_before_stopping() {
let q = JobQueue::new(1);
insert(&q, |builder| {
templates::graceful_rebuild_nodes(builder, "agent-a", true, None);
templates::graceful_rebuild_nodes(builder, "agent-a", None);
});
// Asserted as full rows, not just kinds: the kind list is identical whether
// the quiesce chain runs beside the build or nested under it, so a
@ -515,8 +515,8 @@ fn graceful_rebuild_chain_drains_before_stopping() {
}
/// The non-graceful shape is the default everywhere except the boot sweep:
/// a manual rebuild, a meta-update cascade child and a deploy must NOT spend a
/// drain window, so `StopForUpdate` waits only on `Prebuild`.
/// a manual rebuild and a deploy must NOT spend a drain window, so
/// `StopForUpdate` waits only on `Prebuild`.
#[test]
fn non_graceful_rebuild_has_no_signal_or_drain() {
// Read the shape off the queue rather than out of a node list: a declared
@ -524,7 +524,7 @@ fn non_graceful_rebuild_has_no_signal_or_drain() {
// observable where it matters — in what the scheduler runs.
let q = JobQueue::new(1);
insert(&q, |builder| {
templates::rebuild_nodes(builder, "agent-a", true, None);
templates::rebuild_nodes(builder, "agent-a", None);
});
assert_eq!(
declared_shape(&q)
@ -576,10 +576,10 @@ fn rebuild_chain_declares_its_resources_on_the_brace() {
res("reconcile"),
],
[
// The meta preamble takes the global window and *nothing else* —
// no slot (it does no nix work) and no lease. It stays a sibling
// root so that hive-global window is not held across any build.
vec![Resource::MetaWindow],
// The preamble declares nothing — no slot (it does no nix work) and
// no lease. It stays a sibling root so nothing it holds is held
// across a build.
vec![],
// The brace takes both, atomically, and holds them for its whole
// subtree. Hoisting the slot here is not a widening: it already
// spanned the entire rebuild when `Prebuild` held it, because a unit
@ -782,44 +782,43 @@ fn boot_sweep_nodes_declare_their_own_resources() {
// nothing in this module covered it. When resource derivation moved to the
// construction sites, this path was missed and both kinds silently declared
// nothing — dropping the agent lease a boot `Reconcile` needs to not race
// another DAG's container ops, and letting the sweep `MetaLock` land its
// meta commit inside another node's staged deploy window. Nothing failed to
// compile; only an exhaustive caller list would have caught it.
// another DAG's container ops. Nothing failed to compile; only an
// exhaustive caller list would have caught it.
let q = JobQueue::new(4);
insert(&q, |builder| {
crate::workers::auto_update::boot_nodes(
builder,
true,
vec!["stale-agent".to_owned()],
&["stale-agent".to_owned()],
vec!["drifted-agent".to_owned()],
);
});
let mut lock = declared_resources(&q, node_of(&q, "meta_lock"));
lock.sort_by_key(|r| format!("{r:?}"));
let mut window = declared_resources(&q, node_of(&q, "agent_window"));
window.sort_by_key(|r| format!("{r:?}"));
assert_eq!(
lock,
vec![Resource::BuildSlot, Resource::MetaWindow],
"the sweep MetaLock runs a nix lock bump and commits to meta"
window,
vec![
Resource::Agent("stale-agent".to_owned()),
Resource::BuildSlot
],
"a swept rebuild holds the agent's lease and a build slot"
);
// Two `reconcile` nodes exist: the swept agent's recovery tail and the
// drifted agent's own.
assert_eq!(
declared_resources(&q, node_of(&q, "reconcile")),
vec![Resource::Agent("drifted-agent".to_owned())],
declared_resources_of_kind(&q, "reconcile"),
vec![
vec![Resource::Agent("drifted-agent".to_owned())],
vec![Resource::Agent("stale-agent".to_owned())],
],
"a boot Reconcile touches the container, so it holds that agent's lease"
);
}
/// Crash-watch suppression for a cascade rebuild, which the deleted half of
/// `meta_update_grows_cascade_in_dag` used to assert via a DAG-level
/// `transient` field on the submit-time spec (both the field and the spec type
/// are gone).
///
/// The property is unchanged — a container going down under a rebuild must not
/// read as a crash — but it is no longer a DAG-level declaration: each node
/// answers for itself, so the assertion moves to the nodes a cascade actually
/// runs. Kept as its own test rather than dropped, because it is the *property*
/// that mattered, not the field that used to carry it.
/// Crash-watch suppression for a rebuild: a container going down under a
/// rebuild must not read as a crash. Each node answers for itself, so the
/// assertion is on the nodes a rebuild actually runs.
#[test]
fn rebuild_chain_nodes_suppress_crash_watch() {
for kind in [
@ -990,26 +989,15 @@ fn a_fanned_out_mechanical_node_declares_its_agent_lease() {
);
}
/// A running `MetaLock` grows one rebuild subgraph per agent into **its own
/// DAG**, rooted on itself — not as child DAGs. That is what keeps a boot sweep
/// (or a meta-update cascade) one unit of work, with every rebuild building
/// against the lock the emitter just bumped.
///
/// Replaces `grown_subgraph_roots_on_emitter_and_rebases_local_deps` and
/// `meta_update_grows_cascade_in_dag`, which differed only in which rebuild
/// flavour they grew and each minted a builder by hand to simulate the graft.
/// What they were checking is the `grown_*_rebuilds` templates, so this calls
/// one — the boot sweep's, since that is the caller that grows a graceful one.
///
/// That the grafted work lands under the emitter, and that the emitter parks in
/// `Finishing` until it settles, is `hive_jobq`'s
/// (`a_completing_node_grows_the_work_it_declared`).
/// The boot sweep declares one rebuild subgraph per stale agent into **one
/// DAG**, each an independent root — not as child DAGs. That keeps a boot sweep
/// one unit of work the operator can watch and cancel.
#[test]
fn a_meta_lock_grows_one_rebuild_subgraph_per_agent() {
fn the_boot_sweep_declares_one_rebuild_subgraph_per_stale_agent() {
let q = JobQueue::new(4);
let agents = vec!["alice".to_owned(), "bob".to_owned()];
insert(&q, |builder| {
templates::grown_graceful_rebuilds(builder, &agents, true);
crate::workers::auto_update::boot_nodes(builder, &agents, Vec::new());
});
// One chain per agent, each an independent group root — so the two rebuild

47
hive-c0re/src/lifecycle/git.rs
View file

@ -187,44 +187,6 @@ pub async fn git_rev_parse(dir: &Path, refname: &str) -> Result<String> {
Ok(String::from_utf8_lossy(&out.stdout).trim().to_owned())
}
/// The contents of `path` at commit `rev` in `dir`, `None` when `rev` has no
/// such file. Reads the object database, not the working tree.
pub async fn git_show_file(dir: &Path, rev: &str, path: &str) -> Result<Option<String>> {
let spec = format!("{rev}:{path}");
let listed = run_with_deadline(
git_command()
.current_dir(dir)
.args(["ls-tree", "--name-only", rev, "--", path]),
GIT_LOCAL_DEADLINE,
&format!("git ls-tree {spec} in {}", dir.display()),
)
.await?;
if !listed.status.success() {
bail!(
"git ls-tree {spec} failed ({}): {}",
listed.status,
String::from_utf8_lossy(&listed.stderr).trim()
);
}
if listed.stdout.iter().all(u8::is_ascii_whitespace) {
return Ok(None);
}
let shown = run_with_deadline(
git_command().current_dir(dir).args(["show", &spec]),
GIT_LOCAL_DEADLINE,
&format!("git show {spec} in {}", dir.display()),
)
.await?;
if !shown.status.success() {
bail!(
"git show {spec} failed ({}): {}",
shown.status,
String::from_utf8_lossy(&shown.stderr).trim()
);
}
Ok(Some(String::from_utf8_lossy(&shown.stdout).into_owned()))
}
/// Plant a lightweight tag at `target`. Errors if the tag already
/// exists — we want loud failures on id reuse, not silent
/// overwrites.
@ -241,9 +203,10 @@ pub async fn git_read_tree_reset(dir: &Path, target: &str) -> Result<()> {
git(dir, &["read-tree", "--reset", "-u", target]).await
}
/// Hard-set a ref to `target`. Used to fast-forward `refs/heads/main`
/// to the just-deployed proposal commit. Uses `update-ref`, not
/// `branch -f`, so it works regardless of where HEAD currently sits.
/// Hard-set a ref to `target` with `update-ref`, so it works regardless of
/// where HEAD currently sits. Test fixtures only: production moves refs with
/// [`git_update_ref_cas`].
#[cfg(test)]
pub async fn git_update_ref(dir: &Path, refname: &str, target: &str) -> Result<()> {
git(dir, &["update-ref", refname, target]).await
}
@ -253,7 +216,7 @@ pub async fn git_update_ref(dir: &Path, refname: &str, target: &str) -> Result<(
/// untouched — when the current value differs. That refusal is the whole
/// difference between "advance this branch" and "overwrite whatever is there".
///
/// Prefer this over [`git_update_ref`] whenever the caller already knows the
/// Prefer this over a plain `update-ref` whenever the caller already knows the
/// value it believes the ref holds: a plain `update-ref` that raced another
/// writer discards the other writer's commits without a word.
///

4
hive-c0re/src/lifecycle/host_config.rs
View file

@ -529,8 +529,8 @@ async fn set_nspawn_flags(
}
}
// systemd-nspawn refuses to start a container whose bind
// source doesn't exist. The meta repo is created by the
// startup migration, but make sure the directory is there
// source doesn't exist. The directory is created by the
// startup migration, but make sure it is there
// before the role holder comes up in case set_nspawn_flags
// fires first (e.g. cold start with no agents).
let meta_root = crate::paths::meta_root();

97
hive-c0re/src/lifecycle/mod.rs
View file

@ -7,10 +7,12 @@ mod setup;
#[cfg(test)]
mod tests;
#[cfg(test)]
pub use git::git_update_ref;
pub use git::{
FETCH_DEADLINE, GIT_LOCAL_DEADLINE, git, git_authed, git_command, git_command_authed,
git_is_ancestor, git_read_tree_reset, git_rev_parse, git_show_file, git_tag, git_update_ref,
git_update_ref_cas, run_with_deadline,
git_is_ancestor, git_read_tree_reset, git_rev_parse, git_tag, git_update_ref_cas,
run_with_deadline,
};
pub use host_config::write_dropins;
pub use setup::{
@ -207,10 +209,10 @@ async fn port_collision(self_name: &str) -> Result<Option<String>> {
/// First-spawn pre-create provisioning — the job queue's `Provision`
/// node. Fail fast on a port collision, set up the proposed/applied
/// repos + state subvolume + claude/notes dirs, then register the new
/// agent in the meta flake (`sync_agents`) so the later
/// `nixos-container create --flake meta#<name>` ref resolves. Does NOT
/// agent in the host-side files that follow the agent set
/// ([`crate::agent_set::refresh`]). Does NOT
/// create the container — that's `create_only` / the `Create` node.
pub async fn provision_container(name: &str, hive: &HiveEnv, paths: &AgentPaths) -> Result<()> {
pub async fn provision_container(name: &str, paths: &AgentPaths) -> Result<()> {
validate(name)?;
if let Some(other) = port_collision(name).await? {
bail!(
@ -219,7 +221,7 @@ pub async fn provision_container(name: &str, hive: &HiveEnv, paths: &AgentPaths)
);
}
// Read before anything is seeded: an archetype agent gets no config
// repo, and the meta render below builds it from the archetype.
// repo, and the build takes it from the archetype.
crate::archetypes::refresh(name).await?;
if crate::archetypes::of(name)?.is_none() {
setup_proposed(&paths.proposed, name).await?;
@ -228,19 +230,15 @@ pub async fn provision_container(name: &str, hive: &HiveEnv, paths: &AgentPaths)
ensure_agent_state_subvolume(name).await?;
ensure_claude_dir(&paths.claude)?;
ensure_state_dir(&paths.notes)?;
// Meta flake gets the new agent's input + nixosConfiguration
// before `nixos-container create` so the `--flake meta#<name>`
// ref resolves.
let agents = agents_after_spawn(name).await?;
crate::meta::sync_agents(hive, &agents).await
crate::agent_set::refresh(&agents).await
}
/// The `nixos-container create` proper — the job queue's `Create` node.
/// Assumes `provision_container` already registered the agent in meta.
/// `meta_rev` is the `internal/meta` commit to build from; none builds from the
/// hive's own meta flake (`crate::meta_rev`).
pub async fn create_only(name: &str, meta_rev: Option<&str>) -> Result<()> {
priv_run_inner("create", name, meta_rev, None).await
/// Assumes `provision_container` already registered the agent.
/// `meta_rev` is the `internal/meta` commit to build from (`crate::meta_rev`).
pub async fn create_only(name: &str, meta_rev: &str) -> Result<()> {
priv_run_inner("create", name, Some(meta_rev), None).await
}
/// Rebuild-path preamble, run by the job queue's `Prebuild` node: fail fast on
@ -298,15 +296,15 @@ pub async fn swap_update(
name: &str,
hive: &HiveEnv,
paths: &AgentPaths,
meta_rev: Option<&str>,
meta_rev: &str,
node_id: Option<u64>,
) -> Result<()> {
write_dropins(name, hive, paths).await?;
priv_run_inner("update", name, meta_rev, node_id).await
priv_run_inner("update", name, Some(meta_rev), node_id).await
}
/// Build the `AgentSpec` list for the meta flake from `nixos-container
/// list` + a hypothetical extra name not yet in the list (for spawn
/// Build the `AgentSpec` list for [`crate::agent_set::refresh`] from
/// `nixos-container list` + a hypothetical extra name not yet in the list (for spawn
/// where the new agent's container doesn't exist yet). Pass empty
/// `name_to_add` from rebuild paths where the agent is already in the
/// container list.
@ -314,17 +312,17 @@ pub async fn swap_update(
/// Propagates errors from `list()` rather than swallowing them.
/// Using `.unwrap_or_default()` here would silently produce an empty
/// agent list when `nixos-container list` fails (priv helper down, race),
/// which `sync_agents` would then commit to meta — dropping every agent
/// from `flake.nix`. Callers that can tolerate failures (e.g. migration)
/// which would drop every agent from the topology and the gateway. Callers
/// that can tolerate failures (e.g. migration)
/// handle the `Err` themselves with `.unwrap_or_default()`.
async fn agents_for_meta(name_to_add: Option<&str>) -> Result<Vec<crate::meta::AgentSpec>> {
async fn agents_for_meta(name_to_add: Option<&str>) -> Result<Vec<hive_meta_flake::AgentSpec>> {
let containers = list().await?;
let archetypes = crate::archetypes::read_at(&crate::archetypes::path())?;
let mut out: Vec<crate::meta::AgentSpec> = containers
let mut out: Vec<hive_meta_flake::AgentSpec> = containers
.into_iter()
.filter_map(|c| {
let name = c.strip_prefix(AGENT_PREFIX)?.to_owned();
Some(crate::meta::AgentSpec {
Some(hive_meta_flake::AgentSpec {
port: agent_web_port(&name),
archetype: archetypes.get(&name).cloned(),
pin: None,
@ -335,7 +333,7 @@ async fn agents_for_meta(name_to_add: Option<&str>) -> Result<Vec<crate::meta::A
if let Some(extra) = name_to_add
&& !out.iter().any(|a| a.name == extra)
{
out.push(crate::meta::AgentSpec {
out.push(hive_meta_flake::AgentSpec {
port: agent_web_port(extra),
archetype: archetypes.get(extra).cloned(),
pin: None,
@ -346,7 +344,7 @@ async fn agents_for_meta(name_to_add: Option<&str>) -> Result<Vec<crate::meta::A
Ok(out)
}
async fn agents_after_spawn(name: &str) -> Result<Vec<crate::meta::AgentSpec>> {
async fn agents_after_spawn(name: &str) -> Result<Vec<hive_meta_flake::AgentSpec>> {
agents_for_meta(Some(name)).await
}
@ -354,7 +352,7 @@ async fn agents_after_spawn(name: &str) -> Result<Vec<crate::meta::AgentSpec>> {
/// `nixos-container list` says), sorted, no extras. For callers
/// outside this module that need to reseed meta after lifecycle
/// changes — destroy, startup reconciliation, etc.
pub async fn agents_for_meta_listing() -> Result<Vec<crate::meta::AgentSpec>> {
pub async fn agents_for_meta_listing() -> Result<Vec<hive_meta_flake::AgentSpec>> {
agents_for_meta(None).await
}
@ -970,6 +968,13 @@ fn make_log_callback(
}
}
/// A `create` or `update` asked for without an `internal/meta` commit. The
/// public entry points take the commit, so this guards the string dispatch
/// below only.
fn missing_meta_rev(kind: &str, name: &str) -> anyhow::Error {
anyhow::anyhow!("{name}: {kind} needs an internal/meta commit to build from")
}
/// Execute a container operation via hive-priv and integrate with
/// `build_logs.sqlite`. hive-priv runs as root and forwards output lines
/// to hive-c0re in real time via the streaming priv protocol. Each line
@ -1007,22 +1012,28 @@ async fn priv_run_inner(
// For long-running ops use the streaming protocol so build_logs
// receives lines in real time rather than as a batch at completion.
let result: Result<()> = match kind {
"create" => {
crate::priv_client::create_container_streaming(
name,
meta_rev,
make_log_callback(logs.clone(), log_id, cmdline.clone()),
)
.await
}
"update" => {
crate::priv_client::update_container_streaming(
name,
meta_rev,
make_log_callback(logs.clone(), log_id, cmdline.clone()),
)
.await
}
"create" => match meta_rev {
Some(rev) => {
crate::priv_client::create_container_streaming(
name,
rev,
make_log_callback(logs.clone(), log_id, cmdline.clone()),
)
.await
}
None => Err(missing_meta_rev(kind, name)),
},
"update" => match meta_rev {
Some(rev) => {
crate::priv_client::update_container_streaming(
name,
rev,
make_log_callback(logs.clone(), log_id, cmdline.clone()),
)
.await
}
None => Err(missing_meta_rev(kind, name)),
},
"start" => crate::priv_client::start_container(name).await,
"stop" => crate::priv_client::stop_container(name).await,
"kill" => crate::priv_client::kill_container(name).await,

4
hive-c0re/src/lifecycle/setup.rs
View file

@ -260,8 +260,8 @@ fn initial_agent_nix(name: &str) -> String {
}
/// Module-only flake exposed by every agent's repo. Consumed by the
/// hive-c0re-owned meta flake at `/var/lib/hyperhive/meta/` as a flake
/// input. The wrapper is intentionally permissive:
/// swarm's `internal/meta` flake as a flake input. The wrapper is
/// intentionally permissive:
///
/// - Manager edits `inputs.* = …` to add other flakes (e.g. an MCP
/// server's own flake) — the lock for those lands in the agent's

10
hive-c0re/src/main.rs
View file

@ -15,6 +15,7 @@ use clap::{Parser, Subcommand};
// root so `crate::broker::…` style paths keep resolving unchanged.
mod actions;
mod agent_config;
mod agent_set;
mod container_view;
mod coordinator;
mod dashboard;
@ -241,10 +242,9 @@ async fn cmd_serve(
let coord = Arc::new(Coordinator::open(&db, env, model_prices, build_slots)?);
// Idempotent pre-flight: rewrite pre-meta-layout applied
// repos, ensure proposed repos carry the `applied`
// remote, bootstrap the meta repo, repoint containers at
// `meta#<name>` (one-shot, guarded by a marker file).
// Runs before the boot reconcile, so agents are built against meta.
if let Err(e) = migrate::run(&coord).await {
// remote, refresh the files that follow the agent set.
// Runs before the boot reconcile.
if let Err(e) = migrate::run().await {
tracing::warn!(error = ?e, "startup migration failed");
}
// Before any agent starts, so the bind mounts it gets from the stored
@ -402,7 +402,7 @@ async fn cmd_serve(
// unregister_agent. No recurring poll needed — c0re owns the listeners.
mcp_sockets::sync_on_start(coord.clone()).await;
// Job-queue scheduler: drives the global DAG queue (rebuild /
// meta-update / spawn / power ops). Concurrency comes from the
// spawn / power ops). Concurrency comes from the
// build-slot count + per-agent leases inside the queue, not from
// multiple workers — cheap nodes (graceful signals, drains,
// reconciles) overlap nix-heavy ones structurally. Call sites

1671
hive-c0re/src/meta.rs
View file

File diff suppressed because it is too large Load diff

440
hive-c0re/src/meta_rev.rs
View file

@ -1,15 +1,20 @@
//! The `internal/meta` commit a hive builds an agent from.
//!
//! `internal/meta` is the swarm's meta flake, committed by swarm-controller.
//! A rebuild's `MetaSync` node resolves its head once ([`resolve`]) and
//! records it ([`record`]); `Prebuild` and `Swap` read the record
//! ([`recorded`]), so both build the same flake. A create resolves and uses
//! the rev directly. No rev, whether because the repo or its commit is absent
//! or because the commit does not pin this agent's current config, leaves the
//! build on the hive's own meta flake.
//! `internal/meta` is the swarm's meta flake, committed by swarm-controller;
//! a hive has no meta flake of its own. A rebuild's `MetaSync` node resolves
//! the head ([`resolve`]) and records it ([`record`]); `Prebuild` and `Swap`
//! read the record ([`recorded`]), so both build the same flake. A create
//! resolves and uses the commit directly. When the head does not pin the
//! agent's current config, the build fails rather than falling back: the
//! swarm re-pins before it asks a hive to deploy.
//!
//! The record also names the pins a successful build deploys. [`mark_deployed`]
//! copies them to the agent's markers, which the boot reconcile
//! ([`crate::auto_update`]) and the dashboard's update chip read.
use anyhow::{Context, Result};
use anyhow::{Context, Result, bail};
use hive_priv_sock::{meta_flake_url, valid_meta_rev};
use serde::{Deserialize, Serialize};
use swarm_queue_client::agent_archetype::ArchetypeRef;
/// What the forge says about `internal/meta` and an agent's config repo.
@ -26,118 +31,197 @@ pub(crate) trait MetaSource {
async fn config_head(&self, agent: &str) -> Result<Option<String>>;
}
/// The `internal/meta` commit to build `agent` from, or `None` for the hive's
/// own flake.
/// An `internal/meta` commit and the pins it holds for one agent.
#[derive(Debug, PartialEq, Eq, Serialize, Deserialize)]
pub(crate) struct MetaBuild {
/// The `internal/meta` commit.
pub rev: String,
/// The hyperhive commit its flake pins.
pub hyperhive: String,
/// The agent's config commit its flake pins; none for an archetype agent.
pub config: Option<String>,
}
/// The `internal/meta` commit to build `agent` from.
///
/// `internal/meta` is private, so a hive whose nix cannot authenticate
/// ([`MetaSource::nix_can_fetch`]) builds from its own flake. The head is
/// taken only when its `flake.nix` builds the agent the way it is wanted now:
/// from `archetype` when that is set, otherwise from the config commit the
/// forge's `main` is at. A head that is behind (the swarm's update has not
/// run, or failed), lacks the agent or lists it under another archetype would
/// build an older module than the one being deployed.
/// ([`MetaSource::nix_can_fetch`]) cannot build. The head is taken only when
/// its `flake.nix` builds the agent the way it is wanted now: from
/// `archetype` when that is set, otherwise from the config commit the forge's
/// `main` is at. A head that is behind (the swarm's update has not run, or
/// failed), lacks the agent or lists it under another archetype would build an
/// older module than the one being deployed.
///
/// # Errors
/// A forge read that failed.
/// No usable head, a head that does not pin the agent's current config, or a
/// forge read that failed.
pub(crate) async fn choose(
source: &impl MetaSource,
agent: &str,
archetype: Option<&ArchetypeRef>,
) -> Result<Option<String>> {
) -> Result<MetaBuild> {
if !source.nix_can_fetch() {
return Ok(None);
bail!("{agent}: this hive has no forge credential for nix to fetch internal/meta with");
}
let Some(rev) = source.meta_head().await? else {
bail!("{agent}: the swarm has not published internal/meta yet");
};
if !valid_meta_rev(&rev) {
bail!("{agent}: internal/meta's head {rev:?} is not a commit id");
}
let flake = source.flake_at(&rev).await?;
let config = if archetype.is_some() {
None
} else {
source.config_head(agent).await?
};
if !hive_meta_flake::builds_agent_at(&flake, agent, archetype, config.as_deref()) {
bail!("{agent}: the swarm meta flake does not pin its config yet (internal/meta at {rev})");
}
let Some(hyperhive) = hive_meta_flake::pinned_rev(&flake, "hyperhive") else {
bail!("{agent}: internal/meta at {rev} pins no hyperhive commit");
};
Ok(MetaBuild {
rev,
hyperhive: hyperhive.to_owned(),
config,
})
}
/// The hyperhive commit `internal/meta`'s head pins; `None` when the repo or
/// its commit is absent.
///
/// # Errors
/// A forge read that failed, or a head that pins no hyperhive commit.
pub(crate) async fn hyperhive_pin(source: &impl MetaSource) -> Result<Option<String>> {
let Some(rev) = source.meta_head().await? else {
return Ok(None);
};
if !valid_meta_rev(&rev) {
return Ok(None);
}
let flake = source.flake_at(&rev).await?;
let config_head = if archetype.is_some() {
None
} else {
let Some(head) = source.config_head(agent).await? else {
return Ok(None);
};
Some(head)
};
Ok(
hive_meta_flake::builds_agent_at(&flake, agent, archetype, config_head.as_deref())
.then_some(rev),
)
let pin = hive_meta_flake::pinned_rev(&flake, "hyperhive")
.with_context(|| format!("internal/meta at {rev} pins no hyperhive commit"))?;
Ok(Some(pin.to_owned()))
}
/// [`choose`] against the forge this hive reads. `None`, logged, whenever the
/// hive cannot tell: no forge credential, a failed read, no usable commit.
pub(crate) async fn resolve(agent: &str) -> Option<String> {
let archetype = match crate::archetypes::of(agent) {
Ok(archetype) => archetype,
Err(e) => {
tracing::warn!(%agent, error = ?e, "internal/meta: cannot read the agent's archetype; building from the hive's own flake");
return None;
}
};
/// The hyperhive commit `internal/meta` pins now, as this hive reads it: what
/// an agent's marker is compared with to tell it is stale. `None`, logged,
/// when the hive cannot tell (no forge token, a failed read, no commit yet),
/// which callers treat as nothing to rebuild for.
pub(crate) async fn current_hyperhive_pin() -> Option<String> {
let Some(source) = crate::forge::internal_meta_source().await else {
tracing::info!(%agent, "internal/meta: no forge read token; building from the hive's own flake");
tracing::info!("internal/meta: no forge read token; cannot tell which hyperhive it pins");
return None;
};
match choose(&source, agent, archetype.as_ref()).await {
Ok(Some(rev)) => {
tracing::info!(%agent, %rev, "internal/meta: building from the swarm's meta flake");
Some(rev)
}
Ok(None) => {
tracing::info!(%agent, "internal/meta: absent or not pinning this agent's config; building from the hive's own flake");
None
}
match hyperhive_pin(&source).await {
Ok(pin) => pin,
Err(e) => {
tracing::warn!(%agent, error = %format!("{e:#}"), "internal/meta: cannot read it; building from the hive's own flake");
tracing::warn!(error = %format!("{e:#}"), "internal/meta: cannot read its hyperhive pin");
None
}
}
}
/// Record `rev` as the commit `agent`'s build reads, or clear the record for
/// `None`. The record is in hive-c0re's own state ([`crate::paths::meta_rev_dir`]),
/// where the agent cannot write.
/// [`choose`] against the forge this hive reads.
///
/// # Errors
/// The record file could not be written or removed.
pub(crate) fn record(agent: &str, rev: Option<&str>) -> Result<()> {
record_in(&crate::paths::meta_rev_dir(), agent, rev)
/// As [`choose`], or this hive holds no forge token or cannot read the
/// agent's archetype.
pub(crate) async fn resolve(agent: &str) -> Result<MetaBuild> {
let archetype = crate::archetypes::of(agent)
.with_context(|| format!("{agent}: cannot read the agent's archetype"))?;
let Some(source) = crate::forge::internal_meta_source().await else {
bail!("{agent}: this hive holds no forge token to read internal/meta with");
};
let build = choose(&source, agent, archetype.as_ref()).await?;
tracing::info!(%agent, rev = %build.rev, "internal/meta: building from the swarm's meta flake");
Ok(build)
}
/// The commit [`record`] stored for `agent`, if it is a valid one.
pub(crate) fn recorded(agent: &str) -> Option<String> {
/// Record `build` as what `agent`'s rebuild builds from. The record is in
/// hive-c0re's own state ([`crate::paths::meta_rev_dir`]), where the agent
/// cannot write.
///
/// # Errors
/// The record file could not be written.
pub(crate) fn record(agent: &str, build: &MetaBuild) -> Result<()> {
record_in(&crate::paths::meta_rev_dir(), agent, build)
}
/// What [`record`] stored for `agent`, if it is a valid record.
pub(crate) fn recorded(agent: &str) -> Option<MetaBuild> {
recorded_in(&crate::paths::meta_rev_dir(), agent)
}
/// Written whole to a temp file beside the record and renamed over it, so a
/// reader sees the old record or the new one and a crash leaves no partial
/// file.
fn record_in(dir: &std::path::Path, agent: &str, rev: Option<&str>) -> Result<()> {
fn record_in(dir: &std::path::Path, agent: &str, build: &MetaBuild) -> Result<()> {
std::fs::create_dir_all(dir).with_context(|| format!("create {}", dir.display()))?;
let path = dir.join(agent);
match rev {
let tmp = dir.join(format!(".{agent}.tmp"));
std::fs::write(&tmp, serde_json::to_vec(build)?)
.with_context(|| format!("write {}", tmp.display()))?;
std::fs::rename(&tmp, &path)
.with_context(|| format!("rename {} to {}", tmp.display(), path.display()))
}
fn recorded_in(dir: &std::path::Path, agent: &str) -> Option<MetaBuild> {
let build: MetaBuild = serde_json::from_slice(&std::fs::read(dir.join(agent)).ok()?).ok()?;
let valid = valid_meta_rev(&build.rev)
&& valid_meta_rev(&build.hyperhive)
&& build.config.as_deref().is_none_or(valid_meta_rev);
valid.then_some(build)
}
/// The record of `agent`'s build, or the error naming why a `Prebuild` or
/// `Swap` has nothing to build from.
///
/// # Errors
/// No valid record: its `MetaSync` did not run, or ran in an older hive-c0re.
pub(crate) fn recorded_or_err(agent: &str) -> Result<MetaBuild> {
recorded(agent)
.with_context(|| format!("{agent}: no internal/meta commit was recorded for this build"))
}
/// Mark `agent` as running what `build` pins: the hyperhive commit for the
/// boot reconcile ([`crate::paths::applied_rev_marker`]), the config commit
/// for the dashboard's deployed chip ([`crate::paths::applied_config_marker`]).
///
/// # Errors
/// A marker could not be written or removed.
pub(crate) fn mark_deployed(agent: &str, build: &MetaBuild) -> Result<()> {
mark_deployed_in(
&crate::paths::applied_rev_marker(agent),
&crate::paths::applied_config_marker(agent),
build,
)
}
fn mark_deployed_in(
hyperhive: &std::path::Path,
config: &std::path::Path,
build: &MetaBuild,
) -> Result<()> {
std::fs::write(hyperhive, &build.hyperhive)
.with_context(|| format!("write {}", hyperhive.display()))?;
match &build.config {
Some(rev) => {
std::fs::create_dir_all(dir).with_context(|| format!("create {}", dir.display()))?;
let tmp = dir.join(format!(".{agent}.tmp"));
std::fs::write(&tmp, rev).with_context(|| format!("write {}", tmp.display()))?;
std::fs::rename(&tmp, &path)
.with_context(|| format!("rename {} to {}", tmp.display(), path.display()))
std::fs::write(config, rev).with_context(|| format!("write {}", config.display()))
}
None => match std::fs::remove_file(&path) {
None => match std::fs::remove_file(config) {
Err(e) if e.kind() != std::io::ErrorKind::NotFound => {
Err(e).with_context(|| format!("remove {}", path.display()))
Err(e).with_context(|| format!("remove {}", config.display()))
}
_ => Ok(()),
},
}
}
fn recorded_in(dir: &std::path::Path, agent: &str) -> Option<String> {
let rev = std::fs::read_to_string(dir.join(agent)).ok()?;
valid_meta_rev(rev.trim()).then(|| rev.trim().to_owned())
/// Whether an agent's `stored` marker says it runs the hyperhive commit `pin`
/// names. With no `pin`, because the hive can't read `internal/meta`, nothing
/// is stale: a rebuild would have no flake to build from either.
pub(crate) fn marker_is_fresh(pin: Option<&str>, stored: Option<&str>) -> bool {
pin.is_none_or(|pin| stored == Some(pin))
}
/// The `<flake>#<agent>` ref [`crate::lifecycle::prebuild_toplevel`] builds
@ -233,36 +317,52 @@ mod tests {
}
}
fn reviewer() -> ArchetypeRef {
ArchetypeRef {
source: "hyperhive".to_owned(),
name: "reviewer".to_owned(),
}
}
/// The error a build with no usable pin ends in: it names the agent and
/// says the swarm meta flake does not pin its config yet.
fn assert_unpinned(err: &anyhow::Error, agent: &str) {
let msg = format!("{err:#}");
assert!(
msg.starts_with(&format!("{agent}: "))
&& msg.contains("swarm meta flake does not pin its config yet"),
"{msg}"
);
}
#[tokio::test]
async fn the_head_is_taken_when_it_pins_the_agents_current_config() {
assert_eq!(
choose(&forge(Some(ARGUS_HEAD)), "argus", None)
.await
.unwrap(),
Some(META.to_owned())
MetaBuild {
rev: META.to_owned(),
hyperhive: HYPERHIVE.to_owned(),
config: Some(ARGUS_HEAD.to_owned()),
}
);
}
#[tokio::test]
async fn a_head_behind_the_agents_config_is_not_taken() {
assert_eq!(
choose(&forge(Some(STALE)), "argus", None).await.unwrap(),
None
);
async fn a_head_behind_the_agents_config_is_an_error() {
let err = choose(&forge(Some(STALE)), "argus", None)
.await
.unwrap_err();
assert_unpinned(&err, "argus");
}
#[tokio::test]
async fn a_head_without_the_agent_is_not_taken() {
async fn a_head_without_the_agent_is_an_error() {
let mut source = forge(Some(ARGUS_HEAD));
source.flake = source.flake.replace("argus", "someone");
assert_eq!(choose(&source, "argus", None).await.unwrap(), None);
}
fn reviewer() -> ArchetypeRef {
ArchetypeRef {
source: "hyperhive".to_owned(),
name: "reviewer".to_owned(),
}
let err = choose(&source, "argus", None).await.unwrap_err();
assert_unpinned(&err, "argus");
}
#[tokio::test]
@ -285,84 +385,120 @@ mod tests {
let source = Archetypes(forge(None));
assert_eq!(
choose(&source, "rev", Some(&reviewer())).await.unwrap(),
Some(META.to_owned())
MetaBuild {
rev: META.to_owned(),
hyperhive: HYPERHIVE.to_owned(),
config: None,
},
"an archetype agent pins no config"
);
let other = ArchetypeRef {
name: "planner".to_owned(),
..reviewer()
};
assert_eq!(
choose(&source, "rev", Some(&other)).await.unwrap(),
None,
"the head lists another archetype"
assert_unpinned(
&choose(&source, "rev", Some(&other)).await.unwrap_err(),
"rev",
);
assert_eq!(
choose(&source, "ghost", Some(&reviewer())).await.unwrap(),
None
assert_unpinned(
&choose(&source, "ghost", Some(&reviewer()))
.await
.unwrap_err(),
"ghost",
);
}
/// The head lists argus as a custom agent and rev as an archetype agent;
/// an agent switched either way is not what the head builds.
#[tokio::test]
async fn an_agent_switched_between_custom_and_archetype_is_not_taken() {
async fn an_agent_switched_between_custom_and_archetype_is_an_error() {
let source = forge(Some(ARGUS_HEAD));
assert_eq!(
choose(&source, "argus", Some(&reviewer())).await.unwrap(),
None,
"custom in the head, archetype now"
);
assert_eq!(
choose(&source, "rev", None).await.unwrap(),
None,
"archetype in the head, custom now"
assert_unpinned(
&choose(&source, "argus", Some(&reviewer()))
.await
.unwrap_err(),
"argus",
);
assert_unpinned(&choose(&source, "rev", None).await.unwrap_err(), "rev");
}
/// A hive without the credential nix's git reads builds from its own flake
/// even though its own token reads the repo; the first test is the control.
/// A hive without the credential nix's git reads cannot build from the
/// repo even though its own token reads it; the first test is the control.
#[tokio::test]
async fn a_hive_whose_nix_cannot_authenticate_builds_from_its_own_flake() {
async fn a_hive_whose_nix_cannot_authenticate_cannot_build() {
let mut source = forge(Some(ARGUS_HEAD));
source.fetchable = false;
assert_eq!(choose(&source, "argus", None).await.unwrap(), None);
let err = choose(&source, "argus", None).await.unwrap_err();
assert!(
format!("{err:#}").contains("no forge credential"),
"{err:#}"
);
}
#[tokio::test]
async fn no_repo_or_no_commit_is_the_hives_own_flake() {
async fn no_repo_or_no_commit_is_an_error_naming_the_agent() {
let mut source = forge(Some(ARGUS_HEAD));
source.head = None;
assert_eq!(choose(&source, "argus", None).await.unwrap(), None);
let err = choose(&source, "argus", None).await.unwrap_err();
assert!(
format!("{err:#}").starts_with("argus: the swarm has not published internal/meta"),
"{err:#}"
);
}
#[tokio::test]
async fn a_custom_agent_with_no_config_commit_is_the_hives_own_flake() {
assert_eq!(choose(&forge(None), "argus", None).await.unwrap(), None);
async fn a_custom_agent_with_no_config_commit_is_an_error() {
let err = choose(&forge(None), "argus", None).await.unwrap_err();
assert_unpinned(&err, "argus");
}
#[tokio::test]
async fn the_hyperhive_pin_is_the_one_in_the_head() {
assert_eq!(
hyperhive_pin(&forge(None)).await.unwrap().as_deref(),
Some(HYPERHIVE)
);
let mut source = forge(None);
source.head = None;
assert_eq!(hyperhive_pin(&source).await.unwrap(), None);
let mut source = forge(None);
source.flake = source.flake.replace("hyperhive.url", "other.url");
assert!(hyperhive_pin(&source).await.is_err(), "a head with no pin");
}
fn build() -> MetaBuild {
MetaBuild {
rev: META.to_owned(),
hyperhive: HYPERHIVE.to_owned(),
config: Some(ARGUS_HEAD.to_owned()),
}
}
#[test]
fn a_record_is_read_back_and_cleared() {
fn a_record_is_read_back_and_replaced() {
let dir = tempfile::tempdir().unwrap();
assert_eq!(recorded_in(dir.path(), "argus"), None);
record_in(dir.path(), "argus", Some(META)).unwrap();
assert_eq!(recorded_in(dir.path(), "argus"), Some(META.to_owned()));
record_in(dir.path(), "argus", &build()).unwrap();
assert_eq!(recorded_in(dir.path(), "argus"), Some(build()));
assert_eq!(
recorded_in(dir.path(), "rev"),
None,
"a record is one agent's"
);
record_in(dir.path(), "argus", Some(STALE)).unwrap();
assert_eq!(recorded_in(dir.path(), "argus"), Some(STALE.to_owned()));
record_in(dir.path(), "argus", None).unwrap();
assert_eq!(recorded_in(dir.path(), "argus"), None);
record_in(dir.path(), "argus", None).unwrap();
let next = MetaBuild {
rev: STALE.to_owned(),
config: None,
..build()
};
record_in(dir.path(), "argus", &next).unwrap();
assert_eq!(recorded_in(dir.path(), "argus"), Some(next));
}
#[test]
fn a_record_leaves_no_temp_file_and_creates_its_dir() {
let dir = tempfile::tempdir().unwrap();
let records = dir.path().join("meta-rev");
record_in(&records, "argus", Some(META)).unwrap();
record_in(&records, "argus", &build()).unwrap();
let names: Vec<_> = std::fs::read_dir(&records)
.unwrap()
.map(|e| e.unwrap().file_name())
@ -370,11 +506,33 @@ mod tests {
assert_eq!(names, ["argus"]);
}
/// Anything but a record naming commit ids is no record: a plain commit id
/// (the older format), a branch name, a bad pin.
#[test]
fn a_record_that_is_not_a_commit_id_is_no_record() {
fn a_record_that_does_not_name_commit_ids_is_no_record() {
let dir = tempfile::tempdir().unwrap();
std::fs::write(dir.path().join("argus"), "main\n").unwrap();
assert_eq!(recorded_in(dir.path(), "argus"), None);
for bad in [
META.to_owned(),
"main\n".to_owned(),
serde_json::to_string(&MetaBuild {
rev: "main".to_owned(),
..build()
})
.unwrap(),
serde_json::to_string(&MetaBuild {
hyperhive: "main".to_owned(),
..build()
})
.unwrap(),
serde_json::to_string(&MetaBuild {
config: Some("main".to_owned()),
..build()
})
.unwrap(),
] {
std::fs::write(dir.path().join("argus"), &bad).unwrap();
assert_eq!(recorded_in(dir.path(), "argus"), None, "{bad}");
}
}
/// The record is hive-c0re's own state, never the agent-writable runtime
@ -385,6 +543,32 @@ mod tests {
assert!(crate::paths::meta_rev_dir().starts_with(crate::paths::state_root()));
}
#[test]
fn marking_deployed_writes_both_pins_and_clears_the_config_of_an_archetype_agent() {
let dir = tempfile::tempdir().unwrap();
let hyperhive = dir.path().join("hyperhive");
let config = dir.path().join("config");
mark_deployed_in(&hyperhive, &config, &build()).unwrap();
assert_eq!(std::fs::read_to_string(&hyperhive).unwrap(), HYPERHIVE);
assert_eq!(std::fs::read_to_string(&config).unwrap(), ARGUS_HEAD);
let archetype = MetaBuild {
config: None,
..build()
};
mark_deployed_in(&hyperhive, &config, &archetype).unwrap();
assert!(!config.exists(), "an archetype agent pins no config");
mark_deployed_in(&hyperhive, &config, &archetype).unwrap();
}
#[test]
fn a_marker_is_fresh_only_at_the_pin() {
assert!(marker_is_fresh(Some(HYPERHIVE), Some(HYPERHIVE)));
assert!(!marker_is_fresh(Some(HYPERHIVE), Some(STALE)), "behind");
assert!(!marker_is_fresh(Some(HYPERHIVE), None), "never built");
assert!(marker_is_fresh(None, Some(STALE)), "no pin to compare with");
assert!(marker_is_fresh(None, None));
}
#[test]
fn the_prebuild_ref_is_the_flake_hive_priv_builds() {
assert_eq!(

22
hive-c0re/src/migrate.rs
View file

@ -1,5 +1,5 @@
//! Startup convergence. Three phases, all idempotent and unguarded:
//! harness files, applied + proposed repos, meta repo. They re-run every
//! harness files, applied + proposed repos, the agent-set files. They re-run every
//! boot on purpose — each one is a no-op once its state is already
//! correct.
//!
@ -19,7 +19,6 @@
//! details: `docs/agent-lifecycle/config-changes.md::Startup migrations (older hosts)`.
use std::path::Path;
use std::sync::Arc;
use std::time::Duration;
use anyhow::{Context, Result};
@ -27,7 +26,6 @@ use tokio::process::Command;
use crate::coordinator::Coordinator;
use crate::lifecycle::{self, AGENT_PREFIX};
use crate::meta;
const KILL_SWITCH: &str = "HIVE_SKIP_META_MIGRATION";
@ -47,7 +45,7 @@ const GIT_TIMEOUT: Duration = Duration::from_mins(2);
/// (raw `import ./agent.nix`) get rewritten on next hive-c0re start.
const MODULE_FLAKE_MARKER: &str = "_module.args.flakeInputs";
pub async fn run(coord: &Arc<Coordinator>) -> Result<()> {
pub async fn run() -> Result<()> {
if std::env::var(KILL_SWITCH).is_ok() {
tracing::info!("migration: {KILL_SWITCH} set — skipping");
return Ok(());
@ -107,21 +105,21 @@ pub async fn run(coord: &Arc<Coordinator>) -> Result<()> {
}
}
// Phase 3: meta repo.
tracing::debug!("migration: phase 3 (meta sync_agents)");
// `sync_agents` renders exactly the agents it is given, so an empty
// stand-in for an unreadable list would drop every agent from the flake.
// Phase 3: the files that follow the agent set.
tracing::debug!("migration: phase 3 (agent_set refresh)");
// `refresh` takes the agents it is given as the whole set, so an empty
// stand-in for an unreadable list would drop every agent from the topology.
let agents = match lifecycle::agents_for_meta_listing().await {
Ok(agents) => agents,
Err(e) => {
tracing::warn!(error = ?e, "migration: container list failed; skipping meta sync_agents");
tracing::warn!(error = ?e, "migration: container list failed; skipping agent_set refresh");
return Ok(());
}
};
match tokio::time::timeout(GIT_TIMEOUT, meta::sync_agents(&coord.hive_env(), &agents)).await {
Ok(Err(e)) => tracing::warn!(error = ?e, "migration: meta sync_agents failed"),
match tokio::time::timeout(GIT_TIMEOUT, crate::agent_set::refresh(&agents)).await {
Ok(Err(e)) => tracing::warn!(error = ?e, "migration: agent_set refresh failed"),
Err(_) => {
tracing::warn!(timeout = ?GIT_TIMEOUT, "migration: meta sync_agents timed out — skipping");
tracing::warn!(timeout = ?GIT_TIMEOUT, "migration: agent_set refresh timed out — skipping");
}
Ok(Ok(())) => {}
}

23
hive-c0re/src/paths.rs
View file

@ -168,26 +168,31 @@ pub fn applied_dir(name: &str) -> PathBuf {
applied_root().join(name)
}
/// `applied/.<name>.hyperhive-rev` — marker recording the flake rev an
/// agent was last successfully rebuilt against (auto-update staleness check).
/// `applied/.<name>.hyperhive-rev` — marker recording the hyperhive commit
/// `internal/meta` pinned when an agent was last successfully built (the boot
/// reconcile's staleness check).
#[must_use]
pub fn applied_rev_marker(name: &str) -> PathBuf {
applied_root().join(format!(".{name}.hyperhive-rev"))
}
/// `meta/` — the meta flake working tree (inputs, `flake.lock`, `.git`).
/// `applied/.<name>.config-rev` — marker recording the config commit
/// `internal/meta` pinned when an agent was last successfully built (the
/// dashboard's deployed chip). Absent for an archetype agent.
#[must_use]
pub fn applied_config_marker(name: &str) -> PathBuf {
applied_root().join(format!(".{name}.config-rev"))
}
/// `meta/` — plain state beside the hive's other files (`topology.json`,
/// the swarm limits, the permission seed files) and, on a hive that predates
/// `internal/meta`, a leftover git repo.
// nix: bind-mounted read-only into agent containers as `/meta` (the harness nix modules) — must match.
#[must_use]
pub fn meta_root() -> PathBuf {
state_root().join("meta")
}
/// `meta/flake.lock` — the meta flake lock (read for input rev display).
#[must_use]
pub fn meta_flake_lock() -> PathBuf {
meta_root().join("flake.lock")
}
/// `meta/.git/index.lock` — git index lock; checked before meta git ops
/// so a stale lock from a crashed process can be cleared.
#[must_use]

8
hive-c0re/src/priv_client.rs
View file

@ -235,14 +235,14 @@ pub async fn kill_container(name: &str) -> Result<()> {
/// appending lines to `build_logs` or otherwise capturing the output.
pub async fn update_container_streaming(
name: &str,
meta_rev: Option<&str>,
meta_rev: &str,
on_line: impl FnMut(PrivStream, &str),
) -> Result<()> {
ok(call_streaming(
&PrivRequest::UpdateContainer {
name: name.to_owned(),
stream: true,
meta_rev: meta_rev.map(str::to_owned),
meta_rev: meta_rev.to_owned(),
},
on_line,
)
@ -253,14 +253,14 @@ pub async fn update_container_streaming(
/// arrive. Returns `Ok(())` on success.
pub async fn create_container_streaming(
name: &str,
meta_rev: Option<&str>,
meta_rev: &str,
on_line: impl FnMut(PrivStream, &str),
) -> Result<()> {
ok(call_streaming(
&PrivRequest::CreateContainer {
name: name.to_owned(),
stream: true,
meta_rev: meta_rev.map(str::to_owned),
meta_rev: meta_rev.to_owned(),
},
on_line,
)

2
hive-c0re/src/server.rs
View file

@ -557,7 +557,7 @@ async fn submit_single(coord: &Arc<Coordinator>, name: &str, verb: Verb) -> Host
// so it is a plain template insert rather than a `*_many` gather.
let inserted = coord
.job_queue
.insert_job(|b| crate::job_queue::templates::rebuild(b, name, true));
.insert_job(|b| crate::job_queue::templates::rebuild(b, name));
if inserted.is_ok() {
coord.emit_rebuild_queue_snapshot();
}

8
hive-c0re/src/swarm_status.rs
View file

@ -315,13 +315,13 @@ async fn handle_deploy_request(
}
let inserted = if known {
// The same insert the operator's own `rebuild` verb makes, relock and
// all: "deploy this agent" means here exactly what it already meant, and
// a swarm-triggered rebuild that quietly did something narrower would be
// The same insert the operator's own `rebuild` verb makes: "deploy this
// agent" means here exactly what it already meant, and a
// swarm-triggered rebuild that quietly did something narrower would be
// a second definition of the word.
coord
.job_queue
.insert_job(|b| crate::job_queue::templates::rebuild(b, &agent, true))
.insert_job(|b| crate::job_queue::templates::rebuild(b, &agent))
.map(Some)
} else {
queue_first_deploy(coord, &agent)

6
hive-c0re/src/workers/agent_sockets.rs
View file

@ -175,9 +175,9 @@ pub fn write(names: &[String]) -> Result<()> {
pub fn spawn_poll() {
tokio::spawn(async move {
let mut interval = tokio::time::interval(std::time::Duration::from_secs(10));
// First tick fires immediately; that's fine — meta::sync_agents
// also writes on boot, this just catches up the window before
// the next agent restart.
// First tick fires immediately; that's fine — the migration's
// `agent_set::refresh` also writes on boot, this just catches up the
// window before the next agent restart.
loop {
interval.tick().await;
match crate::lifecycle::agents_for_meta_listing().await {

109
hive-c0re/src/workers/auto_update.rs
View file

@ -1,20 +1,21 @@
//! Boot reconcile: on `hive-c0re serve` boot, (a) run the config path
//! for agents whose per-agent rev marker is stale — one `Boot` DAG
//! (meta hyperhive lock bump) that grows a `Rebuild` subgraph for each
//! stale agent whose `wanted` power intent is `Up` — and (b)
//! converge every other drifted agent to its persisted `wanted` via
//! `Reconcile` DAGs. Two rules keep boot-time nix work minimal:
//! Boot reconcile: on `hive-c0re serve` boot, (a) rebuild the agents whose
//! per-agent rev marker differs from the hyperhive commit the swarm's
//! `internal/meta` pins now — one `Boot` DAG holding a `Rebuild` subgraph for
//! each stale agent whose `wanted` power intent is `Up` — and (b) converge
//! every other drifted agent to its persisted `wanted` via `Reconcile` DAGs.
//! This is how a hive that missed the swarm's fan-out catches up. Three rules
//! keep boot-time nix work minimal:
//!
//! 1. **Stale but wanted-offline agents** get no rebuild — their
//! rebuild happens the first time they're started (the start
//! submit path upgrades a stale start to rebuild+start). The sweep
//! parent still runs whenever *any* marker is stale so the meta
//! hyperhive lock is bumped for those later start-upgrades.
//! 2. **Agents whose rev marker matches** the current hyperhive flake
//! path are skipped — nothing changed, no nix work to do.
//! submit path upgrades a stale start to rebuild+start).
//! 2. **Agents whose rev marker matches** the pin are skipped — nothing
//! changed, no nix work to do.
//! 3. **A pin the hive cannot read** (no forge token, forge down, no
//! `internal/meta` yet) rebuilds nothing: a rebuild would have no flake
//! to build from either.
//!
//! Booting with no config change performs no meta commit — only
//! reconciles. See `docs/scheduler/coordinator.md::Boot reconcile`.
//! See `docs/scheduler/coordinator.md::Boot reconcile`.
use std::path::Path;
use std::sync::Arc;
@ -39,9 +40,10 @@ pub fn current_flake_rev(hyperhive_flake: &str) -> Option<String> {
}
/// Returns true when the applied repo has commits that have not yet been
/// deployed (i.e. the applied HEAD differs from the sha currently locked in
/// meta's flake.lock). This is the semantic the dashboard `needs_update` chip
/// conveys: "there is a config change ready to apply via rebuild."
/// deployed (i.e. the applied HEAD differs from the config commit the agent's
/// last successful build pinned, `paths::applied_config_marker`). This is the
/// semantic the dashboard `needs_update` chip conveys: "there is a config
/// change ready to apply via rebuild."
///
/// Async on purpose: this runs per agent inside `container_view::build_all`,
/// which fires on the ~10s dashboard sweep, every `AgentStatus` request, and
@ -67,10 +69,9 @@ pub async fn agent_config_pending(name: &str, deployed_sha: Option<&str>) -> boo
}
/// Boot reconcile (see the module doc): classify every agent by rev
/// freshness + persisted `wanted` intent, submit one `Boot` DAG
/// (hyperhive lock bump growing an in-DAG rebuild subgraph per stale
/// wanted-up agent) when anything is stale, and `Reconcile` DAGs for
/// agents whose observed power state drifted from `wanted`. Returns Ok even
/// freshness + persisted `wanted` intent, and submit one `Boot` DAG holding a
/// graceful rebuild subgraph per stale wanted-up agent and a `Reconcile` per
/// agent whose observed power state drifted from `wanted`. Returns Ok even
/// if some submissions failed.
pub async fn run(coord: Arc<Coordinator>) -> Result<()> {
let containers = match lifecycle::list().await {
@ -81,7 +82,7 @@ pub async fn run(coord: Arc<Coordinator>) -> Result<()> {
}
};
let current_rev = current_flake_rev(&coord.hyperhive_flake);
let pin = crate::meta_rev::current_hyperhive_pin().await;
// Resolve container names to logical agent names, then sort. The
// parent field this used to depth-sort by is gone; with no
@ -97,7 +98,6 @@ pub async fn run(coord: Arc<Coordinator>) -> Result<()> {
// Classify. `get_or_seed` doubles as the one-time migration: an
// agent without an `agent_power` row is seeded from its observed
// state (running ⇒ Up), after which the DB is authoritative.
let mut any_stale = false;
// stale ∧ wanted=Up → sweep rebuild. The bool is the agent's observed
// running state, used to order running agents first before submit.
let mut fanout: Vec<(String, bool)> = Vec::new();
@ -113,17 +113,11 @@ pub async fn run(coord: Arc<Coordinator>) -> Result<()> {
None
}
};
let fresh = current_rev.as_ref().is_some_and(|rev| {
std::fs::read_to_string(crate::paths::applied_rev_marker(name))
.is_ok_and(|stored| stored == rev.as_str())
});
let stored = std::fs::read_to_string(crate::paths::applied_rev_marker(name)).ok();
let fresh = crate::meta_rev::marker_is_fresh(pin.as_deref(), stored.as_deref());
match boot_action(wanted, fresh, running) {
BootAction::Rebuild => {
any_stale = true;
fanout.push((name.clone(), running));
}
BootAction::Rebuild => fanout.push((name.clone(), running)),
BootAction::Defer { reconcile } => {
any_stale = true;
n_deferred += 1;
tracing::debug!(%name, "boot reconcile: stale but offline — deferring rebuild to on-start");
if reconcile {
@ -158,7 +152,7 @@ pub async fn run(coord: Arc<Coordinator>) -> Result<()> {
fanout.sort_by_key(|(_, running)| !running);
let fanout: Vec<String> = fanout.into_iter().map(|(name, _)| name).collect();
submit_boot_tree(&coord, any_stale, fanout, drifted, n_deferred, n_skipped);
submit_boot_tree(&coord, &fanout, drifted, n_deferred, n_skipped);
submit_startup_sweep_nodes(&coord);
Ok(())
}
@ -166,7 +160,7 @@ pub async fn run(coord: Arc<Coordinator>) -> Result<()> {
/// What the boot sweep does with one agent.
#[derive(Debug, PartialEq, Eq)]
enum BootAction {
/// Stale ∧ wanted up: rebuild against the post-bump lock. The DAG's
/// Stale ∧ wanted up: rebuild from the pinned `internal/meta`. The DAG's
/// tail `Reconcile` brings the agent (back) up, covering both the
/// running-stale and stopped-but-wanted-up cases.
Rebuild,
@ -233,34 +227,22 @@ fn submit_startup_sweep_nodes(coord: &Arc<Coordinator>) {
/// That split is not cosmetic: this path constructs nodes outside
/// `job_queue/`, so it is the one place a resource declaration can be forgotten
/// without any in-module test noticing. It has happened once already — the
/// sweep `MetaLock` and the boot `Reconcile`s silently declared nothing when
/// kind-derived resources were removed, which drops the agent lease a boot
/// reconcile needs to not race another DAG's container ops.
/// boot `Reconcile`s silently declared nothing when kind-derived resources
/// were removed, which drops the agent lease a boot reconcile needs to not
/// race another DAG's container ops.
pub(crate) fn boot_nodes(
builder: &crate::job_queue::JobBuilder,
any_stale: bool,
fanout: Vec<String>,
fanout: &[String],
drifted: Vec<String>,
) {
use crate::job_queue::NodeKind;
use crate::job_queue::resource::Resource;
// Sweep whenever ANY marker is stale — even when every stale agent is
// wanted-offline: the hyperhive lock bump must land now so their later
// start-upgrade rebuilds build against it. No stale agents ⇒ no MetaLock
// ⇒ no meta commit on a no-change boot. The `fanout` list rides the
// MetaLock into `run_meta_lock`, which appends the rebuild subgraphs.
if any_stale {
let _ = builder
.node(NodeKind::MetaLock {
sweep: true,
fanout: Some(fanout),
// A sweep bumps `hyperhive` alone (`lock_update_hyperhive`),
// so it names no inputs.
inputs: Vec::new(),
})
.needs(Resource::BuildSlot)
.needs(Resource::MetaWindow);
// One graceful rebuild subgraph per stale agent, each an independent
// group root: agents that were mid-turn when the host came up drain rather
// than being cut off, and the rebuilds overlap on their own leases.
for agent in fanout {
let _ = crate::job_queue::templates::graceful_rebuild_nodes(builder, agent, None);
}
// One boot Reconcile per drifted agent — independent roots.
for name in drifted {
@ -273,23 +255,18 @@ pub(crate) fn boot_nodes(
}
/// Submit this boot's work as **one DAG** (no anchor node, no per-agent
/// child DAGs). Node 0 is the sweep `MetaLock` (only when
/// something is stale) — its executor bumps the hyperhive lock, then grows
/// one rebuild subgraph per stale agent into *this same* DAG (rooted on the
/// `MetaLock`, so they build against the post-bump lock; see
/// `exec::run_meta_lock`). Every drifted agent gets a boot `Reconcile` as an
/// independent root — a boot reconcile needs no lock bump, so it converges
/// concurrently with the sweep. No-op when there's nothing to do.
/// child DAGs). It holds one graceful rebuild subgraph per stale agent, and
/// every drifted agent gets a boot `Reconcile` as an independent root, so they
/// converge concurrently with the rebuilds. No-op when there's nothing to do.
fn submit_boot_tree(
coord: &Arc<Coordinator>,
any_stale: bool,
fanout: Vec<String>,
fanout: &[String],
drifted: Vec<String>,
n_deferred: usize,
n_skipped: usize,
) {
// Fully-quiet boot (nothing stale, nothing drifted) inserts nothing.
if !any_stale && drifted.is_empty() {
// Fully-quiet boot (nothing to rebuild, nothing drifted) inserts nothing.
if fanout.is_empty() && drifted.is_empty() {
return;
}
// The summary the sweep used to hand the container as its `reason` is a log
@ -309,7 +286,7 @@ fn submit_boot_tree(
// during their `Swap` (applied at claim time); a reconcile-only boot needs
// no transient.
if let Err(e) = coord.job_queue.insert_job(|b| {
boot_nodes(b, any_stale, fanout, drifted);
boot_nodes(b, fanout, drifted);
Vec::new()
}) {
tracing::warn!(error = ?e, "boot: sweep DAG insert failed");