Watch
0
0
Fork
You've already forked hyperhive
0
hyperhive/hive-c0re/src/lifecycle.rs

374 lines
13 KiB
Rust

//! `nixos-container` lifecycle + per-agent config flake generation.
use std::path::Path;
use anyhow::{Context, Result, bail};
use tokio::process::Command;
/// 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.
pub const AGENT_PREFIX: &str = "h-";
pub const MAX_AGENT_NAME: usize = 9;
/// Container name of the manager (a separate slot from sub-agents).
pub const MANAGER_NAME: &str = "hm1nd";
/// Mount point of the per-agent runtime directory inside the container.
pub const CONTAINER_RUNTIME_MOUNT: &str = "/run/hive";
const GIT_NAME: &str = "hive-c0re";
const GIT_EMAIL: &str = "hive-c0re@hyperhive";
/// Sub-agent web UI port range. Deterministic from the agent's name (FNV-1a
/// hash mod range size), so the dashboard can compute the same port without
/// asking hive-c0re.
const WEB_PORT_BASE: u16 = 8100;
const WEB_PORT_RANGE: u16 = 900;
/// Default resource caps applied to every managed container via a systemd
/// drop-in under `/run/systemd/system/container@<NAME>.service.d/`.
const DEFAULT_MEMORY_MAX: &str = "2G";
const DEFAULT_CPU_QUOTA: &str = "50%";
/// Returns the per-agent web UI port. Same hash on both sides — manager,
/// dashboard, and agent harness all agree.
#[must_use]
pub fn agent_web_port(name: &str) -> u16 {
let mut hash: u32 = 2_166_136_261;
for b in name.bytes() {
hash ^= u32::from(b);
hash = hash.wrapping_mul(16_777_619);
}
// Modulo of a u32 by a u16's value is guaranteed < u16::MAX, so try_from never fails.
WEB_PORT_BASE + u16::try_from(hash % u32::from(WEB_PORT_RANGE)).unwrap_or(0)
}
pub fn container_name(name: &str) -> String {
format!("{AGENT_PREFIX}{name}")
}
fn validate(name: &str) -> Result<()> {
if name.is_empty() {
bail!("agent name must not be empty");
}
if name.len() > MAX_AGENT_NAME {
bail!(
"agent name '{name}' is too long ({} chars); max {MAX_AGENT_NAME}",
name.len()
);
}
Ok(())
}
pub async fn spawn(
name: &str,
hyperhive_flake: &str,
agent_dir: &Path,
proposed_dir: &Path,
applied_dir: &Path,
) -> Result<()> {
validate(name)?;
setup_proposed(proposed_dir, name).await?;
setup_applied(applied_dir, name, hyperhive_flake).await?;
let container = container_name(name);
let flake_ref = format!("{}#default", applied_dir.display());
run(&["create", &container, "--flake", &flake_ref]).await?;
set_nspawn_flags(&container, agent_dir)?;
set_resource_limits(&container)?;
systemd_daemon_reload().await?;
run(&["start", &container]).await
}
pub async fn kill(name: &str) -> Result<()> {
validate(name)?;
let container = container_name(name);
run(&["stop", &container]).await
}
pub async fn rebuild(
name: &str,
hyperhive_flake: &str,
agent_dir: &Path,
applied_dir: &Path,
) -> Result<()> {
validate(name)?;
setup_applied(applied_dir, name, hyperhive_flake).await?;
let container = container_name(name);
let flake_ref = format!("{}#default", applied_dir.display());
set_nspawn_flags(&container, agent_dir)?;
set_resource_limits(&container)?;
systemd_daemon_reload().await?;
run(&["update", &container, "--flake", &flake_ref]).await?;
// Restart so any nspawn-level changes (bind mounts, networking, etc.) apply.
run(&["stop", &container]).await?;
run(&["start", &container]).await
}
pub async fn list() -> Result<Vec<String>> {
let out = Command::new("nixos-container")
.arg("list")
.output()
.await
.context("invoke nixos-container list")?;
if !out.status.success() {
bail!(
"nixos-container list exited with status {}: {}",
out.status,
String::from_utf8_lossy(&out.stderr).trim()
);
}
Ok(String::from_utf8_lossy(&out.stdout)
.lines()
.map(str::trim)
.filter(|line| line.starts_with(AGENT_PREFIX) || *line == MANAGER_NAME)
.map(str::to_owned)
.collect())
}
/// Initialize the manager-editable proposed repo. Contains only `agent.nix`
/// (the file the manager edits). Touched by hive-c0re only on first spawn —
/// never again — so the manager can't be surprised by hive-c0re commits or
/// working-tree resets.
pub async fn setup_proposed(proposed_dir: &Path, name: &str) -> Result<()> {
if proposed_dir.join(".git").exists() {
return Ok(());
}
std::fs::create_dir_all(proposed_dir)
.with_context(|| format!("create {}", proposed_dir.display()))?;
let agent_path = proposed_dir.join("agent.nix");
if !agent_path.exists() {
std::fs::write(&agent_path, initial_agent_nix(name))
.with_context(|| format!("write {}", agent_path.display()))?;
}
git(proposed_dir, &["init", "--initial-branch=main"]).await?;
git(proposed_dir, &["add", "agent.nix"]).await?;
git_commit(proposed_dir, "hive-c0re init").await?;
Ok(())
}
/// Maintain the authoritative applied repo. Rewrites `flake.nix` every call
/// (so a new hyperhive flake URL propagates on rebuild); seeds `agent.nix`
/// only on first call. `apply_commit` overwrites `agent.nix` later.
pub async fn setup_applied(applied_dir: &Path, name: &str, hyperhive_flake: &str) -> Result<()> {
std::fs::create_dir_all(applied_dir)
.with_context(|| format!("create {}", applied_dir.display()))?;
let port = agent_web_port(name);
let flake_body = format!(
r#"{{
description = "hyperhive sub-agent {name}";
inputs.hyperhive.url = "{hyperhive_flake}";
outputs =
{{ hyperhive, ... }}:
{{
nixosConfigurations.default = hyperhive.nixosConfigurations.agent-base.extendModules {{
modules = [
./agent.nix
{{
environment.etc."gitconfig".text = ''
[user]
name = {name}
email = {name}@hyperhive
[init]
defaultBranch = main
'';
systemd.services.hive-ag3nt.environment = {{
HIVE_PORT = "{port}";
HIVE_LABEL = "{name}";
}};
}}
];
}};
}};
}}
"#,
);
std::fs::write(applied_dir.join("flake.nix"), flake_body)
.with_context(|| format!("write {}/flake.nix", applied_dir.display()))?;
let agent_path = applied_dir.join("agent.nix");
if !agent_path.exists() {
std::fs::write(&agent_path, initial_agent_nix(name))
.with_context(|| format!("write {}", agent_path.display()))?;
}
if !applied_dir.join(".git").exists() {
git(applied_dir, &["init", "--initial-branch=main"]).await?;
}
git(applied_dir, &["add", "-A"]).await?;
let clean = git_status(applied_dir, &["diff", "--cached", "--quiet"]).await?;
if !clean {
git_commit(applied_dir, "hive-c0re sync").await?;
}
Ok(())
}
/// Apply a manager-proposed commit: read `agent.nix` at `commit_ref` from the
/// proposed repo, write it into the applied repo, commit. Hive-c0re alone
/// advances `applied`'s `main`; the manager only sees `proposed/`.
pub async fn apply_commit(applied_dir: &Path, proposed_dir: &Path, commit_ref: &str) -> Result<()> {
let out = git_command()
.current_dir(proposed_dir)
.args(["show", &format!("{commit_ref}:agent.nix")])
.output()
.await
.with_context(|| format!("git show in {}", proposed_dir.display()))?;
if !out.status.success() {
bail!(
"agent.nix at commit {commit_ref} not found in {}: {}",
proposed_dir.display(),
String::from_utf8_lossy(&out.stderr).trim()
);
}
std::fs::write(applied_dir.join("agent.nix"), &out.stdout)
.with_context(|| format!("write {}/agent.nix", applied_dir.display()))?;
git(applied_dir, &["add", "agent.nix"]).await?;
let clean = git_status(applied_dir, &["diff", "--cached", "--quiet"]).await?;
if !clean {
git_commit(applied_dir, &format!("apply {commit_ref}")).await?;
}
Ok(())
}
fn initial_agent_nix(name: &str) -> String {
format!(
"{{ ... }}:\n{{\n # Per-agent overrides for {name}. The manager edits this\n # file (and commits) to customise the agent's NixOS config.\n}}\n",
)
}
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.
#[must_use]
pub fn git_command() -> Command {
let exe = std::env::var("HYPERHIVE_GIT").unwrap_or_else(|_| "git".into());
Command::new(exe)
}
async fn git(dir: &Path, args: &[&str]) -> Result<()> {
let out = git_command()
.current_dir(dir)
.args(args)
.output()
.await
.with_context(|| format!("git {} in {}", args.join(" "), dir.display()))?;
if !out.status.success() {
bail!(
"git {} failed ({}): {}",
args.join(" "),
out.status,
String::from_utf8_lossy(&out.stderr).trim()
);
}
Ok(())
}
/// Returns true if the command exits 0.
async fn git_status(dir: &Path, args: &[&str]) -> Result<bool> {
let st = git_command()
.current_dir(dir)
.args(args)
.status()
.await
.with_context(|| format!("git {} in {}", args.join(" "), dir.display()))?;
Ok(st.success())
}
/// Write a systemd drop-in for `container@<container>.service` that applies
/// our default resource caps. Goes under `/run/systemd/system/...` so it's
/// ephemeral (regenerated on every spawn / rebuild).
fn set_resource_limits(container: &str) -> Result<()> {
let dir = format!("/run/systemd/system/container@{container}.service.d");
std::fs::create_dir_all(&dir).with_context(|| format!("create {dir}"))?;
let path = format!("{dir}/hyperhive-limits.conf");
let content = format!(
"[Service]\nMemoryMax={DEFAULT_MEMORY_MAX}\nCPUQuota={DEFAULT_CPU_QUOTA}\n",
);
std::fs::write(&path, content).with_context(|| format!("write {path}"))?;
tracing::info!(
%path,
memory_max = DEFAULT_MEMORY_MAX,
cpu_quota = DEFAULT_CPU_QUOTA,
"wrote resource limits drop-in"
);
Ok(())
}
async fn systemd_daemon_reload() -> Result<()> {
let out = Command::new("systemctl")
.arg("daemon-reload")
.output()
.await
.context("invoke systemctl daemon-reload")?;
if !out.status.success() {
bail!(
"systemctl daemon-reload failed ({}): {}",
out.status,
String::from_utf8_lossy(&out.stderr).trim()
);
}
Ok(())
}
/// Idempotently rewrite the `EXTRA_NSPAWN_FLAGS` line in
/// `/etc/nixos-containers/<container>.conf`. The start script expands this
/// variable unquoted into the `systemd-nspawn` command.
fn set_nspawn_flags(container: &str, agent_dir: &Path) -> Result<()> {
let path = format!("/etc/nixos-containers/{container}.conf");
let original = std::fs::read_to_string(&path).with_context(|| format!("read {path}"))?;
let flag = format!(
"EXTRA_NSPAWN_FLAGS=\"--bind={}:{CONTAINER_RUNTIME_MOUNT}\"",
agent_dir.display()
);
let mut lines: Vec<String> = original
.lines()
.filter(|line| !line.trim_start().starts_with("EXTRA_NSPAWN_FLAGS="))
.map(str::to_owned)
.collect();
lines.push(flag);
let mut content = lines.join("\n");
content.push('\n');
std::fs::write(&path, content).with_context(|| format!("write {path}"))?;
tracing::info!(%path, "set EXTRA_NSPAWN_FLAGS");
Ok(())
}
async fn run(args: &[&str]) -> Result<()> {
let out = Command::new("nixos-container")
.args(args)
.output()
.await
.with_context(|| format!("invoke nixos-container {}", args.join(" ")))?;
let stdout = String::from_utf8_lossy(&out.stdout);
let stderr = String::from_utf8_lossy(&out.stderr);
if !stdout.trim().is_empty() {
tracing::info!(target: "nixos-container", "{}", stdout.trim());
}
if !stderr.trim().is_empty() {
tracing::warn!(target: "nixos-container", "{}", stderr.trim());
}
if !out.status.success() {
bail!(
"nixos-container {} failed ({}): {}",
args.join(" "),
out.status,
stderr.trim()
);
}
Ok(())
}