vinzenz/hyperhive
vinzenz/hyperhive
Watch
0
0
Fork
You've already forked hyperhive
 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

refactor(#1202): introduce HiveEnv + AgentPaths to reduce arg repetition

Browse source
This commit is contained in:
damocles 2026-06-03 21:51:45 +02:00 • committed by mara
commit 29c7f64bd3
8 changed files with 131 additions and 235 deletions
Download patch file Download diff file Expand all files Collapse all files

1
Cargo.lock generated
View file

@ -1247,6 +1247,7 @@ version = "0.1.0"
dependencies = [
"anyhow",
"hive-sh4re",
"libc",
"rmcp",
"schemars",
"serde",

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

@ -130,19 +130,9 @@ pub async fn run_approval_apply_commit(
let approval = fetch_approval_for_worker(coord, approval_id, ApprovalKind::ApplyCommit)?;
let agent_dir = coord.ensure_runtime(&approval.agent)?;
let applied_dir = Coordinator::agent_applied_dir(&approval.agent);
let claude_dir = Coordinator::agent_claude_dir(&approval.agent);
let notes_dir = Coordinator::agent_notes_dir(&approval.agent);
coord.set_queue_step(queue_entry_id, "apply commit");
let (result, terminal_tag, is_first_spawn) = run_apply_commit(
coord,
&approval,
&agent_dir,
&applied_dir,
&claude_dir,
&notes_dir,
queue_entry_id,
)
.await;
let (result, terminal_tag, is_first_spawn) =
run_apply_commit(coord, &approval, &agent_dir, &applied_dir, queue_entry_id).await;
coord.set_queue_step(queue_entry_id, "forge push");
if let Err(e) = crate::forge::push_config(&approval.agent).await {
tracing::warn!(agent = %approval.agent, error = ?e, "forge: push_config after apply failed");
@ -215,32 +205,14 @@ pub async fn run_approval_spawn(
) -> Result<()> {
let approval = fetch_approval_for_worker(coord, approval_id, ApprovalKind::Spawn)?;
let agent_dir = coord.ensure_runtime(&approval.agent)?;
let proposed_dir = Coordinator::agent_proposed_dir(&approval.agent);
let applied_dir = Coordinator::agent_applied_dir(&approval.agent);
let claude_dir = Coordinator::agent_claude_dir(&approval.agent);
let notes_dir = Coordinator::agent_notes_dir(&approval.agent);
let hive = coord.hive_env();
let paths = Coordinator::agent_paths(&approval.agent, agent_dir);
// Transient guard keeps the per-container "Spawning" pill lit while
// the worker is doing the actual nixos-container create. Auto-clears
// on the function's scope exit (success or panic).
let _guard = coord.transient_guard(&approval.agent, TransientKind::Spawning);
coord.set_queue_step(queue_entry_id, "lifecycle::spawn");
let result = lifecycle::spawn(
&approval.agent,
&coord.hyperhive_flake,
&coord.nixpkgs_flake,
&coord.nixpkgs_unstable_flake,
&agent_dir,
&proposed_dir,
&applied_dir,
&claude_dir,
&notes_dir,
coord.dashboard_port,
&coord.operator_pronouns,
&coord.context_window_tokens,
&coord.agent_cpu_quota,
&coord.agent_memory_max,
)
.await;
let result = lifecycle::spawn(&approval.agent, &hive, &paths).await;
if result.is_ok() {
coord.set_queue_step(queue_entry_id, "forge user");
if let Err(e) = crate::forge::ensure_user_for(&approval.agent).await {
@ -450,8 +422,6 @@ async fn run_apply_commit(
approval: &hive_sh4re::Approval,
agent_dir: &std::path::Path,
applied_dir: &std::path::Path,
claude_dir: &std::path::Path,
notes_dir: &std::path::Path,
queue_entry_id: Option<u64>,
) -> (Result<()>, Option<String>, bool) {
let id = approval.id;
@ -548,17 +518,7 @@ async fn run_apply_commit(
);
}
};
if let Err(e) = crate::meta::sync_agents(
&coord.hyperhive_flake,
&coord.nixpkgs_flake,
&coord.nixpkgs_unstable_flake,
coord.dashboard_port,
&coord.operator_pronouns,
&coord.context_window_tokens,
&agents,
)
.await
{
if let Err(e) = crate::meta::sync_agents(&coord.hive_env(), &agents).await {
let _ = lifecycle::git_update_ref(applied_dir, "refs/heads/main", &prev_main_sha).await;
let _ = lifecycle::git_read_tree_reset(applied_dir, "refs/heads/main").await;
return (
@ -585,14 +545,12 @@ async fn run_apply_commit(
// Step labels are emitted inside rebuild_no_meta via the callback so
// the dashboard reflects actual phase progress rather than a static
// "nixos-container update" label for the whole multi-minute window.
let hive = coord.hive_env();
let paths = Coordinator::agent_paths(&approval.agent, agent_dir.to_path_buf());
let build_result = lifecycle::rebuild_no_meta(
&approval.agent,
agent_dir,
applied_dir,
claude_dir,
notes_dir,
&coord.agent_cpu_quota,
&coord.agent_memory_max,
&hive,
&paths,
&|step| coord.set_queue_step(queue_entry_id, step),
)
.await;
@ -722,16 +680,7 @@ pub async fn destroy(coord: &Arc<Coordinator>, name: &str, purge: bool) -> Resul
/// destroy). Idempotent — a no-op when nothing changed.
async fn sync_meta_after_lifecycle(coord: &Coordinator) -> Result<()> {
let agents = lifecycle::agents_for_meta_listing().await?;
crate::meta::sync_agents(
&coord.hyperhive_flake,
&coord.nixpkgs_flake,
&coord.nixpkgs_unstable_flake,
coord.dashboard_port,
&coord.operator_pronouns,
&coord.context_window_tokens,
&agents,
)
.await
crate::meta::sync_agents(&coord.hive_env(), &agents).await
}
pub async fn deny(coord: &Coordinator, id: i64, note: Option<&str>) -> Result<()> {

43
hive-c0re/src/auto_update.rs
View file

@ -77,9 +77,8 @@ pub async fn rebuild_agent(
let agent_dir = coord
.ensure_runtime(name)
.with_context(|| format!("ensure_runtime {name}"))?;
let applied_dir = Coordinator::agent_applied_dir(name);
let claude_dir = Coordinator::agent_claude_dir(name);
let notes_dir = Coordinator::agent_notes_dir(name);
let hive = coord.hive_env();
let paths = Coordinator::agent_paths(name, agent_dir);
// Suppress crash_watch during the stop+start window inside
// lifecycle::rebuild. Dashboard rebuilds already do this via
// lifecycle_action; this catches the auto-update scan + any
@ -87,18 +86,8 @@ pub async fn rebuild_agent(
let guard = coord.transient_guard(name, crate::coordinator::TransientKind::Rebuilding);
let result = lifecycle::rebuild(
name,
&coord.hyperhive_flake,
&coord.nixpkgs_flake,
&coord.nixpkgs_unstable_flake,
&agent_dir,
&applied_dir,
&claude_dir,
&notes_dir,
coord.dashboard_port,
&coord.operator_pronouns,
&coord.context_window_tokens,
&coord.agent_cpu_quota,
&coord.agent_memory_max,
&hive,
&paths,
&|step| coord.set_queue_step(queue_entry_id, step),
)
.await;
@ -190,27 +179,9 @@ pub async fn ensure_root_agent(coord: &Arc<Coordinator>) -> Result<()> {
}
tracing::info!("manager container missing — spawning");
let runtime = coord.ensure_runtime(MANAGER_NAME)?;
let proposed = Coordinator::agent_proposed_dir(MANAGER_NAME);
let applied = Coordinator::agent_applied_dir(MANAGER_NAME);
let claude_dir = Coordinator::agent_claude_dir(MANAGER_NAME);
let notes_dir = Coordinator::agent_notes_dir(MANAGER_NAME);
lifecycle::spawn(
MANAGER_NAME,
&coord.hyperhive_flake,
&coord.nixpkgs_flake,
&coord.nixpkgs_unstable_flake,
&runtime,
&proposed,
&applied,
&claude_dir,
&notes_dir,
coord.dashboard_port,
&coord.operator_pronouns,
&coord.context_window_tokens,
&coord.agent_cpu_quota,
&coord.agent_memory_max,
)
.await?;
let hive = coord.hive_env();
let paths = Coordinator::agent_paths(MANAGER_NAME, runtime);
lifecycle::spawn(MANAGER_NAME, &hive, &paths).await?;
if let Some(rev) = current_rev {
let _ = std::fs::write(rev_marker_path(MANAGER_NAME), &rev);
}

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

@ -143,6 +143,44 @@ pub struct Coordinator {
shutdown_tx: watch::Sender<bool>,
}
/// Hive-wide configuration that lifecycle and meta operations need.
/// Extracted from `Coordinator` so callers can pass a single struct
/// instead of repeating the same 6 arguments everywhere.
///
/// Cloned from `Coordinator` via [`Coordinator::hive_env`]. All fields
/// are cheap to clone (small strings + small map); lifecycle ops are
/// infrequent enough that the copy cost is irrelevant.
#[derive(Clone, Debug)]
pub struct HiveEnv {
pub hyperhive_flake: String,
pub nixpkgs_flake: String,
pub nixpkgs_unstable_flake: String,
pub dashboard_port: u16,
pub operator_pronouns: String,
pub context_window_tokens: std::collections::HashMap<String, u64>,
/// Per-agent systemd `CPUQuota=` value (e.g. `"200%"`).
pub agent_cpu_quota: String,
/// Per-agent systemd `MemoryMax=` value (e.g. `"4G"`).
pub agent_memory_max: String,
}
/// Per-agent filesystem paths that lifecycle operations write to.
/// Assembled from `Coordinator`'s static path helpers so callers
/// don't repeat the same `Coordinator::agent_*_dir(name)` calls.
#[derive(Clone, Debug)]
pub struct AgentPaths {
/// Runtime socket dir (tmpfs, recreated per boot).
pub agent_dir: PathBuf,
/// Manager-editable proposed config repo.
pub proposed_dir: PathBuf,
/// Hive-c0re-authoritative applied config repo.
pub applied_dir: PathBuf,
/// Claude OAuth credentials (survives purge boundary).
pub claude_dir: PathBuf,
/// Agent durable notes + forge token (survives purge boundary).
pub notes_dir: PathBuf,
}
/// Per-agent in-progress state that the dashboard surfaces between approve
/// click and container ready.
#[derive(Debug, Clone)]
@ -282,6 +320,42 @@ impl Coordinator {
})
}
/// Snapshot the hive-wide configuration fields as a [`HiveEnv`].
/// Pass the result to `lifecycle::spawn` / `rebuild` / `meta::sync_agents`
/// instead of threading the individual fields separately.
#[must_use]
pub fn hive_env(&self) -> HiveEnv {
HiveEnv {
hyperhive_flake: self.hyperhive_flake.clone(),
nixpkgs_flake: self.nixpkgs_flake.clone(),
nixpkgs_unstable_flake: self.nixpkgs_unstable_flake.clone(),
dashboard_port: self.dashboard_port,
operator_pronouns: self.operator_pronouns.clone(),
context_window_tokens: self.context_window_tokens.clone(),
agent_cpu_quota: self.agent_cpu_quota.clone(),
agent_memory_max: self.agent_memory_max.clone(),
}
}
/// Assemble the per-agent filesystem paths for `name`. The caller
/// must supply `agent_dir` (from `ensure_runtime`) since that
/// creates the tmpfs entry on first call. All other paths are
/// derived statically from `name`.
///
/// ```no_run
/// let paths = Coordinator::agent_paths(name, coord.ensure_runtime(name)?);
/// ```
#[must_use]
pub fn agent_paths(name: &str, agent_dir: PathBuf) -> AgentPaths {
AgentPaths {
agent_dir,
proposed_dir: Self::agent_proposed_dir(name),
applied_dir: Self::agent_applied_dir(name),
claude_dir: Self::agent_claude_dir(name),
notes_dir: Self::agent_notes_dir(name),
}
}
/// Emit a `RebuildQueueChanged` snapshot event. Called from the
/// queue mutation helpers (`enqueue` / `finish` / `cancel`-adjacent
/// wrappers below) and the worker so every state transition

93
hive-c0re/src/lifecycle.rs
View file

@ -6,6 +6,8 @@ use anyhow::{Context, Result, bail};
use hive_sh4re::priv_proto::BindMount;
use tokio::process::Command;
use crate::coordinator::{AgentPaths, HiveEnv};
/// Sub-agent container prefix. `nixos-container` caps the total container name
/// at 11 chars (it gets encoded into network interface names), so the agent
/// name itself can be at most `MAX_AGENT_NAME` chars.
@ -221,23 +223,7 @@ async fn port_collision(self_name: &str) -> Option<String> {
None
}
#[allow(clippy::too_many_arguments, clippy::implicit_hasher)]
pub async fn spawn(
name: &str,
hyperhive_flake: &str,
nixpkgs_flake: &str,
nixpkgs_unstable_flake: &str,
agent_dir: &Path,
proposed_dir: &Path,
applied_dir: &Path,
claude_dir: &Path,
notes_dir: &Path,
dashboard_port: u16,
operator_pronouns: &str,
context_window_tokens: &std::collections::HashMap<String, u64>,
cpu_quota: &str,
memory_max: &str,
) -> Result<()> {
pub async fn spawn(name: &str, hive: &HiveEnv, paths: &AgentPaths) -> Result<()> {
validate(name)?;
if let Some(other) = port_collision(name).await {
bail!(
@ -245,28 +231,19 @@ pub async fn spawn(
agent_web_port(name)
);
}
setup_proposed(proposed_dir, name).await?;
setup_applied(applied_dir, Some(proposed_dir), name).await?;
ensure_claude_dir(claude_dir)?;
ensure_state_dir(notes_dir)?;
setup_proposed(&paths.proposed_dir, name).await?;
setup_applied(&paths.applied_dir, Some(&paths.proposed_dir), name).await?;
ensure_claude_dir(&paths.claude_dir)?;
ensure_state_dir(&paths.notes_dir)?;
// 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(
hyperhive_flake,
nixpkgs_flake,
nixpkgs_unstable_flake,
dashboard_port,
operator_pronouns,
context_window_tokens,
&agents,
)
.await?;
crate::meta::sync_agents(hive, &agents).await?;
let container = container_name(name);
priv_run("create", name).await?;
set_nspawn_flags(&container, agent_dir, claude_dir, notes_dir).await?;
set_resource_limits(&container, cpu_quota, memory_max).await?;
set_nspawn_flags(&container, &paths.agent_dir, &paths.claude_dir, &paths.notes_dir).await?;
set_resource_limits(&container, &hive.agent_cpu_quota, &hive.agent_memory_max).await?;
systemd_daemon_reload().await?;
priv_run("start", name).await
}
@ -389,21 +366,10 @@ pub async fn destroy(name: &str) -> Result<()> {
Ok(())
}
#[allow(clippy::too_many_arguments, clippy::implicit_hasher)]
pub async fn rebuild(
name: &str,
hyperhive_flake: &str,
nixpkgs_flake: &str,
nixpkgs_unstable_flake: &str,
agent_dir: &Path,
applied_dir: &Path,
claude_dir: &Path,
notes_dir: &Path,
dashboard_port: u16,
operator_pronouns: &str,
context_window_tokens: &std::collections::HashMap<String, u64>,
cpu_quota: &str,
memory_max: &str,
hive: &HiveEnv,
paths: &AgentPaths,
on_step: &(dyn Fn(&str) + Send + Sync),
) -> Result<()> {
// Sync the meta flake (idempotent — no-op when the rendered
@ -412,21 +378,12 @@ pub async fn rebuild(
// got added directly via `nixos-container create` outside
// hive-c0re).
let agents = agents_for_meta(None).await?;
crate::meta::sync_agents(
hyperhive_flake,
nixpkgs_flake,
nixpkgs_unstable_flake,
dashboard_port,
operator_pronouns,
context_window_tokens,
&agents,
)
.await?;
crate::meta::sync_agents(hive, &agents).await?;
// Then bump just this agent's input — picks up whatever
// `applied/<n>/main` currently points at (deployed/<latest>).
// Commits the lock if it changed.
crate::meta::lock_update_for_rebuild(name).await?;
rebuild_no_meta(name, agent_dir, applied_dir, claude_dir, notes_dir, cpu_quota, memory_max, on_step).await
rebuild_no_meta(name, hive, paths, on_step).await
}
/// Container-level rebuild without touching the meta repo. Callers
@ -441,12 +398,8 @@ pub async fn rebuild(
/// is not needed.
pub async fn rebuild_no_meta(
name: &str,
agent_dir: &Path,
applied_dir: &Path,
claude_dir: &Path,
notes_dir: &Path,
cpu_quota: &str,
memory_max: &str,
hive: &HiveEnv,
paths: &AgentPaths,
on_step: &(dyn Fn(&str) + Send + Sync),
) -> Result<()> {
validate(name)?;
@ -456,17 +409,17 @@ pub async fn rebuild_no_meta(
agent_web_port(name)
);
}
setup_applied(applied_dir, None, name).await?;
ensure_claude_dir(claude_dir)?;
ensure_state_dir(notes_dir)?;
setup_applied(&paths.applied_dir, None, name).await?;
ensure_claude_dir(&paths.claude_dir)?;
ensure_state_dir(&paths.notes_dir)?;
let container = container_name(name);
let flake_ref = format!("{}#{name}", crate::meta::meta_dir().display());
if container_exists(name).await {
// Rebuild strategy: stop-before-update + pre-build.
// See `docs/coordinator.md::Container lifecycle`.
let was_running = is_running(name).await;
set_nspawn_flags(&container, agent_dir, claude_dir, notes_dir).await?;
set_resource_limits(&container, cpu_quota, memory_max).await?;
set_nspawn_flags(&container, &paths.agent_dir, &paths.claude_dir, &paths.notes_dir).await?;
set_resource_limits(&container, &hive.agent_cpu_quota, &hive.agent_memory_max).await?;
systemd_daemon_reload().await?;
if was_running {
on_step("nix build");
@ -537,8 +490,8 @@ pub async fn rebuild_no_meta(
// See `docs/coordinator.md::Spawn path`.
on_step("nixos-container create");
priv_run("create", name).await?;
set_nspawn_flags(&container, agent_dir, claude_dir, notes_dir).await?;
set_resource_limits(&container, cpu_quota, memory_max).await?;
set_nspawn_flags(&container, &paths.agent_dir, &paths.claude_dir, &paths.notes_dir).await?;
set_resource_limits(&container, &hive.agent_cpu_quota, &hive.agent_memory_max).await?;
systemd_daemon_reload().await?;
on_step("nixos-container start");
priv_run("start", name).await

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

@ -10,6 +10,7 @@ use anyhow::{Context, Result, bail};
use tokio::process::Command;
use tokio::sync::Mutex;
use crate::coordinator::HiveEnv;
use crate::lifecycle;
const META_ROOT: &str = "/var/lib/hyperhive/meta";
@ -50,26 +51,18 @@ pub fn meta_dir() -> PathBuf {
/// rendered contents differ from disk; an unchanged `flake.nix` is a
/// no-op.
#[allow(dead_code, clippy::implicit_hasher)] // first caller lands in a later commit
pub async fn sync_agents(
hyperhive_flake: &str,
nixpkgs_flake: &str,
nixpkgs_unstable_flake: &str,
dashboard_port: u16,
operator_pronouns: &str,
context_window_tokens: &std::collections::HashMap<String, u64>,
agents: &[AgentSpec],
) -> Result<()> {
pub async fn sync_agents(hive: &HiveEnv, agents: &[AgentSpec]) -> Result<()> {
let _guard = META_LOCK.lock().await;
let dir = meta_dir();
std::fs::create_dir_all(&dir).with_context(|| format!("create {}", dir.display()))?;
let new_flake = render_flake(
hyperhive_flake,
nixpkgs_flake,
nixpkgs_unstable_flake,
dashboard_port,
operator_pronouns,
context_window_tokens,
&hive.hyperhive_flake,
&hive.nixpkgs_flake,
&hive.nixpkgs_unstable_flake,
hive.dashboard_port,
&hive.operator_pronouns,
&hive.context_window_tokens,
agents,
);
let flake_path = dir.join("flake.nix");

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

@ -76,17 +76,7 @@ pub async fn run(coord: &Arc<Coordinator>) -> Result<()> {
let agents = lifecycle::agents_for_meta_listing()
.await
.unwrap_or_default();
if let Err(e) = meta::sync_agents(
&coord.hyperhive_flake,
&coord.nixpkgs_flake,
&coord.nixpkgs_unstable_flake,
coord.dashboard_port,
&coord.operator_pronouns,
&coord.context_window_tokens,
&agents,
)
.await
{
if let Err(e) = meta::sync_agents(&coord.hive_env(), &agents).await {
tracing::warn!(error = ?e, "migration: meta sync_agents failed");
}

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

@ -80,27 +80,9 @@ async fn dispatch(req: &HostRequest, coord: Arc<Coordinator>) -> HostResponse {
HostRequest::Spawn { name } => {
tracing::info!(%name, "spawn");
let agent_dir = coord.ensure_runtime(name)?;
let proposed_dir = Coordinator::agent_proposed_dir(name);
let applied_dir = Coordinator::agent_applied_dir(name);
let claude_dir = Coordinator::agent_claude_dir(name);
let notes_dir = Coordinator::agent_notes_dir(name);
match lifecycle::spawn(
name,
&coord.hyperhive_flake,
&coord.nixpkgs_flake,
&coord.nixpkgs_unstable_flake,
&agent_dir,
&proposed_dir,
&applied_dir,
&claude_dir,
&notes_dir,
coord.dashboard_port,
&coord.operator_pronouns,
&coord.context_window_tokens,
&coord.agent_cpu_quota,
&coord.agent_memory_max,
)
.await
let hive = coord.hive_env();
let paths = Coordinator::agent_paths(name, agent_dir);
match lifecycle::spawn(name, &hive, &paths).await
{
Ok(()) => {
coord.notify_manager(&hive_sh4re::HelperEvent::Spawned {
@ -178,26 +160,9 @@ async fn dispatch(req: &HostRequest, coord: Arc<Coordinator>) -> HostResponse {
HostRequest::Rebuild { name } => {
tracing::info!(%name, "rebuild");
let agent_dir = coord.ensure_runtime(name)?;
let applied_dir = Coordinator::agent_applied_dir(name);
let claude_dir = Coordinator::agent_claude_dir(name);
let notes_dir = Coordinator::agent_notes_dir(name);
let result = lifecycle::rebuild(
name,
&coord.hyperhive_flake,
&coord.nixpkgs_flake,
&coord.nixpkgs_unstable_flake,
&agent_dir,
&applied_dir,
&claude_dir,
&notes_dir,
coord.dashboard_port,
&coord.operator_pronouns,
&coord.context_window_tokens,
&coord.agent_cpu_quota,
&coord.agent_memory_max,
&|_| (),
)
.await;
let hive = coord.hive_env();
let paths = Coordinator::agent_paths(name, agent_dir);
let result = lifecycle::rebuild(name, &hive, &paths, &|_| ()).await;
// Mirror auto_update::rebuild_agent — the manager wants
// to know about every rebuild attempt regardless of
// which surface triggered it, especially failures