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.deploy.swarm-controller.enable` on, and serves HTTP over a
//! unix socket that the hive-gateway's nginx proxies to.
//!
//! Configuration is read-only and loaded once at startup from env vars the
//! NixOS module sets (`services.hyperhive.swarm.controller`) — see
//! `load_hives`. A config change means a redeploy, same as every other
//! option this process reads.
//!
//! The one thing it does persist is `webhook-secret` under its
//! `StateDirectory` (see `webhook`), because that key is handed to Forgejo
//! at registration and so cannot be regenerated per boot.
//!
//! 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::{Path, State},
    routing::get,
};
use hive_jobq_wire::GraphWire as _;
use serde::{Deserialize, Serialize};
use swarm_authelia_bridge_sock::BridgeResponse;
use utoipa::{OpenApi, ToSchema};
use utoipa_axum::{router::OpenApiRouter, routes};

mod agent_status;
mod auth;
mod config_pr;
mod forge;
mod issue_report;
mod matrix_account;
mod otel_http_client;
mod read_policy;
mod status;
mod store;
mod vcs_metrics;
mod wanted;
mod webhook;

/// 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 the agent's repo in `forge::CONFIG_ORG` with the operator
    /// merge gate on its default branch. See `forge::Client::create_repo`.
    CreateRepo { agent: String },
    /// Ensure `agent` exists as a Forgejo user account. Independent of
    /// `CreateIdentity`/`CreateRepo` (a forge user needs neither an
    /// authelia subject nor an existing repo) but a prerequisite for
    /// `AddRepoMember`, which fails with "user does not exist" against an
    /// account this node hasn't created yet. See
    /// `forge::Client::ensure_agent_user`.
    CreateForgeUser { agent: String },
    /// Add `agent` as a write collaborator on its own repo. See
    /// `forge::Client::add_repo_member`.
    AddRepoMember { agent: String },
    /// Seed the agent's repo with `agent.nix` + `flake.nix`. See
    /// `forge::Client::seed_agent_config`.
    ///
    /// Deliberately carries no hive: seeding a config repo is the same
    /// work whichever hive the agent is bound for, and an agent's config
    /// states nothing about where it runs. The hive is an address the
    /// swarm routes a deploy message to — it belongs on the node that
    /// sends that message, not on this one.
    InitAgentConfigRepo { agent: String },
    /// Tell `hive` to rebuild `agent`, by publishing on the swarm's deploy
    /// subject. The one node kind whose effect leaves this host.
    ///
    /// Carries the hive, unlike every variant above — this is the node
    /// `InitAgentConfigRepo`'s doc points at when it says the hive belongs
    /// on the node that sends the deploy message. It names the subject the
    /// message goes to, one per hive, so no other hive is woken by it.
    TriggerDeploy { hive: 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::CreateForgeUser { .. } => "create_forge_user".to_owned(),
            SwarmNodeKind::AddRepoMember { .. } => "add_repo_member".to_owned(),
            SwarmNodeKind::InitAgentConfigRepo { .. } => "init_agent_config_repo".to_owned(),
            SwarmNodeKind::TriggerDeploy { .. } => "trigger_deploy".to_owned(),
        }
    }

    fn data(&self, _id: hive_jobq_wire::WireId) -> serde_json::Value {
        // Spelled out as an or-pattern rather than a catch-all on purpose: a
        // new variant then fails to compile here instead of silently
        // rendering as an agent name. That gate has now fired once —
        // `TriggerDeploy` is the first node that is about an agent *and a
        // hive*, and a catch-all would have dropped the hive from the
        // viewer without anyone noticing.
        match self {
            SwarmNodeKind::CreateIdentity { agent }
            | SwarmNodeKind::CreateRepo { agent }
            | SwarmNodeKind::CreateForgeUser { agent }
            | SwarmNodeKind::AddRepoMember { agent }
            | SwarmNodeKind::InitAgentConfigRepo { agent } => {
                serde_json::json!({ "agent": agent })
            }
            SwarmNodeKind::TriggerDeploy { hive, agent } => {
                serde_json::json!({ "agent": agent, "hive": hive })
            }
        }
    }
}

/// 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>>,
    /// The swarm queue connection, for nodes whose effect is a published
    /// event rather than an API call. A handle rather than the status
    /// reader it is cloned from: a node that announces a deploy has no
    /// business reading hive status, and `status.rs`'s own doc says the
    /// connection living there is an accident of construction order, not a
    /// claim that events are a kind of status.
    queue: Option<async_nats::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 {
                // Matched rather than discarded with `Ok(_)`: a heal writes
                // to a subject this job did not create, and that has to
                // reach a human. `Outcome` has no success-carrying variant,
                // so the controller's own log is the only channel a
                // succeeding node has today.
                Ok(BridgeResponse::Healed) => {
                    tracing::warn!(
                        %agent,
                        "swarm jobq: create_identity marked an EXISTING identity as an \
                         agent — verify this name did not belong to a person"
                    );
                    Outcome::Done
                }
                Ok(_) => Outcome::Done,
                Err(e) => Outcome::Failed(format!("{e:#}")),
            },
        },
        SwarmNodeKind::CreateRepo { 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.create_repo(&agent).await {
                Ok(full_name) => {
                    tracing::info!(%full_name, "swarm jobq: create_repo done");
                    Outcome::Done
                }
                Err(e) => Outcome::Failed(format!("{e:#}")),
            },
        },
        SwarmNodeKind::CreateForgeUser { 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.ensure_agent_user(&agent).await {
                Ok(()) => Outcome::Done,
                Err(e) => Outcome::Failed(format!("{e:#}")),
            },
        },
        SwarmNodeKind::AddRepoMember { 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(&agent, &agent).await {
                Ok(()) => Outcome::Done,
                Err(e) => Outcome::Failed(format!("{e:#}")),
            },
        },
        SwarmNodeKind::InitAgentConfigRepo { 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(&agent, &agent).await {
                Ok(()) => Outcome::Done,
                Err(e) => Outcome::Failed(format!("{e:#}")),
            },
        },
        SwarmNodeKind::TriggerDeploy { hive, agent } => match deps.queue {
            None => Outcome::Failed(
                "no swarm queue is configured on this host, so no hive can be told to deploy"
                    .to_owned(),
            ),
            Some(client) => publish_deploy(&client, &hive, &agent).await,
        },
    };
    (builder, outcome)
}

/// Publish one deploy request and report whether it left this process.
///
/// The flush is not belt-and-braces: `publish` hands the message to the
/// client's write buffer and returns, so a node that reported `Done` on
/// that alone would be claiming a delivery it has no evidence for — the
/// same ordering `webhook::announce_knowledge_change` documents.
async fn publish_deploy(
    client: &async_nats::Client,
    hive: &str,
    agent: &str,
) -> hive_jobq::scheduler::Outcome {
    use hive_jobq::scheduler::Outcome;

    // One subject per hive, so the other hives are never woken by this.
    let subject = swarm_queue_client::deploy_subject(hive);
    let request = swarm_queue_client::DeployRequest {
        agent: agent.to_owned(),
    };
    let payload = match serde_json::to_vec(&request) {
        Ok(payload) => payload,
        Err(e) => return Outcome::Failed(format!("encoding the deploy request failed: {e}")),
    };
    if let Err(e) = client.publish(subject.clone(), payload.into()).await {
        return Outcome::Failed(format!("publishing to {subject} failed: {e}"));
    }
    if let Err(e) = client.flush().await {
        return Outcome::Failed(format!(
            "flushing the deploy event to {subject} failed: {e}"
        ));
    }
    tracing::info!(%subject, %hive, %agent, "swarm jobq: deploy requested");
    Outcome::Done
}

/// 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"),
        (name = "webhook", description = "swarm-wide forge webhook receipt"),
        (name = "repos", description = "forge repo browsing (swarm-ui's issue-report page)"),
    )
)]
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>>,
    /// Publishes the agent set this swarm declares for each hive.
    ///
    /// `None` in exactly the state `status` is: no swarm queue was wired
    /// up, so there is nowhere to publish a declaration to. Shares that
    /// reader's connection rather than opening a second one.
    wanted: Option<Arc<wanted::WantedWriter>>,
    /// Per-agent status, sharing `status`'s connection — same "one queue
    /// connection, several consumers" rationale as `wanted` above. `None`
    /// in exactly the state `status` is.
    agent_status: Option<Arc<agent_status::AgentStatusReader>>,
    /// 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.
    jobq: Arc<Mutex<hive_jobq::scheduler::Scheduler<SwarmNodeKind, SwarmResourceKind>>>,
    /// The identity store's only reader, for `GET /api/agents`.
    ///
    /// `None` when no bridge is wired up, which is the one state in which
    /// the roster cannot be answered at all — the same shape as `status`
    /// above, and for the same reason: an empty roster and an unreadable
    /// one are different facts.
    ///
    /// ⚠️ The two `/api/agents` verbs treat this differently on purpose.
    /// **POST does not consult it**: creation only queues a job, and
    /// whether a bridge exists is `run_swarm_node`'s concern (it holds its
    /// own clone from `spawn_jobq_worker`), so a request still queues
    /// cleanly on a bridge-less host and fails loud once claimed. **GET
    /// has nowhere to defer to** — there is no job, only an answer it
    /// either has or does not.
    auth: Option<Arc<auth::AuthBridge>>,
    /// HMAC secret for swarm-wide forge webhooks, loaded once at startup.
    /// `None` when it could not be read or created — the webhook endpoint
    /// then refuses every delivery with 503 rather than admitting one it
    /// cannot verify. Deliberately not fatal to startup: nothing is
    /// registered against that endpoint yet, and the rest of this daemon's
    /// surface is unaffected. See [`webhook::load_or_generate_secret`].
    ///
    /// `Arc<str>`, not `Arc<String>`: the value is never mutated after
    /// startup, and this way `as_deref()` yields the `&str` the verifier
    /// takes without a second hop through `String`.
    webhook_secret: Option<Arc<str>>,
    /// Last successful `agent-configs/*` scan, kept current by
    /// [`config_pr::spawn`]. `None` when this host has no forge configured
    /// — same "absent means don't ask" shape as `status` and `forge` above,
    /// not a startup failure.
    config_prs: Option<Arc<config_pr::ConfigPrCache>>,
    /// The swarm's human display name (`services.hyperhive.swarm.name`),
    /// loaded once at startup (`load_swarm_name`). `None` when the
    /// operator never set it — a swarm without a display name is a
    /// supported, if less friendly, state, not a startup failure. Same
    /// `Arc<str>` rationale as `webhook_secret`: never mutated, so a
    /// clone per request is just a refcount bump.
    swarm_name: Option<Arc<str>>,
    /// The forge client itself, for the read-only repo/issue-report
    /// routes (`GET /api/repos`, `GET /api/repos/{org}/{repo}/issue-report`)
    /// — distinct from `config_prs`, which holds a *cache* built off this
    /// same client rather than the client. `None` under the same
    /// "no forge configured on this host" shape every other
    /// forge-backed field here uses.
    forge: Option<Arc<forge::Client>>,
}

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

/// The swarm's agent roster — every agent the swarm holds an identity for.
///
/// The identity store **is** the roster rather than one source to assemble
/// one from: an agent without a swarm identity is not a swarm agent, so
/// there is no hive-side list to merge in and no reconciliation to do.
///
/// Deliberately says nothing about health. A roster is the set other views
/// are *complete against* — it is what makes "this agent has never reported"
/// expressible at all, and that distinction only survives while the declared
/// set and the reported set stay separate.
///
/// A hive-less swarm answers `[]`; a swarm with no bridge answers 503. Those
/// are different facts and collapsing them would render an unreadable store
/// as an empty one.
#[utoipa::path(
    get,
    path = "/api/agents",
    responses(
        (status = 200, description = "every agent the swarm holds an identity for", body = Vec<String>),
        (status = 503, description = "no identity bridge is configured here, or its store could not be read", body = String),
    ),
    tag = "agents"
)]
async fn get_agents(State(state): State<AppState>) -> Result<Json<Vec<String>>, StatusUnavailable> {
    let Some(bridge) = state.auth.as_ref() else {
        return Err(StatusUnavailable(
            "no identity bridge is configured on this host".to_owned(),
        ));
    };
    match bridge.list_agent_identities().await {
        Ok(agents) => Ok(Json(agents)),
        Err(e) => {
            let detail = format!("{e:#}");
            tracing::warn!(error = %detail, "reading the agent roster failed");
            Err(StatusUnavailable(detail))
        }
    }
}

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

/// Env var the controller's NixOS module sets from
/// `services.hyperhive.swarm.name` — unset (rather than an empty string)
/// when the operator never configured it. Consumed by `GET /api/swarm`.
const NAME_ENV: &str = "SWARM_CONTROLLER_NAME";

/// Reads [`NAME_ENV`]. `None` on absence — no fallback-and-warn shape like
/// `load_hives`/`load_links` because there is nothing to parse and fail:
/// an unset env var and an operator who never named the swarm are the same
/// state, not an error.
fn load_swarm_name() -> Option<String> {
    std::env::var(NAME_ENV).ok()
}

/// Body of `GET /api/swarm`.
#[derive(Clone, Debug, PartialEq, Serialize, Deserialize, ToSchema)]
struct SwarmInfo {
    /// `None` when the operator never set `services.hyperhive.swarm.name`
    /// — a caller (swarm-ui's chrome) falls back to a generic label rather
    /// than treating this as an error.
    name: Option<String>,
}

/// The swarm's own display name, for UI chrome (page title, nav) that
/// wants to say which swarm it's showing — distinct from [`get_hives`],
/// which lists the *hives inside* the swarm, not the swarm itself.
#[utoipa::path(
    get,
    path = "/api/swarm",
    responses((status = 200, description = "the swarm's display name, if the operator set one", body = SwarmInfo)),
    tag = "hives"
)]
async fn get_swarm_info(State(state): State<AppState>) -> Json<SwarmInfo> {
    Json(SwarmInfo {
        name: state.swarm_name.as_deref().map(str::to_owned),
    })
}

/// 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.
///
/// It travels in `detail` of an RFC 9457 `application/problem+json` body
/// rather than as a bare string, so the cause is an addressable field
/// instead of the whole payload — see `error_problem` below.
struct StatusUnavailable(String);

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

/// Every error this daemon returns, in one shape.
///
/// RFC 9457 `application/problem+json` is the hive-wide contract for HTTP
/// error bodies (`docs/process/conventions.md`), and the operator UIs read `detail`
/// for display. A bare string forces the reader to treat the entire body as
/// the message, which is the difference between a UI that can offer "copy the
/// cause" and one that can only dump a response.
fn error_problem(status: axum::http::StatusCode, detail: &str) -> problem_details::ProblemDetails {
    problem_details::ProblemDetails::from_status_code(status).with_detail(detail)
}

/// The state to declare for one agent.
#[derive(Debug, Deserialize, ToSchema)]
struct SetAgentStateRequest {
    /// Typed as a string in the schema only — the parse is the real enum, so
    /// a value this build does not know is a 400 rather than a field that
    /// silently does nothing.
    #[schema(value_type = String, example = "up")]
    state: swarm_queue_client::wanted::AgentState,
}

/// One agent's line in a hive's declaration.
#[derive(Clone, Debug, Serialize, ToSchema)]
struct AgentDeclaration {
    agent: String,
    state: String,
}

fn render(declaration: &swarm_queue_client::wanted::HiveWanted) -> Vec<AgentDeclaration> {
    declaration
        .agents
        .iter()
        .map(|(agent, wanted)| AgentDeclaration {
            agent: agent.clone(),
            state: wanted.state.as_str().to_owned(),
        })
        .collect()
}

/// The declaration writer, sharing the status reader's connection.
///
/// The controller holds exactly one queue connection by design — see
/// `StatusReader::queue_client`. A second connect would double the
/// auth-callout traffic and give the two paths independent reconnect state,
/// so one could be serving while the other was still down.
fn wanted_writer(status: Option<&Arc<status::StatusReader>>) -> Option<Arc<wanted::WantedWriter>> {
    status.map(|s| Arc::new(wanted::WantedWriter::new(s.queue_client())))
}

/// The per-agent status reader, sharing the status reader's connection and
/// staleness threshold — same rationale as [`wanted_writer`], and the same
/// threshold on purpose: both buckets are fed by the same publisher cadence
/// (`hive-c0re`'s `PUBLISH_INTERVAL`), so two separately-configured
/// thresholds would just be two ways to get out of sync with one cadence.
fn agent_status_reader(
    status: Option<&Arc<status::StatusReader>>,
) -> Option<Arc<agent_status::AgentStatusReader>> {
    status.map(|s| {
        Arc::new(agent_status::AgentStatusReader::new(
            s.queue_client(),
            status::StatusReader::stale_after_from_env(),
        ))
    })
}

/// A hive name that is shaped like one and names a hive this swarm has.
///
/// Reports the status and the detail rather than a rendered
/// `ProblemDetails`: that type is 232 bytes, which makes every `Result` in
/// this path pay for the error case it usually does not take. The handlers
/// render at the boundary, where the body is actually needed.
fn swarm_hive(state: &AppState, hive: &str) -> Result<String, (axum::http::StatusCode, String)> {
    let hive = hive_types::Ident::parse(hive)
        .map_err(|reason| (axum::http::StatusCode::BAD_REQUEST, reason.to_owned()))?
        .into_string();
    // Shaped like a hive name, and actually one. Same two checks
    // `create_agent` makes, for the same reason: a typo otherwise publishes a
    // declaration under a key no hive will ever read.
    if !state.hives.iter().any(|h| h.name == hive) {
        return Err((
            axum::http::StatusCode::BAD_REQUEST,
            format!("{hive:?} is not a hive in this swarm"),
        ));
    }
    Ok(hive)
}

/// Both declaration handlers below take the same hive name and reject it the
/// same way, and need the writer besides.
fn declaration_target(
    state: &AppState,
    hive: &str,
) -> Result<(Arc<wanted::WantedWriter>, String), (axum::http::StatusCode, String)> {
    // Before the name checks, so a deployment with no queue answers 503
    // whatever the caller spelled.
    let writer = state.wanted.clone().ok_or_else(|| {
        (
            axum::http::StatusCode::SERVICE_UNAVAILABLE,
            "this deployment wired up no swarm queue, so there is nowhere to publish a declaration"
                .to_owned(),
        )
    })?;
    Ok((writer, swarm_hive(state, hive)?))
}

/// Declare what this swarm wants of one agent on one hive.
///
/// The whole declaration is returned as published, because the value is the
/// hive's entire agent map and a caller that changed one agent still wants to
/// render the rest.
#[utoipa::path(
    put,
    path = "/api/hives/{hive}/agents/{agent}/state",
    params(
        ("hive" = String, Path, description = "hive whose declaration this is"),
        ("agent" = String, Path, description = "agent to declare"),
    ),
    request_body = SetAgentStateRequest,
    responses(
        (status = 200, description = "the declaration as now published", body = Vec<AgentDeclaration>),
        (status = 400, description = "a name is not an identifier, the hive is not in this swarm, or the state is unknown (problem+json)", body = String),
        (status = 409, description = "the agent is already declared destroyed, a terminal state the request tries to move it off of (problem+json)", body = String),
        (status = 503, description = "no swarm queue is wired up (problem+json)", body = String),
        (status = 500, description = "the declaration could not be published (problem+json)", body = String),
    ),
    tag = "agents"
)]
async fn set_agent_state(
    State(state): State<AppState>,
    axum::extract::Path((hive, agent)): axum::extract::Path<(String, String)>,
    Json(req): Json<SetAgentStateRequest>,
) -> Result<Json<Vec<AgentDeclaration>>, problem_details::ProblemDetails> {
    let (writer, hive) =
        declaration_target(&state, &hive).map_err(|(s, d)| error_problem(s, &d))?;
    let agent = hive_types::Ident::parse(&agent)
        .map_err(|reason| error_problem(axum::http::StatusCode::BAD_REQUEST, reason))?
        .into_string();

    let declaration = writer.set(&hive, &agent, req.state).await.map_err(|e| {
        // A `TerminalStateError` is the caller's mistake, but not a malformed
        // request — the request is well-formed, it just conflicts with the
        // target resource's current (terminal) state, which per RFC 9110 is
        // what 409 exists for, not 400. Everything else here is the
        // pre-existing catch-all.
        if let Some(terminal) = e.downcast_ref::<wanted::TerminalStateError>() {
            return error_problem(axum::http::StatusCode::CONFLICT, &terminal.to_string());
        }
        tracing::warn!(hive = %hive, agent = %agent, error = %format!("{e:#}"), "declaring agent state failed");
        error_problem(
            axum::http::StatusCode::INTERNAL_SERVER_ERROR,
            &format!("{e:#}"),
        )
    })?;
    Ok(Json(render(&declaration)))
}

/// What this swarm currently declares for a hive.
///
/// Read back from the bucket rather than from a second copy kept here: the
/// published value is the record.
#[utoipa::path(
    get,
    path = "/api/hives/{hive}/wanted",
    params(("hive" = String, Path, description = "hive whose declaration to read")),
    responses(
        (status = 200, description = "the declaration, empty when nothing is published yet", body = Vec<AgentDeclaration>),
        (status = 400, description = "not an identifier, or not a hive in this swarm (problem+json)", body = String),
        (status = 503, description = "no swarm queue is wired up (problem+json)", body = String),
        (status = 500, description = "the declaration could not be read (problem+json)", body = String),
    ),
    tag = "agents"
)]
async fn get_hive_wanted(
    State(state): State<AppState>,
    axum::extract::Path(hive): axum::extract::Path<String>,
) -> Result<Json<Vec<AgentDeclaration>>, problem_details::ProblemDetails> {
    let (writer, hive) =
        declaration_target(&state, &hive).map_err(|(s, d)| error_problem(s, &d))?;
    let declaration = writer.view(&hive).await.map_err(|e| {
        tracing::warn!(hive = %hive, error = %format!("{e:#}"), "reading the declaration failed");
        error_problem(
            axum::http::StatusCode::INTERNAL_SERVER_ERROR,
            &format!("{e:#}"),
        )
    })?;
    Ok(Json(declaration.as_ref().map(render).unwrap_or_default()))
}

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

/// What each agent last said about itself, its open config PR if any, and
/// its declared wanted state if any, read at request time — the single
/// call swarm-ui's agent roster page fills its whole table from (mara,
/// 2026-09-02: "the view should be filled by a single backend call").
///
/// Every agent in the roster gets a row whether or not it has ever
/// reported — see the `agent_status` module for why, and for why its hive
/// comes from its own bucket key rather than a second lookup. `config_pr`
/// and `wanted` are both merged in here rather than inside
/// `agent_status::AgentStatusReader`: that module has no forge client or
/// wanted-state writer and is not the place to add one just for a single
/// field, whereas this handler already holds `AppState::config_prs` and
/// `AppState::wanted` alongside the roster-shaped status view — the one
/// spot that has all three without adding an avoidable dependency to a
/// reader that stays deliberately narrow. Both fields are absent on rows
/// where the underlying store isn't configured or has nothing on record,
/// same "absence is the answer" shape `get_config_prs` uses — not a
/// reason to fail the whole response over a field that already knows how
/// to be missing.
#[utoipa::path(
    get,
    path = "/api/agents/status",
    responses(
        (status = 200, description = "a row per agent, freshness + config PR derived now", body = Vec<agent_status::AgentStatusRow>),
        (status = 503, description = "no swarm queue is configured here, no identity bridge is configured, or either store could not be read", body = String),
    ),
    tag = "agents"
)]
async fn get_agents_status(
    State(state): State<AppState>,
) -> Result<Json<Vec<agent_status::AgentStatusRow>>, StatusUnavailable> {
    let Some(reader) = state.agent_status.as_ref() else {
        return Err(StatusUnavailable(
            "no swarm queue is configured on this host".to_owned(),
        ));
    };
    let Some(bridge) = state.auth.as_ref() else {
        return Err(StatusUnavailable(
            "no identity bridge is configured on this host".to_owned(),
        ));
    };
    let roster = bridge.list_agent_identities().await.map_err(|e| {
        let detail = format!("{e:#}");
        tracing::warn!(error = %detail, "reading the agent roster failed");
        StatusUnavailable(detail)
    })?;
    match reader.view(&roster, std::time::SystemTime::now()).await {
        Ok(mut rows) => {
            if let Some(cache) = state.config_prs.as_ref() {
                let mut config_prs = cache.snapshot();
                for row in &mut rows {
                    row.config_pr = config_prs.remove(&row.name);
                }
            }
            // `wanted` merges in the same way and for the same reason as
            // `config_pr` above — one read per distinct hive, not one per
            // agent, since a declaration is already a hive's whole agent
            // map.
            if let Some(writer) = state.wanted.as_ref() {
                // Owned `String` keys, not `&str` borrowed from `rows` —
                // `declarations` outlives the loop below that needs `rows`
                // mutably, so it cannot hold a live borrow into it.
                let hives: std::collections::BTreeSet<String> =
                    rows.iter().filter_map(|r| r.hive.clone()).collect();
                let mut declarations: std::collections::HashMap<
                    String,
                    swarm_queue_client::wanted::HiveWanted,
                > = std::collections::HashMap::new();
                for hive in hives {
                    match writer.view(&hive).await {
                        Ok(Some(declaration)) => {
                            declarations.insert(hive, declaration);
                        }
                        Ok(None) => {}
                        Err(e) => {
                            // Missing wanted state for one hive shouldn't
                            // fail the whole roster — same "absence is
                            // survivable" rule `config_pr` follows above.
                            tracing::warn!(
                                hive, error = %format!("{e:#}"),
                                "reading the wanted declaration failed; its agents show no wanted state"
                            );
                        }
                    }
                }
                for row in &mut rows {
                    row.wanted = row.hive.as_deref().and_then(|hive| {
                        declarations
                            .get(hive)?
                            .agents
                            .get(&row.name)
                            .map(|w| w.state.as_str().to_owned())
                    });
                }
            }
            Ok(Json(rows))
        }
        Err(e) => {
            let detail = format!("{e:#}");
            tracing::warn!(error = %detail, "reading the agent-status bucket failed");
            Err(StatusUnavailable(detail))
        }
    }
}

/// Body of `POST /api/agents` — the agent name to create, and the hive the
/// creation is aimed at. The repo name inside `forge::CONFIG_ORG` is the
/// same string as `name`: one repo per agent, named after it, same
/// convention `hive-c0re::forge` already uses for its own single-hive
/// `CreateRepo` path.
///
/// `hive` is an **address**, not an attribute of the agent: it is where a
/// deploy message goes over the queue. Nothing here writes it into the
/// agent's config — a config naming its own hive would be a second
/// statement of where the agent lives, free to drift from the queue that
/// actually delivers to it.
///
/// It is required rather than optional because it is only knowable here,
/// at creation, from the operator making the choice. An optional field
/// would have been the friendlier migration and would have left every
/// creation in the state this endpoint exists to avoid — one nothing can
/// address.
///
/// Validated here and carried no further: none of the nodes this endpoint
/// queues talks to a hive, so none of them needs the address. It reaches
/// its consumer when the node that *sends* a deploy message exists, and
/// that node takes it from this field. Settling the request shape now is
/// the point — it is the breaking half, and doing it once is cheaper for
/// every caller than doing it again later.
#[derive(Clone, Debug, Deserialize, ToSchema)]
struct CreateAgentRequest {
    name: String,
    hive: 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,
    /// Name collisions that were **allowed through**, phrased for the
    /// operator who just chose the name.
    ///
    /// The creation still happened — this is the "loud warning now, refuse
    /// after a grace period" step, so a caller that ignores this field gets
    /// exactly today's behaviour. It rides on the response rather than
    /// living only in the daemon's log because the person who can still fix
    /// the name in one keystroke is the one holding this response, and they
    /// are not reading the journal.
    ///
    /// Empty (and omitted from the JSON) in the normal case, so a client
    /// that never looks at it is not handed an empty array to reason about.
    #[serde(default, skip_serializing_if = "Vec::is_empty")]
    warnings: Vec<String>,
}

/// Queue the whole agent-creation job graph for `name` — two independent
/// roots, `CreateIdentity` and `CreateForgeUser`, since an authelia
/// subject and a forge account need nothing from each other. `CreateRepo`
/// runs once `CreateIdentity` succeeds; once THAT succeeds,
/// `AddRepoMember` and `InitAgentConfigRepo` both run off it — except
/// `AddRepoMember` also waits on `CreateForgeUser`, because adding a
/// collaborator that does not yet exist as a forge user is a Forgejo
/// validation error, not an idempotent no-op. `InitAgentConfigRepo` has no
/// such second dependency: seeding files uses this daemon's own forge
/// token, never the agent's, so it only ever needed the repo (mara, design
/// review: 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 asynchronously off the scheduler loop already running
/// (`spawn_jobq_worker`). The response reports `CreateIdentity`'s id, one
/// of the graph's two roots — a caller watches the whole thing settle via
/// `/api/jobq/graph`, which serves every root, not just this one.
///
/// `name` is validated with [`hive_types::Ident::parse`] before it becomes
/// `agent`/`repo` anywhere downstream — not the *only* gate (`CreateIdentity`
/// validates server-side too), but a few hops removed from where an
/// unvalidated name would do damage (`forge::seed_agent_config`
/// interpolates `agent` into a nix comment line). Fragile if a second
/// caller of these nodes ever appears; validating here closes it locally.
#[utoipa::path(
    post,
    path = "/api/agents",
    request_body = CreateAgentRequest,
    responses(
        (status = 200, description = "job chain queued", body = CreateAgentResponse),
        (status = 400, description = "`name` or `hive` is not a valid identifier, or `hive` is not in this swarm (problem+json)", body = String),
        (status = 500, description = "the job chain could not be queued (problem+json)", body = String),
    ),
    tag = "agents"
)]
async fn create_agent(
    State(state): State<AppState>,
    Json(req): Json<CreateAgentRequest>,
) -> Result<Json<CreateAgentResponse>, problem_details::ProblemDetails> {
    let agent = hive_types::Ident::parse(&req.name)
        .map_err(|reason| error_problem(axum::http::StatusCode::BAD_REQUEST, reason))?
        .into_string();

    // Two checks, and the second is the one that makes the field worth
    // having: `Ident::parse` says the string is *shaped* like a hive name,
    // and the roster says it *is* one. Without the roster check a typo is
    // accepted, the agent gets created, and nothing notices until a deploy
    // message is addressed to a hive that does not exist — by which point
    // the operator who could have corrected it in one keystroke is gone.
    //
    // `state.hives` is the swarm directory loaded from `SWARM_CONTROLLER_HIVES`
    // at startup and never mutated, so this is a scan of a handful of
    // entries against a value an operator just chose.
    // The name is shaped like an identifier; these two ask whether it is
    // *available*. Both WARN rather than refuse: an operator with agents
    // already created under a colliding name would otherwise be unable to
    // re-run creation at all, so the refusal comes after a grace period,
    // once the warning has had time to be seen.
    //
    // Collected rather than logged-and-dropped — see `CreateAgentResponse`.
    let mut warnings = Vec::new();
    // The blacklist comes from nix via `HIVE_RESERVED_NAMES` — one file, read
    // by this daemon, by hive-c0re and by the swarm collector's own owner
    // assertion. An UNSET variable means this process was never told, which
    // is not the same as "no name is reserved": staying quiet there would be
    // a check that reports clean because it could not run.
    let raw = hive_types::reserved_names_raw();
    match raw.as_deref().map(hive_types::parse_reserved_names) {
        None => {
            tracing::error!(
                var = hive_types::RESERVED_NAMES_ENV,
                "create_agent: reserved-name check could not run — variable not set"
            );
            warnings.push(format!(
                "the reserved-name check did not run: {} is unset, so {agent:?} was accepted \
                 without being checked against the protocol literals",
                hive_types::RESERVED_NAMES_ENV
            ));
        }
        Some(reserved) if hive_types::is_reserved_name(&agent, &reserved) => {
            // A protocol literal: the message layer already produces this
            // name as a sender or recipient, so wakes from the component and
            // messages from the agent become the same broker row.
            tracing::warn!(agent = %agent, "create_agent: reserved name");
            warnings.push(format!(
                "agent name {agent:?} is a reserved protocol name — messages from this agent will \
                 be indistinguishable from hyperhive's own; this will become an error"
            ));
        }
        Some(_) => {}
    }

    let hive = hive_types::Ident::parse(&req.hive)
        .map_err(|reason| error_problem(axum::http::StatusCode::BAD_REQUEST, reason))?
        .into_string();

    // The roster is the same source the `hive` field is validated against
    // immediately below — one scan, asked of the other field. A hive and an
    // agent sharing a name collide in the swarm's own directory keys, which
    // is what makes a pipeline and a label vanish with no failed assertion.
    if state.hives.iter().any(|h| h.name == agent) {
        let detail = format!(
            "agent name {agent:?} is also a hive in this swarm — they share the swarm's directory \
             keys; this will become an error"
        );
        tracing::warn!(agent = %agent, "create_agent: name collides with a hive");
        warnings.push(detail);
    }
    if !state.hives.iter().any(|h| h.name == hive) {
        let known: Vec<&str> = state.hives.iter().map(|h| h.name.as_str()).collect();
        // Name the hives that *would* work: the caller is an operator who
        // just picked one, and "not in this swarm" without the roster
        // leaves them guessing at a typo they cannot see.
        let known = if known.is_empty() {
            "(none configured)".to_owned()
        } else {
            known.join(", ")
        };
        let detail = format!("hive {hive:?} is not in this swarm — known hives: {known}");
        return Err(error_problem(axum::http::StatusCode::BAD_REQUEST, &detail));
    }

    let mut sched = state
        .jobq
        .lock()
        .unwrap_or_else(std::sync::PoisonError::into_inner);
    let ids = sched
        .insert_job(None, |b| {
            let create_identity = b.node(SwarmNodeKind::CreateIdentity {
                agent: agent.clone(),
            });
            // A second, independent root: a forge user needs neither an
            // authelia subject nor an existing repo, so it does not chain
            // off `create_identity` (see the doc comment above).
            let create_forge_user = b.node(SwarmNodeKind::CreateForgeUser {
                agent: agent.clone(),
            });
            let create_repo = b
                .node(SwarmNodeKind::CreateRepo {
                    agent: agent.clone(),
                })
                .after_ok(create_identity);
            // `AddRepoMember` needs both parents: the repo to add a
            // collaborator to, and the forge user to add as one — adding a
            // nonexistent user is a Forgejo validation error, not
            // an idempotent no-op. `InitAgentConfigRepo` needs only the
            // repo — see the doc comment above for why.
            let _add_repo_member = b
                .node(SwarmNodeKind::AddRepoMember {
                    agent: agent.clone(),
                })
                .after_ok(create_repo)
                .after_ok(create_forge_user);
            let init_config = b
                .node(SwarmNodeKind::InitAgentConfigRepo {
                    agent: agent.clone(),
                })
                .after_ok(create_repo);
            // Last, and specifically after the config repo is seeded: the
            // hive deploys by reading that repo, so a deploy asked for any
            // earlier would find nothing to build. This is the edge that
            // makes creating an agent at swarm level actually put it on a
            // hive, rather than leaving a provisioned name nobody runs.
            let _trigger_deploy = b
                .node(SwarmNodeKind::TriggerDeploy {
                    hive: hive.clone(),
                    agent,
                })
                .after_ok(init_config);
            vec![create_identity.guid()]
        })
        .map_err(|e| {
            error_problem(
                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(),
        warnings,
    }))
}

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

/// `agent`'s open config PR, from [`config_pr::ConfigPrCache`] — kept
/// current by a webhook nudge (near-real-time) backstopped by a periodic
/// scan (worst case: up to one `config_pr::POLL_INTERVAL` stale if a
/// delivery was ever missed). Not a live forge read either way, so this
/// answers even when the forge itself is momentarily unreachable.
///
/// `200` with a `null` body means "no open PR" (or "no scan has completed
/// yet") — not distinguished, for the same reason
/// [`config_pr::ConfigPrCache::get`] doesn't: both read as "nothing to show"
/// to the swarm-ui config-PR panel this feeds, and a cache is a best-effort
/// read, not a source of truth callers should expect to disambiguate
/// against.
#[utoipa::path(
    get,
    path = "/api/agents/{name}/config-pr",
    params(("name" = String, Path, description = "agent name")),
    responses(
        (status = 200, description = "the agent's open config PR, or null if none / not scanned yet", body = Option<forge::ConfigPrStatus>),
        (status = 503, description = "no forge is configured on this host", body = String),
    ),
    tag = "agents"
)]
async fn get_agent_config_pr(
    State(state): State<AppState>,
    Path(name): Path<String>,
) -> Result<Json<Option<forge::ConfigPrStatus>>, StatusUnavailable> {
    let Some(cache) = state.config_prs.as_ref() else {
        return Err(StatusUnavailable(
            "no forge is configured on this host".to_owned(),
        ));
    };
    Ok(Json(cache.get(&name)))
}

/// Every agent with an open config PR, in one response — the bulk
/// counterpart to [`get_agent_config_pr`]. swarm-ui's config-PR table needs
/// every agent's status to render, and fetching them one at a time doesn't
/// scale and isn't how the rest of swarm-ui's single-fetch pages (jobq,
/// hives status) work.
///
/// Deliberately **not** `/api/agents/config-prs`: agent names are
/// user-specified (any string `hive_types::Ident` accepts), so a literal
/// path segment sitting where a `{name}` capture could plausibly also want
/// to live is a real, not theoretical, clash risk the moment someone names
/// an agent `config-prs` — per mara's review call, closed off by
/// construction rather than relying on axum's static-route-priority
/// tiebreak to paper over it.
///
/// Only agents with a currently-open PR are present — same "absence is the
/// answer" shape [`get_agent_config_pr`]'s `null` uses, just at map-entry
/// granularity instead of per-request.
#[utoipa::path(
    get,
    path = "/api/config-prs",
    responses(
        (status = 200, description = "agent name -> open config PR, only agents with one present", body = std::collections::HashMap<String, forge::ConfigPrStatus>),
        (status = 503, description = "no forge is configured on this host", body = String),
    ),
    tag = "agents"
)]
async fn get_config_prs(
    State(state): State<AppState>,
) -> Result<Json<std::collections::HashMap<String, forge::ConfigPrStatus>>, StatusUnavailable> {
    let Some(cache) = state.config_prs.as_ref() else {
        return Err(StatusUnavailable(
            "no forge is configured on this host".to_owned(),
        ));
    };
    Ok(Json(cache.snapshot()))
}

/// The swarm's own public base URL, as the forge must address it.
///
/// Set by `swarm-controller.nix` **only when this host actually serves the
/// swarm vhost** that carries the `/webhook/forge/` location. Absent means
/// "the endpoint is not reachable from outside", and the right response to
/// that is to register nothing: a hook pointing at a URL nothing answers is
/// worse than no hook, because forgejo records failed deliveries against a
/// registration that looks configured.
const PUBLIC_URL_ENV: &str = "SWARM_CONTROLLER_PUBLIC_URL";

/// Register the swarm-wide forge hooks against this controller, in the
/// background.
///
/// Detached rather than awaited, and never fatal: the forge may be slow or
/// briefly down at boot, and none of the daemon's other routes depend on a
/// hook existing.
///
/// Retried on a backoff rather than left to the next process start. Losing
/// that race is the *common* case, not an edge one — the controller and the
/// forge come up together on a rebuild, and a deploy was measured where both
/// consecutive starts got `502 Bad Gateway` from the gateway because forgejo
/// was not serving yet. Nothing schedules another start, so "the next restart
/// retries" can leave a swarm with no hooks registered for as long as the
/// daemon keeps running — and the failure is silent at both ends, because the
/// forge has nothing to report about a call that never arrived.
///
/// Silently does nothing when any of the three preconditions is missing —
/// each is a legitimate deployment shape (no forge here, no state directory
/// to hold a secret, no public vhost), and each is already logged where it
/// is discovered.
fn register_swarm_webhooks(forge: Option<Arc<forge::Client>>, secret: Option<Arc<str>>) {
    let Some(forge) = forge else { return };
    let Some(secret) = secret else { return };
    let Ok(public_url) = std::env::var(PUBLIC_URL_ENV) else {
        tracing::info!(
            "{PUBLIC_URL_ENV} unset; not registering swarm-wide forge webhooks (the \
             endpoint is not published on this host)"
        );
        return;
    };
    tokio::spawn(async move {
        // Seconds to wait before each retry — a bounded schedule, not a poll
        // loop: it exists to outlast a forge that is slow to start, not to
        // re-register periodically. Registration is idempotent, so a repeat
        // costs one API call and changes nothing.
        const RETRY_DELAYS_S: [u64; 5] = [15, 30, 60, 120, 240];

        for (attempt, delay) in RETRY_DELAYS_S.iter().enumerate() {
            match forge.ensure_swarm_webhooks(&public_url, &secret).await {
                Ok(()) => return,
                Err(e) => tracing::info!(
                    error = %format!("{e:#}"),
                    attempt = attempt + 1,
                    retry_in_s = *delay,
                    "registering swarm-wide forge webhooks failed; retrying"
                ),
            }
            tokio::time::sleep(std::time::Duration::from_secs(*delay)).await;
        }

        // Last attempt after the final wait, so the schedule above reads as
        // "delay before the next try" rather than one entry meaning two things.
        if let Err(e) = forge.ensure_swarm_webhooks(&public_url, &secret).await {
            tracing::warn!(
                error = %format!("{e:#}"),
                attempts = RETRY_DELAYS_S.len() + 1,
                "registering swarm-wide forge webhooks failed; giving up until the next start"
            );
        }
    });
}

/// Connect to the swarm queue when this deployment wired one up, extracted
/// out of `main` purely to keep that function under clippy's line-count
/// lint — no behavior split, every comment below is unchanged from where
/// it used to sit inline.
///
/// 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.
async fn connect_status_reader() -> Result<Option<Arc<status::StatusReader>>> {
    let Some(cfg) = swarm_queue_client::QueueConfig::from_env("SWARM_CONTROLLER")? else {
        tracing::info!("no swarm queue configured; status aggregation is off");
        return Ok(None);
    };
    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");
            Ok(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"
            );
            Ok(None)
        }
    }
}

/// Clones `forge_client` for [`AppState::forge`] before handing the
/// original off to [`register_swarm_webhooks`], which takes it by value —
/// split out of `main` purely to keep that function under clippy's
/// line-count lint (same rationale as [`connect_status_reader`]'s own doc
/// comment), no behavior change.
fn keep_forge_for_state(
    forge_client: Option<Arc<forge::Client>>,
    webhook_secret: Option<Arc<str>>,
) -> Option<Arc<forge::Client>> {
    let state_forge = forge_client.clone();
    register_swarm_webhooks(forge_client, webhook_secret);
    state_forge
}

#[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")),
        )
        // This is a systemd-managed daemon — stdout always goes to journald,
        // never a human terminal, and journald doesn't strip ANSI escapes:
        // they land in victorialogs as raw byte-array spam otherwise.
        .with_ansi(false)
        .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");

    let status = connect_status_reader().await?;

    // 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: auth.clone(),
        forge: forge_client.clone(),
        queue: status.as_ref().map(|s| s.queue_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);
    // Bound to a named variable, not `_` — dropping the provider stops its
    // `PeriodicReader`, so it must live as long as `main` does (which it
    // does here: this binding never goes out of scope before the process
    // exits via `axum::serve(...).await` below).
    // Own `AuthenticatedHttpClient` instance, independent of `vcs_metrics`'s
    // — both read the same env vars, and a client is cheap enough
    // (one `reqwest::blocking::Client`) that sharing one across two
    // otherwise-independent exporters isn't worth the plumbing. `Ok` becomes
    // `Some`, `Err` becomes `None` — same "log and carry on" shape as the
    // rest of this function's optional wiring; a controller that can't
    // authenticate its jobq push still serves every other route.
    let jobq_http_client = crate::vcs_metrics::authenticated_http_client()
        .inspect_err(
            |e| tracing::warn!(error = ?e, "otel jobq-metrics: no authenticated http client"),
        )
        .ok()
        .map(|c| Box::new(c) as Box<dyn opentelemetry_http::HttpClient>);
    let _jobq_metrics_provider =
        hive_jobq_metrics::spawn_exporter(Arc::clone(&jobq), "swarm-controller", jobq_http_client);

    // Same "log and carry on" shape as the queue/bridge/forge wiring above.
    // A controller that cannot hold a webhook secret still serves every
    // other route; the webhook endpoint answers 503, which is the honest
    // answer rather than a silent accept.
    let webhook_secret = match webhook::load_or_generate_secret() {
        Ok(secret) => Some(Arc::from(secret)),
        Err(e) => {
            tracing::warn!(
                error = %format!("{e:#}"),
                "webhook secret unavailable; swarm-wide forge webhooks are off"
            );
            None
        }
    };

    let config_prs = forge_client.clone().map(config_pr::spawn);
    let state_forge = keep_forge_for_state(forge_client, webhook_secret.clone());

    let hives = load_hives();
    // Before serving, because a hive whose role does not exist cannot log in,
    // and one whose policy does not exist logs in able to read nothing —
    // either way it cannot collect what this daemon writes for it. A store
    // that is not up yet is retried in the background instead of being
    // abandoned, since the two of them boot together.
    let hive_names: Vec<String> = hives.iter().map(|h| h.name.clone()).collect();
    read_policy::ensure_hive_access(hive_names).await;

    let state = AppState {
        hives: Arc::new(hives),
        links: Arc::new(load_links()),
        wanted: wanted_writer(status.as_ref()),
        agent_status: agent_status_reader(status.as_ref()),
        status,
        jobq,
        webhook_secret,
        config_prs,
        swarm_name: load_swarm_name().map(Arc::from),
        auth,
        forge: state_forge,
    };

    let app = build_app(state);
    axum::serve(listener, app)
        .await
        .context("serving swarm-controller")
}

/// Every route this daemon serves, wired to its state.
///
/// Split out of `main` because the route list is the part that grows, and
/// `main` sat exactly on the `too_many_lines` limit — adding an endpoint
/// tripped a lint about the startup sequence, which is not where the change
/// was. A new route now costs one line here and none there.
fn build_app(state: AppState) -> axum::Router {
    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_swarm_info))
        .routes(routes!(get_jobq_graph))
        .routes(routes!(get_jobq_rollup))
        .routes(routes!(get_agent_config_pr))
        .routes(routes!(get_config_prs))
        .routes(routes!(create_agent))
        .routes(routes!(get_agents))
        .routes(routes!(get_agents_status))
        .routes(routes!(set_agent_state))
        .routes(routes!(matrix_account::put_matrix_account))
        .routes(routes!(get_hive_wanted))
        .routes(routes!(issue_report::get_repos))
        .routes(routes!(issue_report::get_issue_report_all))
        .routes(routes!(issue_report::get_issue_report))
        .routes(routes!(webhook::post_webhook_forge))
        .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`.
    router
        .route(
            "/api/openapi.json",
            get(move || async move { Json(api.clone()) }),
        )
        .with_state(state)
}

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

    /// The 503 this route returns is what the operator UIs' shared error
    /// component renders, so assert the RENDERED response rather than the
    /// `problem_details` crate: the contract a UI depends on is the content
    /// type plus a `detail` it can address, and a handler that built the
    /// value and returned it as a bare string would satisfy any test
    /// written against the type alone.
    #[tokio::test]
    async fn status_unavailable_renders_problem_json_with_the_cause_in_detail() {
        use axum::response::IntoResponse as _;

        let cause = "listing status bucket keys: timed out";
        let resp = StatusUnavailable(cause.to_owned()).into_response();

        assert_eq!(resp.status(), axum::http::StatusCode::SERVICE_UNAVAILABLE);
        let ct = resp
            .headers()
            .get(axum::http::header::CONTENT_TYPE)
            .and_then(|v| v.to_str().ok())
            .unwrap_or_default()
            .to_owned();
        assert!(
            ct.starts_with("application/problem+json"),
            "RFC 9457 media type, got {ct:?}"
        );

        let bytes = axum::body::to_bytes(resp.into_body(), 64 * 1024)
            .await
            .expect("body reads");
        let v: serde_json::Value = serde_json::from_slice(&bytes).expect("problem+json parses");
        assert_eq!(v["status"], 503);
        // The cause is an addressable field, not the entire payload — that
        // distinction is the point of the change, so it is what is asserted.
        assert_eq!(v["detail"], cause);
    }

    /// 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.
    /// One roster, one agent name, two hives — the accepted one and a
    /// typo. Built per test so neither can see the other's queue.
    /// The scheduler handle these tests hold onto so they can assert on
    /// what the endpoint queued. Aliased because the full type is three
    /// nested generics deep and reads worse inline than named.
    type SharedSched = std::sync::Arc<
        std::sync::Mutex<
            hive_jobq::scheduler::Scheduler<super::SwarmNodeKind, super::SwarmResourceKind>,
        >,
    >;

    fn state_with_roster() -> (super::AppState, SharedSched) {
        let sched =
            std::sync::Arc::new(std::sync::Mutex::new(hive_jobq::scheduler::Scheduler::new(
                hive_jobq::Graph::new(),
                hive_jobq::resources::ResourceTable::new(),
            )));
        let state = super::AppState {
            hives: std::sync::Arc::new(vec![HiveEntry {
                name: "pr1ma".to_owned(),
                domain: "pr1ma.example".to_owned(),
            }]),
            links: std::sync::Arc::new(Vec::new()),
            status: None,
            // No queue, for the same reason as `status`: these tests drive
            // agent creation, which publishes no declaration.
            wanted: None,
            agent_status: None,
            jobq: std::sync::Arc::clone(&sched),
            webhook_secret: None,
            config_prs: None,
            swarm_name: None,
            // No bridge: these tests drive agent *creation*, which queues a
            // job and never consults one. The roster read is the verb that
            // needs it, and it has its own test below.
            auth: None,
            forge: None,
        };
        (state, sched)
    }

    /// A swarm with no identity bridge cannot answer the roster, and must
    /// say so rather than answer `[]`.
    ///
    /// The distinction is the whole reason this endpoint exists: an empty
    /// roster means *no agents*, and a UI that renders "no agents" for a
    /// store it could not read is showing a fact nobody established. 503
    /// rather than 500 for the same reason the status route uses it — a
    /// bridge is wired up per deployment, so a caller retrying is right.
    #[tokio::test]
    async fn a_roster_with_no_bridge_refuses_rather_than_answering_empty() {
        use axum::response::IntoResponse as _;

        let (state, _sched) = state_with_roster();
        assert!(state.auth.is_none(), "the fixture must have no bridge");

        let err = super::get_agents(axum::extract::State(state))
            .await
            .expect_err("a bridge-less controller cannot produce a roster");
        let resp = err.into_response();
        assert_eq!(resp.status(), axum::http::StatusCode::SERVICE_UNAVAILABLE);

        let bytes = axum::body::to_bytes(resp.into_body(), 64 * 1024)
            .await
            .expect("body reads");
        let v: serde_json::Value = serde_json::from_slice(&bytes).expect("problem+json parses");
        // Asserted on the rendered body rather than the error value: what a
        // consumer can distinguish is what matters, and `[]` and this share
        // a type on the Rust side.
        assert_eq!(v["status"], 503);
        assert!(
            v["detail"]
                .as_str()
                .is_some_and(|d| d.contains("identity bridge")),
            "the cause must name what is missing, got {:?}",
            v["detail"]
        );
    }

    /// The roster check is the half that makes the recorded hive worth
    /// having, so assert it by EFFECT rather than by the message: a hive
    /// that is not in this swarm must be refused **before anything is
    /// queued**. A version that queued the graph and then complained would
    /// satisfy an assertion on the status code alone while still creating
    /// the agent — which is the failure this exists to stop.
    #[tokio::test]
    async fn a_hive_outside_the_roster_is_refused_before_anything_is_queued() {
        let (state, sched) = state_with_roster();

        let err = super::create_agent(
            axum::extract::State(state),
            axum::Json(super::CreateAgentRequest {
                name: "atlas".to_owned(),
                hive: "pr1maa".to_owned(),
            }),
        )
        .await
        .expect_err("a hive outside the roster must be refused");

        // Names what would have worked: an operator who just mistyped a
        // hive cannot see the roster from the error otherwise.
        let rendered = format!("{err:?}");
        assert!(
            rendered.contains("pr1ma"),
            "the refusal should name the known hives, got: {rendered}"
        );

        let queued = sched
            .lock()
            .unwrap_or_else(std::sync::PoisonError::into_inner)
            .graph()
            .nodes()
            .count();
        assert_eq!(queued, 0, "a refused creation must queue no work");
    }

    /// The control for the arm above: the same call with a hive that IS in
    /// the roster gets through and queues the graph. Without this, "refused"
    /// could equally mean the endpoint refuses everything.
    #[tokio::test]
    async fn a_hive_in_the_roster_is_accepted_and_queues_the_graph() {
        let (state, sched) = state_with_roster();

        // The response is bound rather than asserted on: node ids start at
        // zero, so every property I reached for first ("> 0") was a claim
        // about the id allocator rather than about this endpoint. What the
        // accept arm is actually for is the queue count below.
        let _queued = super::create_agent(
            axum::extract::State(state),
            axum::Json(super::CreateAgentRequest {
                name: "atlas".to_owned(),
                hive: "pr1ma".to_owned(),
            }),
        )
        .await
        .expect("a hive in the roster must be accepted");

        let queued = sched
            .lock()
            .unwrap_or_else(std::sync::PoisonError::into_inner)
            .graph()
            .nodes()
            .count();
        assert!(queued > 0, "an accepted creation must queue work");
    }

    #[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 {
                    agent: "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,
            queue: 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
        );
    }

    /// Sibling of the `CreateRepo` test above, same shape: drives
    /// `SwarmNodeKind::CreateForgeUser` through the real scheduler with no
    /// forge configured, asserting the graceful-absence-is-failure branch —
    /// this node's happy path needs a live forge, same caveat as
    /// `CreateRepo`'s.
    #[tokio::test]
    async fn create_forge_user_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::CreateForgeUser {
                    agent: "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,
            queue: 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);
        }
    }

    /// Two states, not three like `load_hives`/`load_links` above: there is
    /// nothing to parse here, so no malformed-input branch exists to cover.
    ///
    /// SAFETY: single-threaded mutation of a process env var no other test
    /// in this crate reads; restored (removed) before returning.
    #[test]
    fn load_swarm_name_covers_missing_and_set() {
        unsafe {
            std::env::remove_var(NAME_ENV);
        }
        assert_eq!(
            load_swarm_name(),
            None,
            "unset env var is an unnamed swarm, not a startup failure"
        );

        unsafe {
            std::env::set_var(NAME_ENV, "constellat1on");
        }
        assert_eq!(load_swarm_name(), Some("constellat1on".to_string()));

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

#[cfg(test)]
mod create_agent_warning_tests {
    use super::CreateAgentResponse;

    /// The normal case must not grow a field. A client that has never
    /// heard of warnings should see the response it always saw — that is
    /// what makes "warn now, refuse later" a non-breaking first step.
    #[test]
    fn no_warnings_are_omitted_from_the_json_entirely() {
        let json = serde_json::to_string(&CreateAgentResponse {
            node_id: 7,
            warnings: vec![],
        })
        .unwrap();
        assert_eq!(json, r#"{"node_id":7}"#);
    }

    /// Presence arm — without it, a `skip_serializing_if` that dropped the
    /// field unconditionally would pass the test above.
    #[test]
    fn warnings_are_serialized_when_present() {
        let json = serde_json::to_string(&CreateAgentResponse {
            node_id: 7,
            warnings: vec!["name is reserved".to_owned()],
        })
        .unwrap();
        assert_eq!(json, r#"{"node_id":7,"warnings":["name is reserved"]}"#);
    }
}