//! Minimal privileged helper for hive-c0re.
//!
//! Runs as root. Exposes a narrow unix socket at `/run/hive/priv.sock`
//! that accepts `PrivRequest` JSON lines and executes only the
//! operations that genuinely require root. All coordination logic,
//! broker, HTTP, and scheduling stay in the unprivileged hive-c0re
//! process.
//!
//! **Security model**: every request is validated against a strict
//! container-name allowlist before any filesystem or process operation.
//! Only containers whose names match the hive convention (`h-*`,
//! the manager container, or known sibling service containers) are
//! accepted. Every variant maps to a single known operation — no
//! arbitrary command pass-through.
//!
//! **Socket activation**: when systemd passes the listener socket via
//! `LISTEN_FDS=1` + `LISTEN_PID=<self>`, the inherited fd 3 is used
//! instead of binding a fresh socket.

use std::path::{Path, PathBuf};

use anyhow::{Context as _, Result, bail};
use hive_sh4re::priv_proto::{
    AGENT_PREFIX, AGENT_STATE_ROOT, BindMount, JournalQuery, MANAGER_NAME, META_DIR,
    NetworkIsolation, PRIV_SOCK, PrivEvent, PrivRequest, PrivResponse, PrivStream, PrivStreamLine,
    SIBLING_CONTAINERS,
};
use tokio::io::{AsyncBufReadExt, AsyncWriteExt, BufReader};
use tokio::net::unix::OwnedWriteHalf;
use tokio::net::{UnixListener, UnixStream};
use tokio::process::Command;

/// Root of the per-agent unix-socket dirs on the host.
const SOCKET_DIR_ROOT: &str = "/run/hive-agent";

#[tokio::main]
async fn main() -> Result<()> {
    tracing_subscriber::fmt()
        .with_env_filter(
            tracing_subscriber::EnvFilter::try_from_default_env().unwrap_or_else(|_| "info".into()),
        )
        .init();

    let listener = socket_listener()?;
    tracing::info!("hive-priv listening");

    loop {
        match listener.accept().await {
            Ok((stream, _)) => {
                tokio::spawn(handle(stream));
            }
            Err(e) => {
                tracing::error!(error = %e, "accept failed");
            }
        }
    }
}

fn socket_listener() -> Result<UnixListener> {
    // hive-priv is ALWAYS socket-activated by the `hive-priv.socket` unit
    // (fd 3 via LISTEN_FDS). There is intentionally no self-bind fallback,
    // so dev and prod take the same path; see docs/boundary.md.
    let listen_fds: Option<i32> = std::env::var("LISTEN_FDS")
        .ok()
        .and_then(|s| s.parse().ok());
    let listen_pid: Option<u32> = std::env::var("LISTEN_PID")
        .ok()
        .and_then(|s| s.parse().ok());

    let activated =
        matches!(listen_fds, Some(n) if n >= 1) && listen_pid == Some(std::process::id());
    if !activated {
        bail!(
            "hive-priv requires systemd socket activation (expected LISTEN_FDS>=1 + \
             LISTEN_PID=<self> for {PRIV_SOCK}); run it via the hive-priv.socket unit, \
             not directly"
        );
    }

    // SAFETY: systemd has passed us a ready UnixListener on fd 3.
    let std_listener = unsafe {
        use std::os::unix::io::FromRawFd;
        std::os::unix::net::UnixListener::from_raw_fd(3)
    };
    std_listener
        .set_nonblocking(true)
        .context("set socket non-blocking")?;
    let listener =
        tokio::net::UnixListener::from_std(std_listener).context("wrap systemd socket")?;
    tracing::info!("using systemd-activated socket");
    Ok(listener)
}

async fn handle(stream: UnixStream) {
    let (reader, mut writer) = stream.into_split();
    let mut lines = BufReader::new(reader).lines();
    while let Ok(Some(line)) = lines.next_line().await {
        let resp = dispatch(&line, &mut writer).await;
        // Write the terminal PrivResponse as a PrivEvent::Done. Wire-identical
        // to a bare PrivResponse (untagged), so old hive-c0re callers that
        // deserialise directly to PrivResponse continue to work.
        let event = PrivEvent::Done(resp);
        let mut json = serde_json::to_string(&event).unwrap_or_else(|e| {
            format!("{{\"ok\":false,\"stdout\":\"\",\"stderr\":\"\",\"error\":\"serialise failed: {e}\"}}")
        });
        json.push('\n');
        if let Err(e) = writer.write_all(json.as_bytes()).await {
            tracing::warn!(error = %e, "write response failed");
            break;
        }
    }
}

async fn dispatch(line: &str, writer: &mut OwnedWriteHalf) -> PrivResponse {
    match serde_json::from_str::<PrivRequest>(line) {
        Ok(req) => match exec(req, writer).await {
            Ok((stdout, stderr)) => PrivResponse {
                ok: true,
                stdout,
                stderr,
                error: None,
            },
            Err(e) => PrivResponse {
                ok: false,
                stdout: String::new(),
                stderr: String::new(),
                error: Some(format!("{e:#}")),
            },
        },
        Err(e) => PrivResponse {
            ok: false,
            stdout: String::new(),
            stderr: String::new(),
            error: Some(format!("parse request: {e}")),
        },
    }
}

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

/// Execute a validated `PrivRequest`. Returns `(stdout, stderr)` on success.
/// For streaming ops (`CreateContainer`/`UpdateContainer` with `stream: true`)
/// output lines are forwarded to `writer` as `PrivEvent::Line` messages and
/// the returned strings are empty.
async fn exec(req: PrivRequest, writer: &mut OwnedWriteHalf) -> Result<(String, String)> {
    match req {
        PrivRequest::StartContainer { ref name } => {
            validate_container_name(name)?;
            container_run(&["start", &container_system_name(name)]).await
        }

        PrivRequest::StopContainer { ref name } => {
            validate_container_name(name)?;
            container_run(&["stop", &container_system_name(name)]).await
        }

        PrivRequest::KillContainer { ref name } => {
            validate_container_name(name)?;
            container_run(&["kill", &container_system_name(name)]).await
        }

        PrivRequest::UpdateContainer { ref name, stream } => {
            container_flake_action("update", name, stream, writer).await
        }

        PrivRequest::CreateContainer { ref name, stream } => {
            container_flake_action("create", name, stream, writer).await
        }

        PrivRequest::DestroyContainer { ref name } => {
            validate_container_name(name)?;
            container_run(&["destroy", &container_system_name(name)]).await
        }

        PrivRequest::ListContainers => container_run(&["list"]).await,

        PrivRequest::ReadContainerJournal {
            ref container,
            ref query,
        } => {
            validate_container_system_name(container)?;
            read_container_journal(container, query).await
        }

        PrivRequest::WriteNspawnFlags {
            ref container,
            ref binds,
            ref isolation,
        } => handle_write_nspawn_flags(container, binds, isolation.as_ref()),

        PrivRequest::WriteResourceLimits {
            ref container,
            ref memory_max,
            ref cpu_quota,
        } => write_resource_limits(container, memory_max, cpu_quota),

        PrivRequest::RemoveServiceDropin { ref container } => remove_service_dropin(container),

        PrivRequest::DaemonReload => daemon_reload().await,

        PrivRequest::ReloadGatewayNginx => sync_gateway_nginx().await,

        PrivRequest::ChownSocketDir {
            ref agent_name,
            uid,
            gid,
        } => chown_socket_dir(agent_name, uid, gid),

        PrivRequest::ChmodSocketDir {
            ref agent_name,
            mode,
        } => chmod_socket_dir(agent_name, mode),

        PrivRequest::RunForgeAdmin { ref args } => {
            for arg in args {
                validate_forge_admin_arg(arg)?;
            }
            run_forge_admin(args).await
        }

        PrivRequest::WriteAgentForgeToken {
            ref agent_name,
            ref token,
        } => {
            validate_agent_name(agent_name)?;
            write_agent_state_file(agent_name, "forge-token", &format!("{token}\n"))
        }

        PrivRequest::WriteAgentMatrixToken {
            ref agent_name,
            ref token,
        } => {
            validate_agent_name(agent_name)?;
            write_agent_state_file(agent_name, "matrix-token", &format!("{token}\n"))
        }

        PrivRequest::RestartMatrixDaemon { ref agent_name } => {
            restart_matrix_daemon(agent_name).await
        }
    }
}

/// Shared body for `CreateContainer` / `UpdateContainer`: validate the
/// name, build the `nixos-container <verb> … --flake <ref>` argv, and
/// run it (streaming line events to `writer` when `stream` is set).
async fn container_flake_action(
    verb: &str,
    name: &str,
    stream: bool,
    writer: &mut OwnedWriteHalf,
) -> Result<(String, String)> {
    validate_container_name(name)?;
    let flake_ref = agent_flake_ref(name);
    let args = [verb, &container_system_name(name), "--flake", &flake_ref];
    if stream {
        container_run_streaming(&args, writer).await
    } else {
        container_run(&args).await
    }
}

/// `WriteNspawnFlags` — validate the container + every bind path, then
/// write the container's nspawn flag overrides.
fn handle_write_nspawn_flags(
    container: &str,
    binds: &[BindMount],
    isolation: Option<&NetworkIsolation>,
) -> Result<(String, String)> {
    validate_container_system_name(container)?;
    for bind in binds {
        validate_bind_path(&bind.host_path)?;
        validate_bind_path(&bind.container_path)?;
    }
    write_nspawn_flags(container, binds, isolation)?;
    Ok((String::new(), String::new()))
}

/// `RemoveServiceDropin` — remove the container service's drop-in dir
/// if present (idempotent).
fn remove_service_dropin(container: &str) -> Result<(String, String)> {
    validate_container_system_name(container)?;
    let dir = format!("/run/systemd/system/container@{container}.service.d");
    if Path::new(&dir).exists() {
        std::fs::remove_dir_all(&dir).with_context(|| format!("remove {dir}"))?;
    }
    Ok((String::new(), String::new()))
}

/// `ChownSocketDir` — chown the agent's host socket dir to its
/// container uid/gid.
fn chown_socket_dir(agent_name: &str, uid: u32, gid: u32) -> Result<(String, String)> {
    validate_agent_name(agent_name)?;
    let path = socket_dir_path(agent_name);
    std::os::unix::fs::chown(&path, Some(uid), Some(gid))
        .with_context(|| format!("chown {} to {uid}:{gid}", path.display()))?;
    Ok((String::new(), String::new()))
}

/// `ChmodSocketDir` — set the mode on the agent's host socket dir.
fn chmod_socket_dir(agent_name: &str, mode: u32) -> Result<(String, String)> {
    use std::os::unix::fs::PermissionsExt as _;
    validate_agent_name(agent_name)?;
    let path = socket_dir_path(agent_name);
    std::fs::set_permissions(&path, std::fs::Permissions::from_mode(mode))
        .with_context(|| format!("chmod {:o} {}", mode, path.display()))?;
    Ok((String::new(), String::new()))
}

/// `WriteResourceLimits` — drop a systemd `MemoryMax`/`CPUQuota`
/// override into the container service's drop-in dir.
fn write_resource_limits(
    container: &str,
    memory_max: &str,
    cpu_quota: &str,
) -> Result<(String, String)> {
    validate_container_system_name(container)?;
    let dir = format!("/run/systemd/system/container@{container}.service.d");
    std::fs::create_dir_all(&dir).with_context(|| format!("create {dir}"))?;
    let path = format!("{dir}/hyperhive-limits.conf");
    let content = format!("[Service]\nMemoryMax={memory_max}\nCPUQuota={cpu_quota}\n");
    std::fs::write(&path, content).with_context(|| format!("write {path}"))?;
    Ok((String::new(), String::new()))
}

/// `DaemonReload` — `systemctl daemon-reload` on the host.
async fn daemon_reload() -> Result<(String, String)> {
    let out = Command::new("systemctl")
        .arg("daemon-reload")
        .output()
        .await
        .context("invoke systemctl daemon-reload")?;
    if !out.status.success() {
        bail!(
            "systemctl daemon-reload failed ({}): {}",
            out.status,
            String::from_utf8_lossy(&out.stderr).trim()
        );
    }
    Ok((String::new(), String::new()))
}

/// `RestartMatrixDaemon` — restart the matrix daemon unit inside the
/// agent's container.
async fn restart_matrix_daemon(agent_name: &str) -> Result<(String, String)> {
    validate_agent_name(agent_name)?;
    let machine = format!("--machine=h-{agent_name}");
    let unit = "hive-matrix-daemon.service";
    let out = Command::new("systemctl")
        .args([&machine, "restart", unit])
        .output()
        .await
        .with_context(|| format!("systemctl restart {unit} in container h-{agent_name}"))?;
    if !out.status.success() {
        bail!(
            "systemctl restart {unit} in h-{agent_name} exited {}: {}",
            out.status,
            String::from_utf8_lossy(&out.stderr).trim()
        );
    }
    Ok((
        String::from_utf8_lossy(&out.stdout).into_owned(),
        String::from_utf8_lossy(&out.stderr).into_owned(),
    ))
}

/// Shared helper for `WriteAgentForgeToken` and `WriteAgentMatrixToken`.
/// Writes `content` to `AGENT_STATE_ROOT/<agent_name>/state/<filename>`,
/// chowns to the agent user (derived from the state dir's existing owner),
/// and chmods 0600. Running as root (hive-priv), so this succeeds
/// regardless of the file's prior owner/permissions.
fn write_agent_state_file(
    agent_name: &str,
    filename: &str,
    content: &str,
) -> Result<(String, String)> {
    use std::os::unix::fs::{MetadataExt as _, PermissionsExt as _};

    let state_dir = PathBuf::from(AGENT_STATE_ROOT)
        .join(agent_name)
        .join("state");
    // NOTE: `create_dir_all` is normally a no-op — lifecycle creates and chowns
    // the state dir during spawn. On the rare edge where the dir doesn't exist
    // yet (container being provisioned for the first time), the newly created dir
    // is root:root. The `stat state_dir` chown below will then see uid=0 and
    // leave the file root-owned (0600). The agent won't be able to read it until
    // its lifecycle completes. If that happens, a `systemctl restart hive-c0re`
    // after provisioning will re-mint and re-write the token correctly.
    std::fs::create_dir_all(&state_dir)
        .with_context(|| format!("create state dir {}", state_dir.display()))?;

    let path = state_dir.join(filename);
    std::fs::write(&path, content).with_context(|| format!("write {}", path.display()))?;
    std::fs::set_permissions(&path, std::fs::Permissions::from_mode(0o600))
        .with_context(|| format!("chmod 600 {}", path.display()))?;

    // Chown to the state dir's owner so the agent process can read the file.
    // If stat fails (e.g. dir just created, owner is root), the file stays
    // root-owned and 0600 — still unreadable by others, just not agent-readable.
    // Log a warning so operators can diagnose.
    match std::fs::metadata(&state_dir) {
        Ok(meta) => {
            if let Err(e) = std::os::unix::fs::chown(&path, Some(meta.uid()), Some(meta.gid())) {
                tracing::warn!(
                    agent = %agent_name,
                    path = %path.display(),
                    error = %e,
                    "write_agent_state_file: chown failed"
                );
            }
        }
        Err(e) => {
            tracing::warn!(
                agent = %agent_name,
                error = %e,
                "write_agent_state_file: stat state_dir failed, leaving file root-owned"
            );
        }
    }
    tracing::info!(agent = %agent_name, file = %filename, "wrote agent state file");
    Ok((String::new(), String::new()))
}

/// Validate a single argument destined for `forgejo admin`. Rejects
/// null bytes and newlines (which could corrupt the subprocess args list
/// or log output). Shell metacharacters are harmless since the command
/// is spawned directly (no shell), but we reject them defensively.
fn validate_forge_admin_arg(arg: &str) -> Result<()> {
    if arg.bytes().any(|b| b == 0 || b == b'\n' || b == b'\r') {
        bail!("forge admin arg {arg:?} contains null byte or newline");
    }
    Ok(())
}

/// Run `forgejo admin <args>` inside the `hive-forge` container as the
/// `forgejo` unix user. Requires root (for nsenter into the container's
/// namespaces). Returns `(stdout, stderr)`.
async fn run_forge_admin(args: &[String]) -> Result<(String, String)> {
    let mut cmd_args: Vec<&str> = vec![
        "run",
        "hive-forge",
        "--",
        "runuser",
        "-u",
        "forgejo",
        "--",
        "forgejo",
        "--work-path",
        "/var/lib/forgejo",
        "admin",
    ];
    for a in args {
        cmd_args.push(a.as_str());
    }
    let out = Command::new("nixos-container")
        .args(&cmd_args)
        .output()
        .await
        .context("invoke nixos-container run hive-forge -- forgejo admin")?;
    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: "forgejo-admin", "{line}");
    }
    for line in stderr.lines() {
        tracing::warn!(target: "forgejo-admin", "{line}");
    }
    if !out.status.success() {
        bail!(
            "forgejo admin {} failed ({}): {}",
            args.join(" "),
            out.status,
            stderr.trim()
        );
    }
    Ok((stdout, stderr))
}

/// Invoke `nixos-container` with the given args, log output to journald.
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();
    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 `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()))
}

/// Read a container's journal as root via `journalctl -M`. Returns
/// `(stdout, stderr)`. Unlike `container_run` a non-zero exit is *not* a
/// hard error — journalctl's own diagnostic (folded into `stderr` with
/// the exit status) is what the caller surfaces to the operator, so the
/// helper never bails.
async fn read_container_journal(container: &str, query: &JournalQuery) -> Result<(String, String)> {
    let mut args: Vec<String> = vec![
        "-M".to_owned(),
        container.to_owned(),
        "--no-pager".to_owned(),
        format!("--output={}", query.output.as_journalctl()),
        "-n".to_owned(),
        query.lines.to_string(),
    ];
    if query.boot {
        args.push("-b".to_owned());
    }
    if let Some(u) = &query.unit {
        args.push("-u".to_owned());
        args.push(u.clone());
    }
    if let Some(p) = &query.priority {
        args.push("-p".to_owned());
        args.push(p.clone());
    }
    // `--grep=`/`--since=`/`--until=` use the `=`-joined form so a value
    // can never be parsed as a separate journalctl flag.
    if let Some(g) = &query.grep {
        args.push(format!("--grep={g}"));
    }
    if let Some(s) = &query.since {
        args.push(format!("--since={s}"));
    }
    if let Some(u) = &query.until {
        args.push(format!("--until={u}"));
    }
    let out = Command::new("journalctl")
        .args(&args)
        .output()
        .await
        .context("invoke journalctl -M")?;
    let stdout = String::from_utf8_lossy(&out.stdout).into_owned();
    let stderr = if out.status.success() {
        String::from_utf8_lossy(&out.stderr).into_owned()
    } else {
        format!(
            "journalctl -M {container} exited {}: {}",
            out.status,
            String::from_utf8_lossy(&out.stderr).trim()
        )
    };
    Ok((stdout, stderr))
}

/// Synchronise the nginx unit inside the `hive-gateway` container.
///
/// Queries `ActiveState` via `systemctl --machine=hive-gateway` (requires
/// root — machine-bus transport enters the container namespace), then
/// dispatches:
/// - `active`  → `systemctl reload nginx` (SIGHUP, zero-downtime)
/// - `failed`  → `systemctl reset-failed nginx` + `systemctl start nginx`
/// - otherwise → `systemctl start nginx`
///
/// Returns `(String::new(), String::new())` on success so it fits the
/// `exec` return type directly.
async fn sync_gateway_nginx() -> Result<(String, String)> {
    let state_out = Command::new("systemctl")
        .args([
            "--machine=hive-gateway",
            "show",
            "--property=ActiveState",
            "--value",
            "nginx",
        ])
        .output()
        .await
        .context("query nginx ActiveState in hive-gateway")?;
    if !state_out.status.success() {
        tracing::warn!(
            exit_code = ?state_out.status.code(),
            stderr = %String::from_utf8_lossy(&state_out.stderr).trim(),
            "systemctl show ActiveState exited non-zero — gateway container may be down"
        );
    }
    let state = String::from_utf8_lossy(&state_out.stdout).trim().to_owned();
    // State-aware dispatch: reload when running; reset+start after
    // start-limit failure; plain start when inactive or unknown.
    match state.as_str() {
        "active" => {
            let out = Command::new("systemctl")
                .args(["--machine=hive-gateway", "reload", "nginx"])
                .output()
                .await
                .context("reload nginx in hive-gateway")?;
            if !out.status.success() {
                bail!(
                    "gateway nginx reload failed ({}): {}",
                    out.status,
                    String::from_utf8_lossy(&out.stderr).trim()
                );
            }
        }
        "failed" => {
            // Clear start-limit so the next start can proceed.
            let _ = Command::new("systemctl")
                .args(["--machine=hive-gateway", "reset-failed", "nginx"])
                .status()
                .await;
            let out = Command::new("systemctl")
                .args(["--machine=hive-gateway", "start", "nginx"])
                .output()
                .await
                .context("start nginx after reset-failed in hive-gateway")?;
            if !out.status.success() {
                bail!(
                    "gateway nginx start (after reset-failed) failed ({}): {}",
                    out.status,
                    String::from_utf8_lossy(&out.stderr).trim()
                );
            }
        }
        _ => {
            // inactive, activating, deactivating, unknown — just start.
            let out = Command::new("systemctl")
                .args(["--machine=hive-gateway", "start", "nginx"])
                .output()
                .await
                .context("start nginx in hive-gateway")?;
            if !out.status.success() {
                bail!(
                    "gateway nginx start failed (state={state:?}) ({}): {}",
                    out.status,
                    String::from_utf8_lossy(&out.stderr).trim()
                );
            }
        }
    }
    Ok((String::new(), String::new()))
}

/// Return the system container name for a logical agent name.
/// All agents (including the manager) use the `h-` prefix.
fn container_system_name(name: &str) -> String {
    format!("{AGENT_PREFIX}{name}")
}

/// Path of the per-agent unix-socket dir on the host.
fn socket_dir_path(agent_name: &str) -> PathBuf {
    PathBuf::from(format!("{SOCKET_DIR_ROOT}/{agent_name}"))
}

/// Validate a logical agent name (the name hive-c0re uses internally,
/// before the `h-` container prefix is applied).
fn validate_agent_name(name: &str) -> Result<()> {
    if name == MANAGER_NAME {
        return Ok(());
    }
    validate_name_chars(name)?;
    Ok(())
}

/// Validate a logical agent name and check it maps to a hive-managed container.
fn validate_container_name(name: &str) -> Result<()> {
    if name == MANAGER_NAME {
        return Ok(());
    }
    if SIBLING_CONTAINERS.contains(&name) {
        return Ok(());
    }
    validate_name_chars(name)?;
    Ok(())
}

/// Validate a system-level container name (already has `h-` prefix for
/// all agents including the manager, or is a sibling service name).
fn validate_container_system_name(name: &str) -> Result<()> {
    if SIBLING_CONTAINERS.contains(&name) {
        return Ok(());
    }
    if let Some(suffix) = name.strip_prefix(AGENT_PREFIX) {
        validate_name_chars(suffix)?;
        return Ok(());
    }
    bail!("container name {name:?} is not managed by hive");
}

fn validate_name_chars(name: &str) -> Result<()> {
    if name.is_empty()
        || !name
            .chars()
            .all(|c| c.is_ascii_lowercase() || c.is_ascii_digit() || c == '-')
    {
        bail!("invalid name {name:?}: must be non-empty lowercase ascii + digits + hyphens");
    }
    Ok(())
}

/// Derive the meta-flake ref for an agent by name.
fn agent_flake_ref(name: &str) -> String {
    format!("{META_DIR}#{name}")
}

/// Validate a bind-mount path: must be absolute, non-empty, and contain
/// no newlines, null bytes, or double-quotes (which would break the
/// `EXTRA_NSPAWN_FLAGS="..."` conf line format).
fn validate_bind_path(path: &str) -> Result<()> {
    if path.is_empty()
        || !path.starts_with('/')
        || path
            .bytes()
            .any(|b| b == 0 || b == b'\n' || b == b'"' || b == b':')
    {
        bail!(
            "invalid bind path {path:?}: must be an absolute path with no colons, newlines, null bytes, or double-quotes"
        );
    }
    Ok(())
}

/// Update `/etc/nixos-containers/<container>.conf`: strips network-isolation
/// vars (`PRIVATE_NETWORK`, `HOST_ADDRESS*`, `LOCAL_ADDRESS*`, `HOST_BRIDGE`,
/// Update `/etc/nixos-containers/<container>.conf`: strip old network vars,
/// write network isolation settings, then append `EXTRA_NSPAWN_FLAGS`.
/// When `isolation` is `Some`, writes `PRIVATE_NETWORK=1` + veth wiring;
/// when `None`, writes `PRIVATE_NETWORK=0`.
fn write_nspawn_flags(
    container: &str,
    binds: &[BindMount],
    isolation: Option<&NetworkIsolation>,
) -> Result<()> {
    use std::fmt::Write as _;
    let path = format!("/etc/nixos-containers/{container}.conf");
    let original = std::fs::read_to_string(&path).with_context(|| format!("read {path}"))?;
    let lines: Vec<&str> = original
        .lines()
        .filter(|line| {
            let t = line.trim_start();
            !t.starts_with("EXTRA_NSPAWN_FLAGS=")
                && !t.starts_with("PRIVATE_NETWORK=")
                && !t.starts_with("HOST_ADDRESS=")
                && !t.starts_with("LOCAL_ADDRESS=")
                && !t.starts_with("HOST_ADDRESS6=")
                && !t.starts_with("LOCAL_ADDRESS6=")
                && !t.starts_with("HOST_BRIDGE=")
        })
        .collect();
    let mut out = lines.join("\n");
    if !out.is_empty() {
        out.push('\n');
    }
    if let Some(iso) = isolation {
        out.push_str("PRIVATE_NETWORK=1\n");
        out.push_str("HOST_ADDRESS=\n");
        let _ = writeln!(out, "LOCAL_ADDRESS={}", iso.agent_ip);
        out.push_str("HOST_ADDRESS6=\n");
        out.push_str("LOCAL_ADDRESS6=\n");
        let _ = writeln!(out, "HOST_BRIDGE={}", iso.bridge);
    } else {
        out.push_str("PRIVATE_NETWORK=0\n");
        out.push_str("HOST_ADDRESS=\n");
        out.push_str("LOCAL_ADDRESS=\n");
        out.push_str("HOST_ADDRESS6=\n");
        out.push_str("LOCAL_ADDRESS6=\n");
        out.push_str("HOST_BRIDGE=\n");
    }
    let flags: Vec<String> = binds
        .iter()
        .map(|b| {
            let flag = if b.read_only { "--bind-ro" } else { "--bind" };
            format!("{flag}={}:{}", b.host_path, b.container_path)
        })
        .collect();
    let flags_joined = flags.join(" ");
    let _ = writeln!(out, "EXTRA_NSPAWN_FLAGS=\"{flags_joined}\"");
    std::fs::write(&path, out).with_context(|| format!("write {path}"))
}