hyperhive/swarm-controller/src/main.rs

//! Swarm-level controller daemon. Runs as the unprivileged
//! `swarm-controller` user on whichever host the operator flips
//! `services.hyperhive.swarm.controller.enable` on, and serves HTTP over a
//! unix socket that the hive-gateway's nginx proxies to.
//!
//! Holds one piece of read-only state: the swarm's hive directory, loaded
//! once at startup from an env var the NixOS module sets
//! (`services.hyperhive.swarm.controller`) — see `load_hives`. Still no
//! persistence and no writes; a config change means a redeploy, same as
//! every other option this process reads.
//!
//! Distinct from `hive-c0re`, which is per-hive: c0re owns the agents on
//! one host, this owns what is true across hives.
//!
//! `OpenAPI` spec generation mirrors `hive-c0re/src/dashboard/mod.rs`
//! exactly: `#[utoipa::path(...)]` per handler, an `ApiDoc` root, and a
//! raw JSON route at `/api/openapi.json`. Swagger UI itself is
//! nginx-hosted from the nix store, same shape as the per-hive
//! dashboard's (`nix/host-modules/hive-gateway/vhosts.nix`'s
//! `swarmUiVhost` — a swagger-ui-theme dist under `/api/docs/`, no
//! fallback to this daemon). Only annotated routes appear in the spec;
//! an unannotated one just doesn't show up, nothing breaks.

use std::os::unix::fs::PermissionsExt as _;
use std::path::PathBuf;
use std::sync::{Arc, Mutex};

use anyhow::{Context, Result};
use axum::{Json, extract::State, routing::get};
use hive_jobq_wire::GraphWire as _;
use serde::{Deserialize, Serialize};
use utoipa::{OpenApi, ToSchema};
use utoipa_axum::{router::OpenApiRouter, routes};

mod auth;
mod forge;
mod status;

/// Node payload for the swarm-level job graph. Named `Swarm*` rather than
/// the bare `NodeKind`/`Resource` `hive-c0re::job_queue::model` already
/// uses, so a grep for either doesn't land on both crates.
///
/// `CreateIdentity` was the first real variant — "the minimal shape for
/// agent creation is creating the identity and wiring that in" (the design
/// thread's own framing for why that landed before the forge-node work
/// here). `CreateRepo`/`AddRepoMember`/`InitAgentConfigRepo` are three
/// separate nodes rather than one combined "create the repo" step so each
/// is independently retryable/observable in the job graph, same as every
/// other multi-step provisioning flow in this codebase (`hive-c0re`'s own
/// `NodeKind` never folds unrelated forge calls into one node either).
#[derive(Clone, Debug)]
enum SwarmNodeKind {
    /// Ensure `agent` exists as an authelia subject at the swarm level.
    CreateIdentity { agent: String },
    /// Create `repo` in `forge::AGENTS_ORG` with the operator merge gate
    /// on its default branch. See `forge::Client::create_repo`.
    CreateRepo { repo: String },
    /// Add `agent` as a write collaborator on `repo`. See
    /// `forge::Client::add_repo_member`.
    AddRepoMember { repo: String, agent: String },
    /// Seed `repo` with `agent.nix` + `flake.nix`. See
    /// `forge::Client::seed_agent_config`.
    InitAgentConfigRepo { repo: String, agent: String },
}

impl hive_jobq_wire::WireNode for SwarmNodeKind {
    fn label(&self) -> String {
        match self {
            SwarmNodeKind::CreateIdentity { .. } => "create_identity".to_owned(),
            SwarmNodeKind::CreateRepo { .. } => "create_repo".to_owned(),
            SwarmNodeKind::AddRepoMember { .. } => "add_repo_member".to_owned(),
            SwarmNodeKind::InitAgentConfigRepo { .. } => "init_agent_config_repo".to_owned(),
        }
    }

    fn data(&self, _id: hive_jobq_wire::WireId) -> serde_json::Value {
        match self {
            SwarmNodeKind::CreateIdentity { agent } => {
                serde_json::json!({ "agent": agent })
            }
            SwarmNodeKind::CreateRepo { repo } => {
                serde_json::json!({ "repo": repo })
            }
            SwarmNodeKind::AddRepoMember { repo, agent }
            | SwarmNodeKind::InitAgentConfigRepo { repo, agent } => {
                serde_json::json!({ "repo": repo, "agent": agent })
            }
        }
    }
}

/// Placeholder resource name — same rationale and same "no variants until a
/// real node needs one" shape as [`SwarmNodeKind`].
#[derive(Clone, Debug, PartialEq, Eq, Hash)]
enum SwarmResourceKind {}

impl hive_jobq_wire::WireResource for SwarmResourceKind {
    fn name(&self) -> String {
        match *self {}
    }
}

/// Everything a claimed node's executor arm might need to reach outside
/// this process — bundled into one `Clone` struct rather than growing
/// `run_swarm_node`'s parameter list per node kind (three forge-shaped
/// node kinds landed in one slice; a fourth parameter each would have made
/// the signature the least readable part of this file). Each field is
/// built once at startup (see `main`) and is `None` exactly when that
/// dependency isn't configured on this host — every arm below treats
/// absence as *this node's* failure, not a reason to skip silently.
#[derive(Clone)]
struct WorkerDeps {
    auth: Option<std::sync::Arc<auth::AuthBridge>>,
    forge: Option<std::sync::Arc<forge::Client>>,
}

/// Run a claimed node's actual work. Mirrors `hive-c0re/src/job_queue/
/// exec.rs::run_node`'s role exactly — the one place a `SwarmNodeKind`
/// variant turns into a real effect.
async fn run_swarm_node(
    _id: hive_jobq::NodeId,
    kind: SwarmNodeKind,
    builder: hive_jobq::builder::JobBuilder<SwarmNodeKind, SwarmResourceKind>,
    deps: WorkerDeps,
) -> (
    hive_jobq::builder::JobBuilder<SwarmNodeKind, SwarmResourceKind>,
    hive_jobq::scheduler::Outcome,
) {
    use hive_jobq::scheduler::Outcome;

    let outcome = match kind {
        SwarmNodeKind::CreateIdentity { agent } => match deps.auth {
            None => {
                Outcome::Failed("no swarm-authelia-bridge is configured on this host".to_owned())
            }
            Some(bridge) => match bridge.ensure_agent_identity(&agent).await {
                Ok(_) => Outcome::Done,
                Err(e) => Outcome::Failed(format!("{e:#}")),
            },
        },
        SwarmNodeKind::CreateRepo { repo } => match deps.forge {
            None => Outcome::Failed(
                "no forge configured on this host (SWARM_CONTROLLER_FORGE_URL / \
                 SWARM_CONTROLLER_FORGE_TOKEN_FILE unset)"
                    .to_owned(),
            ),
            Some(client) => match client.create_repo(&repo).await {
                Ok(full_name) => {
                    tracing::info!(%full_name, "swarm jobq: create_repo done");
                    Outcome::Done
                }
                Err(e) => Outcome::Failed(format!("{e:#}")),
            },
        },
        SwarmNodeKind::AddRepoMember { repo, agent } => match deps.forge {
            None => Outcome::Failed(
                "no forge configured on this host (SWARM_CONTROLLER_FORGE_URL / \
                 SWARM_CONTROLLER_FORGE_TOKEN_FILE unset)"
                    .to_owned(),
            ),
            Some(client) => match client.add_repo_member(&repo, &agent).await {
                Ok(()) => Outcome::Done,
                Err(e) => Outcome::Failed(format!("{e:#}")),
            },
        },
        SwarmNodeKind::InitAgentConfigRepo { repo, agent } => match deps.forge {
            None => Outcome::Failed(
                "no forge configured on this host (SWARM_CONTROLLER_FORGE_URL / \
                 SWARM_CONTROLLER_FORGE_TOKEN_FILE unset)"
                    .to_owned(),
            ),
            Some(client) => match client.seed_agent_config(&repo, &agent).await {
                Ok(()) => Outcome::Done,
                Err(e) => Outcome::Failed(format!("{e:#}")),
            },
        },
    };
    (builder, outcome)
}

/// Spawn the swarm-level job-graph scheduler loop. Mirrors `hive-c0re/src/
/// job_queue/scheduler.rs::run_worker`'s shape: claim one runnable node,
/// spawn the future that runs + completes it, loop again immediately if
/// something started (more may now be runnable), otherwise back off
/// briefly before re-polling.
///
/// No shutdown signal to wire in — unlike `hive-c0re`'s `coord.shutdown_rx()`,
/// this daemon has no graceful-shutdown machinery at all yet (`main`'s
/// `axum::serve` runs unconditionally to process exit), so this loop
/// matches that: it rides the runtime down with the process, same as
/// every in-flight HTTP request does.
///
/// Cheap to run with an empty graph: `claim_next` on a graph nothing was
/// ever inserted into just returns `None` every poll.
///
/// `deps` is cloned per iteration (its fields are `Arc` clones, not
/// reconnects) and moved into the closure `claim_next` takes ownership
/// of — `run_swarm_node` needs its own owned copy since the claimed
/// future may outlive this loop iteration.
fn spawn_jobq_worker(
    sched: Arc<Mutex<hive_jobq::scheduler::Scheduler<SwarmNodeKind, SwarmResourceKind>>>,
    deps: WorkerDeps,
) {
    tokio::spawn(async move {
        loop {
            let deps = deps.clone();
            let runner =
                hive_jobq::scheduler::Scheduler::claim_next(&sched, move |id, kind, builder| {
                    run_swarm_node(id, kind, builder, deps)
                });
            match runner {
                Some(runner) => {
                    tokio::spawn(async move {
                        let (id, grew) = runner.await;
                        if let Err(e) = grew {
                            tracing::warn!(
                                node = id.get(),
                                error = %e,
                                "swarm jobq: grown job rejected"
                            );
                        }
                    });
                    // Something just started — more may be runnable right
                    // now, so loop again immediately rather than sleeping.
                }
                None => {
                    // Nothing runnable. Bounded poll rather than an event
                    // wake (unlike hive-c0re's `notify.notify_one()`,
                    // there is no completion-signal channel here yet) —
                    // fine at this daemon's scale (one graph, no
                    // submitters yet); revisit if/when that stops holding.
                    tokio::time::sleep(std::time::Duration::from_millis(200)).await;
                }
            }
        }
    });
}

/// Where the daemon binds, overridable via `SWARM_CONTROLLER_SOCKET`.
///
/// A compiled-in default is legitimate here and is *not* the mistake that
/// a hardcoded remote address would be: this is a path this process
/// **creates**, not an address it hopes to find something at. systemd's
/// `RuntimeDirectory=swarm-controller` makes the parent exist before
/// `ExecStart`, so the default names a directory the unit just produced.
///
/// The directory is its own — deliberately not shared with hive-c0re's
/// `/run/hyperhive`. The socket is `0666`, so its directory is the only
/// access control it has; co-locating it with c0re's admin socket would
/// put both within reach of whatever can reach either. nginx runs on the
/// host, so nothing narrows its reach for you.
const DEFAULT_SOCKET: &str = "/run/swarm-controller/controller.sock";

fn socket_path() -> PathBuf {
    std::env::var_os("SWARM_CONTROLLER_SOCKET")
        .map_or_else(|| PathBuf::from(DEFAULT_SOCKET), PathBuf::from)
}

/// Root of the auto-generated `OpenAPI` spec, served raw at
/// `/api/openapi.json` — see the module doc comment above. Tag list
/// grows alongside the swarm-level surfaces this daemon picks up, same
/// as `hive-c0re::dashboard::ApiDoc`'s tag list did.
#[derive(OpenApi)]
#[openapi(
    info(
        title = "hyperhive swarm-controller API",
        description = "swarm-controller's HTTP surface, served over its unix \
                        socket behind the gateway's swarm-UI vhost."
    ),
    tags(
        (name = "health", description = "liveness probe"),
        (name = "hives", description = "the swarm's hive directory"),
        (name = "links", description = "swarm service quick links"),
        (name = "jobq", description = "the swarm-level job graph"),
        (name = "agents", description = "creating agent identities at swarm level"),
    )
)]
struct ApiDoc;

/// Liveness probe. Returns the build's version so an operator can tell
/// *which* controller answered without shelling onto the host.
#[utoipa::path(
    get,
    path = "/health",
    responses((status = 200, description = "process is up, body is \"swarm-controller <version>\"", body = String)),
    tag = "health"
)]
async fn health() -> &'static str {
    concat!("swarm-controller ", env!("CARGO_PKG_VERSION"), "\n")
}

/// One hive in the swarm's directory — the same `name`/`domain` pair
/// `services.hyperhive.swarm.hives` (nix/host-modules/swarm.nix) declares,
/// carried across unchanged rather than reshaped, so this stays a direct
/// mirror of the nix source of truth instead of a second vocabulary for
/// the same two fields.
#[derive(Clone, Debug, PartialEq, Serialize, Deserialize, ToSchema)]
struct HiveEntry {
    name: String,
    domain: String,
}

#[derive(Clone)]
struct AppState {
    /// Loaded once at startup (`load_hives`); never mutated, so an
    /// `Arc` clone per request is the whole synchronization story.
    hives: Arc<Vec<HiveEntry>>,
    /// Loaded once at startup (`load_links`); same synchronization story
    /// as `hives`.
    links: Arc<Vec<ServiceLink>>,
    /// `None` when this deployment wired up no swarm queue — the only
    /// state in which `/api/hives/status` cannot answer at all. A queue
    /// that is merely *unreachable* still yields a reader, because
    /// `async-nats` reconnects underneath it.
    status: Option<Arc<status::StatusReader>>,
    /// The swarm-level job graph, wrapped in its
    /// [`hive_jobq::scheduler::Scheduler`] now that something drives it
    /// (`spawn_jobq_worker`) — the graph alone was enough for the
    /// read-only endpoints, the scheduler is what a `claim_next` loop
    /// needs. `std::sync::Mutex`, not `tokio`'s — every lock scope below
    /// is synchronous (no `.await` while held). Always present, never
    /// gated on the swarm queue: this is process state, not something
    /// read over the network.
    /// **Not** consulted by `POST /api/agents` — that endpoint only ever
    /// queues the job (see [`create_agent`]); whether a bridge is
    /// configured is `run_swarm_node`'s concern (it holds its own clone,
    /// handed to it by `spawn_jobq_worker`), not this handler's. A request
    /// still queues cleanly on a bridge-less host, then fails loud once
    /// claimed — same "queue now, fail per-job" shape as an unreachable
    /// swarm queue.
    jobq: Arc<Mutex<hive_jobq::scheduler::Scheduler<SwarmNodeKind, SwarmResourceKind>>>,
}

/// Env var the controller's NixOS module sets from
/// `services.hyperhive.swarm.hives`, JSON-encoded — the full directory
/// (this daemon has no "self" to exclude) rather than peers-minus-self
/// (`services.hyperhive.swarm.peerHives`, which other consumers use).
const HIVES_ENV: &str = "SWARM_CONTROLLER_HIVES";

/// Parses [`HIVES_ENV`] into the swarm's hive directory. Unset or
/// unparseable both fall back to an empty list with a warning rather than
/// failing startup — a swarm-controller that can't yet see its own config
/// (dev run, a module not wired up yet) should still serve `/health`.
fn load_hives() -> Vec<HiveEntry> {
    let Some(raw) = std::env::var_os(HIVES_ENV) else {
        return Vec::new();
    };
    match serde_json::from_str(&raw.to_string_lossy()) {
        Ok(hives) => hives,
        Err(e) => {
            tracing::warn!(error = %e, env = HIVES_ENV, "failed to parse hive directory, serving an empty list");
            Vec::new()
        }
    }
}

/// The swarm's hive directory — every hive, including whichever one this
/// controller instance happens to run on (there is no "self" to exclude
/// at the swarm level, unlike `hive-c0re`'s peer list).
#[utoipa::path(
    get,
    path = "/api/hives",
    responses((status = 200, description = "every hive in the swarm", body = Vec<HiveEntry>)),
    tag = "hives"
)]
async fn get_hives(State(state): State<AppState>) -> Json<Vec<HiveEntry>> {
    Json((*state.hives).clone())
}

/// One quick link to a swarm-wide service (authelia, matrix, forge, this
/// daemon's own swagger UI, …). Deliberately generic rather than named
/// fields per service: each service's own nix module contributes its own
/// entry to `services.hyperhive.swarm.controller.links` (same list-merge
/// idiom `services.hyperhive.gateway.localNames` already uses), so adding
/// a new one is a nix-only change — no new field here, no swarm-ui change.
#[derive(Clone, Debug, PartialEq, Serialize, Deserialize, ToSchema)]
struct ServiceLink {
    label: String,
    /// Emoji or short glyph. Empty string, not `Option`, when a
    /// contributing module has none — one fewer null-vs-absent case for
    /// the frontend to handle, and every real contributor sets one today.
    icon: String,
    url: String,
}

/// Env var the controller's NixOS module sets from the merged
/// `services.hyperhive.swarm.controller.links` list, JSON-encoded — same
/// shape/rationale as [`HIVES_ENV`]. Consumed by `GET /api/links`.
const LINKS_ENV: &str = "SWARM_CONTROLLER_LINKS";

/// Parses [`LINKS_ENV`] into the swarm's service-link list. Same
/// fall-back-to-empty rationale as `load_hives`: a daemon that can't yet
/// see this config should still serve `/health` rather than fail startup.
fn load_links() -> Vec<ServiceLink> {
    let Some(raw) = std::env::var_os(LINKS_ENV) else {
        return Vec::new();
    };
    match serde_json::from_str(&raw.to_string_lossy()) {
        Ok(links) => links,
        Err(e) => {
            tracing::warn!(error = %e, env = LINKS_ENV, "failed to parse service links, serving an empty list");
            Vec::new()
        }
    }
}

/// Quick links to swarm-wide services (authelia, matrix, forge, this
/// daemon's own swagger UI, …), as contributed by each service's own nix
/// module. Empty when nothing was configured to contribute — a caller
/// renders 0 links the same way it renders any other count, no special
/// "not configured" case.
#[utoipa::path(
    get,
    path = "/api/links",
    responses((status = 200, description = "swarm service quick links", body = Vec<ServiceLink>)),
    tag = "links"
)]
async fn get_links(State(state): State<AppState>) -> Json<Vec<ServiceLink>> {
    Json((*state.links).clone())
}

/// Why the status route answers 503 rather than an empty list.
///
/// "I cannot reach the store" and "every hive is silent" are different
/// answers, and rendering the second when the first is true is exactly
/// the smoothing this endpoint exists to avoid — a caller would draw a
/// swarm-wide outage out of a local one. The cause is carried in the
/// body because a bare 503 on an operator-facing diagnostic is how a
/// misconfiguration costs an afternoon; it is a queue/JetStream error
/// string, and this surface is already behind the swarm's SSO.
struct StatusUnavailable(String);

impl axum::response::IntoResponse for StatusUnavailable {
    fn into_response(self) -> axum::response::Response {
        (axum::http::StatusCode::SERVICE_UNAVAILABLE, self.0).into_response()
    }
}

/// What each hive last said about itself, read from the swarm queue at
/// request time.
///
/// Every hive in the roster gets a row whether or not it has ever
/// reported — see the `status` module for why absence, not presence, is
/// the case this is built around.
#[utoipa::path(
    get,
    path = "/api/hives/status",
    responses(
        (status = 200, description = "a row per hive, freshness derived now", body = Vec<status::HiveStatus>),
        (status = 503, description = "no swarm queue is configured here, or its store could not be read", body = String),
    ),
    tag = "hives"
)]
async fn get_hives_status(
    State(state): State<AppState>,
) -> Result<Json<Vec<status::HiveStatus>>, StatusUnavailable> {
    let Some(reader) = state.status.as_ref() else {
        return Err(StatusUnavailable(
            "no swarm queue is configured on this host".to_owned(),
        ));
    };
    match reader
        .view(&state.hives, std::time::SystemTime::now())
        .await
    {
        Ok(rows) => Ok(Json(rows)),
        Err(e) => {
            let detail = format!("{e:#}");
            tracing::warn!(error = %detail, "reading the swarm status bucket failed");
            Err(StatusUnavailable(detail))
        }
    }
}

/// Body of `POST /api/agents` — the agent name to create. The repo name
/// inside `forge::AGENTS_ORG` is the same string: one repo per agent,
/// named after it, same convention `hive-c0re::forge` already uses for its
/// own single-hive `CreateRepo` path.
#[derive(Clone, Debug, Deserialize, ToSchema)]
struct CreateAgentRequest {
    name: String,
}

/// Where the queued job landed — a caller polls `/api/jobq/graph` (or
/// `?states=`) with this id to watch it settle, same as every other job
/// kind this daemon will ever queue.
#[derive(Clone, Debug, Serialize, ToSchema)]
struct CreateAgentResponse {
    node_id: u64,
}

/// Queue the whole agent-creation job graph for `name` — `CreateIdentity`
/// then, once that succeeds, `CreateRepo`; once THAT succeeds,
/// `AddRepoMember` and `InitAgentConfigRepo` both run off it — a fan-out,
/// not a chain, since adding a collaborator and seeding config files are
/// independent operations against the same already-created, already-
/// protected repo and have no ordering requirement on each other (mara,
/// design review: `InitAgentConfigRepo` does not depend on
/// `AddRepoMember` — it's a jobq graph, not a linear chain). Returns as
/// soon as the graph is inserted — **not** once any of it has run; `run_swarm_node` does that
/// work asynchronously off the scheduler loop already running
/// (`spawn_jobq_worker`), same as every other node kind. The response
/// reports `CreateIdentity`'s id, the graph's entry point — a caller
/// watches the whole thing settle via `/api/jobq/graph`, which serves
/// every root, not just this one.
///
/// This endpoint is also the first genuine non-test caller all four
/// `SwarmNodeKind` variants have: each node kind's `dead_code` bound is
/// the whole reason the executor arm and this endpoint had to land in the
/// same change as the variant, not as a follow-up.
#[utoipa::path(
    post,
    path = "/api/agents",
    request_body = CreateAgentRequest,
    responses(
        (status = 200, description = "job chain queued", body = CreateAgentResponse),
        (status = 500, description = "the job chain could not be queued", body = String),
    ),
    tag = "agents"
)]
async fn create_agent(
    State(state): State<AppState>,
    Json(req): Json<CreateAgentRequest>,
) -> Result<Json<CreateAgentResponse>, (axum::http::StatusCode, String)> {
    let mut sched = state
        .jobq
        .lock()
        .unwrap_or_else(std::sync::PoisonError::into_inner);
    let agent = req.name;
    let repo = agent.clone();
    let ids = sched
        .insert_job(None, |b| {
            let create_identity = b.node(SwarmNodeKind::CreateIdentity {
                agent: agent.clone(),
            });
            let create_repo = b
                .node(SwarmNodeKind::CreateRepo { repo: repo.clone() })
                .after_ok(create_identity);
            // Both fan out from `create_repo` directly — independent
            // operations on the same repo, no ordering requirement on
            // each other (see the doc comment above).
            let _add_repo_member = b
                .node(SwarmNodeKind::AddRepoMember {
                    repo: repo.clone(),
                    agent: agent.clone(),
                })
                .after_ok(create_repo);
            let _init_config = b
                .node(SwarmNodeKind::InitAgentConfigRepo { repo, agent })
                .after_ok(create_repo);
            vec![create_identity.guid()]
        })
        .map_err(|e| (axum::http::StatusCode::INTERNAL_SERVER_ERROR, e.to_string()))?;
    let [id] = ids[..] else {
        unreachable!("exactly one handle was asked for");
    };
    Ok(Json(CreateAgentResponse { node_id: id.get() }))
}

/// Query params for `GET /api/jobq/graph` — `?states=` narrows to root
/// groups in the named states, same shape `hive_jobq_wire::parse_states`
/// parses.
#[derive(Deserialize, utoipa::IntoParams)]
struct JobqGraphQuery {
    states: Option<String>,
}

/// Every node of every root group in the swarm-level job graph. Nothing
/// filters "done and old" here — with zero nodes ever submitted there is
/// nothing to bound yet, so `graph.roots()` (everything) is the whole
/// roster passed to [`hive_jobq_wire::GraphWire::wire_snapshot`].
#[utoipa::path(
    get,
    path = "/api/jobq/graph",
    params(JobqGraphQuery),
    responses((status = 200, description = "every node of every root group, as generic \
        `hive_jobq` graph nodes. `?states=` narrows to root groups in the named states.",
        body = Vec<hive_jobq_wire::GraphNode>)),
    tag = "jobq"
)]
async fn get_jobq_graph(
    State(state): State<AppState>,
    axum::extract::Query(q): axum::extract::Query<JobqGraphQuery>,
) -> Json<Vec<hive_jobq_wire::GraphNode>> {
    let states = hive_jobq_wire::parse_states(q.states.as_deref());
    let sched = state
        .jobq
        .lock()
        .unwrap_or_else(std::sync::PoisonError::into_inner);
    let graph = sched.graph();
    let roots: Vec<hive_jobq::NodeId> = graph.roots().map(|n| n.id).collect();
    let nodes = graph.wire_snapshot(roots);
    Json(hive_jobq_wire::filter_nodes_by_state(
        nodes,
        states.as_deref(),
    ))
}

/// Counts by lifecycle state over the same groups `/api/jobq/graph` serves.
#[utoipa::path(
    get,
    path = "/api/jobq/rollup",
    responses((status = 200, description = "counts by lifecycle state", body = Vec<hive_jobq_wire::StateCount>)),
    tag = "jobq"
)]
async fn get_jobq_rollup(State(state): State<AppState>) -> Json<Vec<hive_jobq_wire::StateCount>> {
    let sched = state
        .jobq
        .lock()
        .unwrap_or_else(std::sync::PoisonError::into_inner);
    let graph = sched.graph();
    let roots: Vec<hive_jobq::NodeId> = graph.roots().map(|n| n.id).collect();
    Json(hive_jobq_wire::state_rollup(graph, roots))
}

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

    let path = socket_path();

    // `RuntimeDirectoryPreserve=yes` keeps the directory across a restart,
    // so a socket file from the previous run can outlive the process that
    // owned it and `bind` would fail with EADDRINUSE. Unlinking a stale
    // socket is safe precisely because the directory is ours alone: nothing
    // else can have put a file at this path.
    if let Err(e) = std::fs::remove_file(&path)
        && e.kind() != std::io::ErrorKind::NotFound
    {
        return Err(e).with_context(|| format!("clearing stale socket at {}", path.display()));
    }

    let listener = tokio::net::UnixListener::bind(&path)
        .with_context(|| format!("binding {}", path.display()))?;

    // `bind` leaves the socket 0755, and connecting needs write — the
    // gateway's nginx is a different user, so it would be locked out.
    // 0666 matches how hive-c0re publishes the per-agent sockets
    // (`socket_server::start`), and rests on the same argument: **the
    // containing directory is the access control, not the socket mode.**
    // This directory holds one socket and is bind-mounted into exactly
    // one container. That is also why it must not be shared with
    // hive-c0re's `/run/hyperhive` — with a 0666 socket, a directory
    // that carries more than it should is the whole vulnerability.
    std::fs::set_permissions(&path, std::fs::Permissions::from_mode(0o666))
        .with_context(|| format!("chmod {}", path.display()))?;
    tracing::info!(socket = %path.display(), "swarm-controller listening");

    // Connect to the swarm queue when this deployment wired one up.
    //
    // Deliberately NOT fatal on failure: the controller's HTTP surface is
    // useful without the queue, and a hive that cannot be read from renders
    // as `unknown` rather than as an outage of this daemon. What IS fatal is
    // a half-set environment — `QueueConfig::from_env` refuses that, because
    // silently behaving like an unconfigured host is how every hive ends up
    // reading `never_reported` with nothing to point at.
    let status = match swarm_queue_client::QueueConfig::from_env("SWARM_CONTROLLER")? {
        None => {
            tracing::info!("no swarm queue configured; status aggregation is off");
            None
        }
        Some(cfg) => match swarm_queue_client::connect(cfg).await {
            Ok(client) => {
                // NOT "connected": `retry_on_initial_connect` returns a client
                // before any connection has been established, so claiming a
                // connection here would put "connected to the swarm queue" in
                // the journal moments before every request 503s with "not
                // connected" — and a reader would rightly distrust the second
                // line rather than the first. The connection's real state is
                // reported by the status endpoint, which checks it per request.
                tracing::info!("swarm queue configured; connecting in the background");
                Some(Arc::new(status::StatusReader::new(
                    client,
                    status::StatusReader::stale_after_from_env(),
                )))
            }
            Err(e) => {
                // `chain`, not `{:#}`: this is the queue client's own
                // error type, and thiserror's Display ignores the
                // alternate flag — the source would be dropped silently.
                tracing::warn!(
                    error = swarm_queue_client::chain(&e),
                    "swarm queue unreachable"
                );
                None
            }
        },
    };

    // Same "not fatal, log and carry on" shape as the queue connect above:
    // a controller with no bridge wired up still serves everything else,
    // and `run_swarm_node` gives an honest per-job failure instead of this
    // fn refusing to start.
    let auth = match auth::AuthBridge::from_env() {
        Ok(auth) => auth.map(Arc::new),
        Err(e) => {
            tracing::warn!(error = %format!("{e:#}"), "swarm-authelia-bridge misconfigured; agent identity creation is off");
            None
        }
    };
    // Same shape again: a controller with no forge configured still serves
    // everything else, and the forge-shaped node kinds give an honest
    // per-job failure rather than this fn refusing to start.
    let forge_client = match forge::Client::from_env() {
        Ok(client) => client.map(Arc::new),
        Err(e) => {
            tracing::warn!(error = %format!("{e:#}"), "forge misconfigured; repo provisioning is off");
            None
        }
    };
    let deps = WorkerDeps {
        auth,
        forge: forge_client,
    };

    let jobq = Arc::new(Mutex::new(hive_jobq::scheduler::Scheduler::new(
        hive_jobq::Graph::new(),
        hive_jobq::resources::ResourceTable::new(),
    )));
    spawn_jobq_worker(Arc::clone(&jobq), deps);

    let state = AppState {
        hives: Arc::new(load_hives()),
        links: Arc::new(load_links()),
        status,
        jobq,
    };

    let (router, api) = OpenApiRouter::<AppState>::with_openapi(ApiDoc::openapi())
        .routes(routes!(health))
        .routes(routes!(get_hives))
        .routes(routes!(get_hives_status))
        .routes(routes!(get_links))
        .routes(routes!(get_jobq_graph))
        .routes(routes!(get_jobq_rollup))
        .routes(routes!(create_agent))
        .split_for_parts();
    // Just the JSON, not the UI — Swagger UI itself is nginx-hosted from
    // the nix store (see the module doc comment above). `api` is
    // `Clone`; each request gets its own owned copy for `Json` to
    // serialize, same as `hive-c0re::dashboard::serve`.
    let app = router
        .route(
            "/api/openapi.json",
            get(move || async move { Json(api.clone()) }),
        )
        .with_state(state);
    axum::serve(listener, app)
        .await
        .context("serving swarm-controller")
}

#[cfg(test)]
mod tests {
    use super::{
        DEFAULT_SOCKET, HIVES_ENV, HiveEntry, LINKS_ENV, ServiceLink, SwarmNodeKind, WorkerDeps,
        load_hives, load_links, run_swarm_node,
    };
    use std::path::Path;

    /// Drives `SwarmNodeKind::CreateRepo` through the real
    /// `hive_jobq::scheduler::Scheduler` claim → run → complete path,
    /// rather than only through `create_agent`'s endpoint test (there
    /// isn't one — the endpoint itself is thin, insert-and-return; the
    /// interesting behavior is in `run_swarm_node`'s executor arm, which
    /// this exercises directly).
    ///
    /// Deliberately offline: with both forge env vars unset,
    /// `forge::Client::from_env` returns `Ok(None)` (see that module's doc
    /// comment), so this exercises the whole claim → run → complete path
    /// through `hive_jobq::scheduler::Scheduler` without a real forge
    /// server — at the cost of only ever observing the
    /// graceful-absence-is-failure branch here. The happy path needs an
    /// actual forge instance and isn't something a unit test in this crate
    /// can reach.
    ///
    /// SAFETY: single-threaded mutation of the two `forge` env vars this
    /// test itself owns, restored before returning — no other test in this
    /// crate reads them.
    #[tokio::test]
    async fn create_repo_node_runs_end_to_end_and_fails_without_forge_configured() {
        unsafe {
            std::env::remove_var("SWARM_CONTROLLER_FORGE_URL");
            std::env::remove_var("SWARM_CONTROLLER_FORGE_TOKEN_FILE");
        }

        let mut sched = hive_jobq::scheduler::Scheduler::new(
            hive_jobq::Graph::new(),
            hive_jobq::resources::ResourceTable::new(),
        );
        let id = sched
            .append(
                SwarmNodeKind::CreateRepo {
                    repo: "atlas".to_owned(),
                },
                Vec::new(),
                None,
            )
            .expect("insert");
        let sched = std::sync::Arc::new(std::sync::Mutex::new(sched));

        let deps = WorkerDeps {
            auth: None,
            forge: None,
        };
        let runner =
            hive_jobq::scheduler::Scheduler::claim_next(&sched, move |id, kind, builder| {
                run_swarm_node(id, kind, builder, deps)
            })
            .expect("the node just inserted is runnable");
        runner
            .await
            .1
            .expect("no growth declared, nothing to reject");

        let guard = sched.lock().unwrap();
        let node = guard.graph().node(id).expect("node still present");
        assert_eq!(node.state, hive_jobq::State::Failed);
        assert!(
            node.error.as_deref().unwrap_or_default().contains("forge"),
            "expected a forge-not-configured error, got {:?}",
            node.error
        );
    }

    /// The socket must not share a directory with anything else, because
    /// the socket is `0666` and the directory is therefore the only access
    /// control it has. `/run/hyperhive` in particular holds hive-c0re's
    /// **admin** socket, and nginx — a host service — is bounded by
    /// nothing but the directory itself.
    ///
    /// A test rather than a comment: the failure this guards against is a
    /// one-word edit that looks tidier and reads fine in review.
    #[test]
    fn socket_lives_in_its_own_runtime_dir() {
        let parent = Path::new(DEFAULT_SOCKET)
            .parent()
            .expect("socket path has a parent directory");
        assert_eq!(
            parent,
            Path::new("/run/swarm-controller"),
            "the socket's directory is its access control — moving it under a shared \
             directory (notably /run/hyperhive, which holds the host admin socket) \
             exposes everything else in that directory to the gateway's nginx"
        );
    }

    /// One test, not three, deliberately: `HIVES_ENV` is a real process
    /// env var, and `cargo test`'s default parallel runner would race
    /// separate missing/malformed/valid tests against each other. Driving
    /// all three states sequentially inside one test needs no mutex and
    /// still proves each branch of `load_hives`.
    ///
    /// SAFETY: single-threaded mutation of a process env var no other
    /// test in this crate reads; restored (removed) before returning.
    #[test]
    fn load_hives_covers_missing_malformed_and_valid() {
        unsafe {
            std::env::remove_var(HIVES_ENV);
        }
        assert_eq!(
            load_hives(),
            Vec::<HiveEntry>::new(),
            "unset env var is an empty directory, not a startup failure"
        );

        unsafe {
            std::env::set_var(HIVES_ENV, "not json");
        }
        assert_eq!(
            load_hives(),
            Vec::<HiveEntry>::new(),
            "unparseable env var falls back to empty rather than panicking — \
             /health must still answer even if this daemon's own config is wrong"
        );

        unsafe {
            std::env::set_var(
                HIVES_ENV,
                r#"[{"name":"pr1ma","domain":"pr1ma.example.com"},{"name":"umbra","domain":"umbra.example.com"}]"#,
            );
        }
        assert_eq!(
            load_hives(),
            vec![
                HiveEntry {
                    name: "pr1ma".to_string(),
                    domain: "pr1ma.example.com".to_string(),
                },
                HiveEntry {
                    name: "umbra".to_string(),
                    domain: "umbra.example.com".to_string(),
                },
            ]
        );

        unsafe {
            std::env::remove_var(HIVES_ENV);
        }
    }

    /// Same three-state coverage as `load_hives_covers_missing_malformed_and_valid`,
    /// same reason for one test rather than three (a shared process env var).
    ///
    /// SAFETY: single-threaded mutation of a process env var no other test
    /// in this crate reads; restored (removed) before returning.
    #[test]
    fn load_links_covers_missing_malformed_and_valid() {
        unsafe {
            std::env::remove_var(LINKS_ENV);
        }
        assert_eq!(
            load_links(),
            Vec::<ServiceLink>::new(),
            "unset env var is an empty list, not a startup failure"
        );

        unsafe {
            std::env::set_var(LINKS_ENV, "not json");
        }
        assert_eq!(
            load_links(),
            Vec::<ServiceLink>::new(),
            "unparseable env var falls back to empty rather than panicking"
        );

        unsafe {
            std::env::set_var(
                LINKS_ENV,
                r#"[{"label":"Authelia","icon":"🔑","url":"https://auth.example.com/"}]"#,
            );
        }
        assert_eq!(
            load_links(),
            vec![ServiceLink {
                label: "Authelia".to_string(),
                icon: "🔑".to_string(),
                url: "https://auth.example.com/".to_string(),
            }]
        );

        unsafe {
            std::env::remove_var(LINKS_ENV);
        }
    }
}