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

drop the applied repo: the hive keeps no copy of an agent's config, only the deploy markers. Refs #5913

Browse source
This commit is contained in:
damocles 2026-11-01 17:19:41 +01:00
commit 862495530c
19 changed files with 190 additions and 1634 deletions
Download patch file Download diff file Expand all files Collapse all files

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

@ -5,127 +5,16 @@
use std::sync::Arc;
use anyhow::{Context as _, Result, bail};
use crate::coordinator::Coordinator;
use crate::lifecycle;
/// `applied/main`, and whether `target` descends from it. Fast-forwarding
/// `main` to a `target` that does not is a rewind: it drops every commit on
/// `main` that `target` lacks.
async fn applied_main_under(applied_dir: &std::path::Path, target: &str) -> Result<(String, bool)> {
let current_main = lifecycle::git_rev_parse(applied_dir, "refs/heads/main")
.await
.map_err(|e| anyhow::anyhow!("read applied/main: {e:#}"))?;
let descends = lifecycle::git_is_ancestor(applied_dir, &current_main, target)
.await
.map_err(|e| anyhow::anyhow!("ancestry check {current_main}..{target}: {e:#}"))?;
Ok((current_main, descends))
}
/// What [`advance_applied_to_rev`] did to `applied/main`.
#[derive(Debug, PartialEq, Eq)]
pub(crate) enum RevAdvance {
/// `applied/main` already was the rev, so there is nothing to deploy.
AlreadyApplied,
/// `applied/main` was fast-forwarded to the rev.
Advanced,
/// 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
/// 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 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`.
///
/// # Errors
///
/// Returns an error if the forge fetch fails, if `rev` has diverged from
/// `applied/main` (neither descendant nor ancestor), or if `main` moved during the fast-forward. `main` does not
/// move in any of these.
pub(crate) async fn advance_applied_to_rev(agent: &str, rev: &str) -> Result<RevAdvance> {
advance_applied_main(
&crate::paths::applied_dir(agent),
rev,
crate::forge::fetch_forge_main(agent),
)
.await
}
/// [`advance_applied_to_rev`] with the forge fetch passed in, so it is
/// 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 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,
rev: &str,
fetch: impl std::future::Future<Output = Result<T>>,
) -> Result<RevAdvance> {
let main = lifecycle::git_rev_parse(applied_dir, "refs/heads/main")
.await
.map_err(|e| anyhow::anyhow!("read applied/main: {e:#}"))?;
if main == rev {
return Ok(RevAdvance::AlreadyApplied);
}
fetch.await.context("fetch the forge config main")?;
let (main, descends) = applied_main_under(applied_dir, rev).await?;
if main == rev {
return Ok(RevAdvance::AlreadyApplied);
}
if !descends {
let contained = lifecycle::git_is_ancestor(applied_dir, rev, &main)
.await
.map_err(|e| anyhow::anyhow!("ancestry check {rev}..{main}: {e:#}"))?;
if contained {
return Ok(RevAdvance::Superseded);
}
bail!(
"{rev} does not descend from applied/main ({main}); fast-forwarding to it would \
discard commits"
);
}
ff_applied_main(applied_dir, rev, &main).await?;
Ok(RevAdvance::Advanced)
}
/// Fast-forward `applied/main` to `target` and sync the working tree.
///
/// Compare-and-swap, not a bare set: `main` must still be `expected_main`, the
/// sha the caller's ancestry gate ([`applied_main_under`]) ran against. A plain
/// `update-ref` moves the branch to `target` whatever it currently points at,
/// which turns "fast-forward" into "discard anything that landed in the
/// meantime" — the gate only proves `target` safe against the `main` observed
/// *then*, so this is what makes that proof still true *now*.
async fn ff_applied_main(
applied_dir: &std::path::Path,
target: &str,
expected_main: &str,
) -> Result<()> {
lifecycle::git_update_ref_cas(applied_dir, "refs/heads/main", target, expected_main)
.await
.map_err(|e| {
anyhow::anyhow!("ff main {expected_main} -> {target} (concurrent move?): {e:#}")
})?;
lifecycle::git_read_tree_reset(applied_dir, "refs/heads/main")
.await
.map_err(|e| anyhow::anyhow!("read-tree to main: {e:#}"))
}
/// Tear down a sub-agent container. By default this is non-destructive to
/// persistent state: the applied config repo and the Claude
/// credentials dir under `/var/lib/hyperhive/{agents,applied}/<name>/` are
/// kept, so recreating an agent of the same name reuses prior config + creds
/// persistent state: the state dir and the Claude
/// credentials dir under `/var/lib/hyperhive/agents/<name>/` are
/// kept, so recreating an agent of the same name reuses prior notes + creds
/// (no re-login). The ephemeral runtime dir under `/run/hyperhive/agents/`
/// is cleared because its contents (the mcp socket) don't survive restarts
/// anyway. With `purge=true` the persistent trees are also wiped — config
/// history, claude creds, notes — there is no undo.
/// anyway. With `purge=true` the persistent trees are also wiped — claude
/// creds, notes — there is no undo.
/// The bootstrap/root container is destroyable like any other. Nothing
/// recreates it afterwards: ruth stays gone until she is created again from
/// the swarm dashboard.
@ -144,113 +33,3 @@ pub fn destroy(coord: &Arc<Coordinator>, name: &str, purge: bool) {
}
coord.emit_rebuild_queue_snapshot();
}
#[cfg(test)]
mod tests {
use super::{RevAdvance, advance_applied_main};
use crate::lifecycle;
/// A repo with `main` at `b`, child of `a`, and `c`, a sibling of `b` off
/// `a`. Returns the dir and the shas `[a, b, c]`.
async fn repo() -> (tempfile::TempDir, [String; 3]) {
let dir = tempfile::tempdir().expect("tempdir");
let path = dir.path();
let commit = |msg: &'static str| async move {
lifecycle::git(
path,
&[
"-c",
"user.name=t",
"-c",
"user.email=t@t",
"commit",
"-q",
"--allow-empty",
"-m",
msg,
],
)
.await
.expect("commit");
lifecycle::git_rev_parse(path, "HEAD").await.expect("HEAD")
};
lifecycle::git(path, &["init", "-q", "--initial-branch=main"])
.await
.expect("init");
let a = commit("a").await;
let b = commit("b").await;
lifecycle::git(path, &["checkout", "-q", "-b", "side", &a])
.await
.expect("branch");
let c = commit("c").await;
lifecycle::git(path, &["checkout", "-q", "main"])
.await
.expect("checkout main");
(dir, [a, b, c])
}
async fn main_of(dir: &tempfile::TempDir) -> String {
lifecycle::git_rev_parse(dir.path(), "refs/heads/main")
.await
.expect("main")
}
#[tokio::test]
async fn the_applied_rev_is_a_no_op_without_a_fetch() {
let (dir, [_, b, _]) = repo().await;
let fetched = std::cell::Cell::new(false);
let fetch = async {
fetched.set(true);
anyhow::Ok(())
};
let outcome = advance_applied_main(dir.path(), &b, fetch).await;
assert_eq!(outcome.expect("no-op"), RevAdvance::AlreadyApplied);
assert!(!fetched.get(), "an applied rev must not reach the forge");
assert_eq!(main_of(&dir).await, b);
}
#[tokio::test]
async fn a_rev_not_descending_from_main_is_refused_and_main_stays() {
let (dir, [_, b, c]) = repo().await;
let outcome = advance_applied_main(dir.path(), &c, async { anyhow::Ok(()) }).await;
let err = outcome.expect_err("a sibling of main is not a fast-forward");
assert!(
format!("{err:#}").contains("does not descend from applied/main"),
"{err:#}"
);
assert_eq!(main_of(&dir).await, b);
}
#[tokio::test]
async fn an_ancestor_of_main_is_superseded_not_refused() {
let (dir, [a, b, _]) = repo().await;
let outcome = advance_applied_main(dir.path(), &a, async { anyhow::Ok(()) }).await;
assert_eq!(outcome.expect("no-op"), RevAdvance::Superseded);
assert_eq!(main_of(&dir).await, b);
}
#[tokio::test]
async fn a_descendant_rev_fast_forwards_main() {
let (dir, [a, b, _]) = repo().await;
lifecycle::git_update_ref(dir.path(), "refs/heads/main", &a)
.await
.expect("rewind main");
let outcome = advance_applied_main(dir.path(), &b, async { anyhow::Ok(()) }).await;
assert_eq!(outcome.expect("fast-forward"), RevAdvance::Advanced);
assert_eq!(main_of(&dir).await, b);
}
#[tokio::test]
async fn a_failed_fetch_leaves_main_alone() {
let (dir, [a, b, _]) = repo().await;
lifecycle::git_update_ref(dir.path(), "refs/heads/main", &a)
.await
.expect("rewind main");
let outcome = advance_applied_main(dir.path(), &b, async {
Err::<(), _>(anyhow::anyhow!("forge down"))
})
.await;
assert!(outcome.is_err());
assert_eq!(main_of(&dir).await, a);
}
}

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

@ -156,8 +156,7 @@ pub async fn build_all() -> anyhow::Result<Vec<ContainerView>> {
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.as_deref())
.await;
crate::auto_update::agent_config_pending(logical.as_str(), deployed_full.as_deref());
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

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

@ -286,8 +286,6 @@ mod serve_config_tests {
pub struct AgentPaths {
/// Runtime socket dir (tmpfs, recreated per boot).
pub agent: PathBuf,
/// Hive-c0re-authoritative applied config repo.
pub applied: PathBuf,
/// Claude OAuth credentials (survives purge boundary).
pub claude: PathBuf,
/// Agent durable notes + forge token (survives purge boundary).
@ -436,7 +434,6 @@ impl Coordinator {
.map_err(|e| anyhow::anyhow!("invalid agent name {name:?}: {e}"))?;
Ok(AgentPaths {
agent: agent_dir,
applied: crate::paths::applied_dir(name.as_str()),
claude: Self::agent_claude_dir(&name),
notes: Self::agent_notes_dir(&name),
})

14
hive-c0re/src/dashboard/tombstones.rs
View file

@ -175,15 +175,11 @@ pub(super) async fn post_purge_tombstone(
return error_response(&reason);
}
let mut errors = Vec::new();
for dir in [
crate::paths::agent_state_dir(&name),
crate::paths::applied_dir(name.as_str()),
] {
if dir.exists()
&& let Err(e) = std::fs::remove_dir_all(&dir)
{
errors.push(format!("{}: {e}", dir.display()));
}
let dir = crate::paths::agent_state_dir(&name);
if dir.exists()
&& let Err(e) = std::fs::remove_dir_all(&dir)
{
errors.push(format!("{}: {e}", dir.display()));
}
if errors.is_empty() {
tracing::info!(%name, "tombstone purged");

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

@ -6,13 +6,9 @@
mod internal_meta;
mod reader;
mod reconcile;
mod repos;
pub(crate) use internal_meta::internal_meta_source;
pub use reader::reader_token;
pub use reconcile::{fetch_forge_main, reconcile_config_apply, reconcile_config_status};
pub use repos::fetch_config_main_into_applied;
use std::future::IntoFuture;
use std::sync::OnceLock;

158
hive-c0re/src/forge/reconcile.rs
View file

@ -1,158 +0,0 @@
//! Reconcile an agent's local applied config checkout against its forge
//! `agent-configs/<agent>` `main`. Backs `hivectl forge reconcile-config`
//! via the host-socket `ReconcileConfig{Status,Apply}` requests. hive-c0re
//! owns the applied checkout + the forge credential, so the reconcile op
//! lives here rather than in the CLI.
//!
//! `Status` is read-only (fetches forge `main` into a scratch ref, reports
//! the divergence). `Apply(Forge)` resets the local applied checkout to
//! forge `main`; `internal/meta` pins forge `main` itself, not this tree. It
//! does NOT auto-deploy. `Apply(Local)` is not supported: forge `main` is
//! branch-protected, so it only advances through a config-PR merge.
use std::path::{Path, PathBuf};
use anyhow::{Context, Result};
use hive_host_sock::{HostResponse, ReconcileDirection};
use super::{CONFIG_ORG, forge_git_url, reader_token};
/// Scratch ref the forge `main` is fetched into — outside the normal
/// branch/tag namespace so it never collides with real refs.
pub(crate) const FORGE_MAIN_REF: &str = "refs/hyperhive/forge-config-main";
/// Fetch `agent-configs/<agent>` `main` into the applied repo's scratch
/// ref (read-only, no working-tree change) and return the applied dir.
/// Also how a swarm deploy of a merged commit gets that commit's objects.
pub async fn fetch_forge_main(agent: &str) -> Result<PathBuf> {
let Some(reader) = reader_token().await else {
anyhow::bail!(
"no forge read token: the swarm store holds none for this hive and there is no core \
token file"
);
};
let dir = crate::paths::applied_dir(agent);
if !dir.join(".git").exists() {
anyhow::bail!("agent `{agent}` has no applied config checkout");
}
let url = forge_git_url(&format!("{CONFIG_ORG}/{agent}"));
crate::lifecycle::git_authed(
&dir,
&["fetch", "--force", &url, &format!("main:{FORGE_MAIN_REF}")],
&reader.auth_header(),
)
.await
.context("fetch forge config main (does forge main exist yet?)")?;
Ok(dir)
}
/// Run a git command in `dir` and capture stdout, erroring on non-zero.
async fn git_out(dir: &Path, args: &[&str]) -> Result<String> {
let out = crate::lifecycle::git_command()
.current_dir(dir)
.args(args)
.output()
.await
.with_context(|| format!("invoke git {args:?}"))?;
if !out.status.success() {
anyhow::bail!(
"git {args:?} failed: {}",
String::from_utf8_lossy(&out.stderr).trim()
);
}
Ok(String::from_utf8_lossy(&out.stdout).into_owned())
}
/// Append `lines` (from a git text block) to `messages`, each indented.
fn push_indented(messages: &mut Vec<String>, block: &str) {
messages.extend(block.lines().map(|l| format!(" {l}")));
}
/// Report the divergence between the local applied checkout and forge
/// `main` (host request `ReconcileConfigStatus`). Read-only.
pub async fn reconcile_config_status(agent: &str, verbose: bool) -> Result<HostResponse> {
let dir = fetch_forge_main(agent).await?;
let counts = git_out(
&dir,
&[
"rev-list",
"--left-right",
"--count",
&format!("HEAD...{FORGE_MAIN_REF}"),
],
)
.await?;
let mut parts = counts.split_whitespace();
let local_ahead = parts.next().unwrap_or("0");
let forge_ahead = parts.next().unwrap_or("0");
let mut messages = vec![
format!("agent {agent}: local applied main <-> forge agent-configs/{agent} main"),
format!(" local ahead by {local_ahead}, forge ahead by {forge_ahead} commit(s)"),
];
if local_ahead == "0" && forge_ahead == "0" {
messages.push(" in sync — nothing to reconcile".to_owned());
return Ok(HostResponse::messages(messages));
}
if forge_ahead != "0" {
messages.push(" forge-only commits (applied by --from forge):".to_owned());
push_indented(
&mut messages,
&git_out(
&dir,
&["log", "--oneline", &format!("HEAD..{FORGE_MAIN_REF}")],
)
.await?,
);
}
if local_ahead != "0" {
messages.push(" local-only commits (discarded by --from forge):".to_owned());
push_indented(
&mut messages,
&git_out(
&dir,
&["log", "--oneline", &format!("{FORGE_MAIN_REF}..HEAD")],
)
.await?,
);
}
messages.push(" diff --stat (local HEAD -> forge main):".to_owned());
push_indented(
&mut messages,
&git_out(&dir, &["diff", "--stat", "HEAD", FORGE_MAIN_REF]).await?,
);
if verbose {
messages.push(" full diff:".to_owned());
push_indented(
&mut messages,
&git_out(&dir, &["diff", "HEAD", FORGE_MAIN_REF]).await?,
);
}
Ok(HostResponse::messages(messages))
}
/// Apply a reconcile in `direction` (host request `ReconcileConfigApply`).
/// `Forge` resets the local applied checkout to forge `main`; `Local` is
/// unsupported and returns a clear error.
pub async fn reconcile_config_apply(
agent: &str,
direction: ReconcileDirection,
) -> Result<HostResponse> {
match direction {
ReconcileDirection::Local => Ok(HostResponse::error(
"reconciling forge from local is not supported yet: forge main is \
branch-protected. Resolve via a config PR, or use \
`--from forge` to reset the local checkout to forge main.",
)),
ReconcileDirection::Forge => {
let dir = fetch_forge_main(agent).await?;
crate::lifecycle::git(&dir, &["reset", "--hard", FORGE_MAIN_REF])
.await
.context("reset applied main to forge main")?;
Ok(HostResponse::messages(vec![
format!("reconciled applied/{agent} to forge agent-configs/{agent} main"),
format!("effective on next deploy — rebuild {agent} to apply now"),
]))
}
}
}

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

@ -1,122 +0,0 @@
//! Repo plumbing on the local Forgejo: seeding an agent's `applied` repo from
//! `agent-configs/<name>`. The typed API-client constructor + org-name
//! constants live in the module root (`super`).
use std::path::Path;
use super::{CONFIG_ORG, forge_git_url, reader_token};
/// Seed an agent's `applied` repo, missing its `.git` entirely (first spawn,
/// or a revival after `destroy --purge`), by fetching `agent-configs/<name>`'s
/// `main` from the forge. Forge `main` advances with every operator-merged
/// config PR, so this reflects the last merged config. A config deployed
/// through hive-c0re without a forge merge never reaches it.
///
/// Best-effort: returns `false` (not an error) rather than bailing when there
/// is no [`reader_token`], `applied` already has a `.git` (nothing to do), the
/// forge has no `main` for the agent yet (the normal case right after
/// `swarm-controller` creates it, before it has pushed anything), or any git
/// step fails. Callers fall back to the template or the original hard error. A
/// failed seed leaves the directory empty, so a later "is it seeded" check reads
/// the same as before the attempt. Only `true` once `applied` is seeded and
/// tagged `deployed/0`.
pub async fn fetch_config_main_into_applied(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 seeded = async {
if let Err(e) = std::fs::create_dir_all(&applied) {
tracing::warn!(%name, error = ?e, "forge: applied reseed mkdir failed");
return false;
}
if let Err(e) = crate::lifecycle::git(&applied, &["init", "--initial-branch=main"]).await {
tracing::warn!(%name, error = ?e, "forge: applied reseed init failed");
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",
"--update-head-ok",
&url,
"refs/heads/main:refs/heads/main",
])
.output()
.await;
match out {
Ok(o) if o.status.success() => {}
Ok(o) => {
// `info`, not `warn`: the normal case right after
// `swarm-controller` creates the agent, before it has pushed
// anything into the repo it just made, and nothing is wrong.
tracing::info!(
%name,
stderr = %String::from_utf8_lossy(&o.stderr).trim(),
"forge: no config to fetch for this agent yet"
);
return false;
}
Err(e) => {
tracing::warn!(%name, error = ?e, "forge: applied reseed fetch 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 reseed read-tree failed");
return false;
}
if let Err(e) = crate::lifecycle::git_tag(&applied, "deployed/0", "refs/heads/main").await {
tracing::warn!(%name, error = ?e, "forge: applied reseed tag failed");
return false;
}
tracing::info!(%name, "forge: seeded applied repo from agent-configs main");
true
}
.await;
if !seeded {
clear_dir(&applied);
}
seeded
}
/// Empty `dir`, keeping the directory itself: a failed seed may have written
/// the working tree as well as the `.git`, and the template fallback must start
/// from nothing rather than adopt files that were meant for the forge's commit.
fn clear_dir(dir: &Path) {
let Ok(entries) = std::fs::read_dir(dir) else {
return;
};
for entry in entries.flatten() {
let path = entry.path();
let _ = if path.is_dir() {
std::fs::remove_dir_all(path)
} else {
std::fs::remove_file(path)
};
}
}
#[cfg(test)]
mod tests {
use super::clear_dir;
#[test]
fn clear_dir_empties_a_dir_but_keeps_it() {
let dir = tempfile::tempdir().expect("tempdir");
std::fs::create_dir(dir.path().join(".git")).expect("mkdir .git");
std::fs::write(dir.path().join(".git/HEAD"), "ref").expect("write HEAD");
std::fs::write(dir.path().join("agent.nix"), "{}").expect("write agent.nix");
clear_dir(dir.path());
assert!(dir.path().is_dir(), "the directory itself stays");
assert_eq!(std::fs::read_dir(dir.path()).expect("read_dir").count(), 0);
}
}

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

@ -68,7 +68,7 @@ pub(super) trait Host: Sync {
async fn notify(&self, key: String, summary: String);
/// Re-read the agent's archetype from the swarm.
async fn refresh_archetype(&self, name: &str) -> Result<()>;
/// Rebuild preamble: port check, applied repo and state dirs.
/// Rebuild preamble: port check and state dirs.
async fn prepare_rebuild_dirs(&self, name: &str, paths: &AgentPaths) -> Result<()>;
/// Rewrite the host-side files that follow the agent set.
async fn refresh_agent_set(&self) -> Result<()>;
@ -216,13 +216,11 @@ impl Host for RealHost {
fn purge_dirs(&self, name: &str) -> Vec<PathBuf> {
// A malformed name can't have a persistent state tree (the state dir is
// only ever created under a validated Ident), so only the applied dir
// is left to clear.
// only ever created under a validated Ident), so there is nothing to clear.
hive_types::Ident::parse(name)
.ok()
.map(|id| crate::paths::agent_state_dir(&id))
.into_iter()
.chain([crate::paths::applied_dir(name)])
.collect()
}
@ -426,7 +424,7 @@ async fn run_purge_state(host: &impl Host, agent: &str) {
// The state root may be a btrfs subvolume: a subvolume root can't be
// removed with rmdir/`remove_dir_all`, so delete it via hive-priv (root)
// first. No-op for plain-dir agents — the loop below then handles the
// plain-dir state root plus the applied dir.
// plain-dir state root.
if let Err(e) = host.delete_state_subvolume(agent).await {
tracing::warn!(error = ?e, %agent, "purge: delete state subvolume failed");
}
@ -731,8 +729,8 @@ async fn run_rebuild_bookkeeping(
Ok(())
}
/// First-spawn pre-create provisioning: the applied repo, state
/// subvolume, and the agent-set refresh.
/// First-spawn pre-create provisioning: the state subvolume and the
/// agent-set refresh.
async fn run_provision(host: &impl Host, name: &str) -> Result<()> {
let agent_dir = crate::paths::agent_runtime_dir(name);
let paths = Coordinator::agent_paths(name, agent_dir)?;

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

@ -51,7 +51,7 @@ pub enum NodeKind {
DestroyContainer { agent: String },
/// The `purge = true` half of a destroy: delete the agent's state
/// subvolume (via hive-priv, since a subvolume root defeats
/// `remove_dir_all`) plus its state and applied dirs.
/// `remove_dir_all`) plus its state dir.
PurgeState { agent: String },
/// The post-destroy bookkeeping tail.
DestroyBookkeeping { agent: String },

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

@ -282,7 +282,7 @@ pub fn rebuild(builder: &JobBuilder, agent: &str) -> Vec<hive_jobq::NodeGuid> {
}
/// First deploy of an agent this hive has never seen, asked for by the swarm:
/// `Provision` (applied repo, state subvolume, agent-set registration)
/// `Provision` (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`

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

@ -1,92 +1,8 @@
//! Git shellout helpers for the per-agent applied repo: run
//! `git` with the hive-c0re identity, resolve/plant refs and tags, and
//! fetch proposal commits into the applied repo.
//! Spawning `git` from hive-c0re: the binary lookup and the credential
//! handoff.
use std::path::Path;
use std::process::{Output, Stdio};
use std::time::Duration;
use anyhow::{Context, Result, bail};
use tokio::io::{AsyncRead, AsyncReadExt};
use tokio::process::Command;
/// Deadline for a child that talks to a remote (forge fetch/push): a healthy
/// one takes seconds, so ten minutes only ever fires on a hang.
pub const FETCH_DEADLINE: Duration = Duration::from_mins(10);
/// Deadline for a git child that touches only the local repo.
pub const GIT_LOCAL_DEADLINE: Duration = Duration::from_mins(2);
/// Run `cmd` to completion and capture its output, failing once `deadline`
/// passes. `what` names the command in the error. The child leads its own
/// process group and the whole group is `SIGKILL`ed on timeout, so a
/// grandchild (a `git` or credential helper it spawned) dies with it instead
/// of outliving the lock.
pub async fn run_with_deadline(
cmd: &mut Command,
deadline: Duration,
what: &str,
) -> Result<Output> {
// `kill_on_drop` covers the caller dropping this future; `process_group(0)`
// makes the child's pid the group id the timeout path kills.
let mut child = cmd
.stdin(Stdio::null())
.stdout(Stdio::piped())
.stderr(Stdio::piped())
.kill_on_drop(true)
.process_group(0)
.spawn()
.with_context(|| what.to_owned())?;
// Read before waiting: a reaped child has no id.
let pgid = child.id().and_then(|pid| i32::try_from(pid).ok());
let (stdout, stderr) = (child.stdout.take(), child.stderr.take());
let io = async { tokio::join!(child.wait(), drain(stdout), drain(stderr)) };
if let Ok((status, stdout, stderr)) = tokio::time::timeout(deadline, io).await {
return Ok(Output {
status: status.with_context(|| what.to_owned())?,
stdout: stdout.with_context(|| what.to_owned())?,
stderr: stderr.with_context(|| what.to_owned())?,
});
}
if let Some(pgid) = pgid {
// SAFETY: plain `kill(2)`; `-pgid` targets the child's own process
// group, which `process_group(0)` made it lead.
unsafe {
libc::kill(-pgid, libc::SIGKILL);
}
}
child.wait().await.with_context(|| what.to_owned())?;
bail!("{what} timed out after {}s", deadline.as_secs_f64())
}
/// Everything `pipe` yields until EOF.
async fn drain<R: AsyncRead + Unpin>(pipe: Option<R>) -> std::io::Result<Vec<u8>> {
let mut buf = Vec::new();
if let Some(mut p) = pipe {
p.read_to_end(&mut buf).await?;
}
Ok(buf)
}
const GIT_NAME: &str = "c0re";
const GIT_EMAIL: &str = "c0re@hyperhive.local";
pub(super) async fn git_commit(dir: &Path, message: &str) -> Result<()> {
git(
dir,
&[
"-c",
&format!("user.name={GIT_NAME}"),
"-c",
&format!("user.email={GIT_EMAIL}"),
"commit",
"-m",
message,
],
)
.await
}
/// Spawn `git` honoring the `HYPERHIVE_GIT` env var (absolute path baked in
/// by the NixOS module), falling back to bare `git` (PATH lookup) otherwise.
///
@ -121,125 +37,3 @@ pub fn git_command_authed(auth_header: &str) -> Command {
.env("GIT_CONFIG_VALUE_0", auth_header);
cmd
}
/// Run `cmd` as `git <args>` in `dir` within `deadline`, erroring on a
/// non-zero exit.
async fn run(mut cmd: Command, dir: &Path, args: &[&str], deadline: Duration) -> Result<()> {
let what = format!("git {} in {}", args.join(" "), dir.display());
let out = run_with_deadline(cmd.current_dir(dir).args(args), deadline, &what).await?;
if !out.status.success() {
bail!(
"git {} failed ({}): {}",
args.join(" "),
out.status,
String::from_utf8_lossy(&out.stderr).trim()
);
}
Ok(())
}
pub async fn git(dir: &Path, args: &[&str]) -> Result<()> {
run(git_command(), dir, args, GIT_LOCAL_DEADLINE).await
}
/// [`git`] against an authenticated remote — for the paths that talk to the
/// forge. The credential rides the environment, so `args` (and therefore the
/// error above) stay free of it.
pub async fn git_authed(dir: &Path, args: &[&str], auth_header: &str) -> Result<()> {
run(git_command_authed(auth_header), dir, args, FETCH_DEADLINE).await
}
/// Resolve `refname` (a tag, branch, or sha) in `dir` to its full sha.
pub async fn git_rev_parse(dir: &Path, refname: &str) -> Result<String> {
let out = run_with_deadline(
git_command().current_dir(dir).args(["rev-parse", refname]),
GIT_LOCAL_DEADLINE,
&format!("git rev-parse {refname} in {}", dir.display()),
)
.await?;
if !out.status.success() {
bail!(
"git rev-parse {refname} failed ({}): {}",
out.status,
String::from_utf8_lossy(&out.stderr).trim()
);
}
Ok(String::from_utf8_lossy(&out.stdout).trim().to_owned())
}
/// Plant a lightweight tag at `target`. Errors if the tag already
/// exists — we want loud failures on id reuse, not silent
/// overwrites.
pub async fn git_tag(dir: &Path, name: &str, target: &str) -> Result<()> {
git(dir, &["tag", name, target]).await
}
/// Replace working tree + index with the tree at `target` without
/// moving HEAD. `applied/main` stays pointing at the last known-good
/// `deployed/*` while we let `nixos-container update` evaluate the
/// candidate. On build failure callers reset back to HEAD; on
/// success they fast-forward main to `target`.
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` 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
}
/// Compare-and-swap a ref: move `refname` to `new` only if it currently points
/// at `old`. `git update-ref <ref> <new> <old>` refuses — leaving the ref
/// untouched — when the current value differs. That refusal is the whole
/// difference between "advance this branch" and "overwrite whatever is there".
///
/// 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.
///
/// # Errors
///
/// Returns an error if the ref does not currently point at `old` (the CAS lost)
/// or if the `git` invocation itself fails.
pub async fn git_update_ref_cas(dir: &Path, refname: &str, new: &str, old: &str) -> Result<()> {
git(dir, &["update-ref", refname, new, old]).await
}
/// True when `ancestor` is reachable from `descendant` — i.e. moving a branch
/// from `ancestor` to `descendant` is a genuine fast-forward that discards
/// nothing.
///
/// Not the same question as "did this branch land upstream": a squash-merge
/// rewrites the commit, so `--is-ancestor` correctly answers `false` for a
/// branch whose *contents* were merged. This helper compares two commits with
/// real shared ancestry inside one repo, which is precisely what it decides.
///
/// # Errors
///
/// Returns an error if `git` fails to run or either revision can't be resolved.
/// A clean "no, not an ancestor" is `Ok(false)`, not an error.
pub async fn git_is_ancestor(dir: &Path, ancestor: &str, descendant: &str) -> Result<bool> {
let out = git_command()
.current_dir(dir)
.args(["merge-base", "--is-ancestor", ancestor, descendant])
.output()
.await
.with_context(|| {
format!(
"git merge-base --is-ancestor {ancestor} {descendant} in {}",
dir.display()
)
})?;
match out.status.code() {
Some(0) => Ok(true),
Some(1) => Ok(false),
_ => bail!(
"git merge-base --is-ancestor {ancestor} {descendant} failed ({}): {}",
out.status,
String::from_utf8_lossy(&out.stderr).trim()
),
}
}

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

@ -7,19 +7,9 @@ mod setup;
#[cfg(test)]
mod tests;
#[cfg(test)]
pub use git::git_update_ref;
#[cfg(test)]
pub use git::run_with_deadline;
pub use git::{
git, git_authed, git_command, git_command_authed, git_is_ancestor, git_read_tree_reset,
git_rev_parse, git_tag, git_update_ref_cas,
};
pub use git::{git_command, git_command_authed};
pub use host_config::write_dropins;
pub use setup::{
ensure_agent_state_subvolume, ensure_claude_dir, ensure_state_dir, initial_flake_nix,
setup_applied,
};
pub use setup::{ensure_agent_state_subvolume, ensure_claude_dir, ensure_state_dir};
use anyhow::{Context, Result, bail};
use tokio::process::Command;
@ -205,8 +195,8 @@ 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 applied
/// repo + state subvolume + claude/notes dirs, then register the new
/// node. Fail fast on a port collision, set up the state
/// subvolume + claude/notes dirs, then register the new
/// 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.
@ -218,12 +208,9 @@ pub async fn provision_container(name: &str, paths: &AgentPaths) -> Result<()> {
agent_web_port(name)
);
}
// Read before anything is seeded: an archetype agent gets no config
// repo, and the build takes it from the archetype.
// Read before the build: an archetype agent gets no config repo, and
// the build takes it from the archetype.
crate::archetypes::refresh(name).await?;
if crate::archetypes::of(name)?.is_none() {
setup_applied(&paths.applied, name).await?;
}
ensure_agent_state_subvolume(name).await?;
ensure_claude_dir(&paths.claude)?;
ensure_state_dir(&paths.notes)?;
@ -239,19 +226,7 @@ pub async fn create_only(name: &str, toplevel: &str) -> Result<()> {
}
/// Rebuild-path preamble, run by the job queue's `Prebuild` node: fail fast on
/// a port collision, then make sure the applied repo + state dirs exist.
/// Container untouched.
///
/// `applied` is normally already the right shape here (this runs against a
/// previously-provisioned agent), but a `destroy --purge` removes it
/// (`job_queue::exec::run_purge_state`) — a later revive of the *same*
/// agent name is still a rebuild, not a first-spawn, so it lands here, not
/// in `provision_container`. Recover from `agent-configs/<name>` on the
/// forge when that's the case (forge `main` is the last operator-merged
/// config — see `forge::fetch_config_main_into_applied`'s own doc comment)
/// rather than bailing outright. Deliberately **no template fallback**: a
/// template would silently revive the *wrong* config rather than recovering
/// the right one, which is worse than the clear bail this replaces.
/// a port collision, then make sure the state dirs exist. Container untouched.
pub async fn prepare_rebuild_dirs(name: &str, paths: &AgentPaths) -> Result<()> {
validate(name)?;
if let Some(other) = port_collision(name).await? {
@ -260,20 +235,6 @@ pub async fn prepare_rebuild_dirs(name: &str, paths: &AgentPaths) -> Result<()>
agent_web_port(name)
);
}
// An archetype agent has no applied repo to check or recover.
if crate::archetypes::of(name)?.is_none() {
if !paths.applied.join(".git").exists()
&& !crate::forge::fetch_config_main_into_applied(name).await
{
bail!(
"applied repo at {} is missing its .git directory and could not be reseeded from \
the forge (agent-configs/{name} unreachable, has no `main` — never merged or \
never seeded — or absent); destroy --purge and re-spawn this agent.",
paths.applied.display()
);
}
setup_applied(&paths.applied, name).await?;
}
ensure_agent_state_subvolume(name).await?;
ensure_claude_dir(&paths.claude)?;
ensure_state_dir(&paths.notes)?;

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

@ -1,90 +1,9 @@
//! First-spawn provisioning: seed the hive-c0re-owned applied repo, and
//! ensure the per-agent state / claude-credentials dirs (btrfs subvolume when
//! available) exist.
//!
//! Seeding is the only write the repo gets from here. Later config changes
//! arrive as forge PRs authored in a clone - nothing edits a working tree in
//! place.
//! First-spawn provisioning: ensure the per-agent state / claude-credentials
//! dirs (btrfs subvolume when available) exist.
use std::path::Path;
use anyhow::{Context, Result, bail};
use super::git::{git, git_commit, git_rev_parse, git_tag};
/// Seed an agent's applied repo on first spawn, preferring the config the
/// forge already has. `agent-configs/<name>` is authored by whoever created
/// the agent (at swarm level the controller, which writes the config before
/// any hive is told to deploy it), so this takes the forge's `main` when
/// there is one and writes a template only when there is not. Two authors for
/// one file is the failure mode being avoided: whichever wrote second would
/// win, and neither knows about the other.
///
/// The template ([`seed_template`]) is the fallback when there is genuinely
/// nothing to take yet: `swarm-controller` creates `agent-configs/<name>`
/// before telling a hive to deploy, but hasn't necessarily pushed `main` to
/// it, or the forge is unreachable, or this hive has no forge read token.
///
/// Seeding is the only write hive-c0re makes here. Later config changes
/// arrive as forge PRs and are fetched from the forge's `main` (see
/// `actions::advance_applied_to_rev`). `flake.nix` is tracked in the seed
/// and rides along on every fetch.
///
/// A repo that is already there is only checked for its shape (a rebuild
/// finds it already seeded): a missing `deployed/0` tag means it predates the
/// meta-flake layout and is an error.
pub async fn setup_applied(applied_dir: &Path, name: &str) -> Result<()> {
std::fs::create_dir_all(applied_dir)
.with_context(|| format!("create {}", applied_dir.display()))?;
if !applied_dir.join(".git").exists() {
if crate::forge::fetch_config_main_into_applied(name).await {
return Ok(());
}
seed_template(applied_dir, name).await?;
git_tag(applied_dir, "deployed/0", "refs/heads/main").await?;
} else if git_rev_parse(applied_dir, "refs/tags/deployed/0")
.await
.is_err()
{
// Pre-overhaul applied repo - no deployed/* tag scheme,
// flake.nix may be untracked, agent.nix possibly authored by
// hive-c0re directly. The startup auto-migration fixes this
// in place; if it didn't run (or got skipped), surface a
// clear error.
bail!(
"applied repo at {} predates the meta-flake layout. \
Restart hive-c0re to let the auto-migration run, or \
destroy --purge {name} and re-spawn.",
applied_dir.display()
);
}
Ok(())
}
/// Author the initial config in place: `agent.nix` (the agent's own module)
/// and `flake.nix` (the boilerplate that lets the meta flake import this repo
/// as an input: meta locks at a specific sha and reads
/// `nixosModules.default`, so `flake.nix` must be in the commit). `flake.nix`
/// isn't meant to be edited, but it's tracked so it can be read.
///
/// The fallback half of [`setup_applied`]: reached only when the forge has
/// no config for this agent to take.
async fn seed_template(dir: &Path, name: &str) -> Result<()> {
let agent_path = dir.join("agent.nix");
if !agent_path.exists() {
std::fs::write(&agent_path, initial_agent_nix(name))
.with_context(|| format!("write {}", agent_path.display()))?;
}
let flake_path = dir.join("flake.nix");
if !flake_path.exists() {
std::fs::write(&flake_path, initial_flake_nix())
.with_context(|| format!("write {}", flake_path.display()))?;
}
git(dir, &["init", "--initial-branch=main"]).await?;
git(dir, &["add", "agent.nix", "flake.nix"]).await?;
git_commit(dir, "hive-c0re init").await
}
use anyhow::{Context, Result};
/// Create the per-agent Claude credentials dir if missing. Mode 0755 — hive-core
/// needs read+execute to list the directory so `claude_has_session` can detect a
@ -170,27 +89,3 @@ pub async fn ensure_agent_state_subvolume(name: &str) -> Result<()> {
.await
.with_context(|| format!("ensure btrfs subvolume for agent {name}"))
}
fn initial_agent_nix(name: &str) -> String {
format!(
"{{ config, pkgs, lib, ... }}:\n{{\n # Per-agent overrides for {name}. This is a regular NixOS module\n # — add packages, services, modules, imports as needed.\n #\n # imports = [ ./extra-module.nix ];\n # environment.systemPackages = with pkgs; [ ];\n}}\n",
)
}
/// Module-only flake exposed by every agent's repo. Consumed by the
/// 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
/// own `flake.lock` and rolls up into meta's lock transitively.
/// - The outputs block forwards every input (minus `self`) into
/// `agent.nix` as the `flakeInputs` module argument, so the
/// manager just references `flakeInputs.<name>.packages.${pkgs.system}.default`
/// without further plumbing.
///
/// Identity injection (`HIVE_PORT` / `HIVE_LABEL` / dashboard port /
/// git committer) still lives in the meta flake's wrapper.
pub fn initial_flake_nix() -> &'static str {
"{\n description = \"hyperhive agent\";\n inputs = { };\n outputs =\n { self, ... }@inputs:\n {\n nixosModules.default = {\n imports = [ ./agent.nix ];\n _module.args.flakeInputs = builtins.removeAttrs inputs [ \"self\" ];\n };\n };\n}\n"
}

212
hive-c0re/src/lifecycle/tests.rs
View file

@ -3,48 +3,6 @@
use super::*;
/// Regression test: `setup_applied` must seed both agent.nix and flake.nix
/// in the initial commit, and tag it `deployed/0`. Before commit 5b5a93e
/// flake.nix was missing from the scaffold, requiring manual creation (seen
/// with the damocles agent).
///
/// Exercises the template arm: there is no forge to take a config from in a
/// test (no reader token), so `setup_applied` falls through to
/// `seed_template` — which is the arm this asserts about.
#[tokio::test]
async fn setup_applied_seeds_flake_nix() {
let dir = tempfile::tempdir().expect("tempdir");
let applied = dir.path().join("applied");
setup_applied(&applied, "test-agent")
.await
.expect("setup_applied");
git_rev_parse(&applied, "refs/tags/deployed/0")
.await
.expect("the seed commit is tagged deployed/0");
// Both files must exist on disk.
assert!(applied.join("agent.nix").exists(), "agent.nix missing");
assert!(applied.join("flake.nix").exists(), "flake.nix missing");
// flake.nix must export nixosModules.default (the meta-flake contract).
let flake = std::fs::read_to_string(applied.join("flake.nix")).unwrap();
assert!(
flake.contains("nixosModules.default"),
"flake.nix does not export nixosModules.default"
);
// Both files must be tracked in the initial git commit.
let out = git_command()
.current_dir(&applied)
.args(["show", "--name-only", "--format=", "HEAD"])
.output()
.await
.expect("git show");
let tracked = String::from_utf8_lossy(&out.stdout);
assert!(tracked.contains("agent.nix"), "agent.nix not committed");
assert!(tracked.contains("flake.nix"), "flake.nix not committed");
}
#[test]
fn bridge_gateway_ip_extracts_verbatim_address() {
// HIVE_NETWORK_SUBNET carries the bridge IP verbatim, not the
@ -132,122 +90,6 @@ fn network_isolation_errors_name_the_offending_variable() {
);
}
/// `setup_applied` is idempotent: calling it on an already seeded repo is a
/// no-op (the shape check passes and nothing is written).
#[tokio::test]
async fn setup_applied_idempotent() {
let dir = tempfile::tempdir().expect("tempdir");
let applied = dir.path().join("applied");
setup_applied(&applied, "test-agent")
.await
.expect("first call");
// Second call must not error even though .git already exists.
setup_applied(&applied, "test-agent")
.await
.expect("second call");
// Still one commit.
let out = git_command()
.current_dir(&applied)
.args(["rev-list", "--count", "HEAD"])
.output()
.await
.expect("git rev-list");
let count = String::from_utf8_lossy(&out.stdout).trim().to_owned();
assert_eq!(
count, "1",
"expected exactly one commit after idempotent call"
);
}
/// Build a two-commit repo and return `(dir, repo_path, first, second)`.
async fn two_commit_repo() -> (tempfile::TempDir, std::path::PathBuf, String, String) {
let dir = tempfile::tempdir().expect("tempdir");
let repo = dir.path().join("applied");
setup_applied(&repo, "test-agent")
.await
.expect("setup_applied");
let first = git_rev_parse(&repo, "HEAD").await.expect("rev-parse first");
std::fs::write(repo.join("second.txt"), "second").expect("write second.txt");
git(&repo, &["add", "second.txt"]).await.expect("git add");
git(
&repo,
&[
"-c",
"user.name=test",
"-c",
"user.email=test@example.com",
"commit",
"-m",
"second",
],
)
.await
.expect("git commit");
let second = git_rev_parse(&repo, "HEAD")
.await
.expect("rev-parse second");
assert_ne!(first, second, "second commit did not advance HEAD");
(dir, repo, first, second)
}
/// `git_is_ancestor` answers reachability in both directions. This is the
/// check that decides whether moving a branch is a real fast-forward or a
/// rewind that discards commits.
#[tokio::test]
async fn is_ancestor_distinguishes_direction() {
let (_dir, repo, first, second) = two_commit_repo().await;
assert!(
git_is_ancestor(&repo, &first, &second)
.await
.expect("ancestor check forward"),
"first must be an ancestor of second"
);
assert!(
!git_is_ancestor(&repo, &second, &first)
.await
.expect("ancestor check reverse"),
"second must NOT be an ancestor of first"
);
}
/// Regression test for the deploy path that ate a committed agent config:
/// `git_update_ref_cas` must refuse to move a ref whose current value is not
/// the expected one, and must leave the ref untouched when it refuses. A bare
/// `update-ref` accepts that move and silently drops everything in between.
#[tokio::test]
async fn update_ref_cas_refuses_stale_expectation() {
let (_dir, repo, first, second) = two_commit_repo().await;
// Park a ref at `first`, then CAS it forward with the correct old value.
git_update_ref(&repo, "refs/heads/cas", &first)
.await
.expect("plant ref");
git_update_ref_cas(&repo, "refs/heads/cas", &second, &first)
.await
.expect("CAS with the correct old value must succeed");
assert_eq!(
git_rev_parse(&repo, "refs/heads/cas").await.unwrap(),
second
);
// Now retry with the SAME (now stale) expectation, as a racing deploy
// holding a pre-merge sha would: it must fail rather than rewind.
assert!(
git_update_ref_cas(&repo, "refs/heads/cas", &first, &first)
.await
.is_err(),
"CAS accepted a stale old value"
);
assert_eq!(
git_rev_parse(&repo, "refs/heads/cas").await.unwrap(),
second,
"ref moved even though the CAS failed"
);
}
/// `systemctl is-active` prints the state with a trailing newline, which is
/// the whole reason this parser trims rather than comparing raw.
#[test]
@ -416,60 +258,6 @@ async fn a_systemctl_answer_inside_the_bound_passes_through() {
assert_eq!(got, Some(7));
}
/// A child that outlives the deadline yields an error naming the command and
/// the deadline, and the child and its grandchild are both killed rather than
/// orphaned.
#[tokio::test]
async fn run_with_deadline_times_out_and_kills_the_process_group() {
let dir = tempfile::tempdir().expect("tempdir");
let (shpid, sleeppid) = (dir.path().join("sh.pid"), dir.path().join("sleep.pid"));
let mut cmd = tokio::process::Command::new("sh");
cmd.arg("-c").arg(format!(
"echo $$ > {}; sleep 300 & echo $! > {}; wait",
shpid.display(),
sleeppid.display()
));
let err = run_with_deadline(
&mut cmd,
std::time::Duration::from_millis(500),
"fake fetch",
)
.await
.expect_err("a sleeping child must hit the deadline");
let msg = format!("{err:#}");
assert!(msg.contains("fake fetch timed out after 0.5s"), "{msg}");
for file in [&shpid, &sleeppid] {
// The pid file proves the process existed before the kill.
let pid = std::fs::read_to_string(file)
.expect("process wrote its pid before the deadline")
.trim()
.to_owned();
let proc = std::path::PathBuf::from("/proc").join(&pid);
let mut gone = false;
for _ in 0..100 {
if !proc.exists() {
gone = true;
break;
}
tokio::time::sleep(std::time::Duration::from_millis(50)).await;
}
assert!(gone, "pid {pid} ({}) still running", file.display());
}
}
/// Presence control for the timeout test: a child that finishes in time
/// returns its output.
#[tokio::test]
async fn run_with_deadline_returns_output_of_a_fast_child() {
let mut cmd = tokio::process::Command::new("echo");
cmd.arg("hi");
let out = run_with_deadline(&mut cmd, std::time::Duration::from_secs(30), "echo")
.await
.expect("echo finishes");
assert_eq!(String::from_utf8_lossy(&out.stdout).trim(), "hi");
}
/// The per-host values reach the container as facts: the dashboard port and
/// pinned claude from the hive's config, the OTEL labels and the queue
/// address from the daemon's environment.

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

@ -1,5 +1,5 @@
//! Startup convergence. Three phases, all idempotent and unguarded:
//! harness files, the applied repos, the agent-set files, plus the
//! harness files, the leftover applied repos, the agent-set files, plus the
//! removal of the legacy meta repo's leftovers. They re-run every
//! boot on purpose — each one is a no-op once its state is already
//! correct.
@ -23,29 +23,21 @@ use std::path::Path;
use std::time::Duration;
use anyhow::{Context, Result};
use tokio::process::Command;
use crate::coordinator::Coordinator;
use crate::lifecycle::{self, AGENT_PREFIX};
const KILL_SWITCH: &str = "HIVE_SKIP_META_MIGRATION";
/// Per-shellout timeout for the blocking startup convergence. `run` is
/// Timeout for the shellout in the blocking startup convergence. `run` is
/// awaited *before* the daemon starts serving (main.rs), so any child
/// process that wedges here freezes the whole daemon — admin socket +
/// dashboard included — with no diagnostics: a git shellout was observed
/// blocked for 86min under a concurrent `nixos-rebuild`. Every shellout
/// runs under a timeout that kills the child on elapse, so a stuck phase
/// degrades to a logged warning instead of a hung boot.
/// blocked for 86min under a concurrent `nixos-rebuild`. The shellout
/// runs under a timeout, so a stuck phase degrades to a logged warning
/// instead of a hung boot.
const GIT_TIMEOUT: Duration = Duration::from_mins(2);
/// Substring that identifies the *current* agent flake boilerplate.
/// Bumped whenever the template changes so the startup migration
/// re-renders existing agents onto the new shape. Today the marker
/// is the `flakeInputs` module-arg forwarding line — older templates
/// (raw `import ./agent.nix`) get rewritten on next hive-c0re start.
const MODULE_FLAKE_MARKER: &str = "_module.args.flakeInputs";
pub async fn run() -> Result<()> {
if std::env::var(KILL_SWITCH).is_ok() {
tracing::info!("migration: {KILL_SWITCH} set — skipping");
@ -80,26 +72,16 @@ pub async fn run() -> Result<()> {
Err(e) => tracing::warn!(error = ?e, "migration: removing the legacy meta .git failed"),
}
// Phase 1 + 2: per-agent applied repo.
tracing::debug!("migration: phase 1+2 (applied repo)");
// An archetype agent has no applied repo, and seeding one here would hand
// it a config it is not built from.
let archetypes = crate::archetypes::read_at(&crate::archetypes::path()).unwrap_or_else(|e| {
tracing::warn!(error = ?e, "migration: stored archetypes unreadable; treating every agent as custom");
crate::archetypes::Stored::new()
});
for name in &names {
if archetypes.contains_key(name.as_str()) {
continue;
}
tracing::debug!(%name, "migration: applied");
if let Err(e) = migrate_applied_repo(name.as_str()).await {
tracing::warn!(%name, error = ?e, "migration: applied repo rewrite failed");
}
// Phase 1: the applied repos an earlier hive kept per agent.
tracing::debug!("migration: phase 1 (leftover applied repos)");
match remove_applied_repos(&crate::paths::applied_root()) {
Ok(0) => {}
Ok(n) => tracing::info!(count = n, "migration: removed leftover applied repos"),
Err(e) => tracing::warn!(error = ?e, "migration: removing the applied repos failed"),
}
// Phase 3: the files that follow the agent set.
tracing::debug!("migration: phase 3 (agent_set refresh)");
// Phase 2: the files that follow the agent set.
tracing::debug!("migration: phase 2 (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 {
@ -185,89 +167,27 @@ async fn enumerate_agents() -> Result<Vec<hive_types::Ident>> {
.collect())
}
async fn migrate_applied_repo(name: &str) -> Result<()> {
migrate_applied_dir(name, &crate::paths::applied_dir(name)).await
}
/// Rewrites the applied repo at `dir` to the module-only flake. A missing
/// `flake.nix` counts as "marker absent"; any other read error fails the
/// migration so the agent's file is never overwritten unread.
async fn migrate_applied_dir(name: &str, dir: &Path) -> Result<()> {
if !dir.join(".git").exists() {
return Ok(());
}
let flake_path = dir.join("flake.nix");
let cur = match std::fs::read_to_string(&flake_path) {
Ok(cur) => cur,
Err(e) if e.kind() == std::io::ErrorKind::NotFound => String::new(),
Err(e) => return Err(e).with_context(|| format!("read {}", flake_path.display())),
/// Remove the per-agent `applied/<name>` repos an earlier hive kept in `root`,
/// destroyed agents' included. The forge's `agent-configs/<name>` is the only
/// copy of a config now, so nothing reads them. The dot-files beside them (the
/// deploy markers) stay. Returns how many it removed.
fn remove_applied_repos(root: &Path) -> Result<usize> {
let entries = match std::fs::read_dir(root) {
Ok(entries) => entries,
Err(e) if e.kind() == std::io::ErrorKind::NotFound => return Ok(0),
Err(e) => return Err(e).with_context(|| format!("reading {}", root.display())),
};
if cur.contains(MODULE_FLAKE_MARKER) {
return Ok(());
let mut removed = 0;
for entry in entries {
let entry = entry.with_context(|| format!("reading {}", root.display()))?;
if !entry.file_type()?.is_dir() {
continue;
}
let path = entry.path();
std::fs::remove_dir_all(&path).with_context(|| format!("removing {}", path.display()))?;
removed += 1;
}
let want = lifecycle::initial_flake_nix();
std::fs::write(&flake_path, want).with_context(|| format!("write {}", flake_path.display()))?;
raw_git(
dir,
&[
"-c",
"user.name=c0re",
"-c",
"user.email=c0re@hyperhive.local",
"add",
"flake.nix",
],
)
.await?;
raw_git(
dir,
&[
"-c",
"user.name=c0re",
"-c",
"user.email=c0re@hyperhive.local",
"commit",
"-m",
"migration: module-only flake",
],
)
.await?;
// Relocate deployed/0 to the migration commit so
// setup_applied's existence check passes.
raw_git(dir, &["tag", "-f", "deployed/0", "HEAD"]).await?;
tracing::info!(%name, "migration: applied repo migrated to module-only flake");
Ok(())
}
/// Run a command to completion under a timeout, capturing its output. On
/// timeout the child is killed (`kill_on_drop`) and an error is returned,
/// so a wedged shellout can never freeze startup migration. `what` is a
/// human label surfaced in the timeout error + `with_context`.
async fn output_with_timeout(
mut cmd: Command,
timeout: Duration,
what: &str,
) -> Result<std::process::Output> {
cmd.kill_on_drop(true);
match tokio::time::timeout(timeout, cmd.output()).await {
Ok(r) => r.with_context(|| format!("run {what}")),
Err(_) => anyhow::bail!("{what} timed out after {timeout:?} (child killed)"),
}
}
async fn raw_git(dir: &Path, args: &[&str]) -> Result<()> {
let mut cmd = lifecycle::git_command();
cmd.current_dir(dir).args(args);
let label = format!("git {} in {}", args.join(" "), dir.display());
let out = output_with_timeout(cmd, GIT_TIMEOUT, &label).await?;
if !out.status.success() {
anyhow::bail!(
"git {} failed: {}",
args.join(" "),
String::from_utf8_lossy(&out.stderr).trim()
);
}
Ok(())
Ok(removed)
}
#[cfg(test)]
@ -312,17 +232,29 @@ mod tests {
assert!(!remove_core_token(dir.path()).expect("second"), "no-op");
}
#[tokio::test]
async fn unreadable_flake_is_left_untouched() {
let dir = tempfile::tempdir().expect("tempdir");
std::fs::create_dir(dir.path().join(".git")).expect("mkdir .git");
let flake = dir.path().join("flake.nix");
let bytes = [0xff_u8, 0xfe, b'x'];
std::fs::write(&flake, bytes).expect("write flake");
#[test]
fn applied_repos_go_and_the_markers_beside_them_stay() {
let root = tempfile::tempdir().expect("tempdir");
for name in ["alice", "bob"] {
std::fs::create_dir_all(root.path().join(name).join(".git")).expect("mkdir repo");
std::fs::write(root.path().join(name).join("agent.nix"), "{}").expect("write config");
std::fs::write(root.path().join(format!(".{name}.config-rev")), "abc").expect("marker");
}
let res = migrate_applied_dir("t", dir.path()).await;
assert!(res.is_err(), "non-UTF-8 flake.nix must fail the migration");
assert_eq!(std::fs::read(&flake).expect("read back"), bytes);
assert_eq!(remove_applied_repos(root.path()).expect("first"), 2);
assert!(!root.path().join("alice").exists());
assert!(
root.path().join(".alice.config-rev").exists(),
"marker stays"
);
assert_eq!(
remove_applied_repos(root.path()).expect("second"),
0,
"no-op"
);
assert_eq!(
remove_applied_repos(&root.path().join("missing")).expect("absent root"),
0
);
}
}

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

@ -114,19 +114,14 @@ pub fn agents_root() -> PathBuf {
PathBuf::from(AGENTS_ROOT)
}
/// `applied/` — per-agent *applied* (deployed) config repos + rev markers.
/// `applied/` — per-agent deploy rev markers (and, on a hive that predates
/// their removal, leftover config repos the startup migration clears).
// nix: read by hive-c0re only, but paired with `agents/` in the deploy flow.
#[must_use]
pub fn applied_root() -> PathBuf {
state_root().join("applied")
}
/// `applied/<name>` — one agent's applied config working tree.
#[must_use]
pub fn applied_dir(name: &str) -> PathBuf {
applied_root().join(name)
}
/// `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).

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

@ -203,12 +203,6 @@ async fn dispatch(req: &HostRequest, coord: Arc<Coordinator>) -> HostResponse {
// federation peer-config block) and open the web surfaces
// (`hivectl open`).
HostRequest::Urls => HostResponse::urls(hive_urls()),
HostRequest::ReconcileConfigStatus { agent, verbose } => {
crate::forge::reconcile_config_status(agent.as_str(), *verbose).await?
}
HostRequest::ReconcileConfigApply { agent, direction } => {
crate::forge::reconcile_config_apply(agent.as_str(), *direction).await?
}
HostRequest::QuotaEnable => handle_quota_enable().await?,
HostRequest::QuotaLimit { name, limit } => {
handle_quota_limit(name.as_str(), *limit).await?

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

@ -309,9 +309,9 @@ async fn handle_deploy_request(
}
};
// Only a rebuild applies `rev`. A first deploy ignores it and builds the
// `applied` repo it seeds from the forge's `main`, or the one it finds.
// An archetype agent has no `applied` repo, so its rebuild ignores it too.
// Only a rebuild looks at `rev`. A first deploy ignores it: the build takes
// whatever commit `internal/meta` pins. An archetype agent has no config
// repo, so its rebuild ignores it too.
let archetype = match crate::archetypes::of(&agent) {
Ok(archetype) => archetype,
Err(e) => {
@ -322,8 +322,9 @@ async fn handle_deploy_request(
if known
&& archetype.is_none()
&& let Some(rev) = rev.as_deref()
&& !apply_rev(&agent, rev).await
&& rev_already_deployed(&agent, rev)
{
tracing::info!(%agent, %rev, "swarm events: deploy requested for the deployed rev; nothing queued");
return;
}
@ -351,31 +352,14 @@ async fn handle_deploy_request(
}
}
/// Move `agent`'s `applied/main` to `rev` ahead of its rebuild, and say
/// whether that rebuild should be queued. A `rev` already applied is taken as
/// deployed or deploying, since every path that moves `applied/main` queues a
/// rebuild with it. The exception is `hivectl forge reconcile-config`, which
/// does not deploy, so a rev it put in place is skipped too.
///
/// The refusal's message is matched verbatim by the `hh-config-rev-refused`
/// alert rule in `nix/host-modules/swarm-grafana.nix`; change both together.
/// A failed rebuild surfaces where every rebuild failure does, on this hive.
async fn apply_rev(agent: &str, rev: &str) -> bool {
match crate::actions::advance_applied_to_rev(agent, rev).await {
Ok(crate::actions::RevAdvance::Advanced) => true,
Ok(crate::actions::RevAdvance::AlreadyApplied) => {
tracing::info!(%agent, %rev, "swarm events: deploy requested for the applied rev; nothing queued");
false
}
Ok(crate::actions::RevAdvance::Superseded) => {
tracing::info!(%agent, %rev, "swarm events: deploy requested for a rev applied/main already contains; nothing queued");
false
}
Err(e) => {
tracing::error!(%agent, %rev, error = %format!("{e:#}"), "swarm events: rev not applied; nothing queued");
false
}
}
/// Whether `rev` is the config commit `agent`'s last successful build pinned
/// (`paths::applied_config_marker`), so a rebuild would deploy nothing new. A
/// request for a rev that is still building, or whose build failed, is not
/// deployed yet and queues again: the rebuild resolves its pins from
/// `internal/meta` either way, so a second one is redundant but harmless.
fn rev_already_deployed(agent: &str, rev: &str) -> bool {
std::fs::read_to_string(crate::paths::applied_config_marker(agent))
.is_ok_and(|deployed| deployed.trim() == rev)
}
/// Queue the first deploy of an agent this hive does not have yet: seed its

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

@ -39,31 +39,21 @@ pub fn current_flake_rev(hyperhive_flake: &str) -> Option<String> {
.map(|p| p.display().to_string())
}
/// Returns true when the applied repo has commits that have not yet been
/// 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."
/// Returns true when the config commit `internal/meta` pins for the agent's
/// next build (`meta_rev::recorded`) differs from the one its last successful
/// build pinned (`deployed`, the `paths::applied_config_marker` content). This
/// is the semantic the dashboard `needs_update` chip conveys: "there is a
/// config change ready to apply via rebuild." An archetype agent pins no
/// config commit and never has one pending.
///
/// Async on purpose: this runs per agent inside `container_view::build_all`,
/// which fires on the ~10s dashboard sweep, every `AgentStatus` request, and
/// every `rescan_containers_and_emit` after a lifecycle step. A synchronous
/// `git` fork here blocks a tokio worker for the whole exec — under
/// nix-build disk saturation that's long enough that concurrent sweeps
/// starved the runtime and stalled the per-agent sockets.
pub async fn agent_config_pending(name: &str, deployed_sha: Option<&str>) -> bool {
let applied = crate::paths::applied_dir(name);
let applied_head = crate::lifecycle::git_command()
.args(["-C", &applied.to_string_lossy(), "rev-parse", "HEAD"])
.output()
.await
.ok()
.filter(|o| o.status.success())
.and_then(|o| String::from_utf8(o.stdout).ok())
.map(|s| s.trim().to_owned());
match (applied_head.as_deref(), deployed_sha) {
(Some(head), Some(sha)) => !head.starts_with(sha) && !sha.starts_with(head),
/// Reads one small file and the marker the caller already holds, no `git`:
/// this runs per agent inside `container_view::build_all`, on the ~10s
/// dashboard sweep and every `AgentStatus` request.
#[must_use]
pub fn agent_config_pending(name: &str, deployed: Option<&str>) -> bool {
let pinned = crate::meta_rev::recorded(name).and_then(|build| build.config);
match (pinned.as_deref(), deployed) {
(Some(pinned), Some(deployed)) => pinned != deployed.trim(),
_ => false,
}
}