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

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

pub use git::{
    FETCH_DEADLINE, GIT_LOCAL_DEADLINE, git, git_authed, git_command, git_command_authed,
    git_is_ancestor, git_read_tree_reset, git_rev_parse, git_show_file, git_tag, git_update_ref,
    git_update_ref_cas, run_with_deadline,
};
pub use host_config::{hand_permissions, 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;
/// 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";

/// Web UI port range, shared by every agent including the manager.
/// 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/networking/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())
}

/// Build the network-isolation settings every container is configured
/// with, from the variables the hyperhive NixOS module derives from
/// `services.hyperhive.network.*` onto the `hive-c0re` unit.
///
/// Isolation is the only supported mode: the on/off toggle is gone, so
/// there is no non-isolated branch to fall back to and a missing or
/// malformed value means the daemon is misconfigured — not that a
/// container should quietly come up sharing the host's netns. Silently
/// degrading here dropped a security boundary with nothing in the log
/// to say so.
///
/// Split from [`network_isolation_from_env`] so the parsing is testable
/// without touching process environment.
pub fn network_isolation_from_vars(
    bridge: Option<&str>,
    subnet: Option<&str>,
) -> Result<hive_priv_sock::NetworkIsolation> {
    let bridge = bridge.filter(|s| !s.is_empty()).context(
        "HIVE_NETWORK_BRIDGE is unset or empty — the hyperhive NixOS module sets it \
         from services.hyperhive.network.bridgeName on the hive-c0re unit, so this \
         means the daemon is running outside its unit or with a broken module \
         evaluation",
    )?;
    let subnet = subnet.filter(|s| !s.is_empty()).context(
        "HIVE_NETWORK_SUBNET is unset or empty — the hyperhive NixOS module sets it \
         from services.hyperhive.network.bridgeIp and .bridgePrefixLength on the \
         hive-c0re unit, so this means the daemon is running outside its unit or \
         with a broken module evaluation",
    )?;
    let gateway_ip = bridge_gateway_ip(subnet).with_context(|| {
        format!(
            "HIVE_NETWORK_SUBNET={subnet} is not a valid <ipv4>/<prefix> pair; \
             it comes from services.hyperhive.network.bridgeIp + bridgePrefixLength"
        )
    })?;
    Ok(hive_priv_sock::NetworkIsolation {
        bridge: bridge.to_owned(),
        gateway_ip,
    })
}

/// [`network_isolation_from_vars`] over the real process environment.
///
/// Called once at daemon startup so a bad value fails the unit loudly,
/// and again per container — the variables are process-global, so the
/// second call cannot start failing once the first has passed.
pub fn network_isolation_from_env() -> Result<hive_priv_sock::NetworkIsolation> {
    let bridge = std::env::var("HIVE_NETWORK_BRIDGE").ok();
    let subnet = std::env::var("HIVE_NETWORK_SUBNET").ok();
    network_isolation_from_vars(bridge.as_deref(), subnet.as_deref())
}

/// The hive's facts every container is handed at runtime, from hive-c0re's
/// own unit environment (`nix/host-modules/hive-c0re/environment.nix`). An
/// unset or empty variable is `None`.
fn hive_facts_from_env() -> hive_priv_sock::HiveFacts {
    let read = |name: &str| std::env::var(name).ok().filter(|v| !v.is_empty());
    hive_priv_sock::HiveFacts {
        name: read("HYPERHIVE_HIVE_NAME"),
        domain: read("HYPERHIVE_HIVE_DOMAIN"),
    }
}

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) -> Result<Option<String>> {
    let port = agent_web_port(self_name);
    let raw = list()
        .await
        .context("listing containers for the port-collision check")?;
    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 Ok(Some(other.to_owned()));
        }
    }
    Ok(None)
}

/// 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)
        );
    }
    // Read before anything is seeded: an archetype agent gets no config
    // repo, and the meta render below builds it from the archetype.
    crate::archetypes::refresh(name).await?;
    if crate::archetypes::of(name)?.is_none() {
        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
}

/// Rebuild-path preamble, run by the job queue's `Prebuild` node: fail fast on
/// a port collision, then make sure the applied repo + state dirs exist.
/// Container untouched.
///
/// `applied` is normally already the right shape here (this runs against a
/// previously-provisioned agent), but a `destroy --purge` removes it
/// (`job_queue::exec::run_purge_state`) — a later revive of the *same*
/// agent name is still a rebuild, not a first-spawn, so it lands here, not
/// in `provision_container`. Recover from `agent-configs/<name>` on the
/// forge when that's the case (forge `main` is the last operator-merged
/// config — see `forge::fetch_config_main_into_applied`'s own doc comment)
/// rather than bailing outright. Deliberately **no fallback to
/// `proposed`**: that repo is seeded once at first spawn and never touched
/// again (`setup_proposed`'s own doc comment), so it can be arbitrarily
/// stale for a long-lived agent — restoring from it would silently revive
/// the *wrong* config rather than recovering the right one, which is worse
/// than the clear bail this replaces.
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)
        );
    }
    // An archetype agent has no applied repo to check or recover.
    if crate::archetypes::of(name)?.is_none() {
        if !paths.applied.join(".git").exists()
            && !crate::forge::fetch_config_main_into_applied(name).await
        {
            bail!(
                "applied repo at {} is missing its .git directory and could not be reseeded from \
             the forge (agent-configs/{name} unreachable, has no `main` — never merged or \
             never seeded — or absent); destroy --purge and re-spawn this agent.",
                paths.applied.display()
            );
        }
        setup_applied(&paths.applied, 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,
    node_id: Option<u64>,
) -> Result<()> {
    write_dropins(name, hive, paths).await?;
    priv_run_inner("update", name, node_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 archetypes = crate::archetypes::read_at(&crate::archetypes::path())?;
    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 {
                port: agent_web_port(&name),
                archetype: archetypes.get(&name).cloned(),
                pin: None,
                name,
            })
        })
        .collect();
    if let Some(extra) = name_to_add
        && !out.iter().any(|a| a.name == extra)
    {
        out.push(crate::meta::AgentSpec {
            port: agent_web_port(extra),
            archetype: archetypes.get(extra).cloned(),
            pin: None,
            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
}

/// Per-agent "stop failed" banner guards, keyed by agent name. Cleared the
/// next time that agent's container stops cleanly — a `Mutex<HashMap<...>>`
/// rather than a bare `set_boot_warning` because this condition (unlike a
/// one-shot boot step) genuinely resolves later, once an operator has
/// intervened, and the banner should clear with it instead of surviving
/// until the next hive-c0re restart.
fn stop_failed_guards()
-> &'static std::sync::Mutex<std::collections::HashMap<String, crate::warnings::WarningGuard>> {
    static REG: std::sync::OnceLock<
        std::sync::Mutex<std::collections::HashMap<String, crate::warnings::WarningGuard>>,
    > = std::sync::OnceLock::new();
    REG.get_or_init(|| std::sync::Mutex::new(std::collections::HashMap::new()))
}

/// Leak a per-agent warning `kind`. Agent names are a small, bounded set —
/// leaking one short string per distinct agent that ever hits a stop
/// failure is cheap, same reasoning as `forge::static_kind`.
fn static_stop_kind(name: &str) -> &'static str {
    Box::leak(format!("agent_stop_failed_{name}").into_boxed_str())
}

/// Stop `name`'s container. See `hive-priv`'s `stop_and_release` for why
/// a failure here means the container is likely wedged rather than just
/// slow — SIGKILL doesn't recover that case in practice, so this
/// surfaces it on the dashboard instead of silently retrying.
pub async fn kill(name: &str) -> Result<()> {
    validate(name)?;
    let result = priv_run("stop", name).await;
    // Lock only for the synchronous map update — never held across the
    // `.await` above.
    let mut guards = stop_failed_guards()
        .lock()
        .unwrap_or_else(std::sync::PoisonError::into_inner);
    match result {
        Ok(()) => {
            // A clean stop clears any earlier "this looked wedged" banner
            // for the same agent.
            guards.remove(name);
            Ok(())
        }
        Err(e) => {
            tracing::error!(%name, error = %e, "container stop failed");
            guards.insert(
                name.to_owned(),
                crate::warnings::set_warning(
                    static_stop_kind(name),
                    "crit",
                    format!("{name}: stop failed — {e}"),
                ),
            );
            Err(e)
        }
    }
}

/// 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.
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/scheduler/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}"))
}

/// The part of a container unit's systemd state we act on.
///
/// Deliberately not the full set: everything outside `Active`/`Failed` is
/// one bucket, because the only distinction anything downstream makes is
/// *gave up* versus *anything else*.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum UnitState {
    /// Running.
    Active,
    /// systemd gave up on it — most often by exhausting the bounded start
    /// limit. Its own variant precisely because it is the difference
    /// between **gave up** and **stopped on purpose**, which every other
    /// non-running state collapses into.
    Failed,
    /// `inactive` (the deliberate stop), `activating`, `deactivating`, an
    /// unknown unit, or a `systemctl` we could not run at all.
    Other,
}

impl UnitState {
    /// Parse `systemctl is-active`'s stdout.
    ///
    /// Split out from the shell-out so the mapping is testable without a
    /// systemd to ask — the states we care about are the two we name, and
    /// everything else must land in `Other` rather than being guessed at.
    fn from_is_active(stdout: &str) -> Self {
        match stdout.trim() {
            "active" => Self::Active,
            "failed" => Self::Failed,
            _ => Self::Other,
        }
    }
}

/// The container unit's state, in one `systemctl` call.
///
/// `is-active` **prints** the state and exits 0 only for `active`, so
/// dropping `--quiet` yields both facts from the call [`is_running`]
/// already makes. That matters: the status path runs this per agent (see
/// the spawn-count note in `socket_server::lifecycle_handlers`), so a
/// caller wanting *running* and *failed* must not pay for two subprocesses.
pub async fn unit_state(name: &str) -> UnitState {
    let container = container_name(name);
    let unit = format!("container@{container}.service");
    // Not "the unit is fine" and not "the unit failed" — we simply do not
    // know, and saying `Failed` here would report a gave-up agent every time
    // `systemctl` itself was unavailable or hung.
    systemctl_is_active(&[&unit])
        .await
        .map_or(UnitState::Other, |out| {
            UnitState::from_is_active(&String::from_utf8_lossy(&out.stdout))
        })
}

/// True when the container's systemd unit is active. Used by the dashboard
/// to gate stop/restart buttons.
///
/// Kept boolean on purpose: nearly every caller is reconcile/power logic
/// asking "is it up? if not, start it", and that question has two answers.
/// A caller that needs to tell *failed* from *stopped* wants
/// [`unit_state`] instead.
///
/// Unbounded: its callers record or act on the answer, so a hung `systemctl`
/// must not read as "not running". Periodic workers use [`running_state`].
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())
}

/// Whether the container's systemd unit is active, for periodic workers that
/// must not stall on a hung `systemctl`; `None` when it could not be run or
/// did not answer within [`IS_ACTIVE_TIMEOUT`].
pub async fn running_state(name: &str) -> Option<bool> {
    let container = container_name(name);
    let unit = format!("container@{container}.service");
    systemctl_is_active(&["--quiet", &unit])
        .await
        .map(|out| out.status.success())
}

/// How long one `systemctl is-active` may run before the unit's state counts
/// as unreadable. Periodic workers make this call per agent inline.
const IS_ACTIVE_TIMEOUT: std::time::Duration = std::time::Duration::from_secs(10);

/// Run `systemctl is-active <args>`, killed after [`IS_ACTIVE_TIMEOUT`].
async fn systemctl_is_active(args: &[&str]) -> Option<std::process::Output> {
    let output = Command::new("systemctl")
        .arg("is-active")
        .args(args)
        .kill_on_drop(true)
        .output();
    output_within(IS_ACTIVE_TIMEOUT, output, args).await
}

/// Await a child's output for at most `limit`. `None` when it failed to run
/// or timed out; the timeout drops `fut`, which kills a child spawned with
/// `kill_on_drop(true)`.
async fn output_within<T>(
    limit: std::time::Duration,
    fut: impl std::future::Future<Output = std::io::Result<T>>,
    args: &[&str],
) -> Option<T> {
    if let Ok(result) = tokio::time::timeout(limit, fut).await {
        result.ok()
    } else {
        tracing::warn!(
            ?args,
            timeout_s = limit.as_secs(),
            "systemctl is-active timed out"
        );
        None
    }
}

/// 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.
///
/// Fails when the container survives the destroy. The destroy DAG purges the
/// agent's state only after this returns `Ok`, so a swallowed failure here
/// deletes the state of a container that still exists.
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. When it errors, the container list
    // decides: a container that is already gone is still a successful destroy.
    if let Err(e) = priv_run("destroy", name).await {
        confirm_gone_after_failed_destroy(e, &container, list().await)?;
    }
    // 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(())
}

/// Settle a failed `nixos-container destroy` against the container list taken
/// after it. `Ok` only when the list is readable and no longer names the
/// container; a list that cannot be read proves nothing, so it fails too.
fn confirm_gone_after_failed_destroy(
    destroy_err: anyhow::Error,
    container: &str,
    listed: Result<Vec<String>>,
) -> Result<()> {
    match listed {
        Ok(names) if !names.iter().any(|n| n == container) => {
            tracing::warn!(
                error = ?destroy_err,
                %container,
                "nixos-container destroy returned an error, but the container is gone"
            );
            Ok(())
        }
        Ok(_) => Err(destroy_err.context(format!("container {container} still exists"))),
        Err(list_err) => Err(destroy_err.context(format!(
            "cannot confirm whether container {container} still exists: {list_err:#}"
        ))),
    }
}

/// Wall-clock bound on [`prebuild_toplevel`]'s `nix build`. 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. A stalled download is bounded by nix's own
/// `stalled-download-timeout`; this covers the rest, such as a wedged
/// nix-daemon holding the rebuild's build slot forever.
const PREBUILD_TIMEOUT: std::time::Duration = std::time::Duration::from_hours(1);

/// Wait for `child` until `limit`. `None` means it was still running, and its
/// whole process group was `SIGKILL`ed and the child reaped, so a grandchild
/// holding the daemon connection dies with it. The child must have been
/// spawned with `process_group(0)`.
async fn wait_bounded(
    child: &mut tokio::process::Child,
    limit: std::time::Duration,
) -> std::io::Result<Option<std::process::ExitStatus>> {
    // Group leader, so its pid is the pgid. Read before waiting: a reaped
    // child has no id.
    let pgid = child.id().and_then(|pid| i32::try_from(pid).ok());
    if let Ok(status) = tokio::time::timeout(limit, child.wait()).await {
        return status.map(Some);
    }
    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(None)
}

/// 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/scheduler/coordinator.md::Rebuild path` for why
/// the prebuild happens before stop, and `docs/scheduler/coordinator.md::Prebuild
/// attr path` for why the explicit nixosConfigurations attr is required.
///
/// `node_id` is the queue node this build belongs to, when there is one —
/// it is stored on the `build_logs` row so the dashboard can find the log
/// from the node. Pass `None` for builds that run outside the queue.
pub async fn prebuild_toplevel(name: &str, flake_ref: &str, node_id: Option<u64>) -> 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, node_id)
            .map_err(|e| {
                tracing::warn!(error = ?e, "build_logs: start failed (prebuild log dropped)");
            })
            .ok()
    });

    let mut child = Command::new("nix")
        .args(&args)
        .stdout(std::process::Stdio::piped())
        .stderr(std::process::Stdio::piped())
        .process_group(0)
        .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 {
            if is_actionable_nix_line(&line) {
                tracing::warn!(target: "nix-prebuild", cmdline = %stderr_cmdline, "{line}");
            } else {
                tracing::info!(target: "nix-prebuild", cmdline = %stderr_cmdline, "{line}");
            }
            if let (Some(h), Some(id)) = (&stderr_logs, log_id) {
                h.append_stderr(id, &line);
            }
        }
    });

    let Some(status) = wait_bounded(&mut child, PREBUILD_TIMEOUT)
        .await
        .with_context(|| format!("wait {cmdline}"))?
    else {
        // A process that left the group can still hold the pipes open.
        pump_stdout.abort();
        pump_stderr.abort();
        if let (Some(h), Some(id)) = (&logs, log_id) {
            h.finish(id, crate::build_logs::BuildStatus::Fail);
        }
        let secs = PREBUILD_TIMEOUT.as_secs();
        match log_id {
            Some(id) => {
                bail!("prebuild {cmdline} timed out after {secs}s; killed it; see build log #{id}")
            }
            None => bail!("prebuild {cmdline} timed out after {secs}s; killed it"),
        }
    };
    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())
}

/// Logical agent names from a [`list`] result. An unreadable list stays an
/// error rather than becoming "no agents": a caller fanning work out over
/// every agent would otherwise report success after doing nothing.
pub fn agent_names(listed: Result<Vec<String>>) -> Result<Vec<String>> {
    Ok(listed?
        .into_iter()
        .filter_map(|c| c.strip_prefix(AGENT_PREFIX).map(str::to_owned))
        .collect())
}

/// Ensure the per-agent runtime directory `/run/hyperhive/agents/<name>`
/// exists. It is a bind source and `/run` is a tmpfs, so every start path
/// creates it before `nixos-container start`.
///
/// 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()))
}

/// Whether a `nix`/`nixos-container` stderr line is worth WARN (nix's own
/// warnings and errors) rather than routine progress chatter (`copying N
/// paths…`, `fetching git input…`, `structuredAttrs is enabled`, …) — nix
/// writes the latter to stderr too, and it dwarfs the former in volume.
/// Matches nix's own prefixes (`warning:`, `evaluation warning:`,
/// `error:`) rather than only `warning:`, so a real build failure
/// (`error: builder for '...' failed`, `error: 1 dependencies ... failed
/// to build`) never gets buried at INFO alongside the chatter.
fn is_actionable_nix_line(line: &str) -> bool {
    let trimmed = line.trim_start();
    trimmed.starts_with("warning:")
        || trimmed.starts_with("evaluation warning:")
        || trimmed.starts_with("error:")
}

/// Build the per-line callback for `create_container_streaming` /
/// `update_container_streaming`. Both ops share identical dispatch logic
/// (stdout → info + `append_stdout`, stderr → warn for a nix warning/error
/// and info otherwise + `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 => {
            if is_actionable_nix_line(line) {
                tracing::warn!(target: "nixos-container", cmdline = %cmdline, "{line}");
            } else {
                tracing::info!(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 stamps `node_id` onto the build-log row it opens, so
/// the dashboard can find the log from the queue node (and link to
/// `/api/build-logs/id/{id}/stream`).
///
/// This used to be a `Fn(i64)` callback that handed the row id *back* to the
/// queue, which then held it in a side map. The row carries the link itself
/// now, so the id only ever travels one way.
async fn priv_run_inner(kind: &str, name: &str, node_id: Option<u64>) -> 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, node_id)
            .map_err(|e| {
                tracing::warn!(error = ?e, "build_logs: start failed (priv_run log dropped)");
            })
            .ok()
    });

    // 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) => match log_id {
            Some(id) => bail!("{e:#}; see build log #{id}"),
            None => bail!("{e:#}"),
        },
    }
}