//! Container config writes: `WriteNspawnFlags` (the container's
//! `/etc/nixos-containers/<name>.conf`, plus the bridge-DNS marker and
//! hive-facts file in its rootfs), `WriteResourceLimits` /
//! `RemoveServiceDropin` (the `container@<name>.service` drop-in) and
//! `DaemonReload`.
//!
//! Every op that names a container takes its system name and refuses anything
//! [`validate_container_system_name`] does not accept.

use std::os::fd::{AsRawFd as _, FromRawFd as _, OwnedFd};
use std::path::{Path, PathBuf};

use anyhow::{Context as _, Result, bail};
use hive_priv_sock::{
    AGENT_PREFIX, AGENT_RUNTIME_ROOT, BindMount, CredentialMount, HiveFacts, NetworkIsolation,
};
use tokio::process::Command;

use crate::shared::{StagedFile, open_dir, validate_credential_name, validate_name_chars};

/// `WriteNspawnFlags` — validate the container + every bind path + every
/// credential entry, then write the container's nspawn flag overrides.
pub(crate) fn handle_write_nspawn_flags(
    container: &str,
    binds: &[BindMount],
    isolation: &NetworkIsolation,
    load_credentials: &[CredentialMount],
    hive_facts: &HiveFacts,
) -> Result<(String, String)> {
    let agent = 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)?;
    }
    let hive_facts = render_hive_facts(hive_facts)?;
    // A root that cannot be salvaged keeps its persistent start: the flag is
    // only ever passed for a root the salvage has dealt with.
    let ephemeral = crate::rootfs::prepare(container, agent).unwrap_or_else(|e| {
        tracing::error!(
            %container,
            error = %format!("{e:#}"),
            "root salvage failed; the container starts with its persistent root"
        );
        false
    });
    write_nspawn_flags(
        container,
        binds,
        isolation,
        load_credentials,
        &hive_facts,
        ephemeral,
    )?;
    Ok((String::new(), String::new()))
}

/// [`handle_write_nspawn_flags`] on the blocking pool. The root salvage inside
/// it copies files for as long as the root is big, and would hold a runtime
/// worker for that time.
///
/// # Errors
/// As [`handle_write_nspawn_flags`], or when the task fails to join.
pub(crate) async fn handle_write_nspawn_flags_blocking(
    container: String,
    binds: Vec<BindMount>,
    isolation: NetworkIsolation,
    load_credentials: Vec<CredentialMount>,
    hive_facts: HiveFacts,
) -> Result<(String, String)> {
    tokio::task::spawn_blocking(move || {
        handle_write_nspawn_flags(
            &container,
            &binds,
            &isolation,
            &load_credentials,
            &hive_facts,
        )
    })
    .await
    .context("write nspawn flags task")?
}

/// Validate a system-level container name (`h-` prefix for all agents,
/// the manager included) and return the logical agent name after it.
/// Anything else is refused, the swarm's forge and matrix containers
/// included: they belong to the host that runs them, not to a hive.
pub(crate) fn validate_container_system_name(name: &str) -> Result<&str> {
    let Some(suffix) = name.strip_prefix(AGENT_PREFIX) else {
        bail!("container name {name:?} is not managed by hive");
    };
    validate_name_chars(suffix)?;
    Ok(suffix)
}

/// `RemoveServiceDropin` — remove the container service's drop-in dir
/// if present (idempotent), then the GC root of the build it named.
pub(crate) 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}"))?;
    }
    crate::prestart::unpin_binary(container)?;
    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.
///
/// The drop-in also carries the `ExecStartPre=` that runs
/// `hive-priv check-rootfs <container>` ([`crate::prestart`]) ahead of
/// nixpkgs' start script. It names this process's own binary, so the drop-in
/// and the binary that implements the check are always the same build. The
/// build is rooted against garbage collection for as long as the drop-in
/// names it ([`crate::prestart::pin_binary`]).
///
/// 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 weight of `None` means "not
/// configured" 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
/// hive-c0re creating the dir in its start preamble: if it is missing at
/// start time, the unit idles rather than restart-looping into
/// `start-limit-hit`.
pub(crate) fn write_resource_limits(
    container: &str,
    memory_max: &str,
    cpu_quota: &str,
    cpu_weight: Option<u32>,
    io_weight: Option<u32>,
) -> Result<(String, String)> {
    let logical = validate_container_system_name(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 exe = hive_priv_exe()?;
    // Rooted before the drop-in names it, so the drop-in never names a build a
    // collection can take.
    crate::prestart::pin_binary(container, Path::new(&exe))?;
    let rootfs_check = rootfs_check_command(&exe, container);
    let content = limits_dropin_body(
        &runtime_dir,
        &rootfs_check,
        memory_max,
        cpu_quota,
        cpu_weight,
        io_weight,
    );
    // Published whole: a `daemon-reload` between truncate and write would
    // otherwise load the unit with its limits and start guard missing.
    publish_file(Path::new(&path), content.as_bytes(), 0o644, None)?;
    Ok((String::new(), String::new()))
}

/// How long a window the start-limit counts over, and how many starts it
/// allows inside it.
///
/// `container@.service` sets `Restart=on-failure` and **no** start limit, so
/// systemd's defaults apply: 5 starts per 10s, `RestartSec` 100ms. That
/// makes the bound depend on *how fast* a container dies — one that fails
/// instantly trips the limit in under a second, one that takes longer than
/// ~2s never trips it and restarts forever. Whether an agent gets bounded
/// is not meant to be a function of its failure speed.
///
/// The window has to exceed the worst-case time to burn the burst, or the
/// counter ages out between attempts and the limit is again unreachable:
/// `TimeoutStartSec` is 1min, so `BURST` slow failures plus their backoff
/// can span several minutes. 10min covers that with room.
///
/// Giving up is cheap here **because it is not terminal** — hive-c0re's
/// reconcile sweep retries later, and `reset-failed` (see `StartContainer`)
/// clears the latch first. That is what makes a tight burst safe.
const START_LIMIT_INTERVAL_SEC: u32 = 600;
/// One start plus two retries — the operator's ruling was "retry once or
/// twice", with the reconcile sweep as the slow path after that.
const START_LIMIT_BURST: u32 = 3;
/// Backoff between those retries. The 100ms default is for processes that
/// respawn instantly; a container that just failed to boot gains nothing
/// from being retried a tenth of a second later.
const RESTART_SEC: u32 = 5;

/// Render the body of `hyperhive-limits.conf`.
///
/// `rootfs_check` is the command line of the pre-start root check, the first
/// `[Service]` line: it runs before the nixpkgs start script and a non-zero
/// exit fails the start.
///
/// `[Unit]`: the condition is checked at start time — it skips (not fails)
/// the unit when the MCP socket dir is absent, avoiding restart loops —
/// plus the bounded start limit (see the constants above; `StartLimit*` are
/// `[Unit]` settings since systemd 229 and are silently ignored under
/// `[Service]`).
/// `[Service]`: the restart backoff, the hard caps, 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` — renders no weight lines.
fn limits_dropin_body(
    runtime_dir: &str,
    rootfs_check: &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\
         StartLimitIntervalSec={START_LIMIT_INTERVAL_SEC}\n\
         StartLimitBurst={START_LIMIT_BURST}\n\
         \n\
         [Service]\n\
         ExecStartPre={rootfs_check}\n\
         RestartSec={RESTART_SEC}\n\
         MemoryMax={memory_max}\n\
         CPUQuota={cpu_quota}\n\
         {cpu_weight_line}{io_weight_line}"
    )
}

/// This binary's absolute path, for the drop-in's `ExecStartPre=`.
///
/// The path is the running binary's, not a fixed one, so a drop-in never names
/// a build without the sub-command. systemd expands `%` and `$` in command
/// lines and splits on spaces, so a path with any of them is refused rather
/// than escaped; store paths never have them.
fn hive_priv_exe() -> Result<String> {
    let exe = std::env::current_exe().context("resolve the hive-priv binary path")?;
    let exe = exe
        .to_str()
        .filter(|p| {
            p.starts_with('/')
                && p.bytes()
                    .all(|b| b.is_ascii_alphanumeric() || b"/._+-".contains(&b))
        })
        .with_context(|| {
            format!(
                "hive-priv binary path {} is not safe in a unit file",
                exe.display()
            )
        })?;
    Ok(exe.to_owned())
}

/// The `ExecStartPre=` command of the pre-start root check for `container`:
/// the binary `exe`, then the sub-command and the name.
fn rootfs_check_command(exe: &str, container: &str) -> String {
    format!("{exe} {} {container}", crate::prestart::SUBCOMMAND)
}

/// `DaemonReload` — `systemctl daemon-reload` on the host.
pub(crate) 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()))
}

/// Publish `content` at `path` through a [`StagedFile`], at `mode` and, when
/// given, owned by `owner` (`(uid, gid)`).
fn publish_file(path: &Path, content: &[u8], mode: u32, owner: Option<(u32, u32)>) -> Result<()> {
    let (Some(dir_path), Some(filename)) =
        (path.parent(), path.file_name().and_then(|n| n.to_str()))
    else {
        bail!(
            "publish_file: {} has no directory or file name",
            path.display()
        );
    };
    let dir = open_dir(dir_path)?;
    let staged = StagedFile::write(&dir, dir_path, filename, content, mode)?;
    if let Some((uid, gid)) = owner {
        std::os::unix::fs::fchown(staged.file(), Some(uid), Some(gid))
            .with_context(|| format!("chown {}", path.display()))?;
    }
    staged.publish()
}

/// Validate a bind-mount path: must be absolute, non-empty, and contain
/// no newlines, null bytes, double-quotes, or colons.
///
/// The first three would break the `EXTRA_NSPAWN_FLAGS="..."` conf line
/// format. The colon is the one that matters most and had been left out
/// of this list: `--bind=SRC:DST` is colon-separated, so a path carrying
/// one does not corrupt the line — it silently becomes a *different
/// mount* than the caller asked for.
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(())
}

/// `--tmpfs=<mount>/.git` for every bound git repo, hiding its metadata
/// from inside the container.
///
/// Two kinds of mount qualify, for one reason: **the agent is given a
/// working tree, never a repository.** `/knowledge` is the hive's shared
/// docs — whose `.git/config` has held a credential the host-side worker
/// embedded — and `/agents/<name>/config` is a config repo, an agent's own
/// or a parent's read-only view of a child's. In both cases `.git` carries
/// every branch and the full history of a document whose *currently
/// deployed* value is the only thing a reader may act on, and an abandoned
/// branch is indistinguishable from a live one.
///
/// An overlay rather than an exported copy: there is no second tree to
/// keep in sync, so nothing can go stale, and no code path has to remember
/// to refresh it.
///
/// ⚠️ Ordering matters — these must be appended **after** the `--bind`
/// flags so nspawn mounts them on top of the already-mounted trees.
/// Config mounts are matched by shape, not by a name list: the set is
/// dynamic, growing with each child bound into a parent.
fn git_overlay_flags(binds: &[BindMount]) -> Vec<String> {
    binds
        .iter()
        .map(|b| b.container_path.as_str())
        .filter(|p| *p == "/knowledge" || (p.starts_with("/agents/") && p.ends_with("/config")))
        .map(|p| format!("--tmpfs={p}/.git"))
        .collect()
}

/// Update `/etc/nixos-containers/<container>.conf`: strip old network vars
/// (`PRIVATE_NETWORK`, `HOST_ADDRESS*`, `LOCAL_ADDRESS*`, `HOST_BRIDGE`),
/// write the current network-isolation settings, then append
/// `EXTRA_NSPAWN_FLAGS`. Always writes `PRIVATE_NETWORK=1` + veth
/// wiring — isolation is the only mode, so there is no branch that
/// leaves a container on the host's network namespace.
fn write_nspawn_flags(
    container: &str,
    binds: &[BindMount],
    isolation: &NetworkIsolation,
    load_credentials: &[CredentialMount],
    hive_facts: &str,
    ephemeral: bool,
) -> Result<()> {
    use std::fmt::Write as _;
    use std::os::unix::fs::MetadataExt 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');
    }
    {
        let 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: containers take their IP by
        // DHCP from the bridge dnsmasq pool. (Why HOST_ADDRESS is still
        // written: the comment directly above.)
        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);
    }
    let flags_joined = nspawn_flags(binds, load_credentials, ephemeral).join(" ");
    let _ = writeln!(out, "EXTRA_NSPAWN_FLAGS=\"{flags_joined}\"");
    // Keeps the owner and mode `nixos-container` created the file with; this
    // helper only rewrites its lines.
    let meta = std::fs::metadata(&path).with_context(|| format!("stat {path}"))?;
    publish_file(
        Path::new(&path),
        out.as_bytes(),
        meta.mode() & 0o7777,
        Some((meta.uid(), meta.gid())),
    )?;

    // DNS marker for the in-container resolver oneshot: the oneshot only
    // rewrites the container's resolv.conf when this marker exists, and the
    // marker carries the gateway IP so the container need not re-derive it.
    // Always written — every container is isolated, so there is no mode in
    // which the marker should be absent. Why the rewrite is needed at all,
    // and which unit does it: `docs/networking/network.md` § *How the isolated
    // container gets its resolver*.
    write_bridge_dns_marker(container, isolation)?;
    // Read by the agent's environment generator at boot
    // (`nix/agent-modules/hive-facts.nix`).
    write_container_etc_file_in(&container_rootfs(container), HIVE_FACTS_FILE, hive_facts)?;
    Ok(())
}

/// The flags `EXTRA_NSPAWN_FLAGS` carries, in the order nspawn needs them.
///
/// `--ephemeral` runs the container on a snapshot of its root and drops the
/// snapshot at stop; `crate::rootfs` has made the root fit for that before
/// the flag is asked for. The git overlays and credentials follow the binds
/// they refer to.
fn nspawn_flags(
    binds: &[BindMount],
    load_credentials: &[CredentialMount],
    ephemeral: bool,
) -> Vec<String> {
    let mut flags: Vec<String> = ephemeral
        .then(|| "--ephemeral".to_owned())
        .into_iter()
        .collect();
    flags.extend(binds.iter().map(|b| {
        let flag = if b.read_only { "--bind-ro" } else { "--bind" };
        format!("{flag}={}:{}", b.host_path, b.container_path)
    }));
    flags.extend(git_overlay_flags(binds));
    // 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.
    flags.extend(
        load_credentials
            .iter()
            .map(|cred| format!("--load-credential={}:{}", cred.name, cred.host_path)),
    );
    flags
}

/// Filename of the in-container DNS marker, in the container's own `/etc`.
const BRIDGE_DNS_MARKER: &str = "hyperhive-bridge-dns";

/// Filename of the hive-facts environment file, in the container's own `/etc`.
const HIVE_FACTS_FILE: &str = "hyperhive-hive-facts";

/// The container's `/` on the host.
pub(crate) fn container_rootfs(container: &str) -> PathBuf {
    PathBuf::from(format!("/var/lib/nixos-containers/{container}"))
}

/// Render [`HiveFacts`] as the systemd environment file the agent loads,
/// one `KEY=value` line per present fact.
///
/// Each value is refused, never escaped, unless it is made only of the
/// characters its fact can hold. None of those sets admits a newline (a
/// second assignment), whitespace, a quote, `\` or `$`: the agent hands each
/// line to systemd and its shells, which would read those as syntax.
fn render_hive_facts(facts: &HiveFacts) -> Result<String> {
    use std::fmt::Write as _;
    let dns = |b: u8| b.is_ascii_alphanumeric() || matches!(b, b'.' | b'-' | b'_');
    let port = facts.dashboard_port.map(|p| p.to_string());
    let mut out = String::new();
    for (key, value, max_len, allowed) in [
        (
            "HYPERHIVE_HIVE_NAME",
            facts.name.as_ref(),
            253,
            &dns as &dyn Fn(u8) -> bool,
        ),
        ("HYPERHIVE_HIVE_DOMAIN", facts.domain.as_ref(), 253, &dns),
        ("HIVE_DASHBOARD_PORT", port.as_ref(), 5, &|b: u8| {
            b.is_ascii_digit()
        }),
        // A store path's name characters, plus the `/` of its prefix; the
        // prefix itself is checked below.
        (
            "HYPERHIVE_CLAUDE_CODE_PATH",
            facts.claude_code_path.as_ref(),
            256,
            &|b: u8| dns(b) || matches!(b, b'+' | b'?' | b'=' | b'/'),
        ),
        // `k=v,k=v`, values percent-encoded as the OTEL spec has them.
        (
            "HYPERHIVE_OTEL_EXTRA_RESOURCE_ATTRIBUTES",
            facts.otel_extra_resource_attributes.as_ref(),
            1024,
            &|b: u8| dns(b) || matches!(b, b'=' | b',' | b'/' | b':' | b'%' | b'+' | b'@'),
        ),
        // `<scheme>://<host>:<port>`.
        (
            "HIVE_AGENT_NATS_URL",
            facts.agent_nats_url.as_ref(),
            253,
            &|b: u8| dns(b) || matches!(b, b':' | b'/'),
        ),
        // `<ipv4>:<port>`.
        (
            "HIVE_SQUID_PROXY",
            facts.squid_proxy.as_ref(),
            21,
            &|b: u8| b.is_ascii_digit() || matches!(b, b'.' | b':'),
        ),
        // `host,host`: bare names and addresses.
        (
            "HIVE_EGRESS_NO_PROXY",
            facts.egress_no_proxy.as_ref(),
            4096,
            &|b: u8| dns(b) || b == b',',
        ),
    ] {
        let Some(value) = value else { continue };
        if value.is_empty() || value.len() > max_len || !value.bytes().all(allowed) {
            bail!(
                "invalid {key} value {value:?}: must be 1-{max_len} chars of its fact's character set"
            );
        }
        if key == "HYPERHIVE_CLAUDE_CODE_PATH"
            && !value
                .strip_prefix("/nix/store/")
                .is_some_and(|name| !name.is_empty() && !name.contains('/'))
        {
            bail!("invalid {key} value {value:?}: must be a top-level /nix/store path");
        }
        let _ = writeln!(out, "{key}={value}");
    }
    Ok(out)
}

/// Write the bridge-DNS marker the `hyperhive-isolated-dns` oneshot keys
/// off. The marker file contains just the gateway IP. Always written:
/// every container is isolated, so there is no host-netns case that
/// wants the marker absent.
fn write_bridge_dns_marker(container: &str, isolation: &NetworkIsolation) -> Result<()> {
    write_bridge_dns_marker_in(&container_rootfs(container), &isolation.gateway_ip)
}

/// Write `<rootfs>/etc/hyperhive-bridge-dns`; see [`write_container_etc_file_in`].
fn write_bridge_dns_marker_in(rootfs: &Path, gateway_ip: &str) -> Result<()> {
    write_container_etc_file_in(rootfs, BRIDGE_DNS_MARKER, &format!("{gateway_ip}\n"))
}

/// Write `<rootfs>/etc/<name>` without following a symlink.
///
/// Root inside the container owns its `/etc`, and this write runs as host
/// root, where an absolute symlink planted in the container resolves against
/// the host's `/`. So this opens `etc` with `O_DIRECTORY|O_NOFOLLOW` (a
/// symlinked `etc` fails with `ELOOP`) and publishes the file through a
/// [`StagedFile`] relative to that fd: whatever the container planted at the
/// leaf, symlink or FIFO or device node, is replaced by the rename, never
/// written through or opened.
///
/// On a fresh install the container's `/etc` may not exist yet (nixos-container
/// materialises the rootfs on the first start), so this creates it first; the
/// dir and the file persist through that start. Mode 0644: nothing written
/// here is a secret.
pub(crate) fn write_container_etc_file_in(rootfs: &Path, name: &str, contents: &str) -> Result<()> {
    use std::os::unix::fs::OpenOptionsExt as _;

    let etc_path = rootfs.join("etc");
    // The rootfs dir is the container's `/`, which the container cannot
    // replace, so creating it by path is safe.
    std::fs::create_dir_all(rootfs)
        .with_context(|| format!("create container rootfs {}", rootfs.display()))?;
    let root = std::fs::OpenOptions::new()
        .read(true)
        .custom_flags(libc::O_DIRECTORY | libc::O_NOFOLLOW)
        .open(rootfs)
        .with_context(|| format!("open container rootfs {}", rootfs.display()))?;
    // SAFETY: `root` is an open directory fd and the name a NUL-terminated
    // literal.
    if unsafe { libc::mkdirat(root.as_raw_fd(), c"etc".as_ptr(), 0o755) } != 0 {
        let e = std::io::Error::last_os_error();
        if e.kind() != std::io::ErrorKind::AlreadyExists {
            return Err(e).with_context(|| format!("create {}", etc_path.display()));
        }
    }
    // SAFETY: as above; this checks the result before wrapping it.
    let etc_fd = unsafe {
        libc::openat(
            root.as_raw_fd(),
            c"etc".as_ptr(),
            libc::O_RDONLY | libc::O_DIRECTORY | libc::O_NOFOLLOW | libc::O_CLOEXEC,
        )
    };
    if etc_fd < 0 {
        return Err(std::io::Error::last_os_error())
            .with_context(|| format!("open (no-follow) {}", etc_path.display()));
    }
    // SAFETY: `etc_fd` is a fresh, valid fd that nothing else owns.
    let etc = std::fs::File::from(unsafe { OwnedFd::from_raw_fd(etc_fd) });
    StagedFile::write(&etc, &etc_path, name, contents.as_bytes(), 0o644)?.publish()
}
#[cfg(test)]
mod tests {
    use super::{
        BindMount, HiveFacts, NetworkIsolation, git_overlay_flags, handle_write_nspawn_flags,
        hive_priv_exe, limits_dropin_body, nspawn_flags, publish_file, remove_service_dropin,
        render_hive_facts, rootfs_check_command, validate_container_system_name,
        write_bridge_dns_marker_in, write_resource_limits,
    };
    use crate::shared::test_support::{entries, scratch};

    /// A stand-in for the pre-start check's command line.
    const CHECK: &str = "/store/hive-priv/bin/hive-priv check-rootfs h-iris";

    fn bind(container_path: &str) -> BindMount {
        BindMount {
            host_path: "/var/lib/hyperhive/whatever".to_owned(),
            container_path: container_path.to_owned(),
            read_only: true,
        }
    }

    /// Every bound git repo gets its `.git` overlaid — the knowledge tree
    /// and *each* config mount, an agent's own plus every child's.
    ///
    /// The child case is the one worth pinning: that set grows at runtime
    /// as agents gain children, so a rule written as a list of names would
    /// silently stop covering new ones.
    #[test]
    fn every_bound_git_repo_gets_its_dot_git_hidden() {
        let flags = git_overlay_flags(&[
            bind("/knowledge"),
            bind("/agents/atlas/config"),
            bind("/agents/kiddo/config"),
        ]);
        assert_eq!(
            flags,
            [
                "--tmpfs=/knowledge/.git",
                "--tmpfs=/agents/atlas/config/.git",
                "--tmpfs=/agents/kiddo/config/.git",
            ]
        );
    }

    /// ...and nothing else does. A blanket "overlay .git on every bind"
    /// would mask a real `.git` under `state/`, where an agent legitimately
    /// keeps working clones of its own.
    #[test]
    fn non_repo_mounts_are_left_alone() {
        let flags = git_overlay_flags(&[
            bind("/agents/atlas/state"),
            bind("/work"),
            bind("/applied"),
            bind("/agents/atlas/config-notes"),
        ]);
        assert!(flags.is_empty(), "overlaid a non-repo mount: {flags:?}");
    }

    /// The whole rendered body, pinned literally — including the values,
    /// so changing the restart policy is a visible test edit rather than a
    /// silent one.
    ///
    /// Placement is the part most worth pinning: `StartLimit*` are `[Unit]`
    /// settings and systemd **silently ignores** them under `[Service]`, so
    /// a bound that moved sections would look configured and do nothing.
    /// An unset weight still emits no line at all, which is what keeps a
    /// hive-c0re older than that field from changing what lands on disk.
    #[test]
    fn dropin_body_is_pinned_exactly() {
        assert_eq!(
            limits_dropin_body(
                "/run/hyperhive/agents/iris",
                CHECK,
                "4G",
                "200%",
                None,
                None
            ),
            "[Unit]\n\
             ConditionPathIsDirectory=/run/hyperhive/agents/iris\n\
             StartLimitIntervalSec=600\n\
             StartLimitBurst=3\n\
             \n\
             [Service]\n\
             ExecStartPre=/store/hive-priv/bin/hive-priv check-rootfs h-iris\n\
             RestartSec=5\n\
             MemoryMax=4G\n\
             CPUQuota=200%\n"
        );
    }

    /// The bound is hive-wide **policy**, not a per-agent parameter: it is
    /// rendered from constants and no caller-supplied value can omit or
    /// alter it. This is the property that justifies keeping it out of the
    /// wire protocol — if it ever varies by request, that argument is gone.
    #[test]
    fn the_start_limit_is_present_whatever_the_caller_passes() {
        for (mem, cpu, cw, iw) in [
            ("4G", "200%", None, None),
            ("512M", "50%", Some(10), Some(10)),
            ("infinity", "infinity", Some(10_000), None),
        ] {
            let body = limits_dropin_body("/rt/x", CHECK, mem, cpu, cw, iw);
            let unit = body
                .split("[Service]")
                .next()
                .expect("the body always has a [Unit] section before [Service]");
            assert!(
                unit.contains("StartLimitIntervalSec=600") && unit.contains("StartLimitBurst=3"),
                "start limit missing from [Unit] for ({mem}, {cpu}): {body}"
            );
            assert!(
                body.contains("\nRestartSec=5\n"),
                "restart backoff missing for ({mem}, {cpu}): {body}"
            );
        }
    }

    /// 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", CHECK, "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", CHECK, "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", CHECK, "4G", "200%", None, Some(80));
        assert!(
            io_only.ends_with("CPUQuota=200%\nIOWeight=80\n"),
            "{io_only}"
        );
        assert!(!io_only.contains("CPUWeight"), "{io_only}");
    }

    /// The pre-start check is the first `[Service]` line, whatever else the
    /// caller sets, and a drop-in without it is what an old hive-priv wrote.
    #[test]
    fn the_rootfs_check_leads_the_service_section_whatever_the_caller_passes() {
        for (cw, iw) in [(None, None), (Some(80), Some(80))] {
            let body = limits_dropin_body("/rt/x", CHECK, "4G", "200%", cw, iw);
            let service = body
                .split("[Service]\n")
                .nth(1)
                .expect("the body always has a [Service] section");
            assert_eq!(
                service.lines().next(),
                Some("ExecStartPre=/store/hive-priv/bin/hive-priv check-rootfs h-iris")
            );
            assert_eq!(body.matches("ExecStartPre=").count(), 1, "{body}");
        }
    }

    /// The command names an absolute path to this very binary, the
    /// sub-command that binary implements and the container.
    #[test]
    fn the_check_command_names_this_binary_the_subcommand_and_the_container() {
        let exe = std::env::current_exe().unwrap();
        let cmd = rootfs_check_command(&hive_priv_exe().unwrap(), "h-iris");
        assert_eq!(cmd, format!("{} check-rootfs h-iris", exe.display()));
        assert!(cmd.starts_with('/'), "{cmd}");
    }

    /// The mode lands exactly as asked, whatever the umask would have made
    /// of it (0666 under the usual 022 would otherwise come out 0644), and
    /// the requested owner is on the file.
    #[test]
    fn publish_file_applies_mode_and_owner() {
        use std::os::unix::fs::{MetadataExt as _, PermissionsExt as _};
        let dir = scratch();
        let path = dir.join("c.conf");
        let owner = std::fs::metadata(&dir).unwrap();

        publish_file(&path, b"x\n", 0o666, Some((owner.uid(), owner.gid()))).unwrap();
        let meta = std::fs::metadata(&path).unwrap();
        assert_eq!(meta.permissions().mode() & 0o7777, 0o666);
        assert_eq!((meta.uid(), meta.gid()), (owner.uid(), owner.gid()));

        publish_file(&path, b"y\n", 0o600, None).unwrap();
        let meta = std::fs::metadata(&path).unwrap();
        assert_eq!(meta.permissions().mode() & 0o7777, 0o600);
        assert_eq!(std::fs::read_to_string(&path).unwrap(), "y\n");
        assert_eq!(entries(&dir), ["c.conf"], "temp left behind");
        std::fs::remove_dir_all(&dir).ok();
    }

    /// The container plants a symlink where the bridge-DNS marker goes; the
    /// rename replaces it without writing through it.
    #[test]
    fn bridge_dns_marker_replaces_symlink_leaf() {
        let dir = scratch();
        let rootfs = dir.join("rootfs");
        std::fs::create_dir_all(rootfs.join("etc")).unwrap();
        let target = dir.join("target");
        std::fs::write(&target, "original").unwrap();
        let leaf = rootfs.join("etc/hyperhive-bridge-dns");
        std::os::unix::fs::symlink(&target, &leaf).unwrap();

        write_bridge_dns_marker_in(&rootfs, "10.0.0.1").unwrap();
        assert_eq!(
            std::fs::read_to_string(&target).unwrap(),
            "original",
            "symlink target must be untouched"
        );
        assert!(
            std::fs::symlink_metadata(&leaf)
                .unwrap()
                .file_type()
                .is_file()
        );
        assert_eq!(std::fs::read_to_string(&leaf).unwrap(), "10.0.0.1\n");
        std::fs::remove_dir_all(&dir).ok();
    }

    /// The container replaces its whole `/etc` with a symlink, so the leaf
    /// name resolves inside a directory of its choosing.
    #[test]
    fn bridge_dns_marker_refuses_symlink_etc() {
        let dir = scratch();
        let rootfs = dir.join("rootfs");
        std::fs::create_dir_all(&rootfs).unwrap();
        let elsewhere = dir.join("elsewhere");
        std::fs::create_dir_all(&elsewhere).unwrap();
        let target = elsewhere.join("hyperhive-bridge-dns");
        std::fs::write(&target, "original").unwrap();
        std::os::unix::fs::symlink(&elsewhere, rootfs.join("etc")).unwrap();

        let res = write_bridge_dns_marker_in(&rootfs, "10.0.0.1");
        assert!(res.is_err(), "O_NOFOLLOW must refuse a symlinked etc");
        assert_eq!(
            std::fs::read_to_string(&target).unwrap(),
            "original",
            "file behind the symlinked etc must be untouched"
        );
        std::fs::remove_dir_all(&dir).ok();
    }

    /// A FIFO at the leaf is replaced, never opened, so it cannot block the
    /// helper.
    #[test]
    fn bridge_dns_marker_replaces_fifo_leaf() {
        use std::os::unix::ffi::OsStrExt as _;
        let dir = scratch();
        let rootfs = dir.join("rootfs");
        std::fs::create_dir_all(rootfs.join("etc")).unwrap();
        let fifo = rootfs.join("etc/hyperhive-bridge-dns");
        let c_fifo = std::ffi::CString::new(fifo.as_os_str().as_bytes()).unwrap();
        // SAFETY: `c_fifo` is a valid NUL-terminated path.
        assert_eq!(unsafe { libc::mkfifo(c_fifo.as_ptr(), 0o644) }, 0);

        write_bridge_dns_marker_in(&rootfs, "10.0.0.1").unwrap();
        assert!(
            std::fs::symlink_metadata(&fifo)
                .unwrap()
                .file_type()
                .is_file()
        );
        assert_eq!(std::fs::read_to_string(&fifo).unwrap(), "10.0.0.1\n");
        std::fs::remove_dir_all(&dir).ok();
    }

    /// The control: a plain rootfs with no `etc` yet (a fresh install) gets
    /// the dir and a 0644 marker, and a second write replaces it.
    #[test]
    fn bridge_dns_marker_written_into_plain_etc() {
        use std::os::unix::fs::PermissionsExt as _;
        let dir = scratch();
        let rootfs = dir.join("rootfs");
        std::fs::create_dir_all(&rootfs).unwrap();

        write_bridge_dns_marker_in(&rootfs, "10.0.0.1").unwrap();
        let path = rootfs.join("etc/hyperhive-bridge-dns");
        assert_eq!(std::fs::read_to_string(&path).unwrap(), "10.0.0.1\n");
        let mode = std::fs::metadata(&path).unwrap().permissions().mode() & 0o777;
        assert_eq!(mode, 0o644, "marker must stay readable in the container");

        write_bridge_dns_marker_in(&rootfs, "10.9.8.7").unwrap();
        assert_eq!(std::fs::read_to_string(&path).unwrap(), "10.9.8.7\n");
        std::fs::remove_dir_all(&dir).ok();
    }

    /// Each present fact is one environment-file line; an absent one writes
    /// nothing, so the agent sees the variable unset rather than empty.
    #[test]
    fn hive_facts_render_one_line_per_present_fact() {
        let both = HiveFacts {
            name: Some("pr1ma".to_owned()),
            domain: Some("pr1ma.darkest.space".to_owned()),
            ..HiveFacts::default()
        };
        assert_eq!(
            render_hive_facts(&both).unwrap(),
            "HYPERHIVE_HIVE_NAME=pr1ma\nHYPERHIVE_HIVE_DOMAIN=pr1ma.darkest.space\n"
        );
        let domain_only = HiveFacts {
            domain: Some("h1.example.test".to_owned()),
            ..HiveFacts::default()
        };
        assert_eq!(
            render_hive_facts(&domain_only).unwrap(),
            "HYPERHIVE_HIVE_DOMAIN=h1.example.test\n"
        );
        assert_eq!(render_hive_facts(&HiveFacts::default()).unwrap(), "");
    }

    /// A value that could end its line or be read as syntax is refused, not
    /// escaped: a newline would add an assignment of the caller's choosing.
    #[test]
    fn hive_facts_refuse_values_outside_dns_characters() {
        for bad in ["", "pr1ma\nLD_PRELOAD=/x", "a b", "a\"b", "$(x)"] {
            let facts = HiveFacts {
                name: Some(bad.to_owned()),
                ..HiveFacts::default()
            };
            assert!(
                render_hive_facts(&facts).is_err(),
                "{bad:?} must be refused"
            );
        }
    }

    /// The per-host values agents used to get at build time arrive here, one
    /// line each.
    #[test]
    fn hive_facts_render_the_per_host_values() {
        let facts = HiveFacts {
            dashboard_port: Some(7123),
            claude_code_path: Some("/nix/store/0abc-claude-code-2.1.220".to_owned()),
            otel_extra_resource_attributes: Some(
                "deployment.environment=prod,team=a%2Fb".to_owned(),
            ),
            agent_nats_url: Some("tls://nats.example.test:4222".to_owned()),
            ..HiveFacts::default()
        };
        assert_eq!(
            render_hive_facts(&facts).unwrap(),
            "HIVE_DASHBOARD_PORT=7123\n\
             HYPERHIVE_CLAUDE_CODE_PATH=/nix/store/0abc-claude-code-2.1.220\n\
             HYPERHIVE_OTEL_EXTRA_RESOURCE_ATTRIBUTES=deployment.environment=prod,team=a%2Fb\n\
             HIVE_AGENT_NATS_URL=tls://nats.example.test:4222\n"
        );
    }

    /// The egress facts: the proxy as `<ip>:<port>`, and the hosts that skip it.
    #[test]
    fn hive_facts_render_the_egress_proxy() {
        let facts = HiveFacts {
            squid_proxy: Some("10.42.0.1:3128".to_owned()),
            egress_no_proxy: Some("10.42.0.1,pr1ma.example.test,forge.example.test".to_owned()),
            ..HiveFacts::default()
        };
        assert_eq!(
            render_hive_facts(&facts).unwrap(),
            "HIVE_SQUID_PROXY=10.42.0.1:3128\n\
             HIVE_EGRESS_NO_PROXY=10.42.0.1,pr1ma.example.test,forge.example.test\n"
        );
    }

    /// Neither egress fact may carry a second assignment, a space, a quote or a
    /// URL: the agent builds its proxy URL and `NO_PROXY` from them.
    #[test]
    fn hive_facts_refuse_bad_egress_values() {
        for bad in [
            "",
            "10.42.0.1:3128\nHIVE_AGENT_NATS_URL=x",
            "http://10.42.0.1:3128",
            "10.42.0.1:3128 ",
            "u:p@10.42.0.1:3128",
        ] {
            let facts = HiveFacts {
                squid_proxy: Some(bad.to_owned()),
                ..HiveFacts::default()
            };
            assert!(
                render_hive_facts(&facts).is_err(),
                "{bad:?} must be refused"
            );
        }
        for bad in [
            "",
            "a.test,b.test\nPATH=/x",
            "a.test b.test",
            "\"a.test\"",
            "$(x)",
        ] {
            let facts = HiveFacts {
                egress_no_proxy: Some(bad.to_owned()),
                ..HiveFacts::default()
            };
            assert!(
                render_hive_facts(&facts).is_err(),
                "{bad:?} must be refused"
            );
        }
    }

    /// The claude path is a top-level store path or nothing, and neither
    /// value may carry generator syntax.
    #[test]
    fn hive_facts_refuse_bad_claude_paths_and_resource_attributes() {
        for bad in [
            "/nix/store/",
            "/nix/store/abc-claude/bin/claude",
            "/tmp/claude",
            "nix/store/abc-claude",
            "/nix/store/abc-claude\nLD_PRELOAD=/x",
        ] {
            let facts = HiveFacts {
                claude_code_path: Some(bad.to_owned()),
                ..HiveFacts::default()
            };
            assert!(
                render_hive_facts(&facts).is_err(),
                "{bad:?} must be refused"
            );
        }
        for bad in [
            "a=b\nPATH=/x",
            "a=${HOME}",
            "a=b c",
            "a=\"b\"",
            "a=b\\c",
            "",
        ] {
            let facts = HiveFacts {
                otel_extra_resource_attributes: Some(bad.to_owned()),
                ..HiveFacts::default()
            };
            assert!(
                render_hive_facts(&facts).is_err(),
                "{bad:?} must be refused"
            );
        }
    }

    /// `--ephemeral` is asked for, not assumed: present exactly when the root has
    /// been dealt with, ahead of the binds it applies to. The binds are the
    /// control that the lookup finds what is there.
    #[test]
    fn ephemeral_is_passed_only_when_asked() {
        let binds = [bind("/agents/atlas/work")];
        let with = nspawn_flags(&binds, &[], true);
        assert_eq!(with[0], "--ephemeral", "{with:?}");
        assert!(with.iter().any(|f| f.starts_with("--bind-ro=")), "{with:?}");
        let without = nspawn_flags(&binds, &[], false);
        assert!(!without.iter().any(|f| f == "--ephemeral"), "{without:?}");
        assert_eq!(without.len() + 1, with.len());
    }

    /// The swarm's forge and matrix containers are not the hive's: every
    /// request that names a container as a string refuses them before
    /// touching the filesystem. Control: an agent container name passes.
    #[test]
    fn container_name_requests_refuse_swarm_service_containers() {
        let isolation = NetworkIsolation {
            bridge: "hive0".to_owned(),
            gateway_ip: "10.42.0.1".to_owned(),
        };
        let facts = HiveFacts::default();
        for name in ["hive-forge", "hive-matrix"] {
            let errs = [
                handle_write_nspawn_flags(name, &[], &isolation, &[], &facts).unwrap_err(),
                write_resource_limits(name, "4G", "200%", None, None).unwrap_err(),
                remove_service_dropin(name).unwrap_err(),
            ];
            for e in errs {
                assert!(
                    e.to_string().contains("is not managed by hive"),
                    "{name}: {e:#}"
                );
            }
        }
        assert_eq!(validate_container_system_name("h-atlas").unwrap(), "atlas");
    }
}