hyperhive/hive-c0re/src/meta.rs

//! Single hive-c0re-owned flake at `/var/lib/hyperhive/meta/` that
//! exports one `nixosConfiguration` per agent and drives the system-wide
//! deploy audit trail. Flow (`sync_agents`, two-phase `prepare_deploy` /
//! `finalize_deploy` / `abort_deploy`, `lock_update_hyperhive`):
//! `docs/approvals.md::Meta flake`.

use std::path::Path;

use anyhow::{Context, Result, bail};
use tokio::process::Command;
use tokio::sync::Mutex;

use crate::coordinator::HiveEnv;
use crate::lifecycle;

const GIT_NAME: &str = "c0re";
const GIT_EMAIL: &str = "c0re@hyperhive.local";

/// Single-writer lock around every meta-repo operation. Git isn't
/// safe to drive from concurrent processes against the same `.git/`
/// — two simultaneous `git add` / `commit` invocations race on
/// `.git/index.lock`; if either dies before releasing, the lock
/// sticks and the next operation hits "another git process seems to
/// be running" until somebody `rm`s it manually. Holding this mutex
/// across each public function's git+nix calls makes parallel
/// rebuilds (`auto_update` + dashboard-triggered + apply-commit)
/// take turns instead of colliding.
static META_LOCK: Mutex<()> = Mutex::const_new(());

// Exclusivity for meta-repo *windows* that span multiple `META_LOCK`
// acquisitions — above all the two-phase deploy (`prepare_deploy` stages
// `flake.lock` uncommitted for the whole container build;
// `finalize_deploy` / `abort_deploy` resolve it) — is **not** a mutex in
// this module. `META_LOCK` above serializes individual git ops but cannot
// keep another op out of that staged window; that window is owned by the
// job queue instead, as `Resource::MetaWindow`, declared by every
// meta-mutating node kind (`NodeKind::needs_meta_window`). A resource can
// be held by a subtree root across its children, which a `MutexGuard`
// (bounded by one executor fn) cannot — that's what lets the deploy be
// modelled as sub-nodes rather than one opaque node.

/// Where the manager sees this directory inside its container (RO bind).
pub const CONTAINER_MANAGER_META_MOUNT: &str = "/meta";

#[derive(Debug, Clone)]
pub struct AgentSpec {
    pub name: String,
    pub is_manager: bool,
    pub port: u16,
}

/// Idempotently reconcile the meta repo with the current agent set.
/// First call inits the git repo, runs `nix flake lock`, and lands a
/// seed commit. Subsequent calls only touch `flake.nix` when the
/// rendered contents differ from disk; an unchanged `flake.nix` is a
/// no-op.
pub async fn sync_agents(hive: &HiveEnv, agents: &[AgentSpec]) -> Result<()> {
    let _guard = META_LOCK.lock().await;
    let dir = crate::paths::meta_root();
    std::fs::create_dir_all(&dir).with_context(|| format!("create {}", dir.display()))?;

    let new_flake = render_flake(
        &hive.hyperhive_flake,
        &hive.hyperhive_docs_flake,
        &hive.nixpkgs_flake,
        hive.dashboard_port,
        &hive.operator_pronouns,
        &hive.context_window_tokens,
        agents,
    );
    let flake_path = dir.join("flake.nix");
    let on_disk = std::fs::read_to_string(&flake_path).unwrap_or_default();
    let initial = !dir.join(".git").exists();

    // Embedded-CA list (self-signed hive CA + peer CAs): keep the
    // `./hive-ca.pem` / `./peer-ca-<N>.pem` files at the meta root in
    // lockstep with their host sources so the build-time `certificateFiles`
    // list render_flake emits always resolves. Empty when neither a
    // self-signed hive CA nor any peer CA is configured.
    let (ca_files, ca_changed) = ca_embed_state(&dir);

    // Skip only when both the flake AND the embedded CA are unchanged — a
    // CA rotation with an otherwise-identical flake must still re-commit.
    if !initial && on_disk == new_flake && !ca_changed {
        return Ok(());
    }

    // Safety guard: refuse to write an empty agent list over a non-empty
    // on-disk flake. An empty `agents` slice is never intentional — it
    // means `nixos-container list` failed and the caller got an empty
    // fallback. Overwriting here would drop every agent from the meta
    // flake and trigger unnecessary (and potentially destructive) cascade
    // rebuilds. Callers that genuinely need to clear the agent list
    // (there are none today) must handle this case explicitly.
    if agents.is_empty() && !initial && !on_disk.is_empty() {
        tracing::warn!(
            "sync_agents: refusing to overwrite non-empty meta flake with empty agent list \
             (nixos-container list may have failed)"
        );
        return Ok(());
    }

    std::fs::write(&flake_path, &new_flake)
        .with_context(|| format!("write {}", flake_path.display()))?;

    // Materialise the embedded CA list next to flake.nix + drop any stale
    // CA file; `ca_touched` is every filename written or removed, staged
    // for commit below. Public CA certs only; no private key is embedded.
    let ca_touched = materialise_ca_files(&dir, &ca_files)?;

    // Reconcile topology.json against the live agent set — adds
    // entries for newly-spawned agents (default: manager as parent,
    // manager itself as root) and drops removed agents. Operator
    // overrides via the write API are preserved because reconcile
    // only fills in missing entries. Idempotent; when nothing changed
    // the file isn't touched.
    let agent_names: Vec<String> = agents.iter().map(|a| a.name.clone()).collect();
    let pending: Vec<String> = crate::coordinator::Coordinator::pending_init_names()
        .into_iter()
        .map(hive_types::Ident::into_string)
        .collect();
    crate::topology::reconcile(&agent_names, &pending)
        .with_context(|| format!("reconcile {}", crate::topology::topology_path().display()))?;

    // Refresh /var/lib/hyperhive/run/agent-sockets.json — drives the
    // gateway's unix-socket upstreams. Every agent binds a unix socket
    // (`HIVE_WEB_SOCKET`); the gateway routes via this map, falling back
    // to a computed TCP loopback port (`lifecycle::agent_web_port`) only
    // while an agent's socket marker is absent. Best-effort + non-fatal.
    // See `docs/gateway.md::Per-agent unix-socket upstream`.
    if let Err(e) = crate::agent_sockets::write(&agent_names) {
        tracing::warn!(error = ?e, "agent_sockets::write failed (non-fatal)");
    }

    // Refresh /var/lib/hyperhive/gateway/agents.conf — the nginx include
    // file the gateway container bind-mounts and nginx reads at runtime.
    // c0re triggers a reload (or start) inside hive-gateway via hive-priv
    // after writing the file. Same best-effort + non-fatal shape.
    if let Err(e) = crate::gateway_nginx::write(&agent_names).await {
        tracing::warn!(error = ?e, "gateway_nginx::write failed (non-fatal)");
    }

    if initial {
        git(&dir, &["init", "--initial-branch=main"]).await?;
    }
    // Stage flake.nix *before* running nix flake lock. When meta is
    // a git repo, nix treats it as a `git+file://` self-reference;
    // its dirty-tree fetcher includes index entries (tracked +
    // staged) but skips untracked files, so without the stage step
    // an untracked flake.nix surfaces as "source tree does not
    // contain '/flake.nix'". Lock then commit once with both
    // flake.nix and flake.lock — single commit per change.
    git(&dir, &["add", "flake.nix"]).await?;
    // Stage every embedded CA file we wrote or removed (hive CA + peer
    // CAs). `git add <path>` stages a deletion when the path is tracked
    // and now gone; best-effort so the never-tracked-and-absent case
    // (pathspec mismatch) is a harmless no-op.
    for name in &ca_touched {
        let _ = git(&dir, &["add", "--", name]).await;
    }
    // Stage topology.json on every sync (regenerated by reconcile
    // above when the agent set changed). git add is a no-op when the
    // file content is unchanged.
    if crate::topology::topology_path().exists() {
        git(&dir, &["add", "topology.json"]).await?;
    }
    // Stage tool-groups.json when it exists. Created on first
    // `set_groups` call (operator-driven); absent = all agents on
    // their role defaults, no file needed. git add is a no-op when
    // the file is unchanged.
    if crate::tool_groups::tool_groups_path().exists() {
        git(&dir, &["add", "tool-groups.json"]).await?;
    }
    // Stage capabilities.json when it exists. Created on first
    // `set_caps` call; absent = no agents have extra capabilities.
    if crate::capabilities::capabilities_path().exists() {
        git(&dir, &["add", "capabilities.json"]).await?;
    }
    // Stage roles.json when it exists. Written by topology::write_roles /
    // reconcile_roles on first role assignment or manager default seeding.
    // Without this, roles.json appears as untracked in the meta repo
    // (visible in `git status`) which can confuse nix dirty-tree fetches.
    if crate::topology::roles_path().exists() {
        git(&dir, &["add", "roles.json"]).await?;
    }
    nix(&dir, &["flake", "lock"]).await?;
    if std::path::Path::new(&dir).join("flake.lock").exists() {
        git(&dir, &["add", "flake.lock"]).await?;
    }
    // Build the commit message from what's actually staged so it
    // reflects reality — and skip the commit entirely when nothing
    // changed (avoids "nothing to commit" errors on redundant syncs).
    let staged = git_staged_names(&dir).await?;
    if staged.is_empty() {
        return Ok(());
    }
    let msg = if initial {
        format!("seed meta from {} agent(s)", agents.len())
    } else {
        // Compose a message that names every file that actually changed,
        // mapping the on-disk filename to a short human label.
        let labels: Vec<&str> = staged
            .iter()
            .filter_map(|f| match f.as_str() {
                "flake.nix" => Some("flake"),
                "flake.lock" => Some("lock"),
                "hive-ca.pem" => Some("hive-ca"),
                f if f.starts_with("peer-ca-") && has_pem_ext(f) => Some("peer-ca"),
                "topology.json" => Some("topology"),
                "capabilities.json" => Some("capabilities"),
                "tool-groups.json" => Some("tool-groups"),
                "roles.json" => Some("roles"),
                _ => None,
            })
            .collect();
        if labels.is_empty() {
            "meta: update".to_owned()
        } else {
            format!("meta: update {}", labels.join(", "))
        }
    };
    git_commit(&dir, &msg).await?;
    Ok(())
}

/// Phase 1 of an apply-commit deploy. Updates the locked rev of
/// `agent-<name>` to whatever `applied/<name>/main` currently points
/// at and **stages** the lock so `nixos-container update --flake
/// meta#<n>` (which reads via `git+file://`) sees the new rev via
/// the index. Doesn't commit — `finalize_deploy` commits on build
/// success, `abort_deploy` drops the staged change on failure so
/// meta history only carries successful deploys.
pub async fn prepare_deploy(name: &str) -> Result<()> {
    let _guard = META_LOCK.lock().await;
    let dir = crate::paths::meta_root();
    let input = format!("agent-{name}");
    // Re-lock the agent input against the LOCAL applied mirror, not the
    // persistent forge URL declared in the meta flake (see the `## Meta flake`
    // note in docs/approvals.md): the deploy must build the exact reviewed
    // config that `verify_commit` gated and `applied/<n>/main` was
    // fast-forwarded to, and it must keep working when the forge is
    // unreachable (rebuilds fire on crash-restart / meta bumps too, not just
    // config PRs). `--override-input` writes the applied rev into the lock;
    // the forge URL stays the declared, reviewable source of truth.
    let applied = applied_override_url(&crate::paths::applied_dir(name));
    nix_logged(
        &dir,
        &[
            "flake",
            "update",
            &input,
            "--override-input",
            &input,
            &applied,
        ],
        name,
        "prepare-deploy",
    )
    .await?;
    // Stage the new lock — git+file://'s dirty-tree fetcher reads
    // index entries, so the upcoming nixos-container update sees the
    // bumped rev without a commit yet.
    git(&dir, &["add", "flake.lock"]).await
}

/// Phase 2-success. Commit the staged lock with the deployed tag +
/// sha as the message. No-op when the rev was already at the right
/// place (nothing staged → nothing to commit).
pub async fn finalize_deploy(name: &str, sha: &str, tag: &str) -> Result<()> {
    let _guard = META_LOCK.lock().await;
    let dir = crate::paths::meta_root();
    if !paths_dirty(&dir, &["flake.lock"]).await? {
        return Ok(());
    }
    let short = &sha[..sha.len().min(12)];
    git_commit_paths(
        &dir,
        &format!("deploy {name} {tag} {short}"),
        &["flake.lock"],
    )
    .await
}

/// Phase 2-failure. Unstage + restore the lock so meta returns to
/// the previously-committed shas. The failed proposal is still
/// captured in `applied/<n>`'s annotated `failed/<id>` tag.
pub async fn abort_deploy() -> Result<()> {
    let _guard = META_LOCK.lock().await;
    let dir = crate::paths::meta_root();
    git(&dir, &["restore", "--staged", "flake.lock"]).await?;
    git(&dir, &["restore", "flake.lock"]).await
}

/// One-shot used by the manual-rebuild path: relock just one
/// agent's input and commit the lock change if any. Single-phase
/// (no separate finalize) because rebuild has no failure-revert
/// semantics — it always wants the latest main.
pub async fn lock_update_for_rebuild(name: &str) -> Result<()> {
    let _guard = META_LOCK.lock().await;
    let dir = crate::paths::meta_root();
    let input = format!("agent-{name}");
    // Re-lock from the local applied mirror, not the persistent forge URL —
    // same rationale as `prepare_deploy`: build exactly `applied/<n>/main` and
    // stay reproducible when the forge is unreachable.
    let applied = applied_override_url(&crate::paths::applied_dir(name));
    nix(
        &dir,
        &[
            "flake",
            "update",
            &input,
            "--override-input",
            &input,
            &applied,
        ],
    )
    .await?;
    if !paths_dirty(&dir, &["flake.lock"]).await? {
        return Ok(());
    }
    git(&dir, &["add", "flake.lock"]).await?;
    git_commit_paths(
        &dir,
        &format!("rebuild {name}: lock update"),
        &["flake.lock"],
    )
    .await
}

/// Build the `--override-input` value pinning an agent's config repo to
/// an exact revision: `git+file://<applied_dir>?rev=<sha>`. Pure so the
/// URL shape is unit-testable without a nix shellout.
fn agent_input_override(applied_dir: &Path, sha: &str) -> String {
    format!("git+file://{}?rev={sha}", applied_dir.display())
}

/// `--override-input` URL re-locking an agent's config input against its
/// LOCAL applied mirror (`git+file://<applied_dir>`, current `main` head)
/// instead of the persistent forge URL declared in the meta flake. The
/// deploy + rebuild paths re-lock from here: the applied tree was already
/// fast-forwarded to the reviewed head, so this builds exactly that config
/// and stays reproducible when the forge is unreachable. No `?rev` — `main`
/// head is the reviewed head at deploy time. Pure so it's unit-testable.
fn applied_override_url(applied_dir: &Path) -> String {
    format!("git+file://{}", applied_dir.display())
}

/// Non-mutating "would this commit apply?" verify for the PR-based config
/// flow. Evaluates the agent's nixos configuration with its meta
/// input overridden to the exact `sha`, WITHOUT moving `applied/main` or
/// writing the real meta `flake.lock`: `--override-input` pins the input
/// at eval time only and `--no-write-lock-file` guarantees the on-disk
/// lock is never touched (a genuinely-needed lock change surfaces as an
/// error rather than a silent mutation). Confirms the flake at `sha`
/// evaluates and the meta lock resolves around it. `Ok(())` = would
/// apply; `Err` carries the eval/lock failure so the operator's approve
/// is rejected before any live-state change.
///
/// Precondition: `sha` must be reachable in `applied_dir`'s object store
/// — the caller fetches the PR head into `applied_dir` first. Eval-only
/// (no build): catches nix / eval / module-option errors + lock
/// resolution at the same cost profile as the legacy pre-merge check;
/// the real container build still runs (and can still roll back) on the
/// actual apply, so this is the right "would this apply" gate.
pub async fn verify_commit(name: &str, applied_dir: &Path, sha: &str) -> Result<()> {
    let _guard = META_LOCK.lock().await;
    let dir = crate::paths::meta_root();
    let input = format!("agent-{name}");
    let over = agent_input_override(applied_dir, sha);
    let attr = format!(".#nixosConfigurations.{name}.config.system.build.toplevel.drvPath");
    nix_logged(
        &dir,
        &[
            "eval",
            &attr,
            "--override-input",
            &input,
            &over,
            "--no-write-lock-file",
        ],
        name,
        "verify",
    )
    .await
}

/// Update one or more named inputs in the meta flake and commit
/// the resulting lock change with a single combined message.
/// Used by the dashboard's "update meta inputs" form so the
/// operator can bulk-bump `hyperhive` + selected agents in one
/// shot. Each input name is passed verbatim to
/// Run `nix flake update [inputs...]` on the meta flake and commit the
/// resulting lock changes. When `inputs` is empty, updates ALL inputs
/// (bare `nix flake update`). The caller is responsible for picking
/// real input keys (e.g. via `inputs_view()` snapshotted from the lock
/// file) when targeting specific inputs.
pub async fn lock_update(inputs: &[String]) -> Result<()> {
    let _guard = META_LOCK.lock().await;
    let dir = crate::paths::meta_root();
    let mut args: Vec<&str> = vec!["flake", "update"];
    for i in inputs {
        args.push(i.as_str());
    }
    nix(&dir, &args).await?;
    if !paths_dirty(&dir, &["flake.lock"]).await? {
        return Ok(());
    }
    git(&dir, &["add", "flake.lock"]).await?;
    let msg = if inputs.is_empty() {
        "lock update: all inputs".to_string()
    } else if inputs.len() == 1 {
        format!("lock update: {}", inputs[0])
    } else {
        format!("lock update: {}", inputs.join(", "))
    };
    git_commit_paths(&dir, &msg, &["flake.lock"]).await
}

/// One-shot used by the auto-update path: pin the latest hyperhive
/// rev, commit if the lock changed. Cheaper than `sync_agents`
/// because the per-agent inputs aren't touched.
pub async fn lock_update_hyperhive() -> Result<()> {
    let _guard = META_LOCK.lock().await;
    let dir = crate::paths::meta_root();
    nix(&dir, &["flake", "update", "hyperhive"]).await?;
    if !paths_dirty(&dir, &["flake.lock"]).await? {
        return Ok(());
    }
    git(&dir, &["add", "flake.lock"]).await?;
    git_commit_paths(&dir, "bump hyperhive", &["flake.lock"]).await
}

/// Write the tool-groups file for `agent` and commit it atomically
/// under `META_LOCK`. Ensures the JSON change is staged + committed
/// before the next `prepare_deploy` or `sync_agents` runs, so the
/// working tree is never left dirty by an untimely `PermChange` write.
pub async fn commit_tool_groups(agent: &str, groups: &[String]) -> Result<()> {
    let _guard = META_LOCK.lock().await;
    crate::tool_groups::set_groups(agent, groups)?;
    let dir = crate::paths::meta_root();
    if crate::tool_groups::tool_groups_path().exists() {
        git(&dir, &["add", "tool-groups.json"]).await?;
    }
    if paths_dirty(&dir, &["tool-groups.json"]).await? {
        git_commit_paths(
            &dir,
            &format!("set tool-groups for {agent}"),
            &["tool-groups.json"],
        )
        .await?;
    }
    Ok(())
}

/// Write the capabilities file for `agent` and commit it atomically
/// under `META_LOCK`. Same rationale as `commit_tool_groups`.
pub async fn commit_capabilities(agent: &str, caps: &[String]) -> Result<()> {
    let _guard = META_LOCK.lock().await;
    crate::capabilities::set_caps(agent, caps)
        .map_err(|e| anyhow::anyhow!("set capabilities for {agent}: {e}"))?;
    let dir = crate::paths::meta_root();
    if crate::capabilities::capabilities_path().exists() {
        git(&dir, &["add", "capabilities.json"]).await?;
    }
    if paths_dirty(&dir, &["capabilities.json"]).await? {
        git_commit_paths(
            &dir,
            &format!("set capabilities for {agent}"),
            &["capabilities.json"],
        )
        .await?;
    }
    Ok(())
}

/// Write both perm files for `agent` (whichever are `Some`) and commit
/// them in a SINGLE git commit under `META_LOCK` — the batch
/// `POST /api/permissions` path. A `None` field leaves that file
/// untouched. One commit + (caller does) one rebuild means changing an
/// agent's caps and tool-groups together no longer triggers two
/// rebuilds. Mirrors the staging discipline of `commit_tool_groups` /
/// `commit_capabilities`.
///
/// # Errors
///
/// Returns an error if writing either JSON file fails, a capability name
/// is invalid (`set_caps`), or a git stage/commit step fails.
pub async fn commit_perms(
    agent: &str,
    groups: Option<&[String]>,
    caps: Option<&[String]>,
) -> Result<()> {
    let _guard = META_LOCK.lock().await;
    let dir = crate::paths::meta_root();
    let mut parts: Vec<&str> = Vec::new();
    if let Some(groups) = groups {
        crate::tool_groups::set_groups(agent, groups)?;
        if crate::tool_groups::tool_groups_path().exists() {
            git(&dir, &["add", "tool-groups.json"]).await?;
        }
        parts.push("tool-groups");
    }
    if let Some(caps) = caps {
        crate::capabilities::set_caps(agent, caps)
            .map_err(|e| anyhow::anyhow!("set capabilities for {agent}: {e}"))?;
        if crate::capabilities::capabilities_path().exists() {
            git(&dir, &["add", "capabilities.json"]).await?;
        }
        parts.push("capabilities");
    }
    let paths = ["tool-groups.json", "capabilities.json"];
    if paths_dirty(&dir, &paths).await? {
        git_commit_paths(
            &dir,
            &format!("set {} for {agent}", parts.join(" + ")),
            &paths,
        )
        .await?;
    }
    Ok(())
}

/// Write the topology file and commit it atomically under `META_LOCK`.
/// Returns `Err(String)` on validation failure (unknown agent, cycle,
/// etc.) — same shape as `topology::set_parent` — so callers can
/// surface the error as a user-visible message. Git failures are
/// logged as warnings and don't propagate: the topology write already
/// succeeded, and `sync_agents` will pick up any un-committed change
/// on the next run as a safety net.
pub async fn commit_topology(
    child: &str,
    new_parent: Option<&str>,
) -> std::result::Result<(), String> {
    let _guard = META_LOCK.lock().await;
    crate::topology::set_parent(child, new_parent)?;
    let dir = crate::paths::meta_root();
    let stage = async {
        git(&dir, &["add", "topology.json"]).await?;
        if paths_dirty(&dir, &["topology.json"]).await? {
            git_commit_paths(
                &dir,
                &format!("topology: {} → {}", child, new_parent.unwrap_or("<root>")),
                &["topology.json"],
            )
            .await?;
        }
        Ok::<_, anyhow::Error>(())
    };
    if let Err(e) = stage.await {
        tracing::warn!(%child, ?new_parent, error = ?e, "commit_topology: topology written but git commit failed (sync_agents will recover)");
    }
    Ok(())
}

/// Batch variant of [`commit_topology`]: applies every `(child, new_parent)`
/// move under a single `META_LOCK` acquisition and creates **one** git commit
/// for all of them. Moves are applied in the order given; the first
/// validation error short-circuits the whole batch. True atomic write: all
/// moves are pre-validated against a cumulative in-memory state with
/// [`crate::topology::apply_set_parent`] before anything touches disk, then
/// [`crate::topology::write`] is called exactly once. If any move fails
/// validation the topology file is never modified.
///
/// The multi-move commit message uses `moves[0].1` as the destination label.
/// This is intentional: the dashboard bulk-move UI always sends a single
/// destination for all selected agents, so the message is always accurate in
/// practice.
///
/// Returns a `Vec` of `(child, old_parent)` pairs for every move that
/// actually changed the topology (idempotent same-parent moves are skipped),
/// so the caller can send targeted notifications.
///
/// # Errors
///
/// Returns a `String` error if any move fails validation (cycle, unknown
/// agent, etc.) or if the topology file cannot be written. Git-commit failure
/// is logged as a warning and does not propagate — `sync_agents` will recover.
pub async fn bulk_commit_topology(
    moves: &[(&str, Option<&str>)],
) -> std::result::Result<Vec<(String, Option<String>)>, String> {
    if moves.is_empty() {
        return Ok(vec![]);
    }
    let _guard = META_LOCK.lock().await;
    // Snapshot parents before any writes so we can compute the diff.
    let topo_before = crate::topology::read();
    // Validate all moves against a cumulative in-memory state -- no disk
    // writes yet; first error aborts with the topology file untouched.
    let mut next = topo_before.clone();
    for (child, new_parent) in moves {
        next = crate::topology::apply_set_parent(&next, child, *new_parent)?;
    }
    // Only flush to disk if something actually changed.
    if next != topo_before {
        crate::topology::write(&next).map_err(|e| format!("{e:#}"))?;
    }
    // Commit the whole batch as one git operation.
    let dir = crate::paths::meta_root();
    let commit_msg = if moves.len() == 1 {
        let (child, new_parent) = moves[0];
        format!("topology: {} → {}", child, new_parent.unwrap_or("<root>"))
    } else {
        let names: Vec<&str> = moves.iter().map(|(c, _)| *c).collect();
        let dest = moves[0].1.unwrap_or("<root>");
        format!(
            "topology: move {} agents → {} ({})",
            moves.len(),
            dest,
            names.join(", ")
        )
    };
    let stage = async {
        git(&dir, &["add", "topology.json"]).await?;
        if paths_dirty(&dir, &["topology.json"]).await? {
            git_commit_paths(&dir, &commit_msg, &["topology.json"]).await?;
        }
        Ok::<_, anyhow::Error>(())
    };
    if let Err(e) = stage.await {
        tracing::warn!(error = ?e, "bulk_commit_topology: topology written but git commit failed (sync_agents will recover)");
    }
    // Return (child, old_parent) for each move that changed state.
    let changed = moves
        .iter()
        .filter_map(|(child, new_parent)| {
            let old = topo_before.get(*child).cloned().flatten();
            (old.as_deref() != *new_parent).then_some((child.to_string(), old))
        })
        .collect();
    Ok(changed)
}

#[allow(
    clippy::too_many_arguments,
    reason = "many genuine flake inputs (source flakes, port, pronouns, tokens, \
              agents, forge base); a params struct would just move the same fields"
)]
fn render_flake(
    hyperhive_flake: &str,
    docs_flake: &str,
    nixpkgs_flake: &str,
    dashboard_port: u16,
    operator_pronouns: &str,
    context_window_tokens: &std::collections::HashMap<String, u64>,
    agents: &[AgentSpec],
) -> String {
    render_flake_with_lookup(
        hyperhive_flake,
        docs_flake,
        nixpkgs_flake,
        dashboard_port,
        operator_pronouns,
        context_window_tokens,
        agents,
        agent_canonical_inputs,
    )
}

/// Canonical inputs meta knows how to dedup. An agent that declares one
/// of these as a top-level input in its own `flake.nix` will get a
/// `follows = "<name>"` line emitted in meta — collapsing the
/// otherwise-separate-but-identical `nixpkgs_N` nodes into a single
/// meta-level reference.
const CANONICAL_INPUTS: &[&str] = &["nixpkgs"];

/// Env vars hive-c0re forwards from its own systemd unit env into every
/// sub-agent's harness service env. Each entry is `(env_var_name,
/// host_value)`. Empty / unset vars are filtered out so absent options
/// don't render no-op `FOO = ""` lines into the meta flake.
///
/// Returns owned `String` values so the result is `'static`-friendly +
/// trivial to stub from tests (which build their own slice instead of
/// touching process-wide env).
const FORWARDED_VARS: &[&str] = &[
    "HIVE_FORGE_URL",
    "HIVE_FORGE_PUBLIC_URL",
    "HIVE_MATRIX_URL",
    "HYPERHIVE_PEERS",
    "HYPERHIVE_HIVE_DOMAIN",
    "HYPERHIVE_HIVE_NAME",
    "HYPERHIVE_SWARM_NAME",
];

fn forwarded_env_vars() -> Vec<(&'static str, String)> {
    FORWARDED_VARS
        .iter()
        .filter_map(|&name| {
            std::env::var(name)
                .ok()
                .filter(|v| !v.is_empty())
                .map(|v| (name, v))
        })
        .collect()
}

/// Filename the hive's own self-signed CA cert is embedded under at the
/// meta-flake root. One entry of the embedded-CA list `render_flake`
/// references in `security.pki.certificateFiles` (see `embedded_ca_files`);
/// peer CAs sit alongside it as `peer-ca-<N>.pem`.
const HIVE_CA_FILE: &str = "hive-ca.pem";

/// Host path of the hive CA *certificate*, when self-signed TLS is active.
/// `hive-tls.nix` sets `HIVE_TLS_CA_PATH` in hive-c0re's service env (to
/// `<tls.stateDir>/ca.pem`) whenever the gateway serves a self-signed,
/// hive-CA-signed leaf. Returns `Some(path)` only when the var is set AND
/// the cert exists on disk — so render + write stay consistent (we never
/// emit a `certificateFiles` reference to a file we didn't embed). Only the
/// public cert is ever read here; the CA private key never leaves the host.
fn hive_ca_source() -> Option<String> {
    let path = std::env::var("HIVE_TLS_CA_PATH").ok()?;
    if path.is_empty() || !std::path::Path::new(&path).is_file() {
        return None;
    }
    Some(path)
}

/// Host paths of peer-hive root CA certificates, from `HIVE_PEER_CA_PATHS`
/// (colon-separated; set by hive-c0re.nix from `swarm.peers.<d>.caCert`).
/// Each is embedded alongside the hive CA so a peer's CA is trusted
/// everywhere the hive's own internal CA is — i.e. by every agent. Empty
/// segments and paths that don't resolve to a file are dropped, so we
/// never reference a `certificateFiles` entry we couldn't embed.
fn peer_ca_sources() -> Vec<String> {
    let Ok(raw) = std::env::var("HIVE_PEER_CA_PATHS") else {
        return Vec::new();
    };
    raw.split(':')
        .map(str::trim)
        .filter(|p| !p.is_empty() && std::path::Path::new(p).is_file())
        .map(ToOwned::to_owned)
        .collect()
}

/// Hive-wide OTEL config injected into every agent's build, read off
/// hive-c0re's own unit env (set from `services.hyperhive.otel.*` in
/// `nix/modules/hive-c0re.nix`). A present, non-empty
/// `HYPERHIVE_OTEL_ENDPOINT` is the enable signal — the host module
/// asserts the endpoint is set whenever `otel.enable` is true, so
/// "endpoint present" == "OTEL on". The optional fields map to the
/// matching host options and are only carried when set.
struct OtelConfig {
    endpoint: String,
    protocol: String,
    extra_resource_attributes: Option<String>,
    headers_credential: Option<String>,
    metric_interval_ms: Option<u64>,
    /// `HYPERHIVE_OTEL_DEBUG=1` → `hyperhive.otel.debug = true` →
    /// `CLAUDE_CODE_OTEL_DIAG_STDERR=1` in every agent's env.
    debug: bool,
}

/// Read the hive-wide OTEL config from env, or `None` when OTEL is off.
/// Mirrors `hive_ca_source` — host state surfaced to the meta renderer
/// so it can bake build-time `hyperhive.otel.*` config into each agent
/// (the per-agent options the harness modules consume). Returns `None`
/// when the endpoint signal is absent so the renderer emits no
/// `hyperhive.otel.*` lines and agents keep the disabled default.
fn otel_config() -> Option<OtelConfig> {
    let endpoint = std::env::var("HYPERHIVE_OTEL_ENDPOINT")
        .ok()
        .filter(|v| !v.is_empty())?;
    let protocol = std::env::var("HYPERHIVE_OTEL_PROTOCOL")
        .ok()
        .filter(|v| !v.is_empty())
        .unwrap_or_else(|| "http/protobuf".to_owned());
    let extra_resource_attributes = std::env::var("HYPERHIVE_OTEL_EXTRA_RESOURCE_ATTRIBUTES")
        .ok()
        .filter(|v| !v.is_empty());
    let headers_credential = std::env::var("HYPERHIVE_OTEL_HEADERS_CREDENTIAL")
        .ok()
        .filter(|v| !v.is_empty());
    let metric_interval_ms = std::env::var("HYPERHIVE_OTEL_METRIC_INTERVAL_MS")
        .ok()
        .and_then(|v| v.parse::<u64>().ok())
        .filter(|v| *v > 0);
    let debug = std::env::var("HYPERHIVE_OTEL_DEBUG")
        .ok()
        .is_some_and(|v| v == "1");
    Some(OtelConfig {
        endpoint,
        protocol,
        extra_resource_attributes,
        headers_credential,
        metric_interval_ms,
        debug,
    })
}

/// The ordered set of CA certs embedded next to the meta flake, as
/// `(filename, host_source_path)`. The self-signed hive CA (when active)
/// is `hive-ca.pem`; each peer CA is `peer-ca-<N>.pem` in declaration
/// order. `render_flake` emits exactly these filenames into
/// `security.pki.certificateFiles` and `sync_agents` materialises them,
/// so the rendered reference and the embedded files always agree.
fn embedded_ca_files() -> Vec<(String, String)> {
    let mut out = Vec::new();
    if let Some(p) = hive_ca_source() {
        out.push((HIVE_CA_FILE.to_owned(), p));
    }
    for (i, p) in peer_ca_sources().into_iter().enumerate() {
        out.push((format!("peer-ca-{i}.pem"), p));
    }
    out
}

/// Write each desired embedded CA file next to `flake.nix` and remove
/// any stale one (a hive CA turned off, or a peer dropped from config),
/// so the flake never references a file we didn't write. Returns every
/// filename written or removed, for the caller to stage. The public CA
/// certs only; no private key is ever embedded.
fn materialise_ca_files(dir: &Path, ca_files: &[(String, String)]) -> Result<Vec<String>> {
    let desired: std::collections::HashSet<&str> =
        ca_files.iter().map(|(n, _)| n.as_str()).collect();
    let mut touched: Vec<String> = Vec::new();
    if let Ok(entries) = std::fs::read_dir(dir) {
        for e in entries.flatten() {
            let fname = e.file_name();
            let Some(name) = fname.to_str() else { continue };
            if is_embedded_ca_name(name) && !desired.contains(name) {
                let _ = std::fs::remove_file(dir.join(name));
                touched.push(name.to_owned());
            }
        }
    }
    for (name, content) in ca_files {
        let path = dir.join(name);
        std::fs::write(&path, content).with_context(|| format!("write {}", path.display()))?;
        touched.push(name.clone());
    }
    Ok(touched)
}

/// True for a filename `embedded_ca_files` can produce — the hive CA or
/// a `peer-ca-<N>.pem`. Lets `sync_agents` find stale CA files to clean
/// up (a CA dropped from config) without touching unrelated meta files.
fn is_embedded_ca_name(name: &str) -> bool {
    name == HIVE_CA_FILE || (name.starts_with("peer-ca-") && has_pem_ext(name))
}

/// True when `name` ends in a `.pem` extension (case-insensitive). Split
/// out so the embedded-CA filename checks share one spelling and dodge
/// clippy's case-sensitive-extension lint.
fn has_pem_ext(name: &str) -> bool {
    Path::new(name)
        .extension()
        .is_some_and(|ext| ext.eq_ignore_ascii_case("pem"))
}

/// Embedded-CA state for the meta repo: `(desired_files, changed)`.
/// `desired_files` is `(filename, contents)` for every CA that should sit
/// next to flake.nix (the hive CA + each peer CA). `changed` is true when
/// the on-disk set differs in any way — a file's contents changed, a new
/// CA appeared, or a previously-embedded CA (`hive-ca.pem` /
/// `peer-ca-*.pem`) is no longer wanted (stale, to be removed). Drives
/// both the re-commit decision and the materialise/cleanup in
/// `sync_agents`, so a CA rotation or a peer-set change re-commits even
/// when the flake itself is byte-identical.
fn ca_embed_state(dir: &std::path::Path) -> (Vec<(String, String)>, bool) {
    let desired: Vec<(String, String)> = embedded_ca_files()
        .into_iter()
        .filter_map(|(name, path)| std::fs::read_to_string(&path).ok().map(|c| (name, c)))
        .collect();
    let desired_names: std::collections::HashSet<&str> =
        desired.iter().map(|(n, _)| n.as_str()).collect();

    let mut changed = desired.iter().any(|(name, content)| {
        std::fs::read_to_string(dir.join(name)).unwrap_or_default() != *content
    });

    // A previously-embedded CA file no longer wanted → stale (removal is
    // a change even when every desired file already matches on disk).
    if !changed && let Ok(entries) = std::fs::read_dir(dir) {
        changed = entries.flatten().any(|e| {
            e.file_name()
                .to_str()
                .is_some_and(|name| is_embedded_ca_name(name) && !desired_names.contains(name))
        });
    }
    (desired, changed)
}

/// Read an agent's applied `flake.lock` and return the subset of
/// `CANONICAL_INPUTS` it declares as direct (root-level) inputs.
/// Returns an empty vec when the lock is missing or unparsable —
/// safe degradation, the worst case is no dedup for that agent.
fn agent_canonical_inputs(name: &str) -> Vec<&'static str> {
    let path = crate::paths::applied_dir(name).join("flake.lock");
    let Ok(raw) = std::fs::read_to_string(&path) else {
        return Vec::new();
    };
    let Ok(json) = serde_json::from_str::<serde_json::Value>(&raw) else {
        return Vec::new();
    };
    let Some(nodes) = json.get("nodes").and_then(|v| v.as_object()) else {
        return Vec::new();
    };
    let Some(root_name) = json.get("root").and_then(|v| v.as_str()) else {
        return Vec::new();
    };
    let Some(root_inputs) = nodes
        .get(root_name)
        .and_then(|n| n.get("inputs"))
        .and_then(|v| v.as_object())
    else {
        return Vec::new();
    };
    CANONICAL_INPUTS
        .iter()
        .copied()
        .filter(|canon| root_inputs.contains_key(*canon))
        .collect()
}

/// Inner render helper accepting a lookup fn so tests can stub the
/// agent flake-lock introspection.
#[allow(
    clippy::too_many_lines,
    clippy::too_many_arguments,
    reason = "templated string-builder for the meta flake — the length is one \
              contiguous fmt block, splitting it would just hide the shape; the \
              args mirror the host-level config the flake is rendered from"
)]
fn render_flake_with_lookup<F>(
    hyperhive_flake: &str,
    docs_flake: &str,
    nixpkgs_flake: &str,
    dashboard_port: u16,
    operator_pronouns: &str,
    context_window_tokens: &std::collections::HashMap<String, u64>,
    agents: &[AgentSpec],
    lookup: F,
) -> String
where
    F: Fn(&str) -> Vec<&'static str>,
{
    use std::fmt::Write as _;
    let mut out = String::new();
    out.push_str("{\n  description = \"hyperhive deployed agents\";\n  inputs = {\n");
    // `nixpkgs` is a top-level meta input with an explicit store-path URL.
    // `hyperhive` then follows it via `hyperhive.inputs.nixpkgs.follows`.
    // This cascades through to every agent because
    // `agent-<n>.inputs.nixpkgs.follows = "nixpkgs"` resolves to the same
    // top-level node.
    //
    // Why an explicit `path:` URL instead of
    // `nixpkgs.follows = "hyperhive/nixpkgs"`:
    // meta points to hyperhive's *store path* as its flake input, so nix
    // reads hyperhive's own pinned lock when evaluating that input — the
    // host-level `follows` the operator set never propagates. Injecting the
    // evaluated `pkgs.path` directly at nix-module evaluation time is the
    // only reliable way to honour the host's channel choice.
    //
    // Fallback (flake arg empty): legacy `follows` wiring — used when
    // hive-c0re is not built with this option wired up.
    if nixpkgs_flake.is_empty() {
        // Legacy path: meta defers to hyperhive's own lock.
        let _ = writeln!(out, "    hyperhive.url = \"{hyperhive_flake}\";");
        out.push_str("    nixpkgs.follows = \"hyperhive/nixpkgs\";\n");
    } else {
        let _ = writeln!(out, "    nixpkgs.url = \"{nixpkgs_flake}\";");
        let _ = writeln!(out, "    hyperhive.url = \"{hyperhive_flake}\";");
        out.push_str("    hyperhive.inputs.nixpkgs.follows = \"nixpkgs\";\n");
    }
    // Narrow `docs/` source as its own input so a doc edit only
    // re-locks THIS input instead of re-hashing the whole `hyperhive`
    // source. Threaded to each agent below as `hyperhive.docs.source`.
    // Empty = hive-c0re not built with the option wired up (legacy);
    // agents then keep the harness default (`hyperhive.packages.reference-docs`).
    if !docs_flake.is_empty() {
        // `flake = false`: the docs/ tree is a plain source (no flake.nix),
        // so nix must treat it as raw source, not evaluate it as a flake.
        let _ = writeln!(out, "    hyperhive-docs.url = \"{docs_flake}\";");
        out.push_str("    hyperhive-docs.flake = false;\n");
    }
    // Each agent's config input is its LOCAL applied mirror
    // (`git+file://<applied>`), so the meta flake resolves entirely from
    // on-disk state and boot never depends on the forge being reachable.
    // The forge `agent-configs/<name>` repos stay the review/audit surface
    // (config PRs land there) but are not the flake's build input. deploy +
    // rebuild re-lock this input to `applied/<name>`'s current `main` head;
    // `verify_commit` overrides it to a proposed `?rev=<sha>` for eval
    // before that head moves.
    for spec in agents {
        let _ = writeln!(
            out,
            "    agent-{name}.url = \"git+file://{applied}\";",
            name = spec.name,
            applied = crate::paths::applied_dir(&spec.name).display(),
        );
        // For each canonical input the agent declares in its own
        // `flake.nix` (detected by reading its applied `flake.lock`),
        // emit `inputs.agent-<name>.inputs.<canon>.follows = "<canon>"`.
        // Collapses otherwise-separate-but-identical nixpkgs nodes
        // (root + every agent's own nixpkgs) into one. Skipped
        // silently for agents that don't declare the input — emitting
        // follows on a non-existent input would error at
        // `nix flake lock` time.
        for canon in lookup(&spec.name) {
            let _ = writeln!(
                out,
                "    agent-{}.inputs.{canon}.follows = \"{canon}\";",
                spec.name,
            );
        }
    }
    out.push_str("  };\n  outputs =\n    { self, hyperhive, ... }@inputs:\n    let\n");
    // Free-text operator string — escape backslash + double-quote so a
    // pronouns value like `he/him \ "rare"` round-trips into a valid
    // nix string literal without breaking the flake.
    let pronouns_escaped = operator_pronouns.replace('\\', "\\\\").replace('"', "\\\"");
    let _ = writeln!(
        out,
        "      dashboardPort = {dashboard_port};\n      operatorPronouns = \"{pronouns_escaped}\";\n      mkAgent = {{ name, isManager, port, parent ? null, toolGroups ? null, capabilities ? null }}:"
    );
    out.push_str(
        r#"        let
          base = if isManager
                 then hyperhive.nixosConfigurations.ruth
                 else hyperhive.nixosConfigurations.agent-base;
          input = inputs."agent-${name}";
          service = "hive-agent";
          parentEnv = if parent == null then {} else { HIVE_PARENT = parent; };
          toolGroupsEnv = if toolGroups == null then {} else { HIVE_TOOL_GROUPS = toolGroups; };
          capabilitiesEnv = if capabilities == null then {} else { HIVE_CAPABILITIES = capabilities; };
        in
        base.extendModules {
          modules = [
            input.nixosModules.default
            {
"#,
    );
    // Point the in-container docs dir (`$HIVE_DOCS_DIR`) at the narrow
    // `hyperhive-docs` input instead of the harness default
    // (`hyperhive.packages.reference-docs`, built from the now-docs-stripped source).
    // `inputs."hyperhive-docs"` is reachable via the outputs `@inputs`
    // capture. Emitted only when the input exists (docs_flake non-empty).
    if !docs_flake.is_empty() {
        out.push_str("              hyperhive.docs.source = inputs.\"hyperhive-docs\".outPath;\n");
    }
    // CA trust: embed every hive-trusted CA so each agent validates them at
    // build time. The list is the hive's own self-signed CA (when active)
    // plus every peer-hive root CA (`swarm.peers.<d>.caCert`) — a peer CA is
    // trusted everywhere the hive's own internal CA is. `certificateFiles` is
    // build-time, so the certs travel with the flake source: `sync_agents`
    // writes `./hive-ca.pem` + `./peer-ca-<N>.pem` next to flake.nix and
    // stages them. Only public CA certs are embedded; no private key ever
    // leaves the host. The filename list matches `sync_agents` exactly (both
    // derive it from `embedded_ca_files`), so we never reference a file we
    // didn't embed; emitted only when the list is non-empty.
    let ca_refs: Vec<String> = embedded_ca_files()
        .into_iter()
        .map(|(name, _)| format!("./{name}"))
        .collect();
    if !ca_refs.is_empty() {
        let _ = writeln!(
            out,
            "              security.pki.certificateFiles = [ {} ];",
            ca_refs.join(" ")
        );
    }
    // Hive-wide OTEL stats export (`services.hyperhive.otel.*`): inject the
    // build-time `hyperhive.otel.*` config the harness modules consume (its
    // otelEnv + otelExecStart wrapper + LoadCredential). Host-driven, so
    // the same config lands on every agent; emitted only when enabled.
    // Mirrors the CA-cert injection above — host state -> build-time agent
    // module config.
    if let Some(otel) = otel_config() {
        let esc = |s: &str| s.replace('\\', "\\\\").replace('"', "\\\"");
        out.push_str("              hyperhive.otel.enable = true;\n");
        let _ = writeln!(
            out,
            "              hyperhive.otel.endpoint = \"{}\";",
            esc(&otel.endpoint)
        );
        let _ = writeln!(
            out,
            "              hyperhive.otel.protocol = \"{}\";",
            esc(&otel.protocol)
        );
        if let Some(attrs) = &otel.extra_resource_attributes {
            let _ = writeln!(
                out,
                "              hyperhive.otel.extraResourceAttributes = \"{}\";",
                esc(attrs)
            );
        }
        if let Some(cred) = &otel.headers_credential {
            let _ = writeln!(
                out,
                "              hyperhive.otel.headersCredential = \"{}\";",
                esc(cred)
            );
        }
        if let Some(ms) = otel.metric_interval_ms {
            // Int option — emit a bare numeric literal (no quotes). `ms` is a
            // parsed u64, so it can't inject anything into the rendered nix.
            let _ = writeln!(out, "              hyperhive.otel.metricIntervalMs = {ms};");
        }
        if otel.debug {
            out.push_str("              hyperhive.otel.debug = true;\n");
        }
    }
    // GitHub integration is on by default in every agent
    // (`hyperhive.github.enable`); the host turns it off hive-wide via
    // `services.hyperhive.github.enable = false`, surfaced here as the
    // `HYPERHIVE_GITHUB_DISABLED` env on hive-c0re's unit. Only the OFF
    // override is propagated — the enabled default needs no per-agent line.
    if std::env::var_os("HYPERHIVE_GITHUB_DISABLED").is_some() {
        out.push_str("              hyperhive.github.enable = false;\n");
    }
    out.push_str(
        r#"              # The harness service inside the container runs as a
              # non-root unix user named after the agent (`damocles`,
              # `iris`, `root`, …). UID auto-assigned by NixOS; the
              # per-agent override here is what makes
              # `hyperhive.user.name` match the agent's identity
              # instead of the harness default of `"agent"`.
              hyperhive.user.name = name;
              programs.git.config.user = {
                name = name;
                email = "${name}@hyperhive.local";
              };
              # Container-wide env: every service + co-process daemon can
              # resolve the agent's durable state dir without hard-coding it.
              # `environment.variables` only writes /etc/environment (login
              # shells); `systemd.globalEnvironment` is the analogue for
              # systemd units so tea-login / forge-avatar-sync /
              # hive-matrix-daemon etc. can read `$HYPERHIVE_STATE_DIR`
              # without each service having to redeclare it.
              environment.variables = {
                HIVE_LABEL = name;
                HYPERHIVE_STATE_DIR = "/agents/${name}/state";
                HYPERHIVE_HARNESS_DIR = "/agents/${name}/harness";
              };
              systemd.globalEnvironment = {
                HIVE_LABEL = name;
                HYPERHIVE_STATE_DIR = "/agents/${name}/state";
                HYPERHIVE_HARNESS_DIR = "/agents/${name}/harness";
"#,
    );
    // Forwarded vars (HIVE_FORGE_URL etc.) also go into globalEnvironment,
    // not just the harness service env below, so EVERY service + shell in
    // the container inherits them — crucially the bash-task runner (where
    // `hive-forge` + `git` actually run), plus the matrix daemon, tea-login
    // and interactive shells. Scoped to the harness service alone they were
    // invisible to bash tasks: harmless in shared netns (the localhost
    // default works) but broken under isolation, where the in-cluster
    // `forge.<domain>` URL is the only reachable path.
    for (var, val) in forwarded_env_vars() {
        let escaped = val.replace('\\', "\\\\").replace('"', "\\\"");
        let _ = writeln!(out, "                {var} = \"{escaped}\";");
    }
    out.push_str(
        r"              };
              systemd.services.${service}.environment = parentEnv // toolGroupsEnv // capabilitiesEnv // {
                HIVE_PORT = toString port;
                HIVE_LABEL = name;
                HIVE_DASHBOARD_PORT = toString dashboardPort;
                HIVE_OPERATOR_PRONOUNS = operatorPronouns;",
    );
    // Per-model context-window env vars declared in the host-level
    // `services.hive-c0re.contextWindowTokens` option. Use a sorted
    // iterator for deterministic flake output (no spurious git diffs).
    let mut sorted_tokens: Vec<(&String, &u64)> = context_window_tokens.iter().collect();
    sorted_tokens.sort_by_key(|(k, _)| k.as_str());
    for (key, val) in &sorted_tokens {
        let upper_key = key.to_ascii_uppercase();
        let _ = writeln!(
            out,
            "                HIVE_CONTEXT_WINDOW_TOKENS_{upper_key} = \"{val}\";"
        );
    }
    // Forwarded env vars — picked up from hive-c0re's own systemd unit
    // env (`services.hyperhive.*` options flow through nix/modules/
    // hive-c0re.nix into the host process). We copy whatever's set into
    // each sub-agent's harness service env so the in-container surfaces
    // (`identity.rs`, `forge_notify`) see a consistent view across the
    // whole hive. Absent host-side env (option not set) → skip emission
    // → in-container accessors fall back to None / defaults gracefully.
    //
    // - HIVE_FORGE_URL: agents poll this for Forgejo notifications.
    // - HYPERHIVE_HIVE_DOMAIN: machine-readable hive DNS.
    // - HYPERHIVE_HIVE_NAME / HYPERHIVE_SWARM_NAME: human display
    //   names for hive + swarm.
    for (var, val) in forwarded_env_vars() {
        let escaped = val.replace('\\', "\\\\").replace('"', "\\\"");
        let _ = writeln!(out, "                {var} = \"{escaped}\";");
    }
    out.push_str(
        r#"                HYPERHIVE_STATE_DIR = "/agents/${name}/state";
                HYPERHIVE_HARNESS_DIR = "/agents/${name}/harness";
              };
            }
          ];
        };
    in
    {
      nixosConfigurations = {
"#,
    );
    // Pull the topology map once and look up each agent's parent. An
    // empty / absent topology.json yields `parent = null` for everyone
    // (every container at root). `meta::sync_agents` seeds the file
    // on first run with manager as root + everyone else under manager.
    let topology = crate::topology::read();
    let tool_groups_map = crate::tool_groups::read();
    let capabilities_map = crate::capabilities::read();
    for spec in agents {
        let parent_attr = topology
            .get(&spec.name)
            .and_then(|p| p.as_ref())
            .map_or_else(|| "null".to_owned(), |p| format!("\"{p}\""));
        // Emit `toolGroups = "group1,group2"` when the operator has
        // explicitly configured groups for this agent. Absent entry = null
        // = harness falls back to AGENT_DEFAULT (no env var emitted,
        // no rebuild cascade for agents whose groups haven't changed).
        let groups = tool_groups_map.get(&spec.name).cloned().unwrap_or_default();
        let tool_groups_attr = if groups.is_empty() {
            "null".to_owned()
        } else {
            let joined = groups.join(",");
            format!("\"{joined}\"")
        };
        // Emit `capabilities = "cap1,cap2"` when the operator has
        // granted capabilities to this agent. Absent entry = null = no
        // capability env var injected, capability-gated tools hidden.
        let caps = capabilities_map
            .get(&spec.name)
            .cloned()
            .unwrap_or_default();
        let capabilities_attr = if caps.is_empty() {
            "null".to_owned()
        } else {
            let joined = caps.join(",");
            format!("\"{joined}\"")
        };
        let _ = writeln!(
            out,
            "        {} = mkAgent {{ name = \"{}\"; isManager = {}; port = {}; parent = {}; toolGroups = {}; capabilities = {}; }};",
            spec.name,
            spec.name,
            if spec.is_manager { "true" } else { "false" },
            spec.port,
            parent_attr,
            tool_groups_attr,
            capabilities_attr,
        );
    }
    out.push_str("      };\n    };\n}\n");
    out
}

/// Return the list of file names that are currently staged (index differs
/// from HEAD). On the initial commit (`HEAD` doesn't exist yet) falls back
/// to `git diff --cached --name-only HEAD` failing gracefully by using
/// `git status --porcelain` and collecting the `A ` / `M ` prefix lines.
async fn git_staged_names(dir: &Path) -> Result<Vec<String>> {
    // `--diff-filter=ACM` skips deleted entries — we only care about
    // additions and modifications for message-building purposes.
    let out = lifecycle::git_command()
        .current_dir(dir)
        .args(["diff", "--cached", "--name-only", "--diff-filter=ACM"])
        .output()
        .await
        .with_context(|| format!("git diff --cached in {}", dir.display()))?;
    if out.status.success() {
        let names = String::from_utf8_lossy(&out.stdout)
            .lines()
            .map(str::trim)
            .filter(|l| !l.is_empty())
            .map(ToOwned::to_owned)
            .collect();
        return Ok(names);
    }
    // Fallback for initial commit (no HEAD yet): parse `git status --porcelain`.
    let st = lifecycle::git_command()
        .current_dir(dir)
        .args(["status", "--porcelain"])
        .output()
        .await
        .with_context(|| format!("git status in {}", dir.display()))?;
    let names = String::from_utf8_lossy(&st.stdout)
        .lines()
        .filter(|l| l.starts_with("A ") || l.starts_with("M "))
        .filter_map(|l| l.get(3..))
        .map(ToOwned::to_owned)
        .collect();
    Ok(names)
}

async fn git(dir: &Path, args: &[&str]) -> Result<()> {
    let out = lifecycle::git_command()
        .current_dir(dir)
        .args(args)
        .output()
        .await
        .with_context(|| format!("git {} in {}", args.join(" "), dir.display()))?;
    if !out.status.success() {
        bail!(
            "git {} failed ({}): {}",
            args.join(" "),
            out.status,
            String::from_utf8_lossy(&out.stderr).trim()
        );
    }
    Ok(())
}

async fn git_commit(dir: &Path, message: &str) -> Result<()> {
    git(
        dir,
        &[
            "-c",
            &format!("user.name={GIT_NAME}"),
            "-c",
            &format!("user.email={GIT_EMAIL}"),
            "commit",
            "-m",
            message,
        ],
    )
    .await?;
    // Best-effort mirror to the bundled forge. No-op when the forge
    // isn't seeded (no core token on disk); push failures log a warn
    // but don't bubble up — a missing mirror shouldn't fail an
    // otherwise successful deploy.
    if let Err(e) = crate::forge::push_meta(dir).await {
        tracing::warn!(error = ?e, "forge: meta push after commit failed (non-fatal)");
    }
    Ok(())
}

/// Path-limited commit: commits ONLY the given paths, so unrelated
/// staged content — above all a `prepare_deploy`-staged `flake.lock`
/// — can never be swept into someone else's commit. Every targeted
/// meta commit (perm files, topology, lock bumps) goes through this;
/// only `sync_agents` uses the bare [`git_commit`], because its
/// staged set *is* its intentional commit set.
async fn git_commit_paths(dir: &Path, message: &str, paths: &[&str]) -> Result<()> {
    let name = format!("user.name={GIT_NAME}");
    let email = format!("user.email={GIT_EMAIL}");
    let mut args = vec!["-c", &name, "-c", &email, "commit", "-m", message, "--"];
    args.extend_from_slice(paths);
    git(dir, &args).await?;
    if let Err(e) = crate::forge::push_meta(dir).await {
        tracing::warn!(error = ?e, "forge: meta push after commit failed (non-fatal)");
    }
    Ok(())
}

/// True when any of `paths` differs between HEAD and the index or
/// working tree — the path-scoped replacement for whole-tree
/// `git_is_clean` / `has_staged_changes` guards, which a concurrently
/// staged deploy lock would otherwise trip.
async fn paths_dirty(dir: &Path, paths: &[&str]) -> Result<bool> {
    let mut args = vec!["diff", "--quiet", "HEAD", "--"];
    args.extend_from_slice(paths);
    let out = lifecycle::git_command()
        .current_dir(dir)
        .args(&args)
        .output()
        .await
        .with_context(|| format!("git diff --quiet in {}", dir.display()))?;
    // Exit 0 = no differences; 1 = differences; anything else (e.g.
    // no HEAD yet on a fresh repo) → treat as dirty so the commit
    // path runs and surfaces real errors loudly.
    Ok(!out.status.success())
}

/// Full argv for a `nix` invocation, prefixed with the flakes-enabling
/// experimental-features flag. `--extra-experimental-features` is
/// belt-and-suspenders for hosts that haven't set this in nix.conf: the
/// hyperhive module's deploy guide assumes flakes are already enabled,
/// but the cost of being defensive is one extra argv each call.
fn nix_argv<'a>(args: &[&'a str]) -> Vec<&'a str> {
    let mut all = vec!["--extra-experimental-features", "nix-command flakes"];
    all.extend_from_slice(args);
    all
}

/// Run `nix <args>` in `dir` (flakes enabled), capturing combined output.
/// Shared core of [`nix`] and [`nix_logged`].
async fn nix_output(dir: &Path, args: &[&str]) -> Result<std::process::Output> {
    Command::new("nix")
        .current_dir(dir)
        .args(nix_argv(args))
        // Kill the nix child if the caller's future is dropped (e.g. the
        // startup-migration timeout around `sync_agents` fires) rather than
        // orphaning it against an unreachable forge. No-op on normal exit.
        .kill_on_drop(true)
        .output()
        .await
        .with_context(|| format!("nix {} in {}", args.join(" "), dir.display()))
}

/// Turn a finished nix [`Output`](std::process::Output) into `Result<()>`,
/// bailing with the trimmed stderr tail on non-zero exit.
fn nix_check(args: &[&str], out: &std::process::Output) -> Result<()> {
    if !out.status.success() {
        bail!(
            "nix {} failed ({}): {}",
            args.join(" "),
            out.status,
            String::from_utf8_lossy(&out.stderr).trim()
        );
    }
    Ok(())
}

async fn nix(dir: &Path, args: &[&str]) -> Result<()> {
    let out = nix_output(dir, args).await?;
    nix_check(args, &out)
}

/// Like [`nix`] but records the invocation (cmdline + stdout + stderr +
/// terminal status) into a `build_logs.sqlite` row so a config-approval
/// eval/deploy step shows up on the dashboard. This is what makes a
/// failing eval-verify visible: it runs before any container build, so
/// without a row a rejected config approval leaves the operator with
/// zero build logs to look at. Best-effort logging: a missing global
/// handle or a failed `start()` just skips the row — the command still
/// runs and its exit status is still enforced.
async fn nix_logged(dir: &Path, args: &[&str], agent: &str, kind: &str) -> Result<()> {
    let cmdline = format!("nix {}", nix_argv(args).join(" "));
    let logs = crate::build_logs::global();
    let log_id = logs.as_ref().and_then(|h| {
        h.start(agent, kind, &cmdline)
            .map_err(|e| {
                tracing::warn!(error = ?e, %kind, "build_logs: start failed (meta log dropped)");
            })
            .ok()
    });

    let out = nix_output(dir, args).await?;

    if let (Some(h), Some(id)) = (&logs, log_id) {
        for line in String::from_utf8_lossy(&out.stdout).lines() {
            h.append_stdout(id, line);
        }
        for line in String::from_utf8_lossy(&out.stderr).lines() {
            h.append_stderr(id, line);
        }
        h.finish(
            id,
            if out.status.success() {
                crate::build_logs::BuildStatus::Ok
            } else {
                crate::build_logs::BuildStatus::Fail
            },
        );
    }

    nix_check(args, &out)
}

#[cfg(test)]
mod tests {
    use super::*;

    /// The regression the deploy-window bug review surfaced: a
    /// path-limited commit must leave an unrelated staged file (the
    /// prepare_deploy-staged `flake.lock`) untouched, so a later
    /// `abort_deploy` still has something to restore.
    #[tokio::test]
    async fn path_limited_commit_leaves_unrelated_staged_file_alone() {
        let tmp = tempfile::tempdir().expect("tempdir");
        let dir = tmp.path();
        git(dir, &["init", "--initial-branch=main"])
            .await
            .expect("git init");
        std::fs::write(dir.join("tool-groups.json"), "{}").expect("write");
        std::fs::write(dir.join("flake.lock"), "v1").expect("write");
        git(dir, &["add", "-A"]).await.expect("add");
        git_commit(dir, "seed").await.expect("seed commit");
        // A deploy stages a new lock (uncommitted)…
        std::fs::write(dir.join("flake.lock"), "v2-staged-by-deploy").expect("write");
        git(dir, &["add", "flake.lock"]).await.expect("stage lock");
        // …and a perm change commits, path-limited.
        std::fs::write(dir.join("tool-groups.json"), r#"{"alice":[]}"#).expect("write");
        git(dir, &["add", "tool-groups.json"]).await.expect("add");
        git_commit_paths(dir, "set tool-groups for alice", &["tool-groups.json"])
            .await
            .expect("path-limited commit");
        // The perm file is committed; the deploy's staged lock is not.
        assert!(
            !paths_dirty(dir, &["tool-groups.json"])
                .await
                .expect("check"),
            "perm file must be committed"
        );
        assert!(
            paths_dirty(dir, &["flake.lock"]).await.expect("check"),
            "staged deploy lock must survive the perm commit"
        );
    }

    fn sample_spec(name: &str, is_manager: bool, port: u16) -> AgentSpec {
        AgentSpec {
            name: name.to_owned(),
            is_manager,
            port,
        }
    }

    #[test]
    fn agent_input_override_pins_exact_rev() {
        let p = Path::new("/var/lib/hyperhive/agents/iris/applied");
        assert_eq!(
            agent_input_override(p, "abc123def"),
            "git+file:///var/lib/hyperhive/agents/iris/applied?rev=abc123def"
        );
    }

    #[test]
    fn applied_override_url_targets_local_mirror_main_head() {
        // Deploy + rebuild re-lock against the local applied mirror's `main`
        // head (no `?rev`), never the persistent forge URL — so a rebuild
        // survives forge unreachability and builds the fast-forwarded config.
        let p = Path::new("/var/lib/hyperhive/agents/iris/applied");
        let url = applied_override_url(p);
        assert_eq!(url, "git+file:///var/lib/hyperhive/agents/iris/applied");
        assert!(
            !url.contains("?rev="),
            "must lock main head, not a pinned rev: {url}"
        );
    }

    #[test]
    fn render_flake_uses_explicit_nixpkgs_url_when_provided() {
        let out = render_flake(
            "github:example/hyperhive",
            "path:/nix/store/bbbb-hyperhive-docs-source",
            "path:/nix/store/aaaa-nixpkgs-source",
            8000,
            "she/her",
            &std::collections::HashMap::new(),
            &[sample_spec("alice", false, 9001)],
        );
        // nixpkgs is a top-level input with an explicit URL; hyperhive
        // follows it.
        assert!(
            out.contains("nixpkgs.url = \"path:/nix/store/aaaa-nixpkgs-source\""),
            "expected explicit nixpkgs.url:\n{out}"
        );
        assert!(
            out.contains("hyperhive.inputs.nixpkgs.follows = \"nixpkgs\""),
            "expected hyperhive.inputs.nixpkgs.follows:\n{out}"
        );
        assert!(
            !out.contains("nixpkgs.follows = \"hyperhive"),
            "old-style follows must not appear when flake args are set:\n{out}"
        );
        // the narrow docs source is its own non-flake input, and each
        // agent's docs dir resolves from it rather than hyperhive.packages.reference-docs.
        assert!(
            out.contains("hyperhive-docs.url = \"path:/nix/store/bbbb-hyperhive-docs-source\""),
            "expected hyperhive-docs input url:\n{out}"
        );
        assert!(
            out.contains("hyperhive-docs.flake = false;"),
            "docs source is not a flake, must be flake = false:\n{out}"
        );
        assert!(
            out.contains("hyperhive.docs.source = inputs.\"hyperhive-docs\".outPath;"),
            "expected per-agent docs source wired to the input:\n{out}"
        );
    }

    #[test]
    fn render_flake_omits_docs_input_when_docs_flake_empty() {
        // Legacy / not-wired-up: empty docs_flake emits no docs input and
        // leaves each agent on the harness default (hyperhive.packages.reference-docs).
        let out = render_flake(
            "github:example/hyperhive",
            "",
            "path:/nix/store/aaaa-nixpkgs-source",
            8000,
            "she/her",
            &std::collections::HashMap::new(),
            &[sample_spec("alice", false, 9001)],
        );
        assert!(
            !out.contains("hyperhive-docs"),
            "no docs input/source when docs_flake is empty:\n{out}"
        );
    }

    #[test]
    fn render_flake_falls_back_to_follows_when_nixpkgs_flake_empty() {
        // Empty nixpkgs_flake → legacy follows behaviour (backward compat
        // for any code path that can't inject pkgs.path).
        let out = render_flake(
            "github:example/hyperhive",
            "path:/nix/store/bbbb-hyperhive-docs-source",
            "",
            8000,
            "she/her",
            &std::collections::HashMap::new(),
            &[sample_spec("alice", false, 9001)],
        );
        assert!(
            out.contains("nixpkgs.follows = \"hyperhive/nixpkgs\""),
            "expected fallback follows:\n{out}"
        );
        assert!(
            !out.contains("nixpkgs.url ="),
            "no explicit url should be emitted in fallback mode:\n{out}"
        );
    }

    #[test]
    fn render_flake_emits_follows_for_agents_declaring_nixpkgs() {
        // Stub lookup: pretend `bitburner` + `dmatrix` declare `nixpkgs`
        // at their root, while `argus` has no canonical inputs at all.
        let lookup = |name: &str| -> Vec<&'static str> {
            match name {
                "bitburner" | "dmatrix" => vec!["nixpkgs"],
                _ => vec![],
            }
        };
        let out = render_flake_with_lookup(
            "github:example/hyperhive",
            "path:/nix/store/bbbb-hyperhive-docs-source",
            "path:/nix/store/aaaa-nixpkgs-source",
            8000,
            "she/her",
            &std::collections::HashMap::new(),
            &[
                sample_spec("argus", false, 9001),
                sample_spec("bitburner", false, 9002),
                sample_spec("dmatrix", false, 9003),
            ],
            lookup,
        );
        // bitburner declares nixpkgs → follows emitted.
        assert!(
            out.contains("agent-bitburner.inputs.nixpkgs.follows = \"nixpkgs\""),
            "missing bitburner nixpkgs follows:\n{out}"
        );
        // dmatrix declares nixpkgs → follows emitted.
        assert!(out.contains("agent-dmatrix.inputs.nixpkgs.follows = \"nixpkgs\""));
        // argus declares neither → no follows emitted for it. Asserting
        // ABSENCE is the important bit: emitting a follows on a
        // non-existent input errors at `nix flake lock` time.
        assert!(
            !out.contains("agent-argus.inputs.nixpkgs"),
            "argus shouldn't have nixpkgs follows:\n{out}"
        );
    }

    #[test]
    fn render_flake_skips_canonical_follows_when_lookup_returns_empty() {
        let out = render_flake_with_lookup(
            "github:example/hyperhive",
            "path:/nix/store/bbbb-hyperhive-docs-source",
            "path:/nix/store/aaaa-nixpkgs-source",
            8000,
            "she/her",
            &std::collections::HashMap::new(),
            &[sample_spec("alice", false, 9001)],
            |_| Vec::new(),
        );
        // No agent-side follows when the lookup reports nothing
        // declared — protects agents whose flake.lock can't be read
        // (missing / unparsable) from being broken by a follows on a
        // non-existent input.
        assert!(
            !out.contains("agent-alice.inputs."),
            "alice shouldn't have any inputs follows:\n{out}"
        );
    }

    #[test]
    fn render_flake_forwards_env_into_global_environment() {
        // Regression for the isolation breakage where forwarded vars
        // (HIVE_FORGE_URL etc.) landed only on the harness service env,
        // so the bash-task runner / matrix daemon / shells defaulted to
        // localhost and couldn't reach the in-cluster gateway. They must
        // also appear in `systemd.globalEnvironment`, which every unit +
        // shell in the container inherits.
        //
        // SAFETY: single-threaded mutation of a process env var the other
        // tests don't assert the absence of; restored before returning.
        unsafe {
            std::env::set_var("HIVE_FORGE_URL", "http://forge.example.test");
        }
        let out = render_flake(
            "github:example/hyperhive",
            "path:/nix/store/bbbb-hyperhive-docs-source",
            "path:/nix/store/aaaa-nixpkgs-source",
            8000,
            "she/her",
            &std::collections::HashMap::new(),
            &[sample_spec("alice", false, 9001)],
        );
        unsafe {
            std::env::remove_var("HIVE_FORGE_URL");
        }
        // The var must be emitted inside the globalEnvironment block, i.e.
        // before the per-service harness env block that follows it.
        let global_at = out
            .find("systemd.globalEnvironment = {")
            .expect("globalEnvironment block must exist");
        let service_at = out
            .find("systemd.services.${service}.environment")
            .expect("harness service env block must exist");
        let forge_at = out
            .find("HIVE_FORGE_URL = \"http://forge.example.test\"")
            .expect("HIVE_FORGE_URL must be forwarded into the flake");
        assert!(
            forge_at > global_at && forge_at < service_at,
            "HIVE_FORGE_URL must land inside systemd.globalEnvironment, \
             not only the harness service env:\n{out}"
        );
    }

    #[test]
    fn render_flake_agent_input_points_at_local_applied_mirror() {
        // The agent config input references the LOCAL applied mirror
        // (`git+file://<applied>`), NOT the forge — so the meta flake
        // resolves entirely from on-disk state and boot never depends on the
        // forge being reachable. The forge `agent-configs/<n>` repos stay the
        // review surface, but they are not the flake's build input.
        let out = render_flake(
            "github:example/hyperhive",
            "path:/nix/store/bbbb-hyperhive-docs-source",
            "path:/nix/store/aaaa-nixpkgs-source",
            8000,
            "she/her",
            &std::collections::HashMap::new(),
            &[sample_spec("alice", false, 9001)],
        );
        let want = format!(
            "agent-alice.url = \"git+file://{}\"",
            crate::paths::applied_dir("alice").display()
        );
        assert!(
            out.contains(&want),
            "expected the agent input to point at the local applied mirror ({want}):\n{out}"
        );
        assert!(
            !out.contains("git+http"),
            "no forge git+http URL must remain in the rendered flake:\n{out}"
        );
    }

    #[test]
    fn render_flake_embeds_hive_ca_when_signalled() {
        // When hive-tls.nix signals a self-signed hive CA via
        // HIVE_TLS_CA_PATH (and the cert exists), the agent module must
        // trust it at build time via security.pki.certificateFiles. Absent
        // the signal, no reference is emitted (so the flake doesn't point at
        // a file that was never embedded).
        //
        // SAFETY: single-threaded mutation of a process env var the other
        // tests don't assert the absence of; restored before returning.
        let ca_file = std::env::temp_dir().join(format!("hive-ca-test-{}.pem", std::process::id()));
        std::fs::write(
            &ca_file,
            "-----BEGIN CERTIFICATE-----\ntest\n-----END CERTIFICATE-----\n",
        )
        .expect("write temp CA");

        let render = || {
            render_flake(
                "github:example/hyperhive",
                "path:/nix/store/bbbb-hyperhive-docs-source",
                "path:/nix/store/aaaa-nixpkgs-source",
                8000,
                "she/her",
                &std::collections::HashMap::new(),
                &[sample_spec("alice", false, 9001)],
            )
        };

        // Two peer-hive CA temp files for the list cases.
        let peer0 = std::env::temp_dir().join(format!("peer-ca0-test-{}.pem", std::process::id()));
        let peer1 = std::env::temp_dir().join(format!("peer-ca1-test-{}.pem", std::process::id()));
        std::fs::write(
            &peer0,
            "-----BEGIN CERTIFICATE-----\np0\n-----END CERTIFICATE-----\n",
        )
        .expect("write peer CA 0");
        std::fs::write(
            &peer1,
            "-----BEGIN CERTIFICATE-----\np1\n-----END CERTIFICATE-----\n",
        )
        .expect("write peer CA 1");
        let peer_paths = format!("{}:{}", peer0.display(), peer1.display());

        // All env mutations are serialised within this one test (no other
        // test asserts on these vars), restored before returning.
        unsafe {
            std::env::remove_var("HIVE_PEER_CA_PATHS");
            std::env::set_var("HIVE_TLS_CA_PATH", &ca_file);
        }
        let with_ca = render();
        // Hive CA + peer CAs: the list carries all three, hive CA first.
        unsafe {
            std::env::set_var("HIVE_PEER_CA_PATHS", &peer_paths);
        }
        let with_peers = render();
        // Peers only (this hive on ACME, federating with self-signed peers).
        unsafe {
            std::env::remove_var("HIVE_TLS_CA_PATH");
        }
        let peers_only = render();
        unsafe {
            std::env::remove_var("HIVE_PEER_CA_PATHS");
        }
        let without_ca = render();
        let _ = std::fs::remove_file(&ca_file);
        let _ = std::fs::remove_file(&peer0);
        let _ = std::fs::remove_file(&peer1);

        assert!(
            with_ca.contains("security.pki.certificateFiles = [ ./hive-ca.pem ]"),
            "CA cert must be wired into certificateFiles when signalled:\n{with_ca}"
        );
        assert!(
            with_peers.contains(
                "security.pki.certificateFiles = [ ./hive-ca.pem ./peer-ca-0.pem ./peer-ca-1.pem ]"
            ),
            "hive CA + peer CAs must all appear in the certificateFiles list:\n{with_peers}"
        );
        assert!(
            peers_only
                .contains("security.pki.certificateFiles = [ ./peer-ca-0.pem ./peer-ca-1.pem ]"),
            "peer CAs must be trusted even when this hive has no self-signed CA:\n{peers_only}"
        );
        assert!(
            !without_ca.contains("security.pki.certificateFiles"),
            "no certificateFiles reference without any CA signal:\n{without_ca}"
        );
    }

    #[test]
    fn render_flake_injects_otel_when_signalled() {
        // services.hyperhive.otel.* -> HYPERHIVE_OTEL_* on hive-c0re's unit
        // -> injected as build-time hyperhive.otel.* into every agent. With
        // no endpoint signal, no hyperhive.otel lines are emitted (agents
        // keep the the harness modules disabled default).
        //
        // SAFETY: single-threaded mutation of process env vars no other
        // test asserts on; restored before returning.
        let render = || {
            render_flake(
                "github:example/hyperhive",
                "path:/nix/store/bbbb-hyperhive-docs-source",
                "path:/nix/store/aaaa-nixpkgs-source",
                8000,
                "she/her",
                &std::collections::HashMap::new(),
                &[sample_spec("alice", false, 9001)],
            )
        };
        unsafe {
            std::env::remove_var("HYPERHIVE_OTEL_EXTRA_RESOURCE_ATTRIBUTES");
            std::env::remove_var("HYPERHIVE_OTEL_HEADERS_CREDENTIAL");
            std::env::set_var("HYPERHIVE_OTEL_ENDPOINT", "https://c.example/otel");
            std::env::set_var("HYPERHIVE_OTEL_PROTOCOL", "grpc");
        }
        let on_minimal = render();
        unsafe {
            std::env::set_var(
                "HYPERHIVE_OTEL_EXTRA_RESOURCE_ATTRIBUTES",
                "deployment.environment=prod",
            );
            std::env::set_var(
                "HYPERHIVE_OTEL_HEADERS_CREDENTIAL",
                "/run/secrets/otel-headers",
            );
        }
        let on_full = render();
        unsafe {
            std::env::remove_var("HYPERHIVE_OTEL_ENDPOINT");
            std::env::remove_var("HYPERHIVE_OTEL_PROTOCOL");
            std::env::remove_var("HYPERHIVE_OTEL_EXTRA_RESOURCE_ATTRIBUTES");
            std::env::remove_var("HYPERHIVE_OTEL_HEADERS_CREDENTIAL");
        }
        let off = render();

        assert!(
            on_minimal.contains("hyperhive.otel.enable = true;"),
            "otel enable must be injected:\n{on_minimal}"
        );
        assert!(
            on_minimal.contains("hyperhive.otel.endpoint = \"https://c.example/otel\";"),
            "otel endpoint must be injected:\n{on_minimal}"
        );
        assert!(
            on_minimal.contains("hyperhive.otel.protocol = \"grpc\";"),
            "otel protocol must be injected:\n{on_minimal}"
        );
        // Optional fields absent when unset.
        assert!(
            !on_minimal.contains("hyperhive.otel.extraResourceAttributes"),
            "extraResourceAttributes must not appear when unset:\n{on_minimal}"
        );
        assert!(
            !on_minimal.contains("hyperhive.otel.headersCredential"),
            "headersCredential must not appear when unset:\n{on_minimal}"
        );

        assert!(
            on_full.contains(
                "hyperhive.otel.extraResourceAttributes = \"deployment.environment=prod\";"
            ),
            "extraResourceAttributes must be injected when set:\n{on_full}"
        );
        assert!(
            on_full.contains("hyperhive.otel.headersCredential = \"/run/secrets/otel-headers\";"),
            "headersCredential must be injected when set:\n{on_full}"
        );

        assert!(
            !off.contains("hyperhive.otel"),
            "no otel lines when disabled:\n{off}"
        );
    }

    #[test]
    fn render_flake_injects_github_disable_only_when_signalled() {
        // services.hyperhive.github.enable = false -> HYPERHIVE_GITHUB_DISABLED
        // on hive-c0re's unit -> `hyperhive.github.enable = false` injected into
        // every agent. On by default, so nothing is emitted unless disabled.
        //
        // SAFETY: single-threaded mutation of an env var no other test asserts
        // on; restored before returning.
        let render = || {
            render_flake(
                "github:example/hyperhive",
                "path:/nix/store/bbbb-hyperhive-docs-source",
                "path:/nix/store/aaaa-nixpkgs-source",
                8000,
                "she/her",
                &std::collections::HashMap::new(),
                &[sample_spec("alice", false, 9001)],
            )
        };
        unsafe {
            std::env::remove_var("HYPERHIVE_GITHUB_DISABLED");
        }
        let on_default = render();
        unsafe {
            std::env::set_var("HYPERHIVE_GITHUB_DISABLED", "1");
        }
        let disabled = render();
        unsafe {
            std::env::remove_var("HYPERHIVE_GITHUB_DISABLED");
        }
        assert!(
            !on_default.contains("hyperhive.github.enable"),
            "github.enable must not be emitted by default (agents keep the true default):\n{on_default}"
        );
        assert!(
            disabled.contains("hyperhive.github.enable = false;"),
            "github.enable = false must be injected when the host disables it:\n{disabled}"
        );
    }
}