//! 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_identity; mod agent_state_stream; 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 term_stream; 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 }, /// Mint `agent`'s own client certificate for the swarm secret store, /// publish it there, grant it, and log in with it. See /// `agent_identity::mint_and_verify` — including why this node does not /// report success on a write. /// /// Carries the hive for a different reason than `TriggerDeploy` does: /// not as an address, but because an agent's ACL document grants read on /// its hive's shared queue credential, so the document cannot be rendered /// without knowing which hive the agent belongs to. MintAgentIdentity { hive: String, agent: String }, /// Make sure `agent` holds a live forge access token in the swarm secret /// store, minting one with the forge's admin API when it does not. See /// `forge::agent_token` — including why a rotation is a delete then a /// create. /// /// Carries no hive: the token's store path has no hive segment, and the /// agent pulls it from wherever it runs. MintAgentForgeToken { agent: String }, /// Make sure `agent` holds a live token for its own account on the swarm's /// homeserver, creating the account with the swarm's appservice token /// when it does not. See `matrix_account::agent_token` — including why a /// stored token is read back before anything is minted. /// /// Carries no hive: the path it writes has no hive segment, so an agent /// that moves between hives keeps one matrix identity. MintAgentMatrixAccount { agent: String }, /// Declare `agent` on `hive` as `Paused` in the swarm's wanted-state /// store, so a freshly created agent does not start driving turns the /// moment it's deployed — the operator has to explicitly flip it to `Up`. /// Carries the hive for the same reason `TriggerDeploy`/`MintAgentIdentity` /// do: the wanted-state bucket is keyed per hive. SetAgentWanted { hive: String, 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::MintAgentIdentity { .. } => "mint_agent_identity".to_owned(), SwarmNodeKind::MintAgentForgeToken { .. } => "mint_agent_forge_token".to_owned(), SwarmNodeKind::MintAgentMatrixAccount { .. } => "mint_agent_matrix_account".to_owned(), SwarmNodeKind::SetAgentWanted { .. } => "set_agent_wanted".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 } | SwarmNodeKind::MintAgentForgeToken { agent } | SwarmNodeKind::MintAgentMatrixAccount { agent } => { serde_json::json!({ "agent": agent }) } SwarmNodeKind::TriggerDeploy { hive, agent } | SwarmNodeKind::MintAgentIdentity { hive, agent } | SwarmNodeKind::SetAgentWanted { 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>, forge: Option>, /// 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, /// The authority agent client leaves are issued from, loaded once at /// startup because it holds a private key and a per-node re-read would be /// a per-node chance to read one. `None` on a host the operator has not /// given an authority — see `agent_identity::Authority::from_env`. agent_ca: Option>, /// The wanted-state writer, for nodes that declare what a hive should /// converge an agent to. Shares the status reader's queue connection — /// see `wanted_writer`. `None` exactly when no swarm queue is configured /// on this host, same as `queue`. wanted: Option>, /// Client-server API base of the swarm's homeserver, for the node that /// mints agents' accounts on it. `None` when /// `matrix_account::DEFAULT_HOMESERVER_ENV` is unset, which that node /// **fails** on rather than skips, same as every other field here. matrix_homeserver: Option>, } /// What [`SwarmNodeKind::SetAgentWanted`] declares a brand-new agent to be. /// /// A named constant rather than the literal inline, because this *is* the /// decision the node exists to carry — writing a declaration at all is the /// mechanism, `Paused` is the policy — and a test can pin a name where it /// cannot pin a value written into a KV bucket no unit test has a server for. const NEW_AGENT_WANTED_STATE: swarm_queue_client::wanted::AgentState = swarm_queue_client::wanted::AgentState::Paused; /// 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, deps: WorkerDeps, ) -> ( hive_jobq::builder::JobBuilder, 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::MintAgentIdentity { hive, agent } => { mint_identity(deps.agent_ca.as_deref(), &agent, &hive).await } SwarmNodeKind::MintAgentForgeToken { agent } => mint_forge_token(deps.forge, &agent).await, SwarmNodeKind::MintAgentMatrixAccount { agent } => { mint_matrix_account(deps.matrix_homeserver.as_deref(), &agent).await } SwarmNodeKind::SetAgentWanted { hive, agent } => match deps.wanted { None => Outcome::Failed( "no swarm queue is configured on this host, so no wanted-state \ declaration can be written" .to_owned(), ), Some(writer) => declare_new_agent(&writer, &hive, &agent).await, }, 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) } /// The `MintAgentIdentity` arm, lifted out so `run_swarm_node` stays under /// `clippy::too_many_lines`. async fn mint_identity( authority: Option<&agent_identity::Authority>, agent: &str, hive: &str, ) -> hive_jobq::scheduler::Outcome { use hive_jobq::scheduler::Outcome; let Some(authority) = authority else { return Outcome::Failed( "no agent certificate authority is configured on this host \ (SWARM_CONTROLLER_AGENT_CA_FILE / SWARM_CONTROLLER_AGENT_CA_KEY_FILE unset), \ so this agent has no identity at the swarm secret store" .to_owned(), ); }; match agent_identity::mint_and_verify(authority, agent, hive).await { Ok(()) => Outcome::Done, Err(e) => Outcome::Failed(format!("{e:#}")), } } /// The `MintAgentForgeToken` arm, lifted out so `run_swarm_node` stays under /// `clippy::too_many_lines`. async fn mint_forge_token( forge: Option>, agent: &str, ) -> hive_jobq::scheduler::Outcome { use hive_jobq::scheduler::Outcome; let Some(client) = forge else { return Outcome::Failed( "no forge configured on this host (SWARM_CONTROLLER_FORGE_URL / \ SWARM_CONTROLLER_FORGE_TOKEN_FILE unset), so no agent forge token can be minted" .to_owned(), ); }; match client.ensure_agent_forge_token(agent).await { Ok(()) => Outcome::Done, Err(e) => Outcome::Failed(format!("{e:#}")), } } /// The `MintAgentMatrixAccount` arm, lifted out for the same reason. async fn mint_matrix_account( homeserver: Option<&str>, agent: &str, ) -> hive_jobq::scheduler::Outcome { use hive_jobq::scheduler::Outcome; // Loud on purpose: an agent quietly created without a matrix account is // an agent nothing can talk to, and no hive mints one any more. let Some(base) = homeserver else { return Outcome::Failed(format!( "no matrix homeserver configured on this host ({} unset), so no agent \ matrix account can be minted", matrix_account::DEFAULT_HOMESERVER_ENV )); }; match matrix_account::agent_token::ensure_agent_matrix_account(base, agent).await { Ok(()) => Outcome::Done, Err(e) => Outcome::Failed(format!("{e:#}")), } } /// Declare a brand-new agent at [`NEW_AGENT_WANTED_STATE`] — unless it turns /// out not to be new: an agent that already has a declaration (other than /// `Destroyed`, which this treats as reusable) is left alone, so a retried /// or migrating create is idempotent rather than pausing an already-running /// agent. `wanted.rs` itself has no such carve-out, so this node makes the /// call. async fn declare_new_agent( writer: &wanted::WantedWriter, hive: &str, agent: &str, ) -> hive_jobq::scheduler::Outcome { use hive_jobq::scheduler::Outcome; use swarm_queue_client::wanted::AgentState; let existing = match writer.view(hive).await { Ok(declaration) => declaration.and_then(|d| d.agents.get(agent).map(|a| a.state)), Err(e) => return Outcome::Failed(format!("{e:#}")), }; if let Some(state) = existing && state != AgentState::Destroyed { tracing::debug!( hive, agent, ?state, "agent already declared, leaving its wanted state alone" ); return Outcome::Done; } match writer.set(hive, agent, NEW_AGENT_WANTED_STATE).await { Ok(_) => Outcome::Done, Err(e) => Outcome::Failed(format!("{e:#}")), } } /// 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>>, 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)"), (name = "forge", description = "forge site admins (`swarmctl forge make-admin`)"), ) )] 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 \"", 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>, /// Loaded once at startup (`load_links`); same synchronization story /// as `hives`. links: Arc>, /// `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>, /// 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>, /// 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>, /// 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>>, /// 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 need 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. POST reads /// it only for a name that breaks a naming rule, to tell a new agent /// from an existing one, and warns when it cannot (`name_verdict`). **GET /// has nowhere to defer to** — there is no job, only an answer it /// either has or does not. `POST /api/forge/users/{name}/admin` reads /// it for every name and refuses when it cannot (`admin_unless_agent`). auth: Option>, /// 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`, not `Arc`: 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>, /// 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>, /// 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` rationale as `webhook_secret`: never mutated, so a /// clone per request is just a refcount bump. swarm_name: Option>, /// The forge client itself, for the read-only repo/issue-report /// routes (`GET /api/repos`, `GET /api/repos/{org}/{repo}/issue-report`) /// and `POST /api/forge/users/{name}/admin` — 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>, } /// 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 { 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)), tag = "hives" )] async fn get_hives(State(state): State) -> Json> { 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), (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) -> Result>, 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 { 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)), tag = "links" )] async fn get_links(State(state): State) -> Json> { 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 { 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, } /// 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) -> Json { 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 status a [`wanted::WantedWriter`] failure should answer with. /// /// `WantedWriter::view`/`set` call `ensure_connected` internally and /// propagate through `anyhow`, so the concrete /// [`swarm_queue_client::Error`] survives underneath but is not the type in /// hand — `swarm_queue_client`'s own doc says a caller "gets a type it can /// match on", which means downcasting back to it rather than string-matching /// the rendered message. `NotConnected` is a retryable condition (the queue /// is reconnecting, or hasn't yet) and every other queue-backed route here /// already answers 503 for it; anything else is a genuine failure to publish /// or read, which stays 500. fn wanted_error_status(e: &anyhow::Error) -> axum::http::StatusCode { match e.downcast_ref::() { Some(swarm_queue_client::Error::NotConnected(_)) => { axum::http::StatusCode::SERVICE_UNAVAILABLE } _ => axum::http::StatusCode::INTERNAL_SERVER_ERROR, } } /// 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 { declaration .agents .iter() .map(|(agent, wanted)| AgentDeclaration { agent: agent.clone(), state: <&str>::from(wanted.state).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>) -> Option> { 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>, ) -> Option> { 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 { 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, 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), (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 = 503, description = "no swarm queue is wired up, or it is not connected (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, axum::extract::Path((hive, agent)): axum::extract::Path<(String, String)>, Json(req): Json, ) -> Result>, 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| { tracing::warn!(hive = %hive, agent = %agent, error = %format!("{e:#}"), "declaring agent state failed"); error_problem(wanted_error_status(&e), &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), (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, or it is not connected (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, axum::extract::Path(hive): axum::extract::Path, ) -> Result>, 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(wanted_error_status(&e), &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 = 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, ) -> Result>, 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), (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, ) -> Result>, 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 = 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| <&str>::from(w.state).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, } /// 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, `name` is new and reserved or refused by the forge, 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, Json(req): Json, ) -> Result, 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; the checks below ask whether it // is *available*. An operator with agents already created under a // colliding name must still be able to re-run creation, so a name that // breaks a naming rule is refused only when it is new — see // `name_verdict`. The hive collision further down still only warns, on // its grace period, once the warning has had time to be seen. // // Collected rather than logged-and-dropped — see `CreateAgentResponse`. let mut warnings = Vec::new(); let broken = broken_name_rules( &agent, hive_types::reserved_names_raw().as_deref(), &mut warnings, ); if !broken.is_empty() { let existing = in_roster(state.auth.as_deref(), &agent).await; name_verdict(&agent, &broken, existing, &mut warnings) .map_err(|detail| error_problem(axum::http::StatusCode::BAD_REQUEST, &detail))?; } 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| declare_agent_job(b, &agent, &hive)) .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, })) } /// Every naming rule `agent` breaks, each as a sentence naming the rule. /// /// `reserved` is [`hive_types::reserved_names_raw`]: the list nix owns in /// `nix/reserved-names.nix`, read by this daemon, by hive-c0re and by the /// swarm collector's own owner assertion. `None` 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, so it goes /// onto `warnings` instead. fn broken_name_rules( agent: &str, reserved: Option<&str>, warnings: &mut Vec, ) -> Vec { let mut broken = Vec::new(); match reserved.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 reserved names", hive_types::RESERVED_NAMES_ENV )); } Some(reserved) if hive_types::is_reserved_name(agent, &reserved) => { broken.push(format!( "agent name {agent:?} is reserved (nix/reserved-names.nix): hyperhive's message \ layer already uses it, or the forge refuses it as a username" )); } Some(_) => {} } if let Some(rule) = hive_types::forge_username_violation(agent) { broken.push(format!( "agent name {agent:?} {rule}: the forge refuses it as a username" )); } broken } /// Whether `agent` is already in the swarm roster, or why that could not be /// told. `create_agent` asks only of a name that breaks a rule, so an /// ordinary creation still never waits on the bridge; `make_forge_admin` /// asks of every name. async fn in_roster(auth: Option<&auth::AuthBridge>, agent: &str) -> Result { let Some(auth) = auth else { return Err("no identity bridge is configured".to_owned()); }; auth.list_agent_identities() .await .map(|roster| roster.iter().any(|name| name == agent)) .map_err(|e| format!("{e:#}")) } /// Refuse a NEW agent whose name breaks a rule; warn for one that exists. /// /// Re-creating an agent under its own name is a migration path (mara, on /// the duplicate-name refusal), so an agent already in the roster keeps its /// name with a warning. When the roster cannot be read, a new name and an /// existing one look the same, and this warns rather than refuse: refusing /// would block that re-creation, and the forge still refuses the account one /// node in, as it did before. /// /// # Errors /// Every rule `agent` breaks, as one caller-ready sentence, when `agent` is /// not in the roster. fn name_verdict( agent: &str, broken: &[String], existing: Result, warnings: &mut Vec, ) -> Result<(), String> { match existing { Ok(false) => Err(format!("{}; choose another name", broken.join("; "))), Ok(true) => { tracing::warn!(agent, "create_agent: existing agent breaks a naming rule"); warnings.extend( broken .iter() .map(|rule| format!("{rule} — accepted only because {agent:?} already exists")), ); Ok(()) } Err(why) => { tracing::warn!( agent, why, "create_agent: naming rule broken, roster unreadable" ); warnings.extend(broken.iter().map(|rule| { format!( "{rule} — not refused, because whether {agent:?} already exists could not \ be checked: {why}" ) })); Ok(()) } } } /// The authority agent leaves are issued from, or `None` on a host that was /// given none. /// /// Same "log and carry on" shape as `main`'s other optional wiring: a /// controller with no agent authority still serves everything else, and /// `MintAgentIdentity` fails with a named reason rather than this process /// refusing to start. The `Err` arm is worth its own warning — half an /// authority, or a file that will not read, is a host that looks configured /// and mints nothing. fn load_agent_authority() -> Option> { match agent_identity::Authority::from_env() { Ok(authority) => authority.map(Arc::new), Err(e) => { tracing::warn!( error = %format!("{e:#}"), "agent certificate authority unusable; agents get no store identity here" ); None } } } /// The sub-DAG one agent creation is: the nodes, and the edges between them. /// /// A function rather than a closure inside [`create_agent`] so the endpoint's /// validation and the graph's shape can each be read without scrolling past /// the other — and so the one handle the response reports is returned from /// the place that decides which node it is. fn declare_agent_job( b: &hive_jobq::builder::JobBuilder, agent: &str, hive: &str, ) -> Vec { let create_identity = b.node(SwarmNodeKind::CreateIdentity { agent: agent.to_owned(), }); // 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.to_owned(), }); let create_repo = b .node(SwarmNodeKind::CreateRepo { agent: agent.to_owned(), }) .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.to_owned(), }) .after_ok(create_repo) .after_ok(create_forge_user); let init_config = b .node(SwarmNodeKind::InitAgentConfigRepo { agent: agent.to_owned(), }) .after_ok(create_repo); // The agent's own identity at the swarm secret store, minted and // published at swarm level so no hive ever needs the capability to mint // one. After `create_identity` because the certificate names a principal // the swarm has agreed exists — not because anything in the store reads // authelia. let mint_identity = b .node(SwarmNodeKind::MintAgentIdentity { hive: hive.to_owned(), agent: agent.to_owned(), }) .after_ok(create_identity); // The agent's own forge token, in the store before the hive deploys // the container that pulls it. After the forge user, because the token // is minted for that user. let mint_forge_token = b .node(SwarmNodeKind::MintAgentForgeToken { agent: agent.to_owned(), }) .after_ok(create_forge_user); // The agent's account on the swarm's homeserver. A root of its own: // creating it needs neither an authelia subject nor a forge user nor a // repo, and chaining it behind one of those would make an unrelated // failure look like a matrix failure. let mint_matrix = b.node(SwarmNodeKind::MintAgentMatrixAccount { agent: agent.to_owned(), }); // Declared before the deploy trigger so the pause is visible in the // wanted-state store before the hive brings the container up — see the // node's own doc comment for why "before", not just "eventually". Needs // no parent of its own: the declaration is written from the hive/agent // names known at request time, same as `create_forge_user`. let set_wanted = b.node(SwarmNodeKind::SetAgentWanted { hive: hive.to_owned(), agent: agent.to_owned(), }); // 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. // // `after_any` on the mint, not `after_ok`: a hive cannot pass down a // certificate the swarm has not published, so the deploy must not // overtake the mint — but a host with no authority configured must still // create agents exactly as it does today. `after_ok` there would turn an // unconfigured option into an agent nobody runs. // // The forge-token mint gets `after_any` for the same reason: a host with // no forge or no store must still create agents; only that node fails, // by name, and the agent's container fetches the token on its next // timer tick once one is minted. // // `after_ok` on the declaration, though: an agent deployed without its // pause landing first is the race this node exists to close, so a // declaration that did not land must not be deployed past. A host with no // queue at all fails this node — but `TriggerDeploy` has nothing to // publish to on such a host either, so the strict edge cancels a node // that could not have succeeded anyway. let _trigger_deploy = b .node(SwarmNodeKind::TriggerDeploy { hive: hive.to_owned(), agent: agent.to_owned(), }) .after_ok(init_config) .after_any(mint_identity) .after_any(mint_forge_token) // `after_any` for the same reason: the container reads this token // rather than producing it, so the deploy must not overtake the mint — // but a host with no homeserver configured must still create agents, // and only this node fails, by name. .after_any(mint_matrix) .after_ok(set_wanted); vec![create_identity.guid()] } /// 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, } /// 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)), tag = "jobq" )] async fn get_jobq_graph( State(state): State, axum::extract::Query(q): axum::extract::Query, ) -> Json> { 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 = 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)), tag = "jobq" )] async fn get_jobq_rollup(State(state): State) -> Json> { let sched = state .jobq .lock() .unwrap_or_else(std::sync::PoisonError::into_inner); let graph = sched.graph(); let roots: Vec = 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), (status = 503, description = "no forge is configured on this host", body = String), ), tag = "agents" )] async fn get_agent_config_pr( State(state): State, Path(name): Path, ) -> Result>, 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))) } /// Body of `POST /api/agents/{name}/identity`. #[derive(Deserialize, ToSchema)] struct MintAgentIdentityRequest { /// The hive this agent belongs to. /// /// Required, for the same reason [`CreateAgentRequest`]'s is: the /// credentials this mints name a hive, and the controller has nowhere to /// look one up — agents are created on hives at runtime and this daemon /// keeps no roster of which agent is where. An operator naming the wrong /// one would hand the agent subjects on a hive it does not run on, so it /// is asked for rather than guessed at. hive: String, } /// Success body of `POST /api/agents/{name}/identity`. #[derive(Clone, Debug, Serialize, ToSchema)] struct MintAgentIdentityResponse { /// The queued node, so a caller can follow it in the job view. node_id: u64, } /// Re-run the store-identity mint for an agent that already exists. /// /// **This route exists for backfill.** Agent creation is event-driven and /// there is no reconcile sweep at swarm level, so an agent created before a /// credential was part of the mint never gets one — nothing would ever come /// back around to it. Without a way to re-run the node by hand, the only way /// to give an existing agent its queue credential would be to delete and /// recreate the agent. /// /// A job node rather than an inline write, and specifically the *same* node /// `POST /api/agents` declares: a second code path that mints an identity is a /// second place for the four strings in [`agent_identity`] to disagree. /// /// ⚠️ **Not idempotent in one respect**: the node re-mints the agent's mTLS /// leaf, which the agent picks up on its next boot. Its queue secret is left /// alone — see [`agent_identity::mint_and_verify`] for which half is which and /// why. #[utoipa::path( post, path = "/api/agents/{name}/identity", params(("name" = String, Path, description = "agent name")), request_body = MintAgentIdentityRequest, responses( (status = 200, description = "mint queued", body = MintAgentIdentityResponse), (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 could not be queued (problem+json)", body = String), ), tag = "agents" )] async fn mint_agent_identity( State(state): State, Path(name): Path, Json(req): Json, ) -> Result, problem_details::ProblemDetails> { // Both names are interpolated into store paths and policy documents // downstream, so both are validated here as well as there. let agent = hive_types::Ident::parse(&name) .map_err(|reason| error_problem(axum::http::StatusCode::BAD_REQUEST, reason))? .into_string(); let hive = hive_types::Ident::parse(&req.hive) .map_err(|reason| error_problem(axum::http::StatusCode::BAD_REQUEST, reason))? .into_string(); if !state.hives.iter().any(|h| h.name == hive) { let known: Vec<&str> = state.hives.iter().map(|h| h.name.as_str()).collect(); let known = if known.is_empty() { "(none configured)".to_owned() } else { known.join(", ") }; return Err(error_problem( axum::http::StatusCode::BAD_REQUEST, &format!("hive {hive:?} is not in this swarm — known hives: {known}"), )); } // No reserved-name or collision warnings here, unlike `create_agent`: // those answer "is this name available", and this route is only ever // pointed at a name that was already taken. let mut sched = state .jobq .lock() .unwrap_or_else(std::sync::PoisonError::into_inner); let ids = sched .insert_job(None, |b| { vec![ b.node(SwarmNodeKind::MintAgentIdentity { hive: hive.clone(), agent: agent.clone(), }) .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(MintAgentIdentityResponse { node_id: id.get() })) } /// Success body of `POST /api/agents/{name}/forge-token`. #[derive(Clone, Debug, Serialize, ToSchema)] struct MintAgentForgeTokenResponse { /// The queued node, so a caller can follow it in the job view. node_id: u64, } /// Insert one job per agent: `CreateForgeUser`, then `MintAgentForgeToken` /// `after_ok` it — the same edge [`declare_agent_job`] uses. Returns the mint /// nodes' ids. /// /// The one place that node is queued outside agent creation: both the manual /// route below and `forge::agent_token::spawn`'s periodic pass come through /// here, so a backfilled mint is the same node a new agent gets. The user /// node is what lets the pass reach an agent that has a store identity but /// no forge account; for one that has an account it is a no-op, because /// `forge::Client::ensure_agent_user` folds "already exists" into success. fn queue_forge_token_mints( sched: &Mutex>, agents: Vec, ) -> Result> { let mut sched = sched .lock() .unwrap_or_else(std::sync::PoisonError::into_inner); let mut ids = Vec::with_capacity(agents.len()); for agent in agents { let queued = sched .insert_job(None, |b| { let create_forge_user = b.node(SwarmNodeKind::CreateForgeUser { agent: agent.clone(), }); vec![ b.node(SwarmNodeKind::MintAgentForgeToken { agent }) .after_ok(create_forge_user) .guid(), ] }) .map_err(|e| anyhow::anyhow!("{e}"))?; ids.extend(queued); } Ok(ids) } /// Insert one `MintAgentMatrixAccount` job per agent and return the nodes' /// ids. `matrix_account::agent_token::spawn`'s periodic pass comes through /// here, so a backfilled mint is the same node a new agent gets. fn queue_matrix_account_mints( sched: &Mutex>, agents: Vec, ) -> Result> { let mut sched = sched .lock() .unwrap_or_else(std::sync::PoisonError::into_inner); let mut ids = Vec::with_capacity(agents.len()); for agent in agents { let queued = sched .insert_job(None, |b| { vec![ b.node(SwarmNodeKind::MintAgentMatrixAccount { agent }) .guid(), ] }) .map_err(|e| anyhow::anyhow!("{e}"))?; ids.extend(queued); } Ok(ids) } /// The homeserver agents' own matrix accounts are minted on. Unset is this /// controller creating agents with no matrix account, which the mint node /// reports by name rather than `main` refusing to start. fn configured_matrix_homeserver() -> Option> { matrix_account::configured_default_homeserver() .filter(|v| !v.is_empty()) .map(Arc::from) } /// Start the matrix-account backfill when a homeserver is configured. Lifted /// out of `main` for `clippy::too_many_lines`. fn spawn_matrix_account_backfill( jobq: &Arc>>, homeserver: Option>, ) { let Some(base) = homeserver else { return; }; let sched = Arc::clone(jobq); matrix_account::agent_token::spawn(base, move |agents| { if let Err(e) = queue_matrix_account_mints(&sched, agents) { tracing::warn!(error = %format!("{e:#}"), "agent matrix accounts: queueing failed"); } }); } /// Start the forge's periodic passes — the swarm-wide objects and agents' /// tokens — when a forge is configured. Lifted out of `main` for /// `clippy::too_many_lines`. fn spawn_forge_workers( jobq: &Arc>>, forge_client: Option>, ) { let Some(client) = forge_client else { return; }; forge::objects::spawn(Arc::clone(&client)); let sched = Arc::clone(jobq); forge::agent_token::spawn(client, move |agents| { if let Err(e) = queue_forge_token_mints(&sched, agents) { tracing::warn!(error = %format!("{e:#}"), "agent forge tokens: queueing failed"); } }); } /// Check an agent's forge token now, and mint one if it is missing or stale. /// /// The periodic pass (`forge::agent_token::spawn`) does the same every five /// minutes for every agent with a store identity; this is for an operator who /// does not want to wait, or for an agent the pass skipped. Queues the same /// node agent creation does. Idempotent: a current token is left alone. #[utoipa::path( post, path = "/api/agents/{name}/forge-token", params(("name" = String, Path, description = "agent name")), responses( (status = 200, description = "mint queued", body = MintAgentForgeTokenResponse), (status = 400, description = "`name` is not a valid identifier (problem+json)", body = String), (status = 500, description = "the job could not be queued (problem+json)", body = String), ), tag = "agents" )] async fn mint_agent_forge_token( State(state): State, Path(name): Path, ) -> Result, problem_details::ProblemDetails> { let agent = hive_types::Ident::parse(&name) .map_err(|reason| error_problem(axum::http::StatusCode::BAD_REQUEST, reason))? .into_string(); let ids = queue_forge_token_mints(&state.jobq, vec![agent]).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(MintAgentForgeTokenResponse { node_id: id.get() })) } /// What `POST /api/forge/users/{name}/admin` found. #[derive(Clone, Copy, Debug, PartialEq, Eq, Serialize, Deserialize, ToSchema)] #[serde(rename_all = "snake_case")] enum ForgeAdminChange { Promoted, AlreadyAdmin, } /// Success body of `POST /api/forge/users/{name}/admin`. #[derive(Serialize, Deserialize, ToSchema)] struct MakeForgeAdminResponse { change: ForgeAdminChange, } /// Make an existing human forge account a site admin. Idempotent. /// /// Never creates the account: a human's is made by their first SSO login to /// the forge, and until then this answers 404. Refuses an agent's name: a /// site admin can create repos whatever its `max_repo_creation`, and an /// agent's account is locked to 0 so its token cannot create a repo and /// self-merge in it. #[utoipa::path( post, path = "/api/forge/users/{name}/admin", params(("name" = String, Path, description = "forge username")), responses( (status = 200, description = "the account is a site admin", body = MakeForgeAdminResponse), (status = 400, description = "`name` is not a valid identifier (problem+json)", body = String), (status = 404, description = "no such account: the user has not logged in via SSO yet (problem+json)", body = String), (status = 409, description = "`name` is an agent (problem+json)", body = String), (status = 500, description = "the forge refused the read or the edit (problem+json)", body = String), (status = 503, description = "no forge configured, or the agent roster could not be read (problem+json)", body = String), ), tag = "forge" )] async fn make_forge_admin( State(state): State, Path(name): Path, ) -> Result, problem_details::ProblemDetails> { let name = hive_types::Ident::parse(&name) .map_err(|reason| error_problem(axum::http::StatusCode::BAD_REQUEST, reason))? .into_string(); let Some(forge) = state.forge.clone() else { return Err(error_problem( axum::http::StatusCode::SERVICE_UNAVAILABLE, "no forge is configured on this host", )); }; let is_agent = in_roster(state.auth.as_deref(), &name).await; admin_unless_agent(&name, is_agent, || async { forge.make_site_admin(&name).await }) .await } /// Run `promote` only if `name` is known not to be an agent, and answer with /// what it found. /// /// An unreadable roster refuses too: a site admin is the one thing an agent's /// account must never become, so "could not tell" is not "no". async fn admin_unless_agent( name: &str, is_agent: Result, promote: F, ) -> Result, problem_details::ProblemDetails> where F: FnOnce() -> Fut, Fut: std::future::Future>, { match is_agent { Ok(false) => {} Ok(true) => { return Err(error_problem( axum::http::StatusCode::CONFLICT, &format!("{name:?} is an agent; an agent's forge account is never a site admin"), )); } Err(e) => { return Err(error_problem( axum::http::StatusCode::SERVICE_UNAVAILABLE, &format!("cannot tell whether {name:?} is an agent, so refusing: {e}"), )); } } let plan = promote().await.map_err(|e| { error_problem( axum::http::StatusCode::INTERNAL_SERVER_ERROR, &format!("{e:#}"), ) })?; let change = match plan { forge::site_admin::Plan::NoSuchUser => { return Err(error_problem( axum::http::StatusCode::NOT_FOUND, &format!( "the forge has no user {name:?}: the user has not logged in via SSO yet. \ The forge creates the account on their first authelia login; run this \ again after it" ), )); } forge::site_admin::Plan::AlreadyAdmin => ForgeAdminChange::AlreadyAdmin, forge::site_admin::Plan::Promote => ForgeAdminChange::Promoted, }; Ok(Json(MakeForgeAdminResponse { change })) } /// 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), (status = 503, description = "no forge is configured on this host", body = String), ), tag = "agents" )] async fn get_config_prs( State(state): State, ) -> Result>, 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>, secret: Option>) { 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>> { 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>, webhook_secret: Option>, ) -> Option> { let state_forge = forge_client.clone(); register_swarm_webhooks(forge_client, webhook_secret); state_forge } #[tokio::main] async fn main() -> Result<()> { hive_log::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()), agent_ca: load_agent_authority(), wanted: wanted_writer(status.as_ref()), matrix_homeserver: configured_matrix_homeserver(), }; let jobq = Arc::new(Mutex::new(hive_jobq::scheduler::Scheduler::new( hive_jobq::Graph::new(), hive_jobq::resources::ResourceTable::new(), ))); spawn_matrix_account_backfill(&jobq, deps.matrix_homeserver.clone()); 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); 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 } }; // Diagnostic only — `put_matrix_account` does not call this yet (see its // own doc comment), so this cannot change any request's outcome. Logged // so an operator can tell the config landed before the slice that // consults it does. if matrix_account::homeserver_or_configured_default( None, matrix_account::configured_default_homeserver(), ) .is_none() { tracing::debug!( "{}; matrix-account requests still need their own homeserver", matrix_account::DEFAULT_HOMESERVER_ENV ); } spawn_forge_workers(&jobq, forge_client.clone()); 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 = 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::::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!(mint_agent_identity)) .routes(routes!(mint_agent_forge_token)) .routes(routes!(make_forge_admin)) .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!(term_stream::stream_agent_term)) .routes(routes!(agent_state_stream::stream_agent_state)) .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, SetAgentStateRequest, StatusUnavailable, SwarmNodeKind, WorkerDeps, load_hives, load_links, load_swarm_name, run_swarm_node, wanted, }; 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, >, >; 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 build // the creation graph and read its shape, and never run the node // that would write a 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 consults one only for a rule-breaking name. 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"); } /// Forgejo v16.0.5's reserved usernames our charset can spell, plus the /// bare `-` — the forge section of `nix/reserved-names.nix`. Read from /// the list nix exports, so dropping one from the file reds this. const FORGE_REFUSED_LITERALS: [&str; 23] = [ "-", "admin", "api", "assets", "attachments", "avatar", "avatars", "captcha", "explore", "forgejo-actions", "ghost", "gitea-actions", "issues", "login", "metrics", "milestones", "notifications", "org", "pulls", "repo", "repo-avatars", "user", "v2", ]; /// The list nix renders, panicking when absent for the reason /// `hive-sh4re`'s drift test gives: a skip would read as green. fn nix_reserved() -> String { hive_types::reserved_names_raw() .expect("HIVE_RESERVED_NAMES is exported by nix/checks.nix and nix/devshell.nix") } #[test] fn a_new_agent_named_a_forge_refused_literal_is_refused() { let raw = nix_reserved(); for name in FORGE_REFUSED_LITERALS { let mut warnings = Vec::new(); let broken = super::broken_name_rules(name, Some(&raw), &mut warnings); let err = super::name_verdict(name, &broken, Ok(false), &mut warnings).expect_err(name); assert!(err.contains("nix/reserved-names.nix"), "{name:?}: {err}"); } } /// One case per shape rule, each named in the refusal. Checked against a /// list without them, so the literal half cannot carry this test. #[test] fn a_new_agent_breaking_a_forge_shape_rule_is_refused() { let over = "a".repeat(hive_types::FORGE_USERNAME_MAX_LEN + 1); for (name, rule) in [ ("-agent", "must not start with '-'"), ("my--agent", "must not contain '--'"), ("agent-", "must not end with '-'"), (over.as_str(), "must be 40 characters or fewer"), ] { let mut warnings = Vec::new(); let broken = super::broken_name_rules(name, Some("operator"), &mut warnings); let err = super::name_verdict(name, &broken, Ok(false), &mut warnings).expect_err(name); assert!(err.contains(rule), "{name:?}: {err}"); } } /// The control for both tests above, against the real list. #[test] fn an_ordinary_name_and_a_40_char_name_break_no_rule() { let raw = nix_reserved(); let at_cap = "a".repeat(hive_types::FORGE_USERNAME_MAX_LEN); for name in ["my-agent", at_cap.as_str()] { let mut warnings = Vec::new(); let broken = super::broken_name_rules(name, Some(&raw), &mut warnings); assert!(broken.is_empty(), "{name:?}: {broken:?}"); assert!(warnings.is_empty(), "{name:?}: {warnings:?}"); } } /// Re-creating an agent under its own name must keep working: a name /// already in the roster is warned about, never refused. #[test] fn an_existing_agent_keeps_a_rule_breaking_name_with_a_warning() { let mut warnings = Vec::new(); let broken = super::broken_name_rules("admin", Some("admin"), &mut warnings); super::name_verdict("admin", &broken, Ok(true), &mut warnings) .expect("an existing agent must not be refused"); assert_eq!(warnings.len(), 1, "{warnings:?}"); assert!(warnings[0].contains("already exists"), "{warnings:?}"); } /// A roster that cannot be read cannot tell new from existing, so it /// warns — and the warning says the check did not decide. #[test] fn an_unreadable_roster_warns_rather_than_refusing() { let mut warnings = Vec::new(); let broken = super::broken_name_rules("agent-", Some("operator"), &mut warnings); super::name_verdict( "agent-", &broken, Err("bridge down".to_owned()), &mut warnings, ) .expect("an unreadable roster must not refuse"); assert_eq!(warnings.len(), 1, "{warnings:?}"); assert!(warnings[0].contains("bridge down"), "{warnings:?}"); } /// `admin_unless_agent` with a `promote` that records whether it ran and /// answers `plan`. async fn forge_admin_route( is_agent: Result, plan: crate::forge::site_admin::Plan, ) -> ( Result, problem_details::ProblemDetails>, bool, ) { let ran = std::sync::atomic::AtomicBool::new(false); let result = super::admin_unless_agent("atlas", is_agent, || async { ran.store(true, std::sync::atomic::Ordering::SeqCst); Ok(plan) }) .await; (result, ran.into_inner()) } /// An agent's name never reaches the forge. #[tokio::test] async fn the_forge_admin_route_refuses_an_agent() { let (result, ran) = forge_admin_route(Ok(true), crate::forge::site_admin::Plan::Promote).await; let Err(problem) = result else { panic!("an agent's name must be refused"); }; assert!(!ran, "the account must not be touched"); assert_eq!(problem.status, Some(axum::http::StatusCode::CONFLICT)); } /// A roster that can't be read can't vouch for the name, so the route /// refuses rather than guessing "not an agent". #[tokio::test] async fn the_forge_admin_route_refuses_on_an_unreadable_roster() { let (result, ran) = forge_admin_route( Err("bridge down".to_owned()), crate::forge::site_admin::Plan::Promote, ) .await; let Err(problem) = result else { panic!("an unreadable roster must refuse"); }; assert!(!ran, "the account must not be touched"); assert_eq!( problem.status, Some(axum::http::StatusCode::SERVICE_UNAVAILABLE) ); } /// A missing account is a 404 that tells the operator what to wait for. #[tokio::test] async fn the_forge_admin_route_names_the_missing_sso_login() { let (result, ran) = forge_admin_route(Ok(false), crate::forge::site_admin::Plan::NoSuchUser).await; let Err(problem) = result else { panic!("a missing account must not report success"); }; assert!(ran); assert_eq!(problem.status, Some(axum::http::StatusCode::NOT_FOUND)); let rendered = format!("{problem:?}"); assert!( rendered.contains("has not logged in via SSO yet"), "{rendered}" ); } #[tokio::test] async fn the_forge_admin_route_promotes_a_non_agent() { let (result, ran) = forge_admin_route(Ok(false), crate::forge::site_admin::Plan::Promote).await; let Ok(axum::Json(resp)) = result else { panic!("a non-agent must be promoted"); }; assert!(ran); assert_eq!(resp.change, super::ForgeAdminChange::Promoted); } /// Idempotent: a second run succeeds and says nothing changed. #[tokio::test] async fn the_forge_admin_route_succeeds_for_an_admin() { let (result, _) = forge_admin_route(Ok(false), crate::forge::site_admin::Plan::AlreadyAdmin).await; let Ok(axum::Json(resp)) = result else { panic!("an existing admin must succeed"); }; assert_eq!(resp.change, super::ForgeAdminChange::AlreadyAdmin); } /// The backfill route's roster check, asserted by effect for the same /// reason its sibling above is: a refusal that queued first would still /// re-mint the agent's certificate, which every running agent on the /// named hive picks up on its next boot. #[tokio::test] async fn a_backfill_for_a_hive_outside_the_roster_queues_nothing() { let (state, sched) = state_with_roster(); let err = super::mint_agent_identity( axum::extract::State(state), axum::extract::Path("atlas".to_owned()), axum::Json(super::MintAgentIdentityRequest { hive: "pr1maa".to_owned(), }), ) .await .expect_err("a hive outside the roster must be refused"); 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 backfill must queue no work"); } /// A name that is not an identifier is refused before it can reach a /// store path. `create_agent` gets this for free from the roster check /// on the hive; the agent name has no roster to check against, so this /// is the only thing standing between `../beta` and a path built out of /// it. #[tokio::test] async fn a_backfill_for_a_name_that_is_not_an_identifier_queues_nothing() { let (state, sched) = state_with_roster(); super::mint_agent_identity( axum::extract::State(state), axum::extract::Path("../beta".to_owned()), axum::Json(super::MintAgentIdentityRequest { hive: "pr1ma".to_owned(), }), ) .await .expect_err("a traversal in the agent name must be refused"); let queued = sched .lock() .unwrap_or_else(std::sync::PoisonError::into_inner) .graph() .nodes() .count(); assert_eq!(queued, 0, "a refused backfill must queue no work"); } /// The accept arm, and the property that makes this a *backfill* route /// rather than a second creation route: it queues the mint node **and /// nothing else**. `create_agent`'s graph also scaffolds a config repo /// and publishes a deploy; running either of those against an agent that /// already exists is the failure this assertion exists to catch. #[tokio::test] async fn a_backfill_queues_the_mint_node_and_only_the_mint_node() { let (state, sched) = state_with_roster(); let queued = super::mint_agent_identity( axum::extract::State(state), axum::extract::Path("atlas".to_owned()), axum::Json(super::MintAgentIdentityRequest { hive: "pr1ma".to_owned(), }), ) .await .expect("a hive in the roster must be accepted"); let guard = sched .lock() .unwrap_or_else(std::sync::PoisonError::into_inner); let nodes: Vec<_> = guard.graph().nodes().collect(); let [node] = nodes[..] else { panic!("a backfill queues exactly one node, got {}", nodes.len()); }; assert!( matches!( &node.payload, SwarmNodeKind::MintAgentIdentity { hive, agent } if hive == "pr1ma" && agent == "atlas" ), "the one node must be the mint, carrying both names: {:?}", node.payload ); // The id the operator is told to watch has to be the node that was // actually inserted, or `swarmctl` prints a handle to nothing. assert_eq!(queued.node_id, node.id.get()); } #[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, agent_ca: None, wanted: None, matrix_homeserver: 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, agent_ca: None, wanted: None, matrix_homeserver: 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 ); } /// Third sibling of the two above, and the same deliberate caveat: with /// no authority configured this reaches only the /// graceful-absence-is-failure branch. The happy path is a live store /// and a real login, which is exactly why it is `mint_and_verify`'s own /// job to prove it at agent-creation time rather than a unit test's. /// /// What this does pin is the degrade: a host that was never given an /// authority fails this one node with a reason that names the two /// variables, and creates the agent anyway. #[tokio::test] async fn mint_agent_identity_node_runs_end_to_end_and_fails_without_an_authority() { let mut sched = hive_jobq::scheduler::Scheduler::new( hive_jobq::Graph::new(), hive_jobq::resources::ResourceTable::new(), ); let id = sched .append( SwarmNodeKind::MintAgentIdentity { hive: "pr1ma".to_owned(), 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, agent_ca: None, wanted: None, matrix_homeserver: 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); let error = node.error.as_deref().unwrap_or_default(); assert!( error.contains(crate::agent_identity::ENV_AGENT_CA) && error.contains(crate::agent_identity::ENV_AGENT_CA_KEY), "the reason must name both variables an operator has to set, got {error:?}" ); } /// Fourth sibling, same shape and the same caveat once more: with no /// queue configured this reaches only the graceful-absence-is-failure /// branch. Writing the declaration for real needs a `JetStream` server, /// which no test in this crate has — `wanted::apply`'s own unit tests /// cover what gets written, and [`a_new_agent_is_declared_paused`] below /// pins the state this node asks for. #[tokio::test] async fn set_agent_wanted_node_runs_end_to_end_and_fails_without_a_queue() { let mut sched = hive_jobq::scheduler::Scheduler::new( hive_jobq::Graph::new(), hive_jobq::resources::ResourceTable::new(), ); let id = sched .append( SwarmNodeKind::SetAgentWanted { hive: "pr1ma".to_owned(), 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, agent_ca: None, wanted: None, matrix_homeserver: 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("queue"), "expected a no-queue-configured error, got {:?}", node.error ); } /// The whole point of the node: a freshly created agent is declared /// `Paused`, so it does not start driving turns the moment its hive /// brings it up. Any other state here is the bug this fixes. #[test] fn a_new_agent_is_declared_paused() { assert_eq!( crate::NEW_AGENT_WANTED_STATE, swarm_queue_client::wanted::AgentState::Paused ); } /// Same reasoning as `a_mint_node_renders_both_the_agent_and_the_hive`: /// this variant carries a hive too, and has to have joined `data`'s /// two-field or-pattern rather than the agent-only one. #[test] fn a_set_wanted_node_renders_both_the_agent_and_the_hive() { use hive_jobq_wire::WireNode as _; let kind = SwarmNodeKind::SetAgentWanted { hive: "pr1ma".to_owned(), agent: "atlas".to_owned(), }; assert_eq!(kind.label(), "set_agent_wanted"); let data = kind.data(1); assert_eq!(data["agent"], "atlas"); assert_eq!(data["hive"], "pr1ma"); } /// The ordering the fix depends on: the pause has to be in the /// wanted-state store *before* the hive is told to deploy, or a freshly /// created agent runs undeclared for the window between the two. /// /// Asserted on the graph `create_agent` builds, same as the mint edge's /// test above and for the same reason — the edge is one line in a /// builder closure whose absence changes nothing observable until a real /// agent boots and starts taking turns nobody asked for. #[tokio::test] async fn the_deploy_waits_for_the_pause_to_be_declared() { use hive_jobq_wire::WireNode as _; let (state, sched) = state_with_roster(); 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 guard = sched .lock() .unwrap_or_else(std::sync::PoisonError::into_inner); let graph = guard.graph(); let id_of = |label: &str| { graph .nodes() .find(|n| n.payload.label() == label) .unwrap_or_else(|| panic!("the graph holds a {label} node")) .id }; let set_wanted = id_of("set_agent_wanted"); let deploy = graph.node(id_of("trigger_deploy")).expect("just found"); let when = deploy .deps .iter() .find_map(|d| match d { hive_jobq::Dep::Node { id, when } if *id == set_wanted => Some(*when), _ => None, }) .expect("the deploy waits for the declaration"); assert!( !when.accepts(hive_jobq::TerminalState::Failed), "a declaration that never landed must not be deployed past; this \ edge has to be `after_ok`, not `after_any`" ); } /// The node carries the hive, and the viewer has to see it. The `data` /// match is an or-pattern on purpose (see its own comment), and this is /// the assertion that the new variant joined the two-field arm rather /// than the agent-only one — a viewer silently missing the hive is the /// failure that comment describes having already happened once. #[test] fn a_mint_node_renders_both_the_agent_and_the_hive() { use hive_jobq_wire::WireNode as _; let kind = SwarmNodeKind::MintAgentIdentity { hive: "pr1ma".to_owned(), agent: "atlas".to_owned(), }; assert_eq!(kind.label(), "mint_agent_identity"); let data = kind.data(1); assert_eq!(data["agent"], "atlas"); assert_eq!(data["hive"], "pr1ma"); } /// The ordering the operator's ruling requires: a hive cannot pass down /// a certificate the swarm has not published, so the deploy message must /// not leave before the mint is terminal. /// /// Asserted on the graph `create_agent` builds, because the edge is one /// line in a builder closure and its absence changes nothing observable /// until a real agent boots without an identity. #[tokio::test] async fn the_deploy_waits_for_the_mint_and_is_not_cancelled_by_it() { use hive_jobq_wire::WireNode as _; let (state, sched) = state_with_roster(); 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 guard = sched .lock() .unwrap_or_else(std::sync::PoisonError::into_inner); let graph = guard.graph(); let id_of = |label: &str| { graph .nodes() .find(|n| n.payload.label() == label) .unwrap_or_else(|| panic!("the graph holds a {label} node")) .id }; let mint = id_of("mint_agent_identity"); let deploy = graph.node(id_of("trigger_deploy")).expect("just found"); let when = deploy .deps .iter() .find_map(|d| match d { hive_jobq::Dep::Node { id, when } if *id == mint => Some(*when), _ => None, }) .expect("the deploy waits for the mint"); assert!( when.accepts(hive_jobq::TerminalState::Failed), "an unconfigured authority must not cancel the deploy; this edge \ has to be `after_any`, not `after_ok`" ); } /// The forge-token mint sits between the forge user and the deploy: it /// needs the user to exist, and the deploy must not overtake it — but a /// host with no forge or no store must still deploy the agent. #[tokio::test] async fn the_forge_token_mint_follows_the_user_and_does_not_block_the_deploy() { use hive_jobq_wire::WireNode as _; let (state, sched) = state_with_roster(); 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 guard = sched .lock() .unwrap_or_else(std::sync::PoisonError::into_inner); let graph = guard.graph(); let id_of = |label: &str| { graph .nodes() .find(|n| n.payload.label() == label) .unwrap_or_else(|| panic!("the graph holds a {label} node")) .id }; let edge = |from: hive_jobq::NodeId, to: &str| { graph .node(id_of(to)) .expect("just found") .deps .iter() .find_map(|d| match d { hive_jobq::Dep::Node { id, when } if *id == from => Some(*when), _ => None, }) .unwrap_or_else(|| panic!("{to} waits for node {}", from.get())) }; let user_to_mint = edge(id_of("create_forge_user"), "mint_agent_forge_token"); assert!( !user_to_mint.accepts(hive_jobq::TerminalState::Failed), "a token for a user that was never created cannot be minted; this \ edge has to be `after_ok`" ); let mint_to_deploy = edge(id_of("mint_agent_forge_token"), "trigger_deploy"); assert!( mint_to_deploy.accepts(hive_jobq::TerminalState::Failed), "a host with no forge or store must not cancel the deploy; this edge \ has to be `after_any`, not `after_ok`" ); } #[test] fn a_forge_token_node_renders_the_agent() { use hive_jobq_wire::WireNode as _; let kind = SwarmNodeKind::MintAgentForgeToken { agent: "atlas".to_owned(), }; assert_eq!(kind.label(), "mint_agent_forge_token"); assert_eq!(kind.data(1)["agent"], "atlas"); } #[test] fn a_matrix_account_node_renders_the_agent() { use hive_jobq_wire::WireNode as _; let kind = SwarmNodeKind::MintAgentMatrixAccount { agent: "atlas".to_owned(), }; assert_eq!(kind.label(), "mint_agent_matrix_account"); assert_eq!(kind.data(1)["agent"], "atlas"); } /// Agent creation mints the matrix account as a root of its own, and the /// deploy waits for it without being cancelled by it: a host with no /// homeserver configured must still deploy the agent. #[tokio::test] async fn the_matrix_account_mint_is_a_root_and_does_not_block_the_deploy() { use hive_jobq_wire::WireNode as _; let (state, sched) = state_with_roster(); 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 guard = sched .lock() .unwrap_or_else(std::sync::PoisonError::into_inner); let graph = guard.graph(); let find = |label: &str| { graph .nodes() .find(|n| n.payload.label() == label) .unwrap_or_else(|| panic!("the graph holds a {label} node")) }; let mint = find("mint_agent_matrix_account"); assert!(mint.deps.is_empty(), "the mint waits for nothing"); let when = find("trigger_deploy") .deps .iter() .find_map(|d| match d { hive_jobq::Dep::Node { id, when } if *id == mint.id => Some(*when), _ => None, }) .expect("the deploy waits for the matrix mint"); assert!( when.accepts(hive_jobq::TerminalState::Failed), "a host with no homeserver must not cancel the deploy; this edge \ has to be `after_any`, not `after_ok`" ); } /// The backfill's queueing: one mint node per agent, nothing else. #[test] fn a_queued_matrix_mint_is_one_node_per_agent() { use hive_jobq_wire::WireNode as _; let sched = std::sync::Mutex::new(hive_jobq::scheduler::Scheduler::new( hive_jobq::Graph::new(), hive_jobq::resources::ResourceTable::new(), )); let ids = super::queue_matrix_account_mints(&sched, vec!["a".to_owned(), "b".to_owned()]) .expect("two jobs insert"); assert_eq!(ids.len(), 2); let guard = sched .lock() .unwrap_or_else(std::sync::PoisonError::into_inner); let mut agents: Vec = guard .graph() .nodes() .map(|n| { assert_eq!(n.payload.label(), "mint_agent_matrix_account"); n.payload.data(n.id.get())["agent"] .as_str() .expect("agent is a string") .to_owned() }) .collect(); agents.sort(); assert_eq!(agents, ["a", "b"]); } /// The manual route and the periodic pass both go through /// `queue_forge_token_mints`, so this is the assertion that a backfilled /// mint is the same pair of nodes agent creation inserts: the forge user, /// then the mint `after_ok` it, per agent. #[test] fn a_queued_mint_creates_the_forge_user_first() { use hive_jobq_wire::WireNode as _; let sched = std::sync::Mutex::new(hive_jobq::scheduler::Scheduler::new( hive_jobq::Graph::new(), hive_jobq::resources::ResourceTable::new(), )); let ids = super::queue_forge_token_mints(&sched, vec!["a".to_owned(), "b".to_owned()]) .expect("two jobs insert"); assert_eq!(ids.len(), 2, "one returned id per agent: its mint node"); let guard = sched .lock() .unwrap_or_else(std::sync::PoisonError::into_inner); let graph = guard.graph(); let agent_of = |n: &hive_jobq::Node| { n.payload.data(n.id.get())["agent"] .as_str() .expect("agent is a string") .to_owned() }; for agent in ["a", "b"] { let find = |label: &str| { graph .nodes() .find(|n| n.payload.label() == label && agent_of(n) == agent) .unwrap_or_else(|| panic!("{agent} has a {label} node")) }; let user = find("create_forge_user").id; let mint = find("mint_agent_forge_token"); assert!(ids.contains(&mint.id), "the returned id is {agent}'s mint"); let when = mint .deps .iter() .find_map(|d| match d { hive_jobq::Dep::Node { id, when } if *id == user => Some(*when), _ => None, }) .unwrap_or_else(|| panic!("{agent}'s mint waits for its own forge user")); assert!( !when.accepts(hive_jobq::TerminalState::Failed), "a token cannot be minted for a user that was never created; this \ edge has to be `after_ok`" ); } assert_eq!( graph.nodes().count(), 4, "two nodes per agent, nothing else" ); } /// The backfill end to end, minus the IO: an agent the pass observed with /// no forge user is planned, and the job it gets creates that user before /// it mints. #[test] fn an_agent_with_no_forge_user_gets_a_user_and_a_mint() { use crate::forge::agent_token::{Observed, plan}; use hive_jobq_wire::WireNode as _; let agents = plan(&[("ruth".to_owned(), Observed::NoForgeUser)]); let sched = std::sync::Mutex::new(hive_jobq::scheduler::Scheduler::new( hive_jobq::Graph::new(), hive_jobq::resources::ResourceTable::new(), )); super::queue_forge_token_mints(&sched, agents).expect("one job inserts"); let guard = sched .lock() .unwrap_or_else(std::sync::PoisonError::into_inner); let mut labels: Vec = guard.graph().nodes().map(|n| n.payload.label()).collect(); labels.sort(); assert_eq!(labels, ["create_forge_user", "mint_agent_forge_token"]); } /// 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::::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::::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::::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::::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); } } // ── NotConnected maps to 503, everything else stays 500 ───────────── /// `wanted_error_status` as a pure function first: build the exact /// `swarm_queue_client::Error::NotConnected` `ensure_connected` raises, /// wrap it the same way `?` inside `WantedWriter::view`/`set` does, and /// confirm it downcasts back to the status those routes are meant to /// answer. #[test] fn not_connected_downcasts_to_503() { let e: anyhow::Error = swarm_queue_client::Error::NotConnected(async_nats::connection::State::Pending).into(); assert_eq!( super::wanted_error_status(&e), axum::http::StatusCode::SERVICE_UNAVAILABLE ); } /// Invert-proof for the test above: an `anyhow::Error` that does not /// wrap a `swarm_queue_client::Error` at all (the shape a genuine write /// or decode failure takes — `apply`'s "declaration is not decodable" /// context, for instance) must NOT downcast to `NotConnected`, and must /// stay 500. Without this, a `wanted_error_status` that answered 503 /// unconditionally would still pass the test above. #[test] fn an_unrelated_error_stays_500() { let e = anyhow::anyhow!("the hive's current declaration is not decodable"); assert_eq!( super::wanted_error_status(&e), axum::http::StatusCode::INTERNAL_SERVER_ERROR ); } /// Second invert-proof, closer to the actual bug this closes: a /// *different* `swarm_queue_client::Error` variant (a real connect /// failure, not `NotConnected`) also stays 500 — proving the match /// looks at the variant, not just at "was this crate's error type /// involved at all". #[test] fn a_different_queue_client_error_variant_stays_500() { let e: anyhow::Error = swarm_queue_client::Error::Connect { url: "nats://queue.example:4222".to_owned(), source: async_nats::ConnectErrorKind::TimedOut.into(), } .into(); assert_eq!( super::wanted_error_status(&e), axum::http::StatusCode::INTERNAL_SERVER_ERROR ); } /// A client that exists but has never connected — same shape /// `matrix_account`'s tests build, duplicated here rather than shared: /// both are a handful of lines and neither crate has a test-support /// module to put a shared helper in. async fn disconnected_client() -> async_nats::Client { async_nats::ConnectOptions::new() .retry_on_initial_connect() .connect("127.0.0.1:1") .await .expect("retry_on_initial_connect returns without waiting for a real connection") } /// `state_with_roster` plus a `wanted` writer over a client that never /// connected — the shape `set_agent_state`/`get_hive_wanted` see when /// the queue is mid-reconnect. async fn state_with_disconnected_wanted() -> super::AppState { let (state, _sched) = state_with_roster(); super::AppState { wanted: Some(std::sync::Arc::new(wanted::WantedWriter::new( disconnected_client().await, ))), ..state } } /// End to end through the real handler: `PUT .../state` against a /// disconnected queue answers 503, not the 500 every other writer /// error still gets. #[tokio::test] async fn set_agent_state_answers_503_on_a_disconnected_queue() { let state = state_with_disconnected_wanted().await; let result = super::set_agent_state( axum::extract::State(state), axum::extract::Path(("pr1ma".to_owned(), "atlas".to_owned())), axum::Json(SetAgentStateRequest { state: swarm_queue_client::wanted::AgentState::Paused, }), ) .await; let problem = result.expect_err("a disconnected queue must not read as success"); assert_eq!( problem.status, Some(axum::http::StatusCode::SERVICE_UNAVAILABLE), "{problem:?}" ); } /// Same fix, the read side: `GET .../wanted` against a disconnected /// queue also answers 503. #[tokio::test] async fn get_hive_wanted_answers_503_on_a_disconnected_queue() { let state = state_with_disconnected_wanted().await; let result = super::get_hive_wanted( axum::extract::State(state), axum::extract::Path("pr1ma".to_owned()), ) .await; let problem = result.expect_err("a disconnected queue must not read as success"); assert_eq!( problem.status, Some(axum::http::StatusCode::SERVICE_UNAVAILABLE), "{problem:?}" ); } /// Control for both tests above: a hive that is not in the roster is /// still a plain 400, even with the same disconnected writer wired up — /// so the 503 above is `ensure_connected` firing, not every error this /// handler can produce collapsing to 503. #[tokio::test] async fn set_agent_state_still_answers_400_for_an_unknown_hive() { let state = state_with_disconnected_wanted().await; let result = super::set_agent_state( axum::extract::State(state), axum::extract::Path(("not-a-hive".to_owned(), "atlas".to_owned())), axum::Json(SetAgentStateRequest { state: swarm_queue_client::wanted::AgentState::Paused, }), ) .await; let problem = result.expect_err("an unknown hive is still a 400"); assert_eq!( problem.status, Some(axum::http::StatusCode::BAD_REQUEST), "{problem:?}" ); } } #[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"]}"#); } }