hyperhive/hive-priv/src/container.rs

//! Container lifecycle ops: `StartContainer`, `StopContainer`,
//! `KillContainer`, `CreateContainer`, `UpdateContainer`, `DestroyContainer`
//! and `ListContainers`. Every op takes a logical agent name and reaches only
//! the `h-`-prefixed container [`container_system_name`] derives from it.
//!
//! `CreateContainer` / `UpdateContainer` build the toplevel here
//! ([`nix_build_toplevel`]) and stream progress to the caller as
//! [`PrivEvent::Line`]s when asked to.

use anyhow::{Context as _, Result, anyhow, bail};
use hive_priv_sock::{PrivEvent, PrivStream, PrivStreamLine, meta_flake_url, valid_meta_rev};
use tokio::io::{AsyncWriteExt, BufReader};
use tokio::net::unix::OwnedWriteHalf;
use tokio::process::Command;

use crate::shared::{container_system_name, validate_agent_name};

/// Write one `PrivEvent::Line` to the client. Best-effort: a write
/// failure is logged but doesn't abort the running subprocess.
async fn write_line_event(writer: &mut OwnedWriteHalf, stream: PrivStream, data: &str) {
    let event = PrivEvent::Line(PrivStreamLine {
        stream,
        data: data.to_owned(),
    });
    if let Ok(mut json) = serde_json::to_string(&event) {
        json.push('\n');
        if let Err(e) = writer.write_all(json.as_bytes()).await {
            tracing::warn!(error = %e, "write_line_event: write failed");
        }
    }
}

/// `StartContainer`: clear any start-limit lockout left by earlier failures
/// so a now-correct start isn't blocked (`nixos-container start` does not do
/// this itself). Best-effort — if the unit doesn't exist yet (first-time
/// create), `reset-failed` is a no-op and the start proceeds regardless.
pub(crate) async fn start_container(name: &str) -> Result<(String, String)> {
    validate_agent_name(name)?;
    let machine = container_system_name(name);
    // Silenced: a first-time create means the unit doesn't exist yet, and
    // `systemctl`'s own "Unit ... not loaded" complaint for that case would
    // otherwise land in the journal at error priority on every such start —
    // routine, not a signal, since the result is already discarded below.
    let _ = Command::new("systemctl")
        .args(["reset-failed", &format!("container@{machine}.service")])
        .stdout(std::process::Stdio::null())
        .stderr(std::process::Stdio::null())
        .status()
        .await;
    container_run(&["start", &machine]).await
}

/// `StopContainer`.
pub(crate) async fn stop_container(name: &str) -> Result<(String, String)> {
    validate_agent_name(name)?;
    stop_and_release(&container_system_name(name)).await
}

/// `KillContainer`: `nixos-container` has no kill verb, so this uses
/// `machinectl` to send `SIGKILL` to every process in the container — the
/// right semantics for a forced shutdown after a graceful stop has already
/// been attempted.
pub(crate) async fn kill_container(name: &str) -> Result<(String, String)> {
    validate_agent_name(name)?;
    let machine = container_system_name(name);
    machinectl_run(&["kill", &machine, "--signal=SIGKILL"]).await
}

/// `DestroyContainer`.
pub(crate) async fn destroy_container(name: &str) -> Result<(String, String)> {
    validate_agent_name(name)?;
    container_run(&["destroy", &container_system_name(name)]).await
}

/// Shared body for `CreateContainer` / `UpdateContainer`: validate the
/// name, build the toplevel ourselves, and apply it (streaming line
/// events to `writer` when `stream` is set).
///
/// **Both verbs build explicitly now — not just `update`.** The first cut
/// of this fix only rewrote `update`, reasoning that `create` was already
/// safe: it wraps its whole action in an exclusive `flock` before calling
/// `nixos-container`'s own `buildFlake()`, and once `update` stopped
/// writing to `buildFlake()`'s shared `.tmp` out-link, concurrent
/// `create`s were the only remaining writers — mutually excluded by that
/// lock. True, but it leaves `create`'s safety resting on an internal
/// implementation detail of a script we don't own (its current locking
/// behavior, which could change upstream without notice) instead of on
/// something we control. Building here for both verbs removes `buildFlake()`
/// from the picture entirely — there's no shared `.tmp` left to race on,
/// so there's nothing left to reason about staying in sync with.
///
/// **`update` no longer calls `nixos-container update` at all.** That
/// action's own version-compat probe runs unconditionally before
/// `--system-path` is ever honored, dying on every agent's update.
/// [`swap_container_profile`] replicates exactly what `update`'s own
/// action does *past* that probe (confirmed against its source): `nix-env
/// --set` the per-container profile, then `systemctl reload` if the
/// container is running. `create` is untouched — it isn't the failing
/// verb.
pub(crate) async fn container_flake_action(
    verb: &str,
    name: &str,
    meta_rev: &str,
    stream: bool,
    writer: &mut OwnedWriteHalf,
) -> Result<(String, String)> {
    validate_agent_name(name)?;
    // The build is the multi-minute phase of this operation — give it the
    // same live-line treatment `container_run_streaming` gives
    // `nixos-container` itself when the caller asked for it. Without
    // this, moving the build out of the streamed `nixos-container` call
    // (which is the whole point of this fix) would silently regress every
    // UI that shows build progress: nothing until the longest phase
    // finishes, then everything at once.
    let toplevel = nix_build_toplevel(name, meta_rev, stream.then_some(&mut *writer)).await?;
    let system_name = container_system_name(name);
    if verb == "update" {
        return swap_container_profile(&system_name, &toplevel, stream, writer).await;
    }
    let args = [verb, &system_name, "--system-path", &toplevel];
    if stream {
        container_run_streaming(&args, writer).await
    } else {
        container_run(&args).await
    }
}

/// Apply a prebuilt toplevel to an existing container directly, without
/// going through `nixos-container update` (see `container_flake_action`'s
/// doc comment for why). Mirrors `nixos-container.pl`'s own `update`
/// action verbatim, past its version probe: point the per-container `nix-
/// env` profile at the new toplevel, then reload the container unit *if*
/// it's currently running — the container's own next start already reads
/// from the profile, so a stopped container needs nothing further.
///
/// Both steps are near-instant in the happy case, but still stream-
/// forwarded like the rest of this operation: a failure here (e.g. a
/// wedged `systemctl reload`) is exactly when the caller most wants the
/// live line in the build log, not just a summary error afterward.
async fn swap_container_profile(
    system_name: &str,
    toplevel: &str,
    stream: bool,
    writer: &mut OwnedWriteHalf,
) -> Result<(String, String)> {
    let profile = format!("/nix/var/nix/profiles/per-container/{system_name}/system");
    let set_out = Command::new("nix-env")
        .args(["-p", &profile, "--set", toplevel])
        .output()
        .await
        .context("invoke nix-env --set")?;
    let mut stdout = String::from_utf8_lossy(&set_out.stdout).into_owned();
    let mut stderr = String::from_utf8_lossy(&set_out.stderr).into_owned();
    log_and_forward(&stdout, &stderr, stream, writer).await;
    if !set_out.status.success() {
        bail!(
            "nix-env -p {profile} --set failed ({}): {}",
            set_out.status,
            stderr.trim()
        );
    }

    let unit = format!("container@{system_name}");
    let state_out = Command::new("systemctl")
        .args(["show", "--property=ActiveState", "--value", &unit])
        .output()
        .await
        .context("query container ActiveState")?;
    let active = String::from_utf8_lossy(&state_out.stdout).trim() == "active";
    if active {
        let reload_out = Command::new("systemctl")
            .args(["reload", &unit])
            .output()
            .await
            .context("invoke systemctl reload")?;
        let r_stdout = String::from_utf8_lossy(&reload_out.stdout).into_owned();
        let r_stderr = String::from_utf8_lossy(&reload_out.stderr).into_owned();
        log_and_forward(&r_stdout, &r_stderr, stream, writer).await;
        if !reload_out.status.success() {
            bail!(
                "systemctl reload {unit} failed ({}): {}",
                reload_out.status,
                r_stderr.trim()
            );
        }
        stdout.push_str(&r_stdout);
        stderr.push_str(&r_stderr);
    }
    Ok((stdout, stderr))
}

/// Log a command's captured stdout/stderr the same way every other
/// `nixos-container`-adjacent shellout in this file does, and — when
/// `stream` is set — forward each line to the client as a live
/// [`PrivEvent::Line`] too, so a caller watching the build log sees these
/// lines exactly like any other step's, not just a summary on failure.
async fn log_and_forward(stdout: &str, stderr: &str, stream: bool, writer: &mut OwnedWriteHalf) {
    for line in stdout.lines() {
        tracing::info!(target: "nixos-container", "{line}");
        if stream {
            write_line_event(writer, PrivStream::Stdout, line).await;
        }
    }
    for line in stderr.lines() {
        tracing::warn!(target: "nixos-container", "{line}");
        if stream {
            write_line_event(writer, PrivStream::Stderr, line).await;
        }
    }
}

/// The environment variable naming the forge's `http` origin, which
/// `internal/meta` is fetched from. Set by the unit; never taken from a
/// request.
pub(crate) const FORGE_URL_ENV: &str = "HIVE_PRIV_FORGE_URL";

/// The flake agent toplevels are built from: `internal/meta` at `rev`. The URL
/// is `forge_base`, hive-priv's own; only the rev comes from the request, and
/// it is checked here.
fn meta_flake(rev: &str, forge_base: Option<&str>) -> Result<String> {
    if !valid_meta_rev(rev) {
        bail!("meta_rev {rev:?} is not a 40-digit lowercase hex commit id");
    }
    let base = forge_base.filter(|b| !b.is_empty()).with_context(|| {
        format!("{FORGE_URL_ENV} is unset, so there is no internal/meta to build from")
    })?;
    Ok(meta_flake_url(base, rev))
}

/// The explicit `nixosConfigurations.<name>.config.system.build.toplevel`
/// flake attr path — same construction `hive-c0re`'s own
/// `lifecycle::prebuild_toplevel` uses, kept here as a pure function so
/// the exact string shape is unit-tested without needing to run `nix`.
fn toplevel_attr(flake: &str, name: &str) -> String {
    format!("{flake}#nixosConfigurations.{name}.config.system.build.toplevel")
}

/// Wall-clock bound on [`nix_build_toplevel`]. Four times CI's observed
/// cold-cache `nix flake check` (~15 min, `.forgejo/workflows/ci.yml`), which
/// compiles this workspace's crates, the part of an agent toplevel no binary
/// cache holds; the rebuild path has usually warmed this exact attr already
/// (`hive-c0re`'s `prebuild_toplevel`). A stalled download is bounded by nix's
/// own `stalled-download-timeout`; this covers the rest, such as a wedged
/// nix-daemon holding the caller's build slot forever.
const NIX_BUILD_TIMEOUT: std::time::Duration = std::time::Duration::from_hours(1);

/// How a [`run_bounded`] child ended.
enum BoundedRun {
    Exited {
        status: std::process::ExitStatus,
        stdout: String,
        stderr: String,
    },
    /// Still running at the limit; its whole process group was killed.
    TimedOut,
}

/// Run `cmd` to completion or until `limit`, draining both pipes: stdout is
/// captured silently, stderr is logged and, when `writer` is set, forwarded
/// live as [`PrivEvent::Line`]s.
///
/// The child leads its own process group and a timeout `SIGKILL`s the whole
/// group, so a grandchild still holding the pipes (or the daemon connection)
/// dies with it instead of outliving the error.
async fn run_bounded(
    mut cmd: Command,
    limit: std::time::Duration,
    mut writer: Option<&mut OwnedWriteHalf>,
) -> std::io::Result<BoundedRun> {
    use tokio::io::AsyncBufReadExt as _;

    let mut child = cmd
        .stdout(std::process::Stdio::piped())
        .stderr(std::process::Stdio::piped())
        .process_group(0)
        .spawn()?;
    // Group leader, so its pid is the pgid.
    let pgid = child.id().and_then(|pid| i32::try_from(pid).ok());

    let stdout = child.stdout.take().expect("stdout piped");
    let stderr = child.stderr.take().expect("stderr piped");
    let run = async {
        let mut stdout_lines = BufReader::new(stdout).lines();
        let mut stderr_lines = BufReader::new(stderr).lines();
        let mut stdout_buf = String::new();
        let mut stderr_buf = String::new();

        // ⚠️ Both pipes are drained concurrently even though only one is
        // streamed: reading stderr alone would let stdout fill its pipe
        // buffer and deadlock the child on a build with enough stdout output
        // to fill it.
        //
        // ⚠️ Each stream's EOF is tracked separately rather than breaking on
        // the first `None`: `next_line()` on a closed stream returns
        // `Ok(None)` immediately and forever, so a loop that keeps polling a
        // finished stream spins hot until the other one ends too.
        let mut stdout_done = false;
        let mut stderr_done = false;
        while !(stdout_done && stderr_done) {
            tokio::select! {
                line = stdout_lines.next_line(), if !stdout_done => {
                    match line {
                        // Captured, never streamed — this is the store path.
                        Ok(Some(l)) => {
                            stdout_buf.push_str(&l);
                            stdout_buf.push('\n');
                        }
                        Ok(None) => stdout_done = true,
                        Err(e) => {
                            tracing::warn!(error = %e, "nix build stdout read error");
                            stdout_done = true;
                        }
                    }
                }
                line = stderr_lines.next_line(), if !stderr_done => {
                    match line {
                        // Streamed as it arrives, so the dashboard and
                        // `journalctl -f` show build progress live.
                        Ok(Some(l)) => {
                            tracing::info!(target: "nix-build-toplevel", "{l}");
                            if let Some(w) = writer.as_deref_mut() {
                                write_line_event(w, PrivStream::Stderr, &l).await;
                            }
                            if !stderr_buf.is_empty() {
                                stderr_buf.push('\n');
                            }
                            stderr_buf.push_str(&l);
                        }
                        Ok(None) => stderr_done = true,
                        Err(e) => {
                            tracing::warn!(error = %e, "nix build stderr read error");
                            stderr_done = true;
                        }
                    }
                }
            }
        }
        let status = child.wait().await?;
        Ok::<_, std::io::Error>(BoundedRun::Exited {
            status,
            stdout: stdout_buf,
            stderr: stderr_buf,
        })
    };
    if let Ok(exited) = tokio::time::timeout(limit, run).await {
        return exited;
    }
    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?;
    Ok(BoundedRun::TimedOut)
}

/// Build `nixosConfigurations.<name>.config.system.build.toplevel` and
/// return the resulting store path, so `create`/`update` can hand
/// `nixos-container` an explicit `--system-path` instead of letting its
/// own `buildFlake()` build to a racy shared out-link. See "Container
/// toplevel builds" in this crate's README for the full story — the
/// concurrency bug this closes (the "agent container gets closure of
/// other agent" mystery bug) and why stdout/stderr are drained
/// concurrently but handled asymmetrically (stderr streamed live,
/// stdout captured and required to be exactly one line).
///
/// `writer` is `None` for the non-streaming call shape (`stream: false`);
/// stderr still logs to journald either way, just without the
/// `PrivEvent::Line` forwarding. Bounded by [`NIX_BUILD_TIMEOUT`].
async fn nix_build_toplevel(
    name: &str,
    meta_rev: &str,
    writer: Option<&mut OwnedWriteHalf>,
) -> Result<String> {
    let flake = meta_flake(meta_rev, std::env::var(FORGE_URL_ENV).ok().as_deref())?;
    let attr = toplevel_attr(&flake, name);
    let mut cmd = Command::new("nix");
    cmd.args([
        "--extra-experimental-features",
        "nix-command flakes",
        "build",
        "--no-link",
        "--print-out-paths",
        &attr,
    ]);
    let (status, stdout_buf, stderr_buf) = match run_bounded(cmd, NIX_BUILD_TIMEOUT, writer)
        .await
        .with_context(|| format!("run nix build {attr}"))?
    {
        BoundedRun::Exited {
            status,
            stdout,
            stderr,
        } => (status, stdout, stderr),
        BoundedRun::TimedOut => bail!(
            "nix build {attr} timed out after {}s; killed it",
            NIX_BUILD_TIMEOUT.as_secs()
        ),
    };

    // ⚠️ Success is decided by the exit status, not by "we parsed a
    // path" — a build can print to stdout and still fail.
    if !status.success() {
        bail!(
            "nix build {attr} failed ({status}): {}",
            stderr_buf.lines().last().unwrap_or("").trim()
        );
    }
    single_output_path(&stdout_buf)
        .map(str::to_owned)
        .map_err(|count| {
            anyhow!("nix build {attr} produced {count} output path(s), expected exactly 1: {stdout_buf:?}")
        })
}

/// Parse `nix build --print-out-paths`' stdout down to the single output
/// path this function's caller expects. `--print-out-paths` prints one
/// line *per output*, not one line total (`nix build --no-link
/// --print-out-paths nixpkgs#openssl` prints two: `…-bin`, `…-man`);
/// `config.system.build.toplevel` is single-output today, so this is one
/// line in practice — but a bare whole-buffer `.trim()` would silently
/// hand a multi-line string on to `--system-path` the day that ever
/// changes, the same corrupted-argument failure this function exists to
/// avoid. Trims and drops empty lines *before* counting, so a lone
/// `"\n"` (or trailing whitespace on the real line) can't be mistaken
/// for a present-but-blank path — see the unit tests below for the exact
/// table this closes. `Err` carries the surviving line count, for the
/// caller's error message.
fn single_output_path(stdout: &str) -> Result<&str, usize> {
    let lines: Vec<&str> = stdout
        .lines()
        .map(str::trim)
        .filter(|l| !l.is_empty())
        .collect();
    match lines[..] {
        [path] => Ok(path),
        _ => Err(lines.len()),
    }
}

/// Invoke `nixos-container` with the given args, log output to journald.
pub(crate) async fn container_run(args: &[&str]) -> Result<(String, String)> {
    let out = Command::new("nixos-container")
        .args(args)
        .output()
        .await
        .context("invoke nixos-container")?;
    let stdout = String::from_utf8_lossy(&out.stdout).into_owned();
    let stderr = String::from_utf8_lossy(&out.stderr).into_owned();
    // `list` is a read-only enumeration called on the hot path (dashboard
    // rescan, forge + boot sweeps) — its stdout is the return value, not
    // progress, so logging every container name on every call floods the
    // journal. Log stdout only for the mutating ops, where each line is
    // genuine progress. stderr is always logged (errors matter regardless).
    if args.first() != Some(&"list") {
        for line in stdout.lines() {
            tracing::info!(target: "nixos-container", "{line}");
        }
    }
    for line in stderr.lines() {
        tracing::warn!(target: "nixos-container", "{line}");
    }
    if !out.status.success() {
        bail!(
            "nixos-container {} failed ({}): {}",
            args.join(" "),
            out.status,
            stderr.trim()
        );
    }
    Ok((stdout, stderr))
}

/// Invoke `machinectl` with the given args, log output to journald.
/// Used for operations that nixos-container doesn't expose (e.g. sending
/// signals to running containers).
async fn machinectl_run(args: &[&str]) -> Result<(String, String)> {
    let out = Command::new("machinectl")
        .args(args)
        .output()
        .await
        .context("invoke machinectl")?;
    let stdout = String::from_utf8_lossy(&out.stdout).into_owned();
    let stderr = String::from_utf8_lossy(&out.stderr).into_owned();
    for line in stdout.lines() {
        tracing::info!(target: "machinectl", "{line}");
    }
    for line in stderr.lines() {
        tracing::warn!(target: "machinectl", "{line}");
    }
    if !out.status.success() {
        bail!(
            "machinectl {} failed ({}): {}",
            args.join(" "),
            out.status,
            stderr.trim()
        );
    }
    Ok((stdout, stderr))
}

/// How long to wait for machined to drop a machine's registration after a
/// shutdown has been asked for. Generous: a container with slow-stopping
/// units legitimately takes a while, and escalating early would SIGKILL a
/// shutdown that was going to finish on its own.
const NAME_RELEASE_TIMEOUT: std::time::Duration = std::time::Duration::from_secs(20);

/// Poll cadence while waiting for the registration to go away.
const NAME_RELEASE_POLL: std::time::Duration = std::time::Duration::from_millis(500);

/// Cap on a single registration probe. The probe is a D-Bus round trip to
/// machined; if machined itself is wedged the call would otherwise sit on
/// the D-Bus method timeout, which is far longer than the whole stop
/// sequence should take.
const MACHINE_PROBE_TIMEOUT: std::time::Duration = std::time::Duration::from_secs(5);

/// True when machined still holds a registration for `machine`.
///
/// This asks machined the same question the registration itself answers —
/// `machinectl show` resolves the name through `GetMachine`, the very lookup
/// that makes a later boot fail with `Failed to register machine: already
/// exists`. So a positive answer here is exactly the condition that breaks
/// the next start, not a proxy for it.
///
/// Deliberately NOT the container's systemd unit state: the unit can be
/// `inactive` while the registration is still held, and that gap is the
/// whole bug this probe exists to catch.
///
/// Errors and timeouts answer "still registered". Being wrong that way costs
/// a redundant SIGKILL to something already gone; being wrong the other way
/// hands back a stop that silently leaked the name.
async fn machine_registered(machine: &str) -> bool {
    let probe = Command::new("machinectl")
        .args(["show", machine, "--property=Name"])
        // Don't leave a probe behind when the timeout below fires.
        .kill_on_drop(true)
        .output();
    match tokio::time::timeout(MACHINE_PROBE_TIMEOUT, probe).await {
        Ok(Ok(out)) => out.status.success(),
        Ok(Err(e)) => {
            tracing::warn!(%machine, error = %e, "machinectl show failed to run; assuming still registered");
            true
        }
        Err(_) => {
            tracing::warn!(%machine, "machinectl show timed out; assuming still registered");
            true
        }
    }
}

/// Poll [`machine_registered`] until the name is free or the timeout expires.
/// Returns true once the registration is gone.
async fn wait_name_released(machine: &str) -> bool {
    let deadline = tokio::time::Instant::now() + NAME_RELEASE_TIMEOUT;
    loop {
        if !machine_registered(machine).await {
            return true;
        }
        if tokio::time::Instant::now() >= deadline {
            return false;
        }
        tokio::time::sleep(NAME_RELEASE_POLL).await;
    }
}

/// Stop `machine` and only report success once machined has actually released
/// the name.
///
/// `nixos-container stop` exiting 0 does not mean the machine is gone. A
/// process that sits in the machine's cgroup without being a child of the
/// container's init — a shell exec'd in from outside, say — never receives
/// the shutdown's SIGTERM if it has been stopped with SIGSTOP, so the
/// registration outlives the "successful" stop. Every later start of that
/// container then fails with `Failed to register machine: already exists`,
/// and machined re-persists the stale record across its own restart, so
/// there is no cleaning it up after the fact. The only place to catch it is
/// here, in the stop.
///
/// So: ask for the stop, wait for the name, and fail loudly if it is still
/// held — a caller that is told the stop worked will go on to start the
/// container and hit the confusing registration error instead of this one.
///
/// This used to escalate to `machinectl kill --signal=SIGKILL` here. Dropped
/// per real incident data: the case this guards is a container genuinely
/// wedged (e.g. a root-login process survived the stop), and SIGKILL
/// doesn't recover that in practice — only a host-level reboot has.
/// Pretending a kill attempt handled it hides a condition that needs a human
/// to look at the host, so this now just reports the failure instead of
/// quietly (and ineffectually) trying to force it.
///
/// The verify-then-fail lives in the helper rather than at a call site so
/// that every stop gets it: dashboard, reconcile, destroy, cold-start
/// fallback. The start path already distrusts its own exit code the same way;
/// this is the missing half of that pair.
async fn stop_and_release(machine: &str) -> Result<(String, String)> {
    let stop = container_run(&["stop", machine]).await;
    if wait_name_released(machine).await {
        // Happy path, and also the path where a stop that reported failure
        // nonetheless brought the machine down. Either way the caller gets
        // the original result untouched.
        return stop;
    }

    tracing::error!(
        %machine,
        "stop finished but machined still holds the registration — the \
         container is likely wedged (a process outside its init tree \
         survived the stop); this needs a host-level look, not another \
         stop attempt"
    );
    bail!(
        "stop {machine}: machined still holds the machine name after {}s. \
         This container is likely wedged and starting it again will fail to \
         register — needs host-level intervention (a reboot has been the \
         only reliable fix in practice).",
        NAME_RELEASE_TIMEOUT.as_secs()
    )
}

/// Invoke `nixos-container` with the given args and forward output lines
/// to the caller as `PrivEvent::Line` messages in real time, logging each
/// line to journald as it arrives. Returns `(String::new(), String::new())`
/// on success (all output was streamed); the error string includes stderr
/// tail on failure.
async fn container_run_streaming(
    args: &[&str],
    writer: &mut OwnedWriteHalf,
) -> Result<(String, String)> {
    use tokio::io::AsyncBufReadExt as _;
    use tokio::process::Command;

    let mut child = Command::new("nixos-container")
        .args(args)
        .stdout(std::process::Stdio::piped())
        .stderr(std::process::Stdio::piped())
        .spawn()
        .context("invoke nixos-container (streaming)")?;

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

    let mut stdout_lines = BufReader::new(stdout).lines();
    let mut stderr_lines = BufReader::new(stderr).lines();

    // Collect stderr for the error message; stream both to the client.
    let mut stderr_buf = String::new();

    // Drive stdout and stderr concurrently. `tokio::select!` interleaves
    // them without bias — both streams drain at roughly the same rate
    // as the subprocess produces output.
    loop {
        tokio::select! {
            line = stdout_lines.next_line() => {
                match line {
                    Ok(Some(l)) => {
                        tracing::info!(target: "nixos-container", "{l}");
                        write_line_event(writer, PrivStream::Stdout, &l).await;
                    }
                    Ok(None) => break,
                    Err(e) => {
                        tracing::warn!(error = %e, "nixos-container stdout read error");
                        break;
                    }
                }
            }
            line = stderr_lines.next_line() => {
                match line {
                    Ok(Some(l)) => {
                        tracing::warn!(target: "nixos-container", "{l}");
                        write_line_event(writer, PrivStream::Stderr, &l).await;
                        if !stderr_buf.is_empty() {
                            stderr_buf.push('\n');
                        }
                        stderr_buf.push_str(&l);
                    }
                    Ok(None) => {}
                    Err(e) => {
                        tracing::warn!(error = %e, "nixos-container stderr read error");
                    }
                }
            }
        }
    }

    // Drain any remaining stderr after stdout closed.
    while let Ok(Some(l)) = stderr_lines.next_line().await {
        tracing::warn!(target: "nixos-container", "{l}");
        write_line_event(writer, PrivStream::Stderr, &l).await;
        if !stderr_buf.is_empty() {
            stderr_buf.push('\n');
        }
        stderr_buf.push_str(&l);
    }

    let status = child.wait().await.context("wait nixos-container")?;
    if !status.success() {
        // Only the last stderr line is embedded — the full stderr was
        // already forwarded line-by-line as PrivEvent::Line messages and
        // is captured in build_logs.sqlite by the caller. Keeping the
        // error message short avoids bloating the anyhow chain.
        bail!(
            "nixos-container {} failed ({}): {}",
            args.join(" "),
            status,
            stderr_buf.lines().last().unwrap_or("").trim()
        );
    }
    Ok((String::new(), String::new()))
}
#[cfg(test)]
mod tests {
    use super::{
        BoundedRun, FORGE_URL_ENV, meta_flake, run_bounded, single_output_path, toplevel_attr,
    };
    use crate::shared::test_support::scratch;

    /// Pins the exact attr path we hand to `nix build` — it has to match
    /// `hive-c0re`'s own `lifecycle::prebuild_toplevel` construction, since
    /// that step's whole point is warming the store for this later build.
    /// A drifted attr path defeats the cache-warming silently — no error,
    /// just a slower `update`. The flake is `internal/meta` at the rev, and its
    /// URL is the unit's, never the request's.
    #[test]
    fn a_meta_rev_builds_from_internal_meta_at_the_units_forge() {
        let rev = "c".repeat(40);
        let forge = Some("http://forge.t.local");
        assert_eq!(
            toplevel_attr(&meta_flake(&rev, forge).unwrap(), "atlas"),
            format!(
                "git+http://forge.t.local/internal/meta.git?rev={rev}\
                 #nixosConfigurations.atlas.config.system.build.toplevel"
            )
        );
    }

    /// Nothing but a commit id gets into the ref, whatever the request says.
    #[test]
    fn a_meta_rev_that_is_not_a_commit_id_is_refused() {
        let forge = Some("http://forge.t.local");
        for bad in [
            "main",
            "",
            "../../x",
            &format!("{}#evil", "a".repeat(34)),
            &format!("{}&ref=x", "a".repeat(34)),
        ] {
            let err = meta_flake(bad, forge).unwrap_err();
            assert!(
                format!("{err:#}").contains("not a 40-digit"),
                "{bad}: {err:#}"
            );
        }
    }

    #[test]
    fn a_meta_rev_without_a_forge_in_the_unit_is_refused() {
        let rev = "c".repeat(40);
        for forge in [None, Some("")] {
            let err = meta_flake(&rev, forge).unwrap_err();
            assert!(format!("{err:#}").contains(FORGE_URL_ENV), "{err:#}");
        }
    }

    /// `--print-out-paths`' exact per-shape table (measured against real
    /// `rustc` semantics, not just reasoned about) — a lone `"\n"` and a
    /// trailing-space path are the two shapes a bare `.lines().collect()`
    /// gets wrong, both catchable only by trimming + dropping empties
    /// before counting rather than after.
    #[test]
    fn single_output_path_rejects_blank_and_trims_whitespace() {
        assert_eq!(single_output_path(""), Err(0));
        assert_eq!(single_output_path("\n"), Err(0));
        assert_eq!(single_output_path("\n\n"), Err(0));
        assert_eq!(single_output_path("/nix/store/abc\n"), Ok("/nix/store/abc"));
        assert_eq!(
            single_output_path("/nix/store/abc \n"),
            Ok("/nix/store/abc")
        );
        assert_eq!(
            single_output_path("/nix/store/abc\r\n"),
            Ok("/nix/store/abc")
        );
        assert_eq!(
            single_output_path("/nix/store/abc\n/nix/store/def\n"),
            Err(2)
        );
    }

    /// Whether `pid` is gone, a zombie (`Z`), or being reaped (`X`): none of
    /// these runs again, and a process passes through them in that order,
    /// so once this is true it stays true.
    fn is_dead(pid: &str) -> bool {
        match std::fs::read_to_string(format!("/proc/{pid}/stat")) {
            Err(_) => true,
            Ok(stat) => stat_is_dead(&stat),
        }
    }

    /// The state letter of a `/proc/<pid>/stat` line, which follows the
    /// parenthesised command name, is `Z` or `X`.
    fn stat_is_dead(stat: &str) -> bool {
        stat.rsplit_once(") ")
            .is_some_and(|(_, rest)| rest.starts_with(['Z', 'X']))
    }

    #[test]
    fn zombie_and_reaped_states_count_as_dead() {
        assert!(stat_is_dead("42 (sleep) Z 1 42 42"));
        assert!(stat_is_dead("42 (sleep) X 1 42 42"));
        assert!(!stat_is_dead("42 (sleep) S 1 42 42"));
        assert!(!stat_is_dead("42 (sleep) R 1 42 42"));
    }

    /// A child still running at the limit comes back `TimedOut`, and the kill
    /// reaches its whole process group: the backgrounded grandchild dies too,
    /// not just the shell that started it.
    #[tokio::test]
    async fn a_build_past_its_limit_is_killed_with_its_group() {
        use std::time::{Duration, Instant};
        let dir = scratch();
        let pidfile = dir.join("grandchild");
        let mut cmd = tokio::process::Command::new("sh");
        cmd.arg("-c")
            .arg(format!("sleep 600 & echo $! > {}; wait", pidfile.display()));

        let started = Instant::now();
        let run = run_bounded(cmd, Duration::from_secs(2), None)
            .await
            .unwrap();
        assert!(matches!(run, BoundedRun::TimedOut), "must time out");
        assert!(
            started.elapsed() < Duration::from_mins(1),
            "the limit did not bound the run"
        );

        let pid = std::fs::read_to_string(&pidfile).unwrap();
        let pid = pid.trim();
        let deadline = Instant::now() + Duration::from_secs(5);
        while !is_dead(pid) && Instant::now() < deadline {
            tokio::time::sleep(Duration::from_millis(50)).await;
        }
        assert!(is_dead(pid), "grandchild {pid} outlived the timeout");
        std::fs::remove_dir_all(&dir).ok();
    }

    /// The control: a child that exits inside its limit is `Exited`, with its
    /// status and both streams kept apart.
    #[tokio::test]
    async fn a_build_inside_its_limit_returns_its_output() {
        let mut cmd = tokio::process::Command::new("sh");
        cmd.arg("-c")
            .arg("echo progress >&2; echo /nix/store/abc-toplevel");
        match run_bounded(cmd, std::time::Duration::from_mins(1), None)
            .await
            .unwrap()
        {
            BoundedRun::Exited {
                status,
                stdout,
                stderr,
            } => {
                assert!(status.success());
                assert_eq!(stdout, "/nix/store/abc-toplevel\n");
                assert_eq!(stderr, "progress");
            }
            BoundedRun::TimedOut => panic!("a fast child must not time out"),
        }
    }
}