hyperhive/hive-c0re/src/lifecycle/mod.rs

//! `nixos-container` lifecycle + per-agent config flake generation.

mod git;
mod host_config;
mod setup;
#[cfg(test)]
mod tests;

pub use git::{
    git, git_command, git_delete_ref, git_is_ancestor, git_read_tree_reset, git_rev_parse, git_tag,
    git_tag_annotated, git_update_ref, git_update_ref_cas,
};
pub use host_config::write_dropins;
pub use setup::{
    ensure_agent_state_subvolume, ensure_claude_dir, ensure_state_dir, initial_flake_nix,
    setup_applied, setup_proposed,
};

use anyhow::{Context, Result, bail};
use tokio::process::Command;

use crate::coordinator::{AgentPaths, HiveEnv};

// `AGENT_PREFIX` (`h-`) + `container_name` are shared with the host-side
// `hivectl` CLI, so they live in `hive-host-sock`; re-exported here so this
// module stays the daemon's facade for its own callsites.
pub use hive_host_sock::{AGENT_PREFIX, container_name};

/// Max agent-name length. `nixos-container` caps the total container name at
/// 11 chars (it gets encoded into network interface names) and `AGENT_PREFIX`
/// (`h-`) takes 2, so the agent name itself is at most `MAX_AGENT_NAME` chars.
pub const MAX_AGENT_NAME: usize = 9;
/// Logical name of the manager agent (broker recipient, state-dir key,
/// meta flake attribute). All persistent state lives under `ruth/`.
pub const MANAGER_NAME: &str = "ruth";
/// Container name of the manager. Uses the same `h-` prefix as sub-agents
/// so `nixos-container list` output is uniform and the list filter is
/// a single `starts_with(AGENT_PREFIX)` check. Logical name → container
/// name: `ruth` → `h-ruth`.
pub const MANAGER_CONTAINER: &str = "h-ruth";

/// Mount point of the per-agent runtime directory inside the container.
pub const CONTAINER_RUNTIME_MOUNT: &str = "/run/hive";

/// Where the per-agent Claude credentials dir mounts inside the
/// container. The harness service runs as a non-root unix user
/// whose home is `/home/<agent>/`, so the mount path varies per
/// agent — `container_claude_mount(name)` returns
/// `/home/<name>/.claude` for every agent including the manager.
/// `claude` inside the container reads
/// `$HOME/.claude` and the service environment sets `HOME` to the
/// same path, so the OAuth session survives container restarts.
#[must_use]
pub fn container_claude_mount(name: &str) -> String {
    format!("/home/{name}/.claude")
}

/// Mount point of the shared directory accessible to all agents.
/// All agents can read/write here; agents should only put things they're
/// willing to lose (other agents may delete them).
pub const CONTAINER_SHARED_MOUNT: &str = "/shared";

/// 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;

/// FNV-1a hash of a string — used by `agent_web_port`.
fn fnv1a(s: &str) -> u32 {
    let mut hash: u32 = 2_166_136_261;
    for b in s.bytes() {
        hash ^= u32::from(b);
        hash = hash.wrapping_mul(16_777_619);
    }
    hash
}

/// Per-agent web UI port — `WEB_PORT_BASE + FNV-1a(name) %
/// WEB_PORT_RANGE` for every agent including the manager. The port
/// allocation rule reads the same for every name; collisions are
/// possible (birthday paradox at ~30 agents) and the operator
/// resolves them by renaming an agent (different hash → different
/// port). Stable across hosts, restarts, and dashboard renders —
/// no state-file dance.
#[must_use]
pub fn agent_web_port(name: &str) -> u16 {
    // Modulo of a u32 by a u16's value is guaranteed < u16::MAX, so try_from never fails.
    WEB_PORT_BASE + u16::try_from(fnv1a(name) % u32::from(WEB_PORT_RANGE)).unwrap_or(0)
}

/// Extract the bridge gateway IP from `HIVE_NETWORK_SUBNET`.
///
/// `HIVE_NETWORK_SUBNET` carries the host-side bridge address verbatim
/// (e.g. `10.42.0.1/24`), **not** the canonical network address — see
/// the note in `set_nspawn_flags` + `docs/network.md`. The IP part is
/// therefore the bridge IP itself: the host end of the bridge, the
/// default-route target for isolated containers, and the address the
/// hive dnsmasq resolver binds. Returns the dotted-decimal IP with the
/// `/<prefix>` stripped, or `None` if the input isn't a valid
/// `<ipv4>/<prefix>` pair.
///
/// Deliberately returns the operator-configured address verbatim rather
/// than deriving `network + 1`: an operator who sets `bridgeIp` to a
/// non-`.1` host address (e.g. `10.42.0.254`) runs the bridge + resolver
/// there, so that — not `.1` — is the real gateway.
#[must_use]
pub fn bridge_gateway_ip(subnet_cidr: &str) -> Option<String> {
    let (ip_str, prefix_str) = subnet_cidr.split_once('/')?;
    // Validate the prefix is a sane IPv4 CIDR length and the address is
    // dotted-decimal IPv4 so a malformed `HIVE_NETWORK_SUBNET` can't
    // smuggle a bogus HOST_ADDRESS into the nspawn conf.
    let prefix_len: u32 = prefix_str.parse().ok()?;
    if prefix_len > 32 {
        return None;
    }
    let octets: Vec<u8> = ip_str
        .split('.')
        .map(|o| o.parse::<u8>().ok())
        .collect::<Option<Vec<_>>>()?;
    if octets.len() != 4 {
        return None;
    }
    Some(ip_str.to_owned())
}

/// Read the agent user's `(uid, gid)` from the container's nixos-managed
/// `/etc/passwd`. Returns `None` when the container hasn't been built
/// yet, the passwd file is unparseable, or the agent user is missing
/// (e.g. legacy container that still runs as root).
///
/// Used by `forge` + `matrix` after writing per-agent state files so
/// the bind-mounted host file ends up readable by the agent user
/// without waiting for the next container activation to run the chown
/// fixup.
///
/// Notes:
///   - Reads the *container-local* passwd at
///     `/var/lib/nixos-containers/<container>/etc/passwd`, not the host's.
///     The container's user-namespace shares uids with the host (no
///     `PrivateUsers`), so the uid is directly usable in host-side
///     `chown(2)`.
///   - Best-effort: caller treats `None` as "skip the chown".
#[must_use]
pub fn agent_uid_gid(agent_name: &str) -> Option<(u32, u32)> {
    let container = container_name(agent_name);
    let passwd_path = format!("/var/lib/nixos-containers/{container}/etc/passwd");
    let content = std::fs::read_to_string(&passwd_path).ok()?;
    for line in content.lines() {
        let mut parts = line.split(':');
        let user = parts.next()?;
        if user != agent_name {
            continue;
        }
        let _ = parts.next()?; // x (password placeholder)
        let uid: u32 = parts.next()?.parse().ok()?;
        let gid: u32 = parts.next()?.parse().ok()?;
        return Some((uid, gid));
    }
    None
}

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(())
}

/// First name (≠ `self_name`) currently running whose hashed port
/// matches this agent's. The harness inside the colliding container
/// would otherwise loop on `AddrInUse` forever; we surface the
/// conflict here so spawn / rebuild fails loudly with an actionable
/// message instead.
async fn port_collision(self_name: &str) -> Option<String> {
    let port = agent_web_port(self_name);
    let raw = list().await.unwrap_or_default();
    for c in raw {
        let Some(other) = c.strip_prefix(AGENT_PREFIX) else {
            continue;
        };
        if other == self_name {
            continue;
        }
        if agent_web_port(other) == port && is_running(other).await {
            return Some(other.to_owned());
        }
    }
    None
}

pub async fn spawn(name: &str, hive: &HiveEnv, paths: &AgentPaths) -> Result<()> {
    create_container(name, hive, paths).await?;
    // Runtime dir must exist before nixos-container start (nspawn bind-mount
    // source). Create it here so callers don't need a separate preamble step.
    ensure_agent_runtime_dir(name)?;
    write_dropins(name, hive, paths).await?;
    priv_run("start", name).await
}

/// First-spawn pre-create provisioning — the job queue's `Provision`
/// node. Fail fast on a port collision, set up the proposed/applied
/// repos + state subvolume + claude/notes dirs, then register the new
/// agent in the meta flake (`sync_agents`) so the later
/// `nixos-container create --flake meta#<name>` ref resolves. Does NOT
/// create the container — that's `create_only` / the `Create` node.
pub async fn provision_container(name: &str, hive: &HiveEnv, paths: &AgentPaths) -> Result<()> {
    validate(name)?;
    if let Some(other) = port_collision(name).await {
        bail!(
            "port {} is already taken by '{other}' — rename one of them and retry",
            agent_web_port(name)
        );
    }
    setup_proposed(&paths.proposed, name).await?;
    setup_applied(&paths.applied, Some(&paths.proposed), name).await?;
    ensure_agent_state_subvolume(name).await?;
    ensure_claude_dir(&paths.claude)?;
    ensure_state_dir(&paths.notes)?;
    // Meta flake gets the new agent's input + nixosConfiguration
    // before `nixos-container create` so the `--flake meta#<name>`
    // ref resolves.
    let agents = agents_after_spawn(name).await?;
    crate::meta::sync_agents(hive, &agents).await
}

/// The `nixos-container create` proper — the job queue's `Create` node.
/// Assumes `provision_container` already registered the agent in meta.
pub async fn create_only(name: &str) -> Result<()> {
    priv_run("create", name).await
}

/// First-spawn provisioning + `nixos-container create`, without the
/// drop-in write or the start. `spawn` composes this with
/// `write_dropins` + start for direct callers (root-agent bootstrap);
/// the job queue instead runs `provision_container` (`Provision` node)
/// and `create_only` (`Create` node) as separate DAG steps.
pub async fn create_container(name: &str, hive: &HiveEnv, paths: &AgentPaths) -> Result<()> {
    provision_container(name, hive, paths).await?;
    create_only(name).await
}

/// 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.
pub async fn prepare_rebuild_dirs(name: &str, paths: &AgentPaths) -> Result<()> {
    validate(name)?;
    if let Some(other) = port_collision(name).await {
        bail!(
            "port {} is already taken by '{other}' — rename one of them and retry",
            agent_web_port(name)
        );
    }
    setup_applied(&paths.applied, None, name).await?;
    ensure_agent_state_subvolume(name).await?;
    ensure_claude_dir(&paths.claude)?;
    ensure_state_dir(&paths.notes)?;
    Ok(())
}

/// Profile-swap for an existing, stopped container: re-apply the
/// drop-ins, then `nixos-container update`. The job queue's `Swap`
/// node. Requires the container stopped (the queue's `StopForUpdate`
/// upstream); does NOT start it — the DAG's tail `Reconcile` owns
/// bringing the agent back to its wanted power state.
pub async fn swap_update(
    name: &str,
    hive: &HiveEnv,
    paths: &AgentPaths,
    on_build_log_id: &(dyn Fn(i64) + Send + Sync),
) -> Result<()> {
    write_dropins(name, hive, paths).await?;
    priv_run_inner("update", name, Some(on_build_log_id)).await
}

/// Build the `AgentSpec` list for the meta flake from `nixos-container
/// list` + a hypothetical extra name not yet in the list (for spawn
/// where the new agent's container doesn't exist yet). Pass empty
/// `name_to_add` from rebuild paths where the agent is already in the
/// container list.
///
/// Propagates errors from `list()` rather than swallowing them.
/// Using `.unwrap_or_default()` here would silently produce an empty
/// agent list when `nixos-container list` fails (priv helper down, race),
/// which `sync_agents` would then commit to meta — dropping every agent
/// from `flake.nix`. Callers that can tolerate failures (e.g. migration)
/// handle the `Err` themselves with `.unwrap_or_default()`.
async fn agents_for_meta(name_to_add: Option<&str>) -> Result<Vec<crate::meta::AgentSpec>> {
    let containers = list().await?;
    let mut out: Vec<crate::meta::AgentSpec> = containers
        .into_iter()
        .filter_map(|c| {
            let name = c.strip_prefix(AGENT_PREFIX)?.to_owned();
            Some(crate::meta::AgentSpec {
                is_manager: name == MANAGER_NAME,
                port: agent_web_port(&name),
                name,
            })
        })
        .collect();
    if let Some(extra) = name_to_add
        && !out.iter().any(|a| a.name == extra)
    {
        out.push(crate::meta::AgentSpec {
            is_manager: extra == MANAGER_NAME,
            port: agent_web_port(extra),
            name: extra.to_owned(),
        });
    }
    out.sort_by(|a, b| a.name.cmp(&b.name));
    Ok(out)
}

async fn agents_after_spawn(name: &str) -> Result<Vec<crate::meta::AgentSpec>> {
    agents_for_meta(Some(name)).await
}

/// Public enumeration of currently-existing agents (whatever
/// `nixos-container list` says), sorted, no extras. For callers
/// outside this module that need to reseed meta after lifecycle
/// changes — destroy, startup reconciliation, etc.
pub async fn agents_for_meta_listing() -> Result<Vec<crate::meta::AgentSpec>> {
    agents_for_meta(None).await
}

pub async fn kill(name: &str) -> Result<()> {
    validate(name)?;
    priv_run("stop", name).await
}

/// Start a container. Success is defined as the container's unit reaching
/// `active`, **not** the `nixos-container start` exit code — a start-job
/// timeout on a slow boot (e.g. DHCP taking ~10s) exits non-zero while
/// `container@<c>.service` keeps retrying and the container comes up seconds
/// later. So a zero exit returns immediately (already active), and a non-zero
/// exit polls the unit state for [`START_SETTLE_TIMEOUT`] before concluding
/// the start actually failed. Every caller (dashboard restart, reconcile
/// start, the cold-start fallback) gets this truth for free.
pub async fn start(name: &str) -> Result<()> {
    validate(name)?;
    if priv_run("start", name).await.is_ok() {
        return Ok(());
    }
    let container = container_name(name);
    if wait_until_running(name, START_SETTLE_TIMEOUT).await {
        tracing::info!(
            container = %container,
            "start exited non-zero but the container reached active on systemd's retry; treating as success"
        );
        return Ok(());
    }
    Err(anyhow::anyhow!(
        "container {container} did not reach active within {}s of start",
        START_SETTLE_TIMEOUT.as_secs()
    ))
}

/// Opaque token produced by [`converge_start_preamble`].
/// [`start_with_fallback`] requires this as proof that the pre-start
/// preamble (runtime dir + drop-ins) ran. Dropping the token without
/// calling `start_with_fallback` is a no-op.
#[must_use = "call lifecycle::start_with_fallback(token) to start the container"]
pub struct StartableAgent {
    name: String,
}

/// Run the per-agent start preamble: ensure the runtime dir exists and write
/// the nspawn / resource-limits drop-ins. Returns a [`StartableAgent`] token
/// as typed proof that the preamble ran; pass it to [`start_with_fallback`].
/// Callers that omit this step cannot call `start_with_fallback` — the type
/// system makes forgetting the preamble a compile error.
///
/// # Errors
///
/// Returns an error if `ensure_agent_runtime_dir` or `write_dropins` fails.
pub async fn converge_start_preamble(
    name: &str,
    hive: &HiveEnv,
    paths: &AgentPaths,
) -> Result<StartableAgent> {
    ensure_agent_runtime_dir(name)?;
    write_dropins(name, hive, paths).await?;
    Ok(StartableAgent {
        name: name.to_owned(),
    })
}

/// Start with the cold-start fallback: when a plain start fails (the
/// activation-error shape), retry once via stop + kill + start before
/// giving up. Used by the queue's fast-lane `Start` handler and the
/// inline start-after-rebuild path.
/// See `docs/coordinator.md::Cold-start fallback`.
///
/// Requires a [`StartableAgent`] token from [`converge_start_preamble`]
/// to prove the preamble ran. For internal use within this module (where
/// the preamble is already enforced structurally) call
/// `start_with_fallback_inner` directly.
///
/// # Errors
///
/// Propagates the retry's start error (annotated with the original
/// failure) when the fallback also fails.
pub async fn start_with_fallback(token: StartableAgent) -> Result<()> {
    start_with_fallback_inner(&token.name).await
}

/// How long to wait for a container's unit to reach `active` after a
/// `nixos-container start` that returned a non-zero exit. A start-job timeout
/// on a slow boot (e.g. DHCP taking ~10s) returns an error while
/// `container@<c>.service` keeps retrying and the container comes up seconds
/// later — so the exit code is NOT authoritative, the unit state is. 60s
/// comfortably covers an observed slow boot (the reported incident settled
/// ~16s after the "failure") without pinning a queue node on a genuine
/// never-boots failure for too long.
const START_SETTLE_TIMEOUT: std::time::Duration = std::time::Duration::from_mins(1);

/// Poll cadence while waiting for a container to reach `active`.
const START_SETTLE_POLL: std::time::Duration = std::time::Duration::from_secs(2);

/// Poll [`is_running`] until the container's unit is active or `timeout`
/// elapses; returns true as soon as it's active.
async fn wait_until_running(name: &str, timeout: std::time::Duration) -> bool {
    let deadline = tokio::time::Instant::now() + timeout;
    loop {
        if is_running(name).await {
            return true;
        }
        if tokio::time::Instant::now() >= deadline {
            return false;
        }
        tokio::time::sleep(START_SETTLE_POLL).await;
    }
}

/// Internal implementation of the cold-start fallback, behind
/// [`start_with_fallback`] (public, token-gated) — the inner form exists for
/// callers that have already run the drop-in preamble themselves.
///
/// [`start`] already treats unit-active (not the exit code) as success and
/// waits out a slow boot, so this only layers the activation-error recovery on
/// top: if the container is still down after `start`, tear it down and try
/// `start` once more. The teardown prefers a graceful `stop` and only
/// escalates to SIGKILL when that `stop` itself fails.
async fn start_with_fallback_inner(name: &str) -> Result<()> {
    validate(name)?;
    if start(name).await.is_ok() {
        return Ok(());
    }
    tracing::warn!(
        %name,
        "start did not bring the container up; hard-resetting (stop, kill only if that fails) and retrying once"
    );
    // Graceful `stop` first; only escalate to SIGKILL if the stop itself
    // fails (e.g. a wedged container whose `machinectl poweroff` never
    // completes). A clean stop is enough on its own — SIGKILL is the
    // last-resort teardown for when graceful shutdown can't finish.
    if let Err(stop_err) = priv_run("stop", name).await {
        tracing::warn!(
            %name,
            error = %stop_err,
            "graceful stop failed; escalating to SIGKILL"
        );
        priv_run("kill", name).await.unwrap_or_else(|e| {
            tracing::warn!(
                %name,
                error = %e,
                "kill after failed stop also failed (ignored)"
            );
        });
    }
    start(name)
        .await
        .with_context(|| format!("cold-start fallback also failed for {name}"))
}

/// True when the container's systemd unit is active. Used by the dashboard
/// to gate stop/restart buttons.
pub async fn is_running(name: &str) -> bool {
    let container = container_name(name);
    let unit = format!("container@{container}.service");
    Command::new("systemctl")
        .args(["is-active", "--quiet", &unit])
        .status()
        .await
        .is_ok_and(|s| s.success())
}

/// True when a hive infrastructure container's systemd unit is active.
/// Sibling of [`is_running`] for sub-agents, but infra container/unit names
/// (`hive-ci`, …) already have no `h-` prefix to strip, so this queries
/// `container@<unit_name>.service` directly rather than going through
/// [`container_name`]. Used by the dashboard C0R3 page's 1NFR4 sub-tab to
/// show each infra container's live status dot.
pub async fn infra_is_running(container: hive_priv_sock::InfraContainer) -> bool {
    let unit = format!("container@{}.service", container.unit_name());
    Command::new("systemctl")
        .args(["is-active", "--quiet", &unit])
        .status()
        .await
        .is_ok_and(|s| s.success())
}

/// Fully tear down a sub-agent's container: stop + remove via `nixos-container
/// destroy`, then clean our own systemd drop-in. Leaves it to the caller to
/// wipe `/var/lib/hyperhive/...` state and the per-agent runtime dir.
pub async fn destroy(name: &str) -> Result<()> {
    validate(name)?;
    let container = container_name(name);
    // nixos-container destroy handles stop + removal of /var/lib/nixos-containers/<C>
    // and /etc/nixos-containers/<C>.conf. Tolerate "no such container".
    if let Err(e) = priv_run("destroy", name).await {
        tracing::warn!(error = ?e, "nixos-container destroy returned an error; continuing cleanup");
    }
    // Remove the systemd resource-limits drop-in via hive-priv.
    if let Err(e) = crate::priv_client::remove_service_dropin(&container).await {
        tracing::warn!(error = ?e, "remove service drop-in failed (non-fatal)");
    }
    Ok(())
}

/// Pre-build `system.build.toplevel` against `meta#<name>` so the
/// subsequent `nixos-container update` finds the result cached and
/// skips straight to the profile-swap. Store-warming only — container
/// is untouched. See `docs/coordinator.md::Rebuild path` for why
/// the prebuild happens before stop, and `docs/coordinator.md::Prebuild
/// attr path` for why the explicit nixosConfigurations attr is required.
///
/// `on_build_log_id` fires with the `build_logs` row id as soon as the
/// row opens, so queue-side callers can link their node to the live
/// stream. Pass `&|_| ()` when not needed.
pub async fn prebuild_toplevel(
    name: &str,
    flake_ref: &str,
    on_build_log_id: &(dyn Fn(i64) + Send + Sync),
) -> Result<()> {
    use tokio::io::{AsyncBufReadExt, BufReader};
    // Split `<root>#<name>` so we can re-emit with the explicit
    // `nixosConfigurations.<name>` segment. The flake_ref shape is
    // constructed by the caller and always contains exactly one
    // `#`; `split_once` returning None here would be a programmer
    // error we'd want to surface loudly rather than paper over.
    let (flake_root, fragment) = flake_ref
        .split_once('#')
        .with_context(|| format!("flake_ref {flake_ref:?} missing '#<name>' fragment"))?;
    // Sanity-check the fragment matches the agent name we were
    // passed — guards against future calls that pass a divergent
    // pair (no current callsite does, but the pair is redundant
    // and worth checking once).
    if fragment != name {
        anyhow::bail!("prebuild_toplevel: flake_ref fragment '{fragment}' ≠ agent name '{name}'");
    }
    let attr = format!("{flake_root}#nixosConfigurations.{name}.config.system.build.toplevel");
    let args = vec![
        "--extra-experimental-features",
        "nix-command flakes",
        "build",
        "--no-link",
        "--print-out-paths",
        &attr,
    ];
    let cmdline = format!("nix {}", args.join(" "));
    tracing::info!(%name, %cmdline, "prebuild: warming system toplevel");

    // Open a build_logs row for this attempt (best-effort — None when
    // the global handle hasn't been installed, e.g. early startup
    // or standalone tests). Lines pumped from stdout/stderr append
    // into the row; `finish` lands the terminal status before we bail.
    let logs = crate::build_logs::global();
    let log_id = logs.as_ref().and_then(|h| {
        h.start(name, "prebuild", &cmdline)
            .map_err(|e| {
                tracing::warn!(error = ?e, "build_logs: start failed (prebuild log dropped)");
            })
            .ok()
    });
    if let Some(id) = log_id {
        on_build_log_id(id);
    }

    let mut child = Command::new("nix")
        .args(&args)
        .stdout(std::process::Stdio::piped())
        .stderr(std::process::Stdio::piped())
        .spawn()
        .with_context(|| format!("spawn {cmdline}"))?;

    let stdout = child.stdout.take().expect("piped stdout");
    let stderr = child.stderr.take().expect("piped stderr");

    let stdout_cmdline = cmdline.clone();
    let stdout_logs = logs.clone();
    let pump_stdout = tokio::spawn(async move {
        let mut lines = BufReader::new(stdout).lines();
        while let Ok(Some(line)) = lines.next_line().await {
            tracing::info!(target: "nix-prebuild", cmdline = %stdout_cmdline, "{line}");
            if let (Some(h), Some(id)) = (&stdout_logs, log_id) {
                h.append_stdout(id, &line);
            }
        }
    });

    let stderr_cmdline = cmdline.clone();
    let stderr_logs = logs.clone();
    let pump_stderr = tokio::spawn(async move {
        let mut lines = BufReader::new(stderr).lines();
        while let Ok(Some(line)) = lines.next_line().await {
            tracing::warn!(target: "nix-prebuild", cmdline = %stderr_cmdline, "{line}");
            if let (Some(h), Some(id)) = (&stderr_logs, log_id) {
                h.append_stderr(id, &line);
            }
        }
    });

    let status = child
        .wait()
        .await
        .with_context(|| format!("wait {cmdline}"))?;
    let _ = pump_stdout.await;
    let _ = pump_stderr.await;

    let ok = status.success();
    if let (Some(h), Some(id)) = (&logs, log_id) {
        h.finish(
            id,
            if ok {
                crate::build_logs::BuildStatus::Ok
            } else {
                crate::build_logs::BuildStatus::Fail
            },
        );
    }
    if !ok {
        match log_id {
            Some(id) => bail!("prebuild {cmdline} failed ({status}); see build log #{id}"),
            None => bail!("prebuild {cmdline} failed ({status})"),
        }
    }
    Ok(())
}

pub async fn list() -> Result<Vec<String>> {
    let stdout = crate::priv_client::list_containers().await?;
    Ok(stdout
        .lines()
        .map(str::trim)
        .filter(|line| line.starts_with(AGENT_PREFIX))
        .map(str::to_owned)
        .collect())
}

/// Sync `/etc/tmpfiles.d/hyperhive-agents.conf` with the currently-known
/// agent set (from `nixos-container list`). Strips the `h-` prefix to get
/// logical names. Best-effort: errors are logged but never propagated — a
/// failed tmpfiles write shouldn't block a spawn or destroy.
///
/// Called at hive-c0re startup and after each spawn / destroy so the file
/// always reflects the live agent set. `systemd-tmpfiles-setup.service`
/// reads the file at boot (before any container units start), pre-creating
/// bind-mount source dirs so container@h-* units don't race hive-c0re.
pub async fn sync_tmpfiles() {
    let agents = match list().await {
        Ok(containers) => containers
            .into_iter()
            .filter_map(|c| c.strip_prefix(AGENT_PREFIX).map(str::to_owned))
            .collect::<Vec<_>>(),
        Err(e) => {
            tracing::warn!(error = ?e, "sync_tmpfiles: list failed; skipping");
            return;
        }
    };
    if let Err(e) = crate::priv_client::sync_agent_tmpfiles(&agents).await {
        tracing::warn!(error = ?e, "sync_tmpfiles: priv call failed");
    } else {
        tracing::debug!(count = agents.len(), "sync_tmpfiles: ok");
    }
}

/// Ensure the per-agent runtime directory `/run/hyperhive/agents/<name>`
/// exists. The directory is also written by `SyncAgentTmpfiles` (run at
/// boot + spawn/destroy), but explicit creation in start/spawn paths guards
/// against races where hive-c0re starts a container before tmpfiles.d has
/// applied the new entry.
///
/// Pure filesystem op — no `Coordinator` dependency — so callers that only
/// need the dir do not have to hold an `Arc<Coordinator>`.
///
/// # Errors
/// Returns an error if `create_dir_all` fails.
pub fn ensure_agent_runtime_dir(name: &str) -> Result<()> {
    let dir = crate::paths::agent_runtime_dir(name);
    std::fs::create_dir_all(&dir)
        .with_context(|| format!("create agent runtime dir {}", dir.display()))
}

/// Build the per-line callback for `create_container_streaming` /
/// `update_container_streaming`. Both ops share identical dispatch logic
/// (stdout → info + `append_stdout`, stderr → warn + `append_stderr`); this
/// helper avoids duplicating that match body across the two call sites.
fn make_log_callback(
    logs: Option<std::sync::Arc<crate::build_logs::BuildLogs>>,
    log_id: Option<i64>,
    cmdline: String,
) -> impl FnMut(hive_priv_sock::PrivStream, &str) {
    use hive_priv_sock::PrivStream;
    move |stream, line| match stream {
        PrivStream::Stdout => {
            tracing::info!(target: "nixos-container", cmdline = %cmdline, "{line}");
            if let (Some(h), Some(id)) = (&logs, log_id) {
                h.append_stdout(id, line);
            }
        }
        PrivStream::Stderr => {
            tracing::warn!(target: "nixos-container", cmdline = %cmdline, "{line}");
            if let (Some(h), Some(id)) = (&logs, log_id) {
                h.append_stderr(id, line);
            }
        }
    }
}

/// Execute a container operation via hive-priv and integrate with
/// `build_logs.sqlite`. hive-priv runs as root and forwards output lines
/// to hive-c0re in real time via the streaming priv protocol. Each line
/// is appended to the build-log row as it arrives, so the dashboard
/// shows live progress during long `nixos-container create` / `update` runs.
async fn priv_run(kind: &str, name: &str) -> Result<()> {
    priv_run_inner(kind, name, None).await
}

/// Like `priv_run` but calls `on_log_id(log_id)` immediately after the
/// build-log row is opened — before the actual container op starts.
/// This lets callers surface the row id for live streaming (e.g. the
/// rebuild-queue worker sets `build_log_id` on the queue entry so the
/// dashboard can link to `/api/build-logs/id/{id}/stream`).
///
/// The callback fires only when a build-log row is successfully opened
/// (i.e. the global `BuildLogs` handle is installed AND `h.start()`
/// succeeds). No-op when `on_log_id` is `None` — that's the path for
/// all callers that don't need the id.
async fn priv_run_inner(
    kind: &str,
    name: &str,
    on_log_id: Option<&(dyn Fn(i64) + Send + Sync)>,
) -> Result<()> {
    let container = container_name(name);
    let cmdline = format!("nixos-container {kind} {container}");

    let logs = crate::build_logs::global();
    let log_id = logs.as_ref().and_then(|h| {
        h.start(name, kind, &cmdline)
            .map_err(|e| {
                tracing::warn!(error = ?e, "build_logs: start failed (priv_run log dropped)");
            })
            .ok()
    });
    // Notify the caller as soon as the log row exists so it can surface
    // the id for live streaming before the container op even starts.
    if let (Some(id), Some(cb)) = (log_id, on_log_id) {
        cb(id);
    }

    // For long-running ops use the streaming protocol so build_logs
    // receives lines in real time rather than as a batch at completion.
    let result: Result<()> = match kind {
        "create" => {
            crate::priv_client::create_container_streaming(
                name,
                make_log_callback(logs.clone(), log_id, cmdline.clone()),
            )
            .await
        }
        "update" => {
            crate::priv_client::update_container_streaming(
                name,
                make_log_callback(logs.clone(), log_id, cmdline.clone()),
            )
            .await
        }
        "start" => crate::priv_client::start_container(name).await,
        "stop" => crate::priv_client::stop_container(name).await,
        "kill" => crate::priv_client::kill_container(name).await,
        "destroy" => crate::priv_client::destroy_container(name).await,
        other => Err(anyhow::anyhow!("unknown container op: {other}")),
    };

    let succeeded = result.is_ok();
    if let (Some(h), Some(id)) = (&logs, log_id) {
        h.finish(
            id,
            if succeeded {
                crate::build_logs::BuildStatus::Ok
            } else {
                crate::build_logs::BuildStatus::Fail
            },
        );
    }

    match result {
        Ok(()) => Ok(()),
        Err(e) => {
            let journal = if kind == "update" {
                container_journal_tail(&container).await
            } else {
                String::new()
            };
            match log_id {
                Some(id) => bail!("{e:#}; see build log #{id}{journal}"),
                None => bail!("{e:#}{journal}"),
            }
        }
    }
}

/// On a failed `nixos-container update`, the stderr nixos-container
/// itself prints is often terse ("failed to reload container") — the
/// real reason (which unit failed `switch-to-configuration` during
/// the reload phase) lands in the *container's* own journal, not on
/// the host. Fetch the tail of it so a failed rebuild self-documents
/// the failing unit in the error string, no second round-trip.
///
/// Scoped to `update`: that's the reload-phase case, and the
/// container is still up (running the old generation) so
/// `journalctl -M` works. Best-effort — returns "" for other verbs
/// or when the journal can't be read (machine gone, journalctl
/// missing); it never produces an error of its own.
async fn container_journal_tail(container: &str) -> String {
    // `-M` enters the container namespace and needs root, so the read
    // is delegated to hive-priv (hive-c0re itself runs unprivileged).
    let res = crate::priv_client::read_container_journal(
        container,
        hive_priv_sock::JournalQuery {
            lines: 40,
            ..Default::default()
        },
    )
    .await;
    match res {
        Ok((stdout, _)) if !stdout.is_empty() => format!(
            "\n--- last 40 journal lines from container '{container}' ---\n{}",
            stdout.trim_end()
        ),
        _ => String::new(),
    }
}