//! 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_priv_sock::{
    AGENT_PREFIX, AGENT_RUNTIME_ROOT, AGENT_STATE_ROOT, BindMount, CredentialMount, InfraAction,
    InfraContainer, JournalQuery, META_DIR, MIGRATE_STAGING_ROOT, NetworkIsolation,
    PAUSED_MARKER_FILE, 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.
// One match arm per priv op — a flat 1:1 dispatch table. The length tracks
// the op count, not complexity; splitting it would just scatter the mapping.
#[allow(clippy::too_many_lines)]
async fn exec(req: PrivRequest, writer: &mut OwnedWriteHalf) -> Result<(String, String)> {
    match req {
        PrivRequest::StartContainer { ref name } => {
            validate_container_name(name)?;
            let machine = container_system_name(name);
            // 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 we proceed.
            let _ = Command::new("systemctl")
                .args(["reset-failed", &format!("container@{machine}.service")])
                .status()
                .await;
            container_run(&["start", &machine]).await
        }

        PrivRequest::StopContainer { ref name } => {
            validate_container_name(name)?;
            stop_and_release(&container_system_name(name)).await
        }

        PrivRequest::KillContainer { ref name } => {
            validate_container_name(name)?;
            // nixos-container has no kill verb. Use machinectl to send SIGKILL
            // to all processes in the container — the right semantics for a
            // forced shutdown after a graceful stop has already been attempted.
            let machine = container_system_name(name);
            machinectl_run(&["kill", &machine, "--signal=SIGKILL"]).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,
            ref load_credentials,
        } => handle_write_nspawn_flags(container, binds, isolation.as_ref(), load_credentials),

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

        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::SetAgentPaused {
            ref agent_name,
            paused,
        } => {
            validate_agent_name(agent_name)?;
            set_agent_paused(agent_name, paused)
        }

        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,
            ref account,
            ref homeserver,
        } => {
            validate_agent_name(agent_name)?;
            // Build the token filename. `None` → the hive account's
            // `matrix-token`; `Some(a)` → `matrix-token-<a>`. The account
            // suffix MUST be validated as a plain identifier (no `/`, `.`,
            // `..`) before it goes into the filename, or a crafted account
            // could traverse out of the state dir — `write_agent_state_file`
            // trusts its `filename` argument.
            let filename = match account {
                None => "matrix-token".to_owned(),
                Some(a) => {
                    validate_name_chars(a)?;
                    format!("matrix-token-{a}")
                }
            };
            let res = write_agent_state_file(agent_name, &filename, &format!("{token}\n"))?;
            // For an extra account, persist its homeserver in a sidecar
            // (`matrix-account-<a>.json`) so the daemon can auto-discover the
            // account without a static `matrixAccounts` config entry. Only
            // when both `account` and `homeserver` are present; the account
            // suffix is already validated above.
            if let (Some(a), Some(hs)) = (account, homeserver) {
                let meta = serde_json::json!({ "homeserver": hs }).to_string();
                write_agent_state_file(agent_name, &format!("matrix-account-{a}.json"), &meta)?;
            }
            Ok(res)
        }

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

        PrivRequest::WriteAgentExtraForgeAccount {
            ref agent_name,
            ref label,
            ref base_url,
            ref token,
        } => {
            validate_agent_name(agent_name)?;
            validate_name_chars(label)?;
            let res = write_agent_state_file(
                agent_name,
                &format!("forge-{label}-token"),
                &format!("{token}\n"),
            )?;
            // Sidecar carries the base URL — there's no host-side nix config
            // for extra forges, so this is the only place it's persisted.
            let meta = serde_json::json!({ "base_url": base_url }).to_string();
            write_agent_state_file(agent_name, &format!("forge-{label}.json"), &meta)?;
            Ok(res)
        }

        PrivRequest::DeleteAgentExtraForgeAccount {
            ref agent_name,
            ref label,
        } => {
            validate_agent_name(agent_name)?;
            validate_name_chars(label)?;
            delete_agent_state_file(agent_name, &format!("forge-{label}-token"))?;
            delete_agent_state_file(agent_name, &format!("forge-{label}.json"))
        }

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

        PrivRequest::RegisterCiRunner { ref token } => register_ci_runner(token).await,

        PrivRequest::ControlInfraContainer { container, action } => {
            control_infra_container(container, action).await
        }

        PrivRequest::EnsureAgentSubvolume { ref agent_name } => {
            validate_agent_name(agent_name)?;
            ensure_agent_subvolume(agent_name).await
        }

        PrivRequest::DeleteAgentSubvolume { ref agent_name } => {
            validate_agent_name(agent_name)?;
            delete_agent_subvolume(agent_name).await
        }

        PrivRequest::EnsureBtrfsQuota => ensure_btrfs_quota().await,

        PrivRequest::ReadSubvolumeUsage { ref agent_name } => {
            validate_agent_name(agent_name)?;
            read_subvolume_usage(agent_name).await
        }

        PrivRequest::SetSubvolumeQuota {
            ref agent_name,
            limit_bytes,
        } => {
            validate_agent_name(agent_name)?;
            set_subvolume_quota(agent_name, limit_bytes).await
        }

        PrivRequest::UpgradeAgentSubvolume { ref agent_name } => {
            validate_agent_name(agent_name)?;
            upgrade_agent_subvolume(agent_name).await
        }

        PrivRequest::SnapshotAgentSubvolume {
            ref agent_name,
            ref snapshot_name,
        } => {
            validate_agent_name(agent_name)?;
            validate_snapshot_name(snapshot_name)?;
            snapshot_agent_subvolume(agent_name, snapshot_name).await
        }

        PrivRequest::DeleteAgentSnapshot {
            ref agent_name,
            ref snapshot_name,
        } => {
            validate_agent_name(agent_name)?;
            validate_snapshot_name(snapshot_name)?;
            delete_agent_snapshot(agent_name, snapshot_name).await
        }

        PrivRequest::SendAgentSnapshotToFile {
            ref agent_name,
            ref snapshot_name,
            ref parent_snapshot_name,
            ref dest_file_name,
        } => {
            validate_agent_name(agent_name)?;
            validate_snapshot_name(snapshot_name)?;
            if let Some(parent) = parent_snapshot_name {
                validate_snapshot_name(parent)?;
            }
            validate_credential_name(dest_file_name)?;
            send_agent_snapshot_to_file(
                agent_name,
                snapshot_name,
                parent_snapshot_name.as_deref(),
                dest_file_name,
            )
            .await
        }

        PrivRequest::SyncAgentTmpfiles { ref agents } => sync_agent_tmpfiles(agents).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 + every
/// credential entry, then write the container's nspawn flag overrides.
fn handle_write_nspawn_flags(
    container: &str,
    binds: &[BindMount],
    isolation: Option<&NetworkIsolation>,
    load_credentials: &[CredentialMount],
) -> Result<(String, String)> {
    validate_container_system_name(container)?;
    for bind in binds {
        validate_bind_path(&bind.host_path)?;
        validate_bind_path(&bind.container_path)?;
    }
    for cred in load_credentials {
        validate_credential_name(&cred.name)?;
        // Same path rules as binds (absolute, no colon/newline/quote/null):
        // the colon ban is essential since `--load-credential=name:path`
        // uses `:` as the name/path separator.
        validate_bind_path(&cred.host_path)?;
    }
    write_nspawn_flags(container, binds, isolation, load_credentials)?;
    Ok((String::new(), String::new()))
}

/// A btrfs snapshot label must start with `hive-` — this doubles as an
/// allow-list: only names hivectl itself constructs (or an operator who
/// knows the convention) can reach the `btrfs subvolume snapshot`/`delete`
/// shellouts, so an arbitrary caller can't use the snapshot ops to probe or
/// churn unrelated paths under `AGENT_STATE_ROOT`. Beyond the prefix, the
/// same charset restriction as [`validate_credential_name`] applies (it's
/// interpolated straight into a filesystem path).
fn validate_snapshot_name(name: &str) -> Result<()> {
    if !name.starts_with("hive-") {
        bail!("invalid snapshot label {name:?}: must start with \"hive-\"");
    }
    validate_credential_name(name)
}

/// A systemd credential id must be a short token — restrict to
/// `[A-Za-z0-9_-]` (no `.`) so it can't inject extra `--load-credential`
/// argv or break the `name:path` shape. `.` is deliberately excluded, not
/// just a bare `..`: this name gets interpolated into filesystem paths
/// (snapshot labels) and there's no legitimate need for a dot in either a
/// systemd credential id or a `hive-`-prefixed snapshot label — we're
/// defining this token format from scratch, so keep it maximally strict
/// rather than allow-then-patch each traversal-adjacent character
/// (mara: "we are making up the rules here, lets go strict").
fn validate_credential_name(name: &str) -> Result<()> {
    if name.is_empty()
        || !name
            .bytes()
            .all(|b| b.is_ascii_alphanumeric() || matches!(b, b'_' | b'-'))
    {
        bail!("invalid credential name {name:?}: must be non-empty [A-Za-z0-9_-]");
    }
    Ok(())
}

/// `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 the systemd resource settings into the
/// container service's drop-in dir, together with a
/// `ConditionPathIsDirectory=` guard on the agent's MCP runtime dir.
///
/// Two different kinds of setting land in the same file. `MemoryMax=` /
/// `CPUQuota=` are hard caps that throttle even on an idle host;
/// `CPUWeight=` / `IOWeight=` are cgroup v2 relative shares that only
/// decide who yields *under contention*. A zero weight means "not
/// configured" (`None`) and omits the line, so a hive-c0re built before the
/// weights existed keeps producing the old two-line drop-in.
///
/// The condition causes systemd to *skip* (not *fail*) the unit when the
/// bind-mount source dir is absent — result is `condition`, which does not
/// increment the start-limit counter.  This is belt-and-braces on top of
/// the tmpfiles.d entries written by `SyncAgentTmpfiles`: in the unlikely
/// event the dir is missing at start time, the unit idles rather than
/// restart-looping into `start-limit-hit`.
fn write_resource_limits(
    container: &str,
    memory_max: &str,
    cpu_quota: &str,
    cpu_weight: Option<u32>,
    io_weight: Option<u32>,
) -> Result<(String, String)> {
    validate_container_system_name(container)?;
    // Derive the logical agent name (strip h- prefix) to form the runtime
    // dir path. Falls back to the full container name for infra containers
    // that don't use the h- prefix.
    let logical = container.strip_prefix(AGENT_PREFIX).unwrap_or(container);
    let runtime_dir = format!("{AGENT_RUNTIME_ROOT}/{logical}");
    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 = limits_dropin_body(&runtime_dir, memory_max, cpu_quota, cpu_weight, io_weight);
    std::fs::write(&path, content).with_context(|| format!("write {path}"))?;
    Ok((String::new(), String::new()))
}

/// Render the body of `hyperhive-limits.conf`.
///
/// `[Unit]`: the condition is checked at start time — it skips (not fails)
/// the unit when the MCP socket dir is absent, avoiding restart loops.
/// `[Service]`: the hard caps first, then the relative weights. A weight of
/// `None` means "not configured" and omits its line entirely, so a request
/// from a hive-c0re built before the weights existed — or one whose nix
/// option is `null` — reproduces the old two-setting drop-in byte for byte.
fn limits_dropin_body(
    runtime_dir: &str,
    memory_max: &str,
    cpu_quota: &str,
    cpu_weight: Option<u32>,
    io_weight: Option<u32>,
) -> String {
    // Built as two possibly-empty lines rather than pushed onto the
    // string: `format!` appended to a `String` trips clippy::pedantic's
    // `format_push_string`, and a `write!` would need an unwrap.
    let cpu_weight_line = cpu_weight.map_or_else(String::new, |w| format!("CPUWeight={w}\n"));
    let io_weight_line = io_weight.map_or_else(String::new, |w| format!("IOWeight={w}\n"));
    format!(
        "[Unit]\n\
         ConditionPathIsDirectory={runtime_dir}\n\
         \n\
         [Service]\n\
         MemoryMax={memory_max}\n\
         CPUQuota={cpu_quota}\n\
         {cpu_weight_line}{io_weight_line}"
    )
}

/// `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(),
    ))
}

/// `RegisterCiRunner` — write the runner registration token to the host-side
/// `/run/hive-ci/runner-token` env-file, then restart the in-container runner
/// so it re-registers. The forge admin token never enters the container; only
/// the registration token c0re passes here is written, and it lands on a host
/// path bind-mounted read-only into hive-ci.
async fn register_ci_runner(token: &str) -> Result<(String, String)> {
    use std::os::unix::fs::PermissionsExt as _;
    // Reject anything that could corrupt the `KEY=VALUE` env-file or smuggle a
    // second line — a forge registration token is an opaque single-line string.
    if token.is_empty() || token.contains(['\n', '\r', '\0']) {
        bail!("ci runner registration token empty or contains control characters");
    }
    let token_path = "/run/hive-ci/runner-token";
    // In-place truncate+write of the existing inode (mirrors the prefetch's
    // `echo > $FILE`), NOT a temp+rename: nspawn pins this file's inode into
    // hive-ci at container start, so a rename would leave the running runner
    // reading the old content. Format + perms match the tmpfiles seed and the
    // prefetch: `TOKEN=<tok>`, mode 0600, root-owned.
    std::fs::write(token_path, format!("TOKEN={token}\n"))
        .with_context(|| format!("write {token_path}"))?;
    std::fs::set_permissions(token_path, std::fs::Permissions::from_mode(0o600))
        .with_context(|| format!("chmod {token_path}"))?;
    // Restart the in-container runner so it reads the new token and registers.
    let out = Command::new("systemctl")
        .args(["--machine=hive-ci", "restart", "gitea-runner-hive.service"])
        .output()
        .await
        .context("systemctl restart gitea-runner-hive.service in hive-ci")?;
    if !out.status.success() {
        bail!(
            "systemctl restart gitea-runner-hive.service in hive-ci 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(),
    ))
}

/// `ControlInfraContainer` — start/stop/restart a hive infrastructure
/// container via `systemctl <verb> container@<container>.service`. The
/// [`InfraContainer`] enum is the allowlist: serde already rejected any
/// unknown / unsafe name (hive-c0re has no variant, so a stop can't sever
/// the daemon socket) at deserialisation, so no root-side `.contains()`
/// check is needed here. Serves both the hive-wide `hivectl stop`/`start`
/// flow and an `infra_admin` agent's `restart` (action = Restart).
async fn control_infra_container(
    container: InfraContainer,
    action: InfraAction,
) -> Result<(String, String)> {
    let verb = action.systemctl_verb();
    let unit = format!("container@{}.service", container.unit_name());
    let out = Command::new("systemctl")
        .args([verb, &unit])
        .output()
        .await
        .with_context(|| format!("systemctl {verb} {unit}"))?;
    if !out.status.success() {
        bail!(
            "systemctl {verb} {unit} exited {}: {}",
            out.status,
            String::from_utf8_lossy(&out.stderr).trim()
        );
    }
    tracing::info!(target: "infra-control", "{verb} {unit}");
    Ok((
        String::from_utf8_lossy(&out.stdout).into_owned(),
        String::from_utf8_lossy(&out.stderr).into_owned(),
    ))
}

/// Create/overwrite `dir/filename` at 0600 without following a symlink at the
/// leaf, returning the open fd for the caller to `fchown`. `filename` must be a
/// single plain component (no `/`, `.`, `..`) — the leaf sits in an
/// agent-writable dir, so `O_NOFOLLOW` refuses a planted symlink (`ELOOP`)
/// instead of letting this root-privileged write/chmod be redirected at another
/// file; `O_WRONLY` refuses a directory leaf (`EISDIR`); `O_TRUNC` keeps the
/// overwrite semantics for an existing regular file. `.mode(0o600)` sets the
/// create mode; the explicit `fchmod` after (on the fd, not a re-resolved path)
/// tightens an already-existing file and dodges umask. The returned fd is the
/// exact inode the write hit, so the caller's `fchown` is TOCTOU-immune.
fn write_state_file_nofollow(dir: &Path, filename: &str, content: &str) -> Result<std::fs::File> {
    use std::io::Write as _;
    use std::os::unix::fs::{OpenOptionsExt as _, PermissionsExt as _};

    if filename.is_empty() || filename == "." || filename == ".." || filename.contains('/') {
        bail!("write_state_file_nofollow: refusing non-plain filename {filename:?}");
    }
    let path = dir.join(filename);
    let mut file = std::fs::OpenOptions::new()
        .write(true)
        .create(true)
        .truncate(true)
        .mode(0o600)
        .custom_flags(libc::O_NOFOLLOW)
        .open(&path)
        .with_context(|| format!("open (no-follow) {}", path.display()))?;
    file.write_all(content.as_bytes())
        .with_context(|| format!("write {}", path.display()))?;
    file.set_permissions(std::fs::Permissions::from_mode(0o600))
        .with_context(|| format!("chmod 600 {}", path.display()))?;
    Ok(file)
}

/// 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)> {
    let state_dir = PathBuf::from(AGENT_STATE_ROOT)
        .join(agent_name)
        .join("state");
    write_agent_dir_file(agent_name, &state_dir, filename, content)
}

/// Create or remove an agent's pause marker under its harness dir. The
/// marker is written empty and chowned to the harness dir's owner (the
/// agent), matching how the harness itself would have created it.
///
/// Both directions are idempotent: re-pausing truncates the existing empty
/// marker rather than failing, and a `NotFound` on removal is the
/// already-resumed case, not an error.
fn set_agent_paused(agent_name: &str, paused: bool) -> Result<(String, String)> {
    let harness_dir = PathBuf::from(AGENT_STATE_ROOT)
        .join(agent_name)
        .join("harness");
    if paused {
        return write_agent_dir_file(agent_name, &harness_dir, PAUSED_MARKER_FILE, "");
    }
    remove_marker_in(&harness_dir, PAUSED_MARKER_FILE)?;
    tracing::info!(agent = %agent_name, "cleared pause marker");
    Ok((String::new(), String::new()))
}

/// Unlink `dir/filename`, treating "already gone" as success.
///
/// `remove_file` unlinks the leaf itself and never follows a symlink, so an
/// agent-planted link at the marker path cannot redirect this root unlink
/// at another file — the same threat `write_state_file_nofollow` closes on
/// the create side.
fn remove_marker_in(dir: &Path, filename: &str) -> Result<()> {
    let path = dir.join(filename);
    match std::fs::remove_file(&path) {
        Ok(()) => Ok(()),
        Err(e) if e.kind() == std::io::ErrorKind::NotFound => Ok(()),
        Err(e) => Err(e).with_context(|| format!("remove {}", path.display())),
    }
}

/// Write `content` to `dir/filename` as root, chowning the result to `dir`'s
/// owner so the agent process can read it back. Shared by the credential
/// writes (which target `state/`) and the pause marker (which targets
/// `harness/`) — both write into a directory owned by the agent, which is
/// precisely why they need hive-priv at all.
fn write_agent_dir_file(
    agent_name: &str,
    dir: &Path,
    filename: &str,
    content: &str,
) -> Result<(String, String)> {
    use std::os::fd::AsRawFd as _;
    use std::os::unix::fs::MetadataExt as _;

    let state_dir = dir.to_path_buf();
    // 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()))?;

    // Security-critical: refuses a symlink the (state/-owning) agent may have
    // planted at the leaf, so this root-privileged create/write/chmod/chown
    // can't be redirected at an arbitrary file. See `write_state_file_nofollow`.
    let path = state_dir.join(filename);
    let file = write_state_file_nofollow(&state_dir, filename, content)?;

    // Chown to the state dir's owner so the agent process can read the file.
    // fchown on the same fd — TOCTOU-immune (the inode the write hit, never a
    // swapped path). 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) => {
            // SAFETY: `file` is an open, owned fd live for the whole call;
            // `fchown` only mutates that inode's uid/gid.
            let rc = unsafe { libc::fchown(file.as_raw_fd(), meta.uid(), meta.gid()) };
            if rc != 0 {
                let e = std::io::Error::last_os_error();
                tracing::warn!(
                    agent = %agent_name,
                    path = %path.display(),
                    error = %e,
                    "write_agent_state_file: fchown 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,
        dir = %state_dir.display(),
        file = %filename,
        "wrote agent file"
    );
    Ok((String::new(), String::new()))
}

/// Remove `AGENT_STATE_ROOT/<agent_name>/state/<filename>` if present.
/// Idempotent revoke counterpart to [`write_agent_state_file`] — a
/// missing file is success, not an error. `filename` must be a single
/// plain component (no `/`, `.`, `..`); callers pass a pre-validated
/// label into a fixed `forge-<label>-token` shape, same as the write
/// side.
fn delete_agent_state_file(agent_name: &str, filename: &str) -> Result<(String, String)> {
    if filename.is_empty() || filename == "." || filename == ".." || filename.contains('/') {
        bail!("delete_agent_state_file: refusing non-plain filename {filename:?}");
    }
    let path = PathBuf::from(AGENT_STATE_ROOT)
        .join(agent_name)
        .join("state")
        .join(filename);
    match std::fs::remove_file(&path) {
        Ok(()) => {
            tracing::info!(agent = %agent_name, file = %filename, "removed agent state file");
        }
        Err(e) if e.kind() == std::io::ErrorKind::NotFound => {}
        Err(e) => return Err(e).with_context(|| format!("remove {}", path.display())),
    }
    Ok((String::new(), String::new()))
}

/// btrfs superblock magic, as reported by `statfs(2)`'s `f_type`.
const BTRFS_SUPER_MAGIC: i64 = 0x9123_683E;

/// Inode number of a btrfs subvolume root (`BTRFS_FIRST_FREE_OBJECTID`).
/// Every subvolume's top directory has this inode; plain directories do
/// not, so `statfs == btrfs && st_ino == 256` reliably identifies a
/// subvolume root.
const BTRFS_SUBVOL_ROOT_INO: u64 = 256;

/// Whether `path` lives on a btrfs filesystem (via `statfs(2)`).
fn is_on_btrfs(path: &Path) -> Result<bool> {
    use std::os::unix::ffi::OsStrExt as _;
    let c_path = std::ffi::CString::new(path.as_os_str().as_bytes())
        .with_context(|| format!("path {} has an interior null byte", path.display()))?;
    // SAFETY: `c_path` is a valid NUL-terminated C string that outlives the
    // call; `statfs` only writes into the zero-initialised `buf`.
    let mut buf: libc::statfs = unsafe { std::mem::zeroed() };
    let rc = unsafe { libc::statfs(c_path.as_ptr(), &raw mut buf) };
    if rc != 0 {
        return Err(std::io::Error::last_os_error())
            .with_context(|| format!("statfs {}", path.display()));
    }
    Ok(buf.f_type == BTRFS_SUPER_MAGIC)
}

/// Whether `path` is the root of a btrfs subvolume (on btrfs and inode 256).
fn is_btrfs_subvolume(path: &Path) -> bool {
    use std::os::unix::fs::MetadataExt as _;
    let on_btrfs = is_on_btrfs(path).unwrap_or(false);
    let ino_match = std::fs::metadata(path).is_ok_and(|m| m.ino() == BTRFS_SUBVOL_ROOT_INO);
    on_btrfs && ino_match
}

/// `EnsureAgentSubvolume` — make the agent's state root a btrfs subvolume
/// when the FS supports it. Idempotent + progressive: no-op when the root
/// already exists or the FS isn't btrfs. See the wire doc on the variant.
async fn ensure_agent_subvolume(agent_name: &str) -> Result<(String, String)> {
    use std::os::unix::fs::MetadataExt as _;

    let root = PathBuf::from(AGENT_STATE_ROOT);
    let agent_root = root.join(agent_name);

    // Progressive: existing agents (plain dir OR already a subvol) are left
    // untouched — never auto-migrated.
    if agent_root.exists() {
        return Ok((String::new(), String::new()));
    }

    // Only btrfs supports subvolumes; on anything else hive-c0re's normal
    // `create_dir_all` makes a plain dir (the pre-subvolume behaviour). The
    // parent must exist for both statfs and `btrfs subvolume create`.
    std::fs::create_dir_all(&root)
        .with_context(|| format!("create agents root {}", root.display()))?;
    if !is_on_btrfs(&root)? {
        return Ok((String::new(), String::new()));
    }

    let out = Command::new("btrfs")
        .args(["subvolume", "create"])
        .arg(&agent_root)
        .output()
        .await
        .with_context(|| format!("spawn btrfs subvolume create {}", agent_root.display()))?;
    if !out.status.success() {
        bail!(
            "btrfs subvolume create {} failed: {}",
            agent_root.display(),
            String::from_utf8_lossy(&out.stderr).trim()
        );
    }

    // The subvol root is created root-owned, but hive-c0re (the `hive-core`
    // user) must be able to mkdir state/ claude/ harness/ inside it — exactly
    // as it would in a plain dir. Chown it to AGENT_STATE_ROOT's owner
    // (hive-core). This MUST succeed: a root-owned subvol would make the
    // downstream dir creation fail with a confusing permission error, and the
    // c0re-side exists-check would then skip re-running this op on retry,
    // wedging the agent. So on any failure roll the subvol back and bail — the
    // create path surfaces a clear error and a retry starts clean.
    let chown_result = std::fs::metadata(&root)
        .with_context(|| format!("stat agents root {} for ownership", root.display()))
        .and_then(|meta| {
            std::os::unix::fs::chown(&agent_root, Some(meta.uid()), Some(meta.gid())).with_context(
                || format!("chown subvol {} to agents-root owner", agent_root.display()),
            )
        });
    if let Err(e) = chown_result {
        // Best-effort rollback so we never leave a root-owned subvol behind.
        let _ = Command::new("btrfs")
            .args(["subvolume", "delete"])
            .arg(&agent_root)
            .output()
            .await;
        return Err(e.context(format!(
            "rolled back subvolume {} after chown failed",
            agent_root.display()
        )));
    }
    tracing::info!(agent = %agent_name, path = %agent_root.display(), "created agent state subvolume");
    Ok((String::new(), String::new()))
}

/// `DeleteAgentSubvolume` — delete the agent's state root iff it's a btrfs
/// subvolume (purge path only). No-op for plain dirs / missing paths;
/// hive-c0re's own `remove_dir_all` covers those. See the wire doc.
async fn delete_agent_subvolume(agent_name: &str) -> Result<(String, String)> {
    let agent_root = PathBuf::from(AGENT_STATE_ROOT).join(agent_name);
    if !is_btrfs_subvolume(&agent_root) {
        return Ok((String::new(), String::new()));
    }
    let out = Command::new("btrfs")
        .args(["subvolume", "delete"])
        .arg(&agent_root)
        .output()
        .await
        .with_context(|| format!("spawn btrfs subvolume delete {}", agent_root.display()))?;
    if !out.status.success() {
        bail!(
            "btrfs subvolume delete {} failed: {}",
            agent_root.display(),
            String::from_utf8_lossy(&out.stderr).trim()
        );
    }
    tracing::info!(agent = %agent_name, path = %agent_root.display(), "deleted agent state subvolume");
    Ok((String::new(), String::new()))
}

/// `EnsureBtrfsQuota` — enable btrfs qgroup accounting on the filesystem
/// holding `AGENT_STATE_ROOT`. Idempotent + statfs-gated (no-op off btrfs).
/// Operator opt-in only; see the wire doc.
async fn ensure_btrfs_quota() -> Result<(String, String)> {
    let root = PathBuf::from(AGENT_STATE_ROOT);
    std::fs::create_dir_all(&root)
        .with_context(|| format!("create agents root {}", root.display()))?;
    if !is_on_btrfs(&root)? {
        // Non-btrfs host: quota/qgroups don't apply. No-op success so the
        // operator-facing verb degrades cleanly.
        return Ok((
            String::new(),
            "filesystem is not btrfs — quota not applicable".to_owned(),
        ));
    }
    let out = Command::new("btrfs")
        .args(["quota", "enable"])
        .arg(&root)
        .output()
        .await
        .with_context(|| format!("spawn btrfs quota enable {}", root.display()))?;
    if !out.status.success() {
        bail!(
            "btrfs quota enable {} failed: {}",
            root.display(),
            String::from_utf8_lossy(&out.stderr).trim()
        );
    }
    tracing::info!(path = %root.display(), "enabled btrfs qgroup accounting");
    Ok((String::new(), String::new()))
}

/// `ReadSubvolumeUsage` — return an agent subvolume's qgroup rows
/// (`btrfs qgroup show -f --raw <…/agent_name>`) verbatim in stdout for
/// hive-c0re to parse. `-f` lists the qgroups impacting the given path,
/// excluding ancestral qgroups (per btrfs-qgroup-show(8)) — so it scopes
/// to this subvolume and never mixes in other agents'. hive-c0re then
/// selects the level-0 (`0/<subvolid>`) leaf row. See the wire doc.
async fn read_subvolume_usage(agent_name: &str) -> Result<(String, String)> {
    let agent_root = PathBuf::from(AGENT_STATE_ROOT).join(agent_name);
    let out = Command::new("btrfs")
        .args(["qgroup", "show", "-f", "--raw"])
        .arg(&agent_root)
        .output()
        .await
        .with_context(|| format!("spawn btrfs qgroup show {}", agent_root.display()))?;
    if !out.status.success() {
        // The common failure is "quota not enabled" — pass the stderr
        // through so hive-c0re can surface it gracefully.
        bail!(
            "btrfs qgroup show {} failed: {}",
            agent_root.display(),
            String::from_utf8_lossy(&out.stderr).trim()
        );
    }
    Ok((
        String::from_utf8_lossy(&out.stdout).into_owned(),
        String::new(),
    ))
}

/// Best-effort removal of a leftover migration path from a prior aborted
/// upgrade: try `btrfs subvolume delete` (in case it's a half-created
/// subvolume) then a plain recursive remove. Both failures are ignored —
/// the path may simply not exist.
async fn cleanup_stale_path(path: &Path) {
    if path.exists() {
        let _ = Command::new("btrfs")
            .args(["subvolume", "delete"])
            .arg(path)
            .output()
            .await;
        let _ = std::fs::remove_dir_all(path);
    }
}

/// Stage a populated subvolume at `tmp` mirroring `agent_root`: create the
/// subvolume, copy `agent_root`'s contents into it preserving
/// ownership/permissions/xattrs, then match the subvolume root's owner + mode
/// to the original. On any failure the partially-staged `tmp` is cleaned up
/// (so the caller can bail with the original dir still untouched).
async fn stage_upgrade_subvolume(agent_root: &Path, tmp: &Path) -> Result<()> {
    use std::os::unix::fs::{MetadataExt as _, PermissionsExt as _};

    // Fresh subvolume to receive the copy.
    let out = Command::new("btrfs")
        .args(["subvolume", "create"])
        .arg(tmp)
        .output()
        .await
        .with_context(|| format!("spawn btrfs subvolume create {}", tmp.display()))?;
    if !out.status.success() {
        bail!(
            "btrfs subvolume create {} failed: {}",
            tmp.display(),
            String::from_utf8_lossy(&out.stderr).trim()
        );
    }

    // Copy contents preserving everything (`-a` = --preserve=all → mode,
    // ownership, timestamps, links, xattrs); reflink for fast CoW clones on the
    // same btrfs. `<src>/.` copies the directory's contents (incl. dotfiles)
    // into the subvolume rather than nesting it.
    let copy = Command::new("cp")
        .arg("-a")
        .arg("--reflink=auto")
        .arg(format!("{}/.", agent_root.display()))
        .arg(tmp)
        .output()
        .await
        .with_context(|| format!("spawn cp into {}", tmp.display()))?;
    if !copy.status.success() {
        cleanup_stale_path(tmp).await;
        bail!(
            "copy {} -> {} failed (original left untouched): {}",
            agent_root.display(),
            tmp.display(),
            String::from_utf8_lossy(&copy.stderr).trim()
        );
    }

    // Match the new subvolume root's ownership + mode to the original dir.
    // `cp -a <src>/.` copies the *contents* but the subvolume root keeps its
    // create-time root ownership, so set it explicitly — the swapped-in
    // subvolume must be indistinguishable from the original to hive-c0re.
    let apply = std::fs::metadata(agent_root)
        .with_context(|| format!("stat {} for ownership", agent_root.display()))
        .and_then(|m| {
            std::os::unix::fs::chown(tmp, Some(m.uid()), Some(m.gid()))
                .with_context(|| format!("chown {} to match original", tmp.display()))?;
            std::fs::set_permissions(tmp, std::fs::Permissions::from_mode(m.mode()))
                .with_context(|| format!("chmod {} to match original", tmp.display()))?;
            Ok(())
        });
    if let Err(e) = apply {
        cleanup_stale_path(tmp).await;
        return Err(e.context("upgrade aborted before swap; original left untouched"));
    }
    Ok(())
}

/// `UpgradeAgentSubvolume` — convert an existing plain-dir agent state root
/// into a btrfs subvolume in place. Operator opt-in; the caller (hivectl)
/// stops the agent first and restarts it after. See the wire doc.
///
/// Migration: stage a sibling subvolume mirroring the dir
/// ([`stage_upgrade_subvolume`]), then rename the original aside and the
/// subvolume into place, then remove the original. Any failure before the
/// rename-swap leaves the original dir untouched.
async fn upgrade_agent_subvolume(agent_name: &str) -> Result<(String, String)> {
    let root = PathBuf::from(AGENT_STATE_ROOT);
    let agent_root = root.join(agent_name);

    if !agent_root.exists() {
        // A crash between the two swap renames (original → `.<name>.old`
        // succeeded, `.<name>.migrating` → agent_root did not) leaves the
        // agent root missing but the original data intact under `.<name>.old`.
        // Point at the recovery rather than a bare "nothing to do" so the
        // operator isn't left guessing where the data went.
        let old = root.join(format!(".{agent_name}.old"));
        if old.exists() {
            bail!(
                "no state dir at {agent} — but {old} holds the original data from an \
                 interrupted upgrade (host crashed mid-swap). Restore it with \
                 `mv {old} {agent}`, then re-run the upgrade.",
                agent = agent_root.display(),
                old = old.display(),
            );
        }
        bail!(
            "no state dir to upgrade at {} — nothing to do",
            agent_root.display()
        );
    }
    // Idempotent: already a subvolume → nothing to do.
    if is_btrfs_subvolume(&agent_root) {
        return Ok((String::new(), String::new()));
    }
    if !is_on_btrfs(&root)? {
        bail!(
            "{} is not on btrfs — subvolumes are unsupported, cannot upgrade",
            root.display()
        );
    }

    // Sibling temp paths on the same filesystem (so the copy can reflink and
    // the swap renames are atomic). Leading dots keep them out of the agent
    // namespace (`validate_agent_name` rejects dot-prefixed names).
    let tmp = root.join(format!(".{agent_name}.migrating"));
    let old = root.join(format!(".{agent_name}.old"));
    // Clear any debris from a previously interrupted run before starting.
    cleanup_stale_path(&tmp).await;
    cleanup_stale_path(&old).await;

    stage_upgrade_subvolume(&agent_root, &tmp).await?;

    // Swap. `rename` is atomic within a filesystem. The window between the
    //    two renames is the only unsafe point: a crash there leaves the agent
    //    root missing but both `.old` (original) and the new subvolume present
    //    — recoverable by hand, hence the loud logging.
    if let Err(e) = std::fs::rename(&agent_root, &old) {
        cleanup_stale_path(&tmp).await;
        return Err(anyhow::Error::new(e).context(format!(
            "rename {} -> {} failed; original left untouched",
            agent_root.display(),
            old.display()
        )));
    }
    if let Err(e) = std::fs::rename(&tmp, &agent_root) {
        // Restore the original from its renamed-aside copy.
        let restored = std::fs::rename(&old, &agent_root).is_ok();
        cleanup_stale_path(&tmp).await;
        return Err(anyhow::Error::new(e).context(format!(
            "rename {} -> {} failed; original {}",
            tmp.display(),
            agent_root.display(),
            if restored {
                "restored"
            } else {
                "COULD NOT BE RESTORED — manual recovery needed"
            }
        )));
    }

    // 5. Success: drop the original (a plain dir) and report.
    if let Err(e) = std::fs::remove_dir_all(&old) {
        // The migration succeeded; a leftover `.old` is cosmetic. Warn only.
        tracing::warn!(
            agent = %agent_name, path = %old.display(),
            "upgraded subvolume but failed to remove old dir: {e}"
        );
    }
    tracing::info!(
        agent = %agent_name, path = %agent_root.display(),
        "upgraded agent state dir to btrfs subvolume"
    );
    Ok((
        format!("upgraded {} to a btrfs subvolume", agent_root.display()),
        String::new(),
    ))
}

/// Derive a snapshot's path from the agent name + label: a dot-prefixed
/// sibling of the agent's state root so it can never collide with a real
/// agent directory (`validate_agent_name` rejects dot-prefixed names).
fn snapshot_path(agent_name: &str, snapshot_name: &str) -> PathBuf {
    PathBuf::from(AGENT_STATE_ROOT).join(format!(".{agent_name}.snapshot.{snapshot_name}"))
}

/// `SnapshotAgentSubvolume` — create a read-only btrfs snapshot of an
/// agent's state subvolume, for `btrfs send` to stream from during
/// inter-hive migration. See the wire doc.
async fn snapshot_agent_subvolume(
    agent_name: &str,
    snapshot_name: &str,
) -> Result<(String, String)> {
    let agent_root = PathBuf::from(AGENT_STATE_ROOT).join(agent_name);
    if !is_btrfs_subvolume(&agent_root) {
        bail!(
            "{} is not a btrfs subvolume — nothing to snapshot (run `hivectl agents subvol upgrade` first)",
            agent_root.display()
        );
    }
    let snap = snapshot_path(agent_name, snapshot_name);
    if snap.exists() {
        bail!(
            "snapshot {} already exists — delete it first or pick a different name",
            snap.display()
        );
    }
    let out = Command::new("btrfs")
        .args(["subvolume", "snapshot", "-r"])
        .arg(&agent_root)
        .arg(&snap)
        .output()
        .await
        .with_context(|| {
            format!(
                "spawn btrfs subvolume snapshot -r {} {}",
                agent_root.display(),
                snap.display()
            )
        })?;
    if !out.status.success() {
        bail!(
            "btrfs subvolume snapshot -r {} {} failed: {}",
            agent_root.display(),
            snap.display(),
            String::from_utf8_lossy(&out.stderr).trim()
        );
    }
    tracing::info!(
        agent = %agent_name, snapshot = %snap.display(),
        "created read-only agent state snapshot"
    );
    Ok((snap.display().to_string(), String::new()))
}

/// `DeleteAgentSnapshot` — delete a previously-created read-only snapshot.
/// No-op if the path doesn't exist. See the wire doc.
async fn delete_agent_snapshot(agent_name: &str, snapshot_name: &str) -> Result<(String, String)> {
    let snap = snapshot_path(agent_name, snapshot_name);
    if !snap.exists() {
        return Ok((String::new(), String::new()));
    }
    let out = Command::new("btrfs")
        .args(["subvolume", "delete"])
        .arg(&snap)
        .output()
        .await
        .with_context(|| format!("spawn btrfs subvolume delete {}", snap.display()))?;
    if !out.status.success() {
        bail!(
            "btrfs subvolume delete {} failed: {}",
            snap.display(),
            String::from_utf8_lossy(&out.stderr).trim()
        );
    }
    tracing::info!(agent = %agent_name, snapshot = %snap.display(), "deleted agent state snapshot");
    Ok((String::new(), String::new()))
}

/// `SendAgentSnapshotToFile` — stream a read-only snapshot (optionally
/// incremental against `parent_name`) to a file under
/// `MIGRATE_STAGING_ROOT` via `btrfs send`. Local-file half of the
/// inter-hive migration transport; see `PrivRequest::SendAgentSnapshotToFile`
/// for the cross-hive follow-up.
async fn send_agent_snapshot_to_file(
    agent_name: &str,
    snapshot_name: &str,
    parent_name: Option<&str>,
    dest_file_name: &str,
) -> Result<(String, String)> {
    let snap = snapshot_path(agent_name, snapshot_name);
    if !snap.exists() {
        bail!(
            "snapshot {} does not exist — create it with `subvol snapshot create` first",
            snap.display()
        );
    }

    std::fs::create_dir_all(MIGRATE_STAGING_ROOT)
        .with_context(|| format!("create {MIGRATE_STAGING_ROOT}"))?;
    let dest = Path::new(MIGRATE_STAGING_ROOT).join(dest_file_name);
    // `create_new` (O_CREAT|O_EXCL) makes the no-overwrite guarantee atomic
    // instead of a check-then-create race against a concurrent request.
    let dest_file = match std::fs::File::options()
        .write(true)
        .create_new(true)
        .open(&dest)
    {
        Ok(f) => f,
        Err(e) if e.kind() == std::io::ErrorKind::AlreadyExists => bail!(
            "{} already exists — pick a different destination or remove it first \
             (send never overwrites an existing export)",
            dest.display()
        ),
        Err(e) => {
            return Err(e).with_context(|| format!("create {}", dest.display()));
        }
    };

    let mut cmd = Command::new("btrfs");
    cmd.arg("send");
    if let Some(parent) = parent_name {
        let parent_path = snapshot_path(agent_name, parent);
        if !parent_path.exists() {
            bail!(
                "parent snapshot {} does not exist — pick an existing parent or omit it for a full send",
                parent_path.display()
            );
        }
        cmd.arg("-p").arg(&parent_path);
    }
    cmd.arg(&snap);
    cmd.stdout(std::process::Stdio::from(dest_file));
    cmd.stderr(std::process::Stdio::piped());

    let out = cmd
        .spawn()
        .with_context(|| format!("spawn btrfs send {}", snap.display()))?
        .wait_with_output()
        .await
        .with_context(|| format!("wait on btrfs send {}", snap.display()))?;
    if !out.status.success() {
        // Clean up a partial/failed export so a retry doesn't trip the
        // "already exists" guard on garbage. Best-effort: warn (don't fail
        // the whole call over it) if removal itself fails, so a stuck
        // partial file that later masquerades as a completed export is at
        // least visible in the log.
        if let Err(rm_err) = std::fs::remove_file(&dest) {
            tracing::warn!(
                dest = %dest.display(), error = %rm_err,
                "failed to remove partial export after btrfs send failure — \
                 next attempt at this dest will hit the already-exists guard"
            );
        }
        bail!(
            "btrfs send {} failed: {}",
            snap.display(),
            String::from_utf8_lossy(&out.stderr).trim()
        );
    }
    tracing::info!(
        agent = %agent_name, snapshot = %snap.display(), dest = %dest.display(),
        parent = ?parent_name, "exported agent snapshot to file"
    );
    Ok((dest.display().to_string(), String::new()))
}

/// `SetSubvolumeQuota` — set or clear a qgroup size limit on an agent
/// subvolume (`btrfs qgroup limit <bytes|none> <…/agent_name>`). See the
/// wire doc.
async fn set_subvolume_quota(
    agent_name: &str,
    limit_bytes: Option<u64>,
) -> Result<(String, String)> {
    let agent_root = PathBuf::from(AGENT_STATE_ROOT).join(agent_name);
    let limit = limit_bytes.map_or_else(|| "none".to_owned(), |n| n.to_string());
    let out = Command::new("btrfs")
        .args(["qgroup", "limit", &limit])
        .arg(&agent_root)
        .output()
        .await
        .with_context(|| format!("spawn btrfs qgroup limit {}", agent_root.display()))?;
    if !out.status.success() {
        bail!(
            "btrfs qgroup limit {limit} {} failed: {}",
            agent_root.display(),
            String::from_utf8_lossy(&out.stderr).trim()
        );
    }
    tracing::info!(agent = %agent_name, %limit, "set agent subvolume quota");
    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(())
}

/// Redact a line before it hits the (root-readable, but still
/// unnecessarily exposed) host journal. `forgejo admin user create
/// --random-password` prints the generated password straight to stdout —
/// case-insensitive substring match on "password" is deliberately broad
/// (not pinned to forgejo's exact wording, which can change across
/// versions) so any password-bearing line gets caught rather than relying
/// on a phrase that could silently drift out of sync.
fn redact_password_line(line: &str) -> std::borrow::Cow<'_, str> {
    if line.to_ascii_lowercase().contains("password") {
        std::borrow::Cow::Borrowed("[redacted: line mentions a password]")
    } else {
        std::borrow::Cow::Borrowed(line)
    }
}

/// 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", "{}", redact_password_line(line));
    }
    for line in stderr.lines() {
        tracing::warn!(target: "forgejo-admin", "{}", redact_password_line(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();
    // `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 escalate to SIGKILL if it is
/// still held. If even that doesn't free it, fail loudly — 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.
///
/// The verify-then-escalate 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::warn!(
        %machine,
        "stop finished but machined still holds the registration; escalating to SIGKILL"
    );
    let kill = machinectl_run(&["kill", machine, "--signal=SIGKILL"]).await;
    if wait_name_released(machine).await {
        if let Err(e) = &kill {
            // The name is free, which is what was asked for. machinectl
            // complaining on the way there is worth a line, not a failure.
            tracing::warn!(%machine, error = %e, "SIGKILL reported an error but the name was released");
        }
        return Ok(kill.unwrap_or_default());
    }

    let detail = kill.err().map_or_else(
        || "SIGKILL was delivered".to_owned(),
        |e| format!("SIGKILL also failed: {e}"),
    );
    bail!(
        "stop {machine}: machined still holds the machine name after {}s ({detail}). \
         Starting this container again will fail to register.",
        NAME_RELEASE_TIMEOUT.as_secs() * 2
    )
}

/// 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<()> {
    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 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>,
    load_credentials: &[CredentialMount],
) -> 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");
        // HOST_ADDRESS = the bridge gateway IP. nixos-container's
        // container-side setup only installs a default route
        // (`ip route add default via $HOST_ADDRESS`) when HOST_ADDRESS is
        // non-empty; leaving it blank gave the container an address but no
        // route off the bridge subnet (no internet, no api.anthropic.com).
        // In bridge mode (HOST_BRIDGE set) the host-side address/route
        // setup is skipped, so this only affects the container's route —
        // exactly what we want.
        let _ = writeln!(out, "HOST_ADDRESS={}", iso.gateway_ip);
        // LOCAL_ADDRESS is intentionally empty: agent containers receive their
        // IP dynamically via DHCP from the bridge dnsmasq pool.  HOST_ADDRESS
        // (the gateway IP) is still written so nixos-container's container-side
        // init installs a default route before the DHCP lease arrives.
        out.push_str("LOCAL_ADDRESS=\n");
        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 mut 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();
    // Defense-in-depth for the knowledge bind-mount: overlay an empty tmpfs
    // on /knowledge/.git so the repo metadata (including any credentials the
    // host-side git worker embedded in .git/config) is invisible inside agent
    // containers. Agents only need the working-tree documents; .git/ has no
    // legitimate use in-container. The --tmpfs must come after the --bind-ro
    // so nspawn processes it as an overlay on top of the already-mounted tree.
    // `crate::knowledge::CONTAINER_MOUNT` is "/knowledge" (hive-c0re const).
    if binds
        .iter()
        .any(|b| b.container_path.as_str() == "/knowledge")
    {
        flags.push("--tmpfs=/knowledge/.git".to_owned());
    }
    // Credential forwarding: nspawn loads each host secret into the
    // container's credential store under `<name>`; inner units inherit it
    // via `LoadCredential=<name>`. Validated (name charset + bind-path
    // rules) in handle_write_nspawn_flags above.
    for cred in load_credentials {
        flags.push(format!(
            "--load-credential={}:{}",
            cred.name, cred.host_path
        ));
    }
    let flags_joined = flags.join(" ");
    let _ = writeln!(out, "EXTRA_NSPAWN_FLAGS=\"{flags_joined}\"");
    std::fs::write(&path, out).with_context(|| format!("write {path}"))?;

    // DNS marker for the in-container resolver oneshot. nixos-container
    // copies the *host's* /etc/resolv.conf into the container at every
    // start (its host resolver — e.g. 127.0.0.53 — is unreachable from a
    // private netns, and isn't authoritative for the hive's own zones
    // anyway). The `hyperhive-isolated-dns` oneshot in harness-base.nix
    // rewrites resolv.conf to point at the bridge resolver, but only when
    // this marker exists; it carries the gateway IP so the container
    // doesn't have to re-derive it. Written on isolate, removed otherwise,
    // so the same shared container toplevel behaves correctly in both modes.
    write_bridge_dns_marker(container, isolation)?;
    Ok(())
}

/// Path to the in-container DNS marker (the container's own `/etc`).
fn bridge_dns_marker_path(container: &str) -> String {
    format!("/var/lib/nixos-containers/{container}/etc/hyperhive-bridge-dns")
}

/// Write (isolated) or remove (host-netns) the bridge-DNS marker the
/// `hyperhive-isolated-dns` oneshot keys off. The marker file contains
/// just the gateway IP. Best-effort on removal (absence is the goal).
fn write_bridge_dns_marker(container: &str, isolation: Option<&NetworkIsolation>) -> Result<()> {
    let path = bridge_dns_marker_path(container);
    match isolation {
        Some(iso) => {
            // On a fresh install the container's `/etc` may not exist yet
            // (rootfs not fully materialised before the first start), so
            // `write` would fail with ENOENT. Create the parent dir first
            // — it's the container's own `/etc`, which nixos-container
            // populates on start; a pre-created dir + our marker persist.
            if let Some(parent) = std::path::Path::new(&path).parent() {
                std::fs::create_dir_all(parent).with_context(|| {
                    format!("create bridge-DNS marker dir {}", parent.display())
                })?;
            }
            std::fs::write(&path, format!("{}\n", iso.gateway_ip))
                .with_context(|| format!("write bridge-DNS marker {path}"))?;
        }
        None => match std::fs::remove_file(&path) {
            Ok(()) => {}
            Err(e) if e.kind() == std::io::ErrorKind::NotFound => {}
            Err(e) => return Err(e).with_context(|| format!("remove bridge-DNS marker {path}")),
        },
    }
    Ok(())
}

/// `SyncAgentTmpfiles` — write `/etc/tmpfiles.d/hyperhive-agents.conf` for
/// the given agent set and immediately apply it with `systemd-tmpfiles --create`.
///
/// Each call atomically replaces the file with entries for all current agents,
/// then creates any missing dirs on the running host. The file survives reboots
/// and is read by `systemd-tmpfiles-setup.service` (runs in `sysinit.target`,
/// before any container units can start), so bind-mount source dirs are always
/// pre-created regardless of whether hive-c0re has reached `ensure_agent_runtime_dir`.
///
/// Directories written per agent:
/// - `/run/hyperhive/agents/<name>` (MCP socket dir, bind-mounted into container
///   as `/run/hive`)
/// - `/run/hive-agent/<name>` (web socket dir, bind-mounted into container)
const TMPFILES_PATH: &str = "/etc/tmpfiles.d/hyperhive-agents.conf";

async fn sync_agent_tmpfiles(agents: &[String]) -> Result<(String, String)> {
    use std::fmt::Write as _;
    for name in agents {
        validate_agent_name(name)?;
    }

    // Build tmpfiles.d content. Root dirs first, then per-agent.
    let mut content =
        String::from("# managed by hive-c0re — do not edit (regenerated on spawn/destroy)\n");
    // Parent dirs — created with permissive mode so hive-c0re can make subdirs.
    // /run/hyperhive itself is also a RuntimeDirectory of hive-c0re.service; the
    // tmpfiles.d entry here ensures it exists before hive-c0re starts (boot race).
    content.push_str("d /run/hyperhive 0750 hive-core hive-core -\n");
    writeln!(content, "d {AGENT_RUNTIME_ROOT} 0755 hive-core hive-core -").ok();
    writeln!(content, "d {SOCKET_DIR_ROOT} 0755 root root -").ok();
    // Per-agent dirs.
    for name in agents {
        writeln!(
            content,
            "d {AGENT_RUNTIME_ROOT}/{name} 0755 hive-core hive-core -"
        )
        .ok();
        // 0777: agent harness (non-root uid) must bind sockets here.
        // `d` adjusts mode/owner on existing dirs; world-writable matches
        // the chmod_socket_dir(0o777) fallback so a runtime re-sync doesn't
        // break a live agent's socket dir. host_config's chown_socket_dir
        // tightens ownership afterwards when the agent uid is available.
        writeln!(content, "d {SOCKET_DIR_ROOT}/{name} 0777 root root -").ok();
    }

    // Atomic write: write to a tmp file then rename so a concurrent reader
    // always sees a complete file.
    let tmp = format!("{TMPFILES_PATH}.tmp");
    std::fs::write(&tmp, &content).with_context(|| format!("write {tmp}"))?;
    std::fs::rename(&tmp, TMPFILES_PATH)
        .with_context(|| format!("rename {TMPFILES_PATH}.tmp -> {TMPFILES_PATH}"))?;
    tracing::info!(agents = agents.len(), "tmpfiles.d: wrote {TMPFILES_PATH}");

    // Apply immediately so dirs exist on the running host, not just after next boot.
    let out = Command::new("systemd-tmpfiles")
        .args(["--create", TMPFILES_PATH])
        .output()
        .await
        .context("systemd-tmpfiles --create")?;
    if !out.status.success() {
        let stderr = String::from_utf8_lossy(&out.stderr).trim().to_owned();
        anyhow::bail!(
            "systemd-tmpfiles --create failed ({}): {stderr}",
            out.status
        );
    }
    Ok((String::new(), String::new()))
}

#[cfg(test)]
mod tests {
    use super::{
        PAUSED_MARKER_FILE, limits_dropin_body, redact_password_line, remove_marker_in,
        write_state_file_nofollow,
    };
    use std::path::PathBuf;
    use std::sync::atomic::{AtomicU32, Ordering};

    /// An unset weight is "not configured": the drop-in must come out
    /// byte-identical to the pre-weights two-setting body, so a hive-c0re
    /// older than this field can't change what lands on disk.
    #[test]
    fn unset_weights_reproduce_the_pre_weights_dropin() {
        assert_eq!(
            limits_dropin_body("/run/hyperhive/agents/iris", "4G", "200%", None, None),
            "[Unit]\n\
             ConditionPathIsDirectory=/run/hyperhive/agents/iris\n\
             \n\
             [Service]\n\
             MemoryMax=4G\n\
             CPUQuota=200%\n"
        );
    }

    /// Weights are appended to the `[Service]` section, each omitted
    /// independently when `None`.
    #[test]
    fn weights_are_emitted_only_when_set() {
        let both = limits_dropin_body("/rt/x", "4G", "200%", Some(80), Some(80));
        assert!(
            both.ends_with("CPUQuota=200%\nCPUWeight=80\nIOWeight=80\n"),
            "{both}"
        );

        let cpu_only = limits_dropin_body("/rt/x", "4G", "200%", Some(80), None);
        assert!(
            cpu_only.ends_with("CPUQuota=200%\nCPUWeight=80\n"),
            "{cpu_only}"
        );
        assert!(!cpu_only.contains("IOWeight"), "{cpu_only}");

        let io_only = limits_dropin_body("/rt/x", "4G", "200%", None, Some(80));
        assert!(
            io_only.ends_with("CPUQuota=200%\nIOWeight=80\n"),
            "{io_only}"
        );
        assert!(!io_only.contains("CPUWeight"), "{io_only}");
    }

    #[test]
    fn redacts_lines_mentioning_password_case_insensitively() {
        assert_eq!(
            redact_password_line("New password: hunter2"),
            "[redacted: line mentions a password]"
        );
        assert_eq!(
            redact_password_line("PASSWORD=hunter2"),
            "[redacted: line mentions a password]"
        );
        assert_eq!(
            redact_password_line("User \"foo\" was successfully created."),
            "User \"foo\" was successfully created."
        );
    }

    /// Unique scratch dir per test, no external tempfile dep.
    fn scratch() -> PathBuf {
        static CTR: AtomicU32 = AtomicU32::new(0);
        let n = CTR.fetch_add(1, Ordering::Relaxed);
        let dir = std::env::temp_dir().join(format!(
            "hive-priv-nofollow-test-{}-{n}",
            std::process::id()
        ));
        std::fs::create_dir_all(&dir).unwrap();
        dir
    }

    #[test]
    fn rejects_non_plain_filenames() {
        let dir = scratch();
        for bad in ["", ".", "..", "a/b", "/etc/passwd", "../escape", "sub/tok"] {
            assert!(
                write_state_file_nofollow(&dir, bad, "x").is_err(),
                "must reject filename {bad:?}"
            );
        }
        std::fs::remove_dir_all(&dir).ok();
    }

    #[test]
    fn refuses_symlink_leaf_and_leaves_target_untouched() {
        let dir = scratch();
        let target = dir.join("target");
        std::fs::write(&target, "original").unwrap();
        // Agent plants a symlink where the token would be written.
        std::os::unix::fs::symlink(&target, dir.join("forge-token")).unwrap();

        let res = write_state_file_nofollow(&dir, "forge-token", "PWNED");
        assert!(res.is_err(), "O_NOFOLLOW must refuse a symlink leaf");
        // The root-privileged write must NOT have followed the link.
        assert_eq!(
            std::fs::read_to_string(&target).unwrap(),
            "original",
            "symlink target must be untouched"
        );
        std::fs::remove_dir_all(&dir).ok();
    }

    #[test]
    fn writes_plain_file_0600() {
        use std::os::unix::fs::PermissionsExt as _;
        let dir = scratch();
        write_state_file_nofollow(&dir, "forge-token", "secret").unwrap();
        let path = dir.join("forge-token");
        assert_eq!(std::fs::read_to_string(&path).unwrap(), "secret");
        let mode = std::fs::metadata(&path).unwrap().permissions().mode() & 0o777;
        assert_eq!(mode, 0o600, "token file must be 0600");
        // Overwrite truncates cleanly (O_TRUNC), no residue.
        write_state_file_nofollow(&dir, "forge-token", "new").unwrap();
        assert_eq!(std::fs::read_to_string(&path).unwrap(), "new");
        std::fs::remove_dir_all(&dir).ok();
    }

    /// The pause marker round-trips through the same root-only path the
    /// credential writes use, and BOTH directions are idempotent — the
    /// dashboard toggle and `hivectl pause|resume` fire blind, without
    /// reading the current state first.
    #[test]
    fn pause_marker_create_and_remove_are_idempotent() {
        let dir = scratch();
        let path = dir.join(PAUSED_MARKER_FILE);

        for _ in 0..2 {
            write_state_file_nofollow(&dir, PAUSED_MARKER_FILE, "").unwrap();
            assert!(path.exists(), "marker must exist after pause");
            assert_eq!(std::fs::read_to_string(&path).unwrap(), "");
        }
        for _ in 0..2 {
            remove_marker_in(&dir, PAUSED_MARKER_FILE).unwrap();
            assert!(!path.exists(), "marker must be gone after resume");
        }
        std::fs::remove_dir_all(&dir).ok();
    }

    /// A resume must never follow an agent-planted symlink at the marker
    /// path: this unlink runs as root, so following it would let an agent
    /// delete an arbitrary file on the host.
    #[test]
    fn resume_unlinks_the_symlink_not_its_target() {
        let dir = scratch();
        let target = dir.join("target");
        std::fs::write(&target, "original").unwrap();
        let link = dir.join(PAUSED_MARKER_FILE);
        std::os::unix::fs::symlink(&target, &link).unwrap();

        remove_marker_in(&dir, PAUSED_MARKER_FILE).unwrap();
        // `exists()` follows the link, so it can't tell "link removed" from
        // "target removed, dangling link left" — stat the link itself.
        assert!(
            std::fs::symlink_metadata(&link).is_err(),
            "the link itself must be unlinked"
        );
        assert_eq!(
            std::fs::read_to_string(&target).unwrap(),
            "original",
            "symlink target must survive the root unlink"
        );
        std::fs::remove_dir_all(&dir).ok();
    }
}