68b38a9d44
The webhook secret slot is filled in the background, on the forge token's capped unbounded backoff, when the store can't answer at start. The hook registration waits for the secret and retries on the same backoff instead of giving up after five tries. Closes #6101
1013 lines
43 KiB
Rust
1013 lines
43 KiB
Rust
//! Process startup: the config this daemon loads once from env vars, the
|
|
//! background passes it starts, and the router it serves.
|
|
//!
|
|
//! `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-c0re/vhosts.nix`'s
|
|
//! `swaggerUiLocations` — 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::path::PathBuf;
|
|
use std::sync::{Arc, Mutex};
|
|
|
|
use anyhow::Result;
|
|
use axum::{Json, routing::get};
|
|
use utoipa::OpenApi;
|
|
use utoipa_axum::{router::OpenApiRouter, routes};
|
|
|
|
use crate::jobq::{
|
|
Late, SwarmNodeKind, SwarmResourceKind, queue_agent_renewals, queue_builder_forge_token_mint,
|
|
queue_ci_metrics_forge_token_mint, queue_classifier_forge_token_mint, queue_docs_grants,
|
|
queue_forge_token_mints, queue_hive_forge_token_mints, queue_hive_sender_mints,
|
|
queue_matrix_account_mints,
|
|
};
|
|
use crate::routes::hives::{HiveEntry, ServiceLink};
|
|
use crate::routes::{self, AppState, agents, config_prs, forge_admin, hives, jobq_graph};
|
|
use crate::{
|
|
agent_control, agent_custom, agent_icon, agent_input, agent_limits, agent_login,
|
|
agent_permissions, agent_renewal, agent_state_stream, agent_status, agents_md, alerts,
|
|
archetypes, auth, backend_env, builder, claude_login, config_drift, config_pr, egress_allow,
|
|
egress_deny, forge, forge_account, github_account, gitlab_account, inference, issue_report,
|
|
linked_accounts, matrix_account, message_stream, notify_flows, notify_router_api, pulls,
|
|
queue_identity, review_classifier, schedules, status, subagent_term, term_stream, wanted,
|
|
webhook,
|
|
};
|
|
|
|
/// 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";
|
|
|
|
pub(crate) 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`)"),
|
|
(name = "schedules", description = "scheduled prompts (`swarmctl schedule`)"),
|
|
(name = "egress", description = "the destinations agents may reach through their hive's egress proxy"),
|
|
(name = "notify", description = "the notification router's flows: built-ins switched off, and custom flows"),
|
|
(name = "review-classifier", description = "trying review classifications against a PR (swarm-ui's try-out page), and the classifier's swarm-wide settings"),
|
|
(name = "pulls", description = "open pull requests, for the try-out page's dropdowns"),
|
|
(name = "alerts", description = "the swarm Grafana's current alerts (swarm-ui's alerts page)"),
|
|
(name = "builder", description = "the swarm builder's finished builds of agent config PRs, which their commit statuses link to"),
|
|
)
|
|
)]
|
|
pub(crate) struct ApiDoc;
|
|
|
|
/// 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.
|
|
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`.
|
|
pub(crate) fn load_hives() -> Vec<HiveEntry> {
|
|
let Some(raw) = std::env::var_os(HIVES_ENV) else {
|
|
return Vec::new();
|
|
};
|
|
match serde_json::from_str(&raw.to_string_lossy()) {
|
|
Ok(hives) => hives,
|
|
Err(e) => {
|
|
tracing::warn!(error = %e, env = HIVES_ENV, "failed to parse hive directory, serving an empty list");
|
|
Vec::new()
|
|
}
|
|
}
|
|
}
|
|
|
|
/// Env var the controller's NixOS module sets from the merged
|
|
/// `services.hyperhive.deploy.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.
|
|
pub(crate) fn load_links() -> Vec<ServiceLink> {
|
|
let Some(raw) = std::env::var_os(LINKS_ENV) else {
|
|
return Vec::new();
|
|
};
|
|
match serde_json::from_str(&raw.to_string_lossy()) {
|
|
Ok(links) => links,
|
|
Err(e) => {
|
|
tracing::warn!(error = %e, env = LINKS_ENV, "failed to parse service links, serving an empty list");
|
|
Vec::new()
|
|
}
|
|
}
|
|
}
|
|
|
|
/// 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.
|
|
pub(crate) fn load_swarm_name() -> Option<String> {
|
|
std::env::var(NAME_ENV).ok()
|
|
}
|
|
|
|
/// 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.
|
|
pub(crate) fn wanted_writer(
|
|
status: Option<&Arc<status::StatusReader>>,
|
|
) -> Option<Arc<wanted::WantedWriter>> {
|
|
status.map(|s| Arc::new(wanted::WantedWriter::new(s.queue_client())))
|
|
}
|
|
|
|
/// The per-agent status reader, sharing the status reader's connection and
|
|
/// staleness threshold — same rationale as [`wanted_writer`], and the same
|
|
/// threshold on purpose: both buckets are fed by the same publisher cadence
|
|
/// (`hive-c0re`'s `PUBLISH_INTERVAL`), so two separately-configured
|
|
/// thresholds would just be two ways to get out of sync with one cadence.
|
|
pub(crate) fn agent_status_reader(
|
|
status: Option<&Arc<status::StatusReader>>,
|
|
) -> Option<Arc<agent_status::AgentStatusReader>> {
|
|
status.map(|s| {
|
|
Arc::new(agent_status::AgentStatusReader::new(
|
|
s.queue_client(),
|
|
status::StatusReader::stale_after_from_env(),
|
|
))
|
|
})
|
|
}
|
|
|
|
/// The per-agent icon reader, sharing the status reader's connection —
|
|
/// same rationale as [`wanted_writer`]. No staleness threshold, unlike
|
|
/// [`agent_status_reader`]: an icon is a property of the agent rather than
|
|
/// a report about it, so there is no cadence it could be late against.
|
|
///
|
|
/// Also starts creating the bucket, which the agents writing it cannot do.
|
|
pub(crate) fn agent_icon_reader(
|
|
status: Option<&Arc<status::StatusReader>>,
|
|
) -> Option<Arc<agent_icon::AgentIconReader>> {
|
|
let reader = Arc::new(agent_icon::AgentIconReader::new(status?.queue_client()));
|
|
let creating = Arc::clone(&reader);
|
|
tokio::spawn(async move { creating.create_when_connected().await });
|
|
Some(reader)
|
|
}
|
|
|
|
/// The `AGENTS.md` reader, sharing the status reader's connection, and
|
|
/// creating the bucket, as [`agent_icon_reader`] does for icons.
|
|
pub(crate) fn agents_md_reader(
|
|
status: Option<&Arc<status::StatusReader>>,
|
|
) -> Option<Arc<agents_md::AgentsMdReader>> {
|
|
let reader = Arc::new(agents_md::AgentsMdReader::new(status?.queue_client()));
|
|
let creating = Arc::clone(&reader);
|
|
tokio::spawn(async move { creating.create_when_connected().await });
|
|
Some(reader)
|
|
}
|
|
|
|
/// What each agent last reported running on, read on the status reader's
|
|
/// connection — same rationale as [`wanted_writer`].
|
|
pub(crate) fn reported_inference(
|
|
status: Option<&Arc<status::StatusReader>>,
|
|
) -> Option<Arc<inference::Reported>> {
|
|
Some(inference::Reported::spawn(status?.queue_client()))
|
|
}
|
|
|
|
/// Whether each agent last said it needs a login, read on the status
|
|
/// reader's connection — same rationale as [`wanted_writer`].
|
|
pub(crate) fn agent_login(
|
|
status: Option<&Arc<status::StatusReader>>,
|
|
) -> Option<Arc<agent_login::Reported>> {
|
|
Some(agent_login::Reported::spawn(status?.queue_client()))
|
|
}
|
|
|
|
/// 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.
|
|
pub(crate) fn configured_matrix_homeserver() -> Option<Arc<str>> {
|
|
matrix_account::configured_default_homeserver()
|
|
.filter(|v| !v.is_empty())
|
|
.map(Arc::from)
|
|
}
|
|
|
|
/// Start the recovery passphrase pass, which needs only the store; the
|
|
/// matrix-account backfill and the hive sender-token pass when a homeserver is
|
|
/// configured; and the operator admin pass when the bridge is too. Lifted out
|
|
/// of `main` for `clippy::too_many_lines`.
|
|
pub(crate) fn spawn_matrix_account_backfill(
|
|
jobq: &Arc<Mutex<hive_jobq::scheduler::Scheduler<SwarmNodeKind, SwarmResourceKind>>>,
|
|
homeserver: Option<Arc<str>>,
|
|
hives: &[HiveEntry],
|
|
auth: Option<Arc<auth::AuthBridge>>,
|
|
) {
|
|
matrix_account::recovery::spawn();
|
|
let Some(base) = homeserver else {
|
|
return;
|
|
};
|
|
if let Some(bridge) = auth {
|
|
matrix_account::operator_admin::spawn(bridge, Arc::clone(&base));
|
|
}
|
|
let sched = Arc::clone(jobq);
|
|
matrix_account::agent_token::spawn(Arc::clone(&base), move |agents| {
|
|
if let Err(e) = queue_matrix_account_mints(&sched, agents) {
|
|
tracing::warn!(error = %format!("{e:#}"), "agent matrix accounts: queueing failed");
|
|
}
|
|
});
|
|
let sched = Arc::clone(jobq);
|
|
let hive_names = hives.iter().map(|h| h.name.clone()).collect();
|
|
matrix_account::hive_sender::spawn(base, hive_names, move |hives| {
|
|
if let Err(e) = queue_hive_sender_mints(&sched, hives) {
|
|
tracing::warn!(error = %format!("{e:#}"), "hive sender tokens: queueing failed");
|
|
}
|
|
});
|
|
}
|
|
|
|
/// Start the forge's periodic passes — the swarm-wide objects, the runner
|
|
/// registration token, hives' and the builder's read tokens, agents'
|
|
/// `internal/docs` grants, and
|
|
/// agents' tokens (which also
|
|
/// sweeps legacy tokens; see `forge::legacy_tokens`) —
|
|
/// once there is a forge client, and the operators' profiles when the bridge
|
|
/// and the queue are wired up too. Lifted out of `main` for
|
|
/// `clippy::too_many_lines`.
|
|
fn spawn_forge_workers(
|
|
jobq: &Arc<Mutex<hive_jobq::scheduler::Scheduler<SwarmNodeKind, SwarmResourceKind>>>,
|
|
client: Arc<forge::Client>,
|
|
hives: &[HiveEntry],
|
|
auth: Option<Arc<auth::AuthBridge>>,
|
|
queue: Option<async_nats::Client>,
|
|
) {
|
|
// `internal/meta` appears, catches up with every config head that moved
|
|
// while this daemon was down, and takes this daemon's rev as the hyperhive
|
|
// pin when that is newer. The timer's first tick is this boot run.
|
|
crate::jobq::spawn_meta_update_timer(Arc::clone(jobq));
|
|
forge::objects::spawn(Arc::clone(&client));
|
|
if let (Some(bridge), Some(queue)) = (auth, queue.clone()) {
|
|
forge::operator_profiles::spawn(Arc::clone(&client), bridge, queue);
|
|
}
|
|
forge::runner_token::spawn(Arc::clone(&client));
|
|
if let Some(queue) = queue {
|
|
crate::cache_miss::spawn(
|
|
queue.clone(),
|
|
Arc::new(wanted::WantedWriter::new(queue)),
|
|
hives.iter().map(|h| h.name.clone()).collect(),
|
|
Arc::clone(&client),
|
|
);
|
|
}
|
|
let sched = Arc::clone(jobq);
|
|
let hive_names = hives.iter().map(|h| h.name.clone()).collect();
|
|
forge::hive_token::spawn(Arc::clone(&client), hive_names, move |hives| {
|
|
if let Err(e) = queue_hive_forge_token_mints(&sched, hives) {
|
|
tracing::warn!(error = %format!("{e:#}"), "hive forge tokens: queueing failed");
|
|
}
|
|
});
|
|
let sched = Arc::clone(jobq);
|
|
forge::classifier_account::spawn(Arc::clone(&client), move || {
|
|
if let Err(e) = queue_classifier_forge_token_mint(&sched) {
|
|
tracing::warn!(error = %format!("{e:#}"), "review classifier forge token: queueing failed");
|
|
}
|
|
});
|
|
let sched = Arc::clone(jobq);
|
|
forge::builder_account::spawn(Arc::clone(&client), move || {
|
|
if let Err(e) = queue_builder_forge_token_mint(&sched) {
|
|
tracing::warn!(error = %format!("{e:#}"), "builder forge token: queueing failed");
|
|
}
|
|
});
|
|
let sched = Arc::clone(jobq);
|
|
forge::ci_metrics_account::spawn(Arc::clone(&client), move || {
|
|
if let Err(e) = queue_ci_metrics_forge_token_mint(&sched) {
|
|
tracing::warn!(error = %format!("{e:#}"), "ci metrics forge token: queueing failed");
|
|
}
|
|
});
|
|
let sched = Arc::clone(jobq);
|
|
forge::docs_grant::spawn(Arc::clone(&client), move |agents| {
|
|
if let Err(e) = queue_docs_grants(&sched, agents) {
|
|
tracing::warn!(error = %format!("{e:#}"), "docs grants: queueing failed");
|
|
}
|
|
});
|
|
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");
|
|
}
|
|
});
|
|
}
|
|
|
|
/// Start the agent credential renewal pass when a swarm queue is wired up:
|
|
/// the pass reads the wanted-state declarations to tell live agents from
|
|
/// destroyed ones, and without a queue it has none. Lifted out of `main` for
|
|
/// `clippy::too_many_lines`.
|
|
pub(crate) fn spawn_agent_renewal(
|
|
jobq: &Arc<Mutex<hive_jobq::scheduler::Scheduler<SwarmNodeKind, SwarmResourceKind>>>,
|
|
wanted: Option<Arc<wanted::WantedWriter>>,
|
|
hives: &[HiveEntry],
|
|
) {
|
|
let Some(wanted) = wanted else {
|
|
return;
|
|
};
|
|
let hives = hives.iter().map(|h| h.name.clone()).collect();
|
|
let sched = Arc::clone(jobq);
|
|
agent_renewal::spawn(wanted, hives, move |renewals| {
|
|
if let Err(e) = queue_agent_renewals(&sched, renewals) {
|
|
tracing::warn!(error = %format!("{e:#}"), "agent credential renewal: queueing failed");
|
|
}
|
|
});
|
|
}
|
|
|
|
/// Start creating every live agent's subagent and terminal streams when a swarm
|
|
/// queue is wired up. Lifted out of `main` for `clippy::too_many_lines`.
|
|
fn spawn_subagent_streams(status: Option<&Arc<status::StatusReader>>, hives: &[HiveEntry]) {
|
|
let (Some(status), Some(wanted)) = (status, wanted_writer(status)) else {
|
|
return;
|
|
};
|
|
let hives = hives.iter().map(|h| h.name.clone()).collect();
|
|
subagent_term::spawn(status.queue_client(), wanted, hives);
|
|
}
|
|
|
|
/// The per-agent background passes: subagent and terminal streams, and
|
|
/// schedule keys, and the permissions bucket and its import, the allowed
|
|
/// domains derived from backend environments, and the
|
|
/// archetypes agents are created from. Beside them, the
|
|
/// copy of the cache facts to each hive, and the proxy credentials squid checks.
|
|
/// Lifted out of `main` for `clippy::too_many_lines`.
|
|
pub(crate) fn spawn_per_agent_upkeep(
|
|
status: Option<&Arc<status::StatusReader>>,
|
|
hives: &[HiveEntry],
|
|
) {
|
|
spawn_subagent_streams(status, hives);
|
|
spawn_schedule_reconcile(status, hives);
|
|
spawn_agent_permissions(status, hives);
|
|
spawn_backend_egress_backfill(status, hives);
|
|
crate::attic_copy::spawn(hives.iter().map(|h| h.name.clone()).collect());
|
|
crate::proxy_credential::spawn(hives.iter().map(|h| h.name.clone()).collect());
|
|
if let Some(status) = status {
|
|
archetypes::spawn(status.queue_client());
|
|
}
|
|
}
|
|
|
|
/// Create the agent-permissions bucket and answer hives' imports when a swarm
|
|
/// queue is wired up. Same shape as [`spawn_subagent_streams`].
|
|
fn spawn_agent_permissions(status: Option<&Arc<status::StatusReader>>, hives: &[HiveEntry]) {
|
|
let (Some(status), Some(wanted)) = (status, wanted_writer(status)) else {
|
|
return;
|
|
};
|
|
let hives = hives.iter().map(|h| h.name.clone()).collect();
|
|
agent_permissions::spawn(status.queue_client(), wanted, hives);
|
|
}
|
|
|
|
/// Drop the allowed-domain entries derived from backend environments that no
|
|
/// longer name their host, and make the entries derived from linked accounts
|
|
/// match the stored accounts, when a swarm queue is wired up.
|
|
fn spawn_backend_egress_backfill(status: Option<&Arc<status::StatusReader>>, hives: &[HiveEntry]) {
|
|
let (Some(status), Some(wanted)) = (status, wanted_writer(status)) else {
|
|
return;
|
|
};
|
|
let hives = hives.iter().map(|h| h.name.clone()).collect();
|
|
crate::backend_egress::spawn_backfill(status.queue_client(), wanted, hives);
|
|
}
|
|
|
|
/// Start rewriting every agent's schedule key from the definitions when a
|
|
/// swarm queue is wired up. Same shape as [`spawn_subagent_streams`].
|
|
fn spawn_schedule_reconcile(status: Option<&Arc<status::StatusReader>>, hives: &[HiveEntry]) {
|
|
let (Some(status), Some(wanted)) = (status, wanted_writer(status)) else {
|
|
return;
|
|
};
|
|
let hives = hives.iter().map(|h| h.name.clone()).collect();
|
|
schedules::spawn_reconcile(status.queue_client(), wanted, hives);
|
|
}
|
|
|
|
/// 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.
|
|
pub(crate) 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 capped backoff for as long as it takes, 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.
|
|
///
|
|
/// Runs once there is a forge client, and silently does nothing when the
|
|
/// public URL is missing: no public vhost is a legitimate deployment shape,
|
|
/// and it is logged here. A webhook secret that is not loaded yet is waited
|
|
/// for, as a forge that is not answering is.
|
|
fn register_swarm_webhooks(forge: Arc<forge::Client>, secret: Late<Arc<str>>) {
|
|
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 {
|
|
register_when_ready(&secret, |secret| {
|
|
let forge = Arc::clone(&forge);
|
|
let public_url = public_url.clone();
|
|
async move { forge.ensure_swarm_webhooks(&public_url, &secret).await }
|
|
})
|
|
.await;
|
|
});
|
|
}
|
|
|
|
/// How long the first retry of the registration waits.
|
|
const REGISTRATION_FIRST: std::time::Duration = std::time::Duration::from_secs(15);
|
|
|
|
/// Run `ensure` with the webhook secret until it succeeds, waiting for
|
|
/// `secret` to be filled first. Registration is idempotent, so a repeat costs
|
|
/// one API call and changes nothing.
|
|
async fn register_when_ready<F, Fut>(secret: &Late<Arc<str>>, mut ensure: F)
|
|
where
|
|
F: FnMut(Arc<str>) -> Fut,
|
|
Fut: std::future::Future<Output = Result<()>>,
|
|
{
|
|
forge::retry_unbounded("the swarm-wide forge webhooks", REGISTRATION_FIRST, || {
|
|
let secret = secret.get().cloned();
|
|
let registering = secret.map(&mut ensure);
|
|
async move {
|
|
match registering {
|
|
Some(registering) => registering.await,
|
|
None => anyhow::bail!("the webhook secret is not loaded yet"),
|
|
}
|
|
}
|
|
})
|
|
.await;
|
|
tracing::info!("registered the swarm-wide forge webhooks");
|
|
}
|
|
|
|
/// 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
|
|
/// an identity that could not be resolved, which `queue_identity::init`
|
|
/// reports in `main` before this runs: silently behaving like an
|
|
/// unconfigured host is how every hive ends up reading `never_reported`
|
|
/// with nothing to point at.
|
|
pub(crate) async fn connect_status_reader() -> Result<Option<Arc<status::StatusReader>>> {
|
|
let Some(cfg) = queue_identity::get().cloned() 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)
|
|
}
|
|
}
|
|
}
|
|
|
|
/// Fill the forge slots and start everything that runs off the forge client —
|
|
/// its periodic passes, the config-PR poll and the webhook registration — now
|
|
/// if the token is in the store, otherwise when `forge::enable`'s background
|
|
/// re-read finds it. Nothing to start when no forge URL is configured. Lifted
|
|
/// out of `main` for `clippy::too_many_lines`.
|
|
pub(crate) async fn start_forge(
|
|
forge_client: Late<Arc<forge::Client>>,
|
|
config_prs: Late<Arc<config_pr::ConfigPrCache>>,
|
|
jobq: Arc<Mutex<hive_jobq::scheduler::Scheduler<SwarmNodeKind, SwarmResourceKind>>>,
|
|
hives: Vec<HiveEntry>,
|
|
webhook_secret: Late<Arc<str>>,
|
|
auth: Option<Arc<auth::AuthBridge>>,
|
|
queue: Option<async_nats::Client>,
|
|
) {
|
|
let url = match forge::url_from_env() {
|
|
Ok(Some(url)) => url,
|
|
Ok(None) => return,
|
|
Err(e) => {
|
|
tracing::warn!(error = %format!("{e:#}"), "forge misconfigured; repo provisioning is off");
|
|
return;
|
|
}
|
|
};
|
|
forge::enable(url, forge_client, move |client| {
|
|
let drift = queue.clone().map(|queue| config_drift::Deps {
|
|
jobq: Arc::clone(&jobq),
|
|
wanted: Arc::new(wanted::WantedWriter::new(queue.clone())),
|
|
hives: hives.iter().map(|h| h.name.clone()).collect(),
|
|
queue,
|
|
});
|
|
spawn_forge_workers(&jobq, Arc::clone(&client), &hives, auth, queue);
|
|
config_prs.get_or_init(|| config_pr::spawn(Arc::clone(&client), drift));
|
|
register_swarm_webhooks(client, webhook_secret);
|
|
})
|
|
.await;
|
|
}
|
|
|
|
/// The routes that write, delete and list an agent's linked accounts and
|
|
/// backend environment, kept apart so `build_app` stays under the line limit.
|
|
fn linked_account_routes(router: OpenApiRouter<AppState>) -> OpenApiRouter<AppState> {
|
|
router
|
|
.routes(routes!(
|
|
matrix_account::put_matrix_account,
|
|
linked_accounts::delete_matrix_account
|
|
))
|
|
.routes(routes!(
|
|
forge_account::put_forge_account,
|
|
linked_accounts::delete_forge_account
|
|
))
|
|
.routes(routes!(
|
|
gitlab_account::put_gitlab_account,
|
|
linked_accounts::delete_gitlab_account
|
|
))
|
|
.routes(routes!(
|
|
github_account::put_github_account,
|
|
linked_accounts::delete_github_account
|
|
))
|
|
.routes(routes!(
|
|
backend_env::put_backend_env,
|
|
linked_accounts::delete_backend_env
|
|
))
|
|
.routes(routes!(linked_accounts::get_linked_accounts))
|
|
}
|
|
|
|
/// 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.
|
|
pub(crate) fn build_app(state: AppState) -> axum::Router {
|
|
let (router, api) =
|
|
linked_account_routes(OpenApiRouter::<AppState>::with_openapi(ApiDoc::openapi()))
|
|
.routes(routes!(routes::health))
|
|
.routes(routes!(alerts::get_alerts))
|
|
.routes(routes!(hives::get_hives))
|
|
.routes(routes!(hives::get_hives_status))
|
|
.routes(routes!(hives::get_links))
|
|
.routes(routes!(hives::get_swarm_info))
|
|
.routes(routes!(jobq_graph::get_jobq_graph))
|
|
.routes(routes!(jobq_graph::get_jobq_rollup))
|
|
.routes(routes!(config_prs::get_agent_config_pr))
|
|
.routes(routes!(agents::get_agent_icon))
|
|
.routes(routes!(agents_md::get_agents_md, agents_md::put_agents_md))
|
|
.routes(routes!(
|
|
inference::get_agent_inference,
|
|
inference::put_agent_inference,
|
|
inference::delete_agent_inference
|
|
))
|
|
.routes(routes!(
|
|
agent_limits::get_agent_limits,
|
|
agent_limits::put_agent_limits,
|
|
agent_limits::delete_agent_limits
|
|
))
|
|
.routes(routes!(
|
|
agent_permissions::get_agent_permissions,
|
|
agent_permissions::put_agent_permissions,
|
|
agent_permissions::delete_agent_permissions
|
|
))
|
|
.routes(routes!(agent_permissions::get_permission_catalogue))
|
|
.routes(routes!(
|
|
egress_allow::get_allowed_domains,
|
|
egress_allow::add_allowed_domain,
|
|
egress_allow::remove_allowed_domain
|
|
))
|
|
.routes(routes!(
|
|
egress_deny::get_denied_domains,
|
|
egress_deny::add_denied_domain,
|
|
egress_deny::remove_denied_domain
|
|
))
|
|
.routes(routes!(archetypes::get_archetypes))
|
|
.routes(routes!(config_prs::get_config_prs))
|
|
.routes(routes!(agents::create_agent))
|
|
.routes(routes!(forge_admin::mint_agent_forge_token))
|
|
.routes(routes!(forge_admin::make_forge_admin))
|
|
.routes(routes!(agents::get_agents))
|
|
.routes(routes!(agents::get_agents_status))
|
|
.routes(routes!(agents::set_agent_state))
|
|
.routes(routes!(agents::get_hive_wanted))
|
|
.routes(routes!(term_stream::stream_agent_term))
|
|
.routes(routes!(message_stream::stream_messages))
|
|
.routes(routes!(subagent_term::list_subagents))
|
|
.routes(routes!(subagent_term::stream_subagent_term))
|
|
.routes(routes!(agent_state_stream::stream_agent_state))
|
|
.routes(routes!(agent_input::post_agent_input))
|
|
.routes(routes!(agent_input::post_agents_broadcast))
|
|
.routes(routes!(agent_control::post_agent_cancel))
|
|
.routes(routes!(agent_custom::post_agent_custom))
|
|
.routes(routes!(
|
|
claude_login::post_claude_login,
|
|
claude_login::delete_claude_login
|
|
))
|
|
.routes(routes!(claude_login::post_claude_login_code))
|
|
.routes(routes!(schedules::list_schedules, schedules::add_schedule))
|
|
.routes(routes!(
|
|
schedules::edit_schedule,
|
|
schedules::remove_schedule
|
|
))
|
|
.routes(routes!(issue_report::get_repos))
|
|
.routes(routes!(issue_report::get_issue_report_all))
|
|
.routes(routes!(issue_report::get_issue_report))
|
|
.routes(routes!(pulls::get_pull_repos))
|
|
.routes(routes!(pulls::get_pulls))
|
|
.routes(routes!(review_classifier::post_try_out))
|
|
.routes(routes!(review_classifier::get_try_out))
|
|
.routes(routes!(review_classifier::get_rules))
|
|
.routes(routes!(review_classifier::open_pr::post_open_pr))
|
|
.routes(routes!(review_classifier::runs::get_runs))
|
|
.routes(routes!(builder::runs::get_runs))
|
|
.routes(routes!(builder::runs::get_run))
|
|
.routes(routes!(
|
|
review_classifier::settings::get_settings,
|
|
review_classifier::settings::put_settings
|
|
))
|
|
.routes(routes!(notify_flows::get_flows, notify_flows::put_flows))
|
|
.routes(routes!(notify_router_api::get_deliveries))
|
|
.routes(routes!(
|
|
notify_router_api::get_dry_run,
|
|
notify_router_api::post_dry_run
|
|
))
|
|
.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 std::path::Path;
|
|
|
|
use super::*;
|
|
use crate::routes::tests::state_with_roster;
|
|
|
|
/// The socket must not share a directory with anything else, because
|
|
/// the socket is `0666` and the directory is therefore the only access
|
|
/// control it has. `/run/hyperhive` in particular holds hive-c0re's
|
|
/// **admin** socket, and nginx — a host service — is bounded by
|
|
/// nothing but the directory itself.
|
|
///
|
|
/// A test rather than a comment: the failure this guards against is a
|
|
/// one-word edit that looks tidier and reads fine in review.
|
|
#[test]
|
|
fn socket_lives_in_its_own_runtime_dir() {
|
|
let parent = Path::new(DEFAULT_SOCKET)
|
|
.parent()
|
|
.expect("socket path has a parent directory");
|
|
assert_eq!(
|
|
parent,
|
|
Path::new("/run/swarm-controller"),
|
|
"the socket's directory is its access control — moving it under a shared \
|
|
directory (notably /run/hyperhive, which holds the host admin socket) \
|
|
exposes everything else in that directory to the gateway's nginx"
|
|
);
|
|
}
|
|
|
|
/// One test, not three, deliberately: `HIVES_ENV` is a real process
|
|
/// env var, and `cargo test`'s default parallel runner would race
|
|
/// separate missing/malformed/valid tests against each other. Driving
|
|
/// all three states sequentially inside one test needs no mutex and
|
|
/// still proves each branch of `load_hives`.
|
|
///
|
|
/// SAFETY: single-threaded mutation of a process env var no other
|
|
/// test in this crate reads; restored (removed) before returning.
|
|
#[test]
|
|
fn load_hives_covers_missing_malformed_and_valid() {
|
|
unsafe {
|
|
std::env::remove_var(HIVES_ENV);
|
|
}
|
|
assert_eq!(
|
|
load_hives(),
|
|
Vec::<HiveEntry>::new(),
|
|
"unset env var is an empty directory, not a startup failure"
|
|
);
|
|
|
|
unsafe {
|
|
std::env::set_var(HIVES_ENV, "not json");
|
|
}
|
|
assert_eq!(
|
|
load_hives(),
|
|
Vec::<HiveEntry>::new(),
|
|
"unparseable env var falls back to empty rather than panicking — \
|
|
/health must still answer even if this daemon's own config is wrong"
|
|
);
|
|
|
|
unsafe {
|
|
std::env::set_var(
|
|
HIVES_ENV,
|
|
r#"[{"name":"pr1ma","domain":"pr1ma.example.com","dashboard":"pr1ma.hives.example"},{"name":"umbra","domain":"umbra.example.com","dashboard":"umbra.hives.example"}]"#,
|
|
);
|
|
}
|
|
assert_eq!(
|
|
load_hives(),
|
|
vec![
|
|
HiveEntry {
|
|
name: "pr1ma".to_string(),
|
|
domain: "pr1ma.example.com".to_string(),
|
|
dashboard: "pr1ma.hives.example".to_string(),
|
|
},
|
|
HiveEntry {
|
|
name: "umbra".to_string(),
|
|
domain: "umbra.example.com".to_string(),
|
|
dashboard: "umbra.hives.example".to_string(),
|
|
},
|
|
]
|
|
);
|
|
|
|
unsafe {
|
|
std::env::remove_var(HIVES_ENV);
|
|
}
|
|
}
|
|
|
|
/// Same three-state coverage as `load_hives_covers_missing_malformed_and_valid`,
|
|
/// same reason for one test rather than three (a shared process env var).
|
|
///
|
|
/// SAFETY: single-threaded mutation of a process env var no other test
|
|
/// in this crate reads; restored (removed) before returning.
|
|
#[test]
|
|
fn load_links_covers_missing_malformed_and_valid() {
|
|
unsafe {
|
|
std::env::remove_var(LINKS_ENV);
|
|
}
|
|
assert_eq!(
|
|
load_links(),
|
|
Vec::<ServiceLink>::new(),
|
|
"unset env var is an empty list, not a startup failure"
|
|
);
|
|
|
|
unsafe {
|
|
std::env::set_var(LINKS_ENV, "not json");
|
|
}
|
|
assert_eq!(
|
|
load_links(),
|
|
Vec::<ServiceLink>::new(),
|
|
"unparseable env var falls back to empty rather than panicking"
|
|
);
|
|
|
|
unsafe {
|
|
std::env::set_var(
|
|
LINKS_ENV,
|
|
r#"[{"label":"Authelia","icon":"🔑","url":"https://auth.example.com/"}]"#,
|
|
);
|
|
}
|
|
assert_eq!(
|
|
load_links(),
|
|
vec![ServiceLink {
|
|
label: "Authelia".to_string(),
|
|
icon: "🔑".to_string(),
|
|
url: "https://auth.example.com/".to_string(),
|
|
}]
|
|
);
|
|
|
|
unsafe {
|
|
std::env::remove_var(LINKS_ENV);
|
|
}
|
|
}
|
|
|
|
/// Two states, not three like `load_hives`/`load_links` above: there is
|
|
/// nothing to parse here, so no malformed-input branch exists to cover.
|
|
///
|
|
/// SAFETY: single-threaded mutation of a process env var no other test
|
|
/// in this crate reads; restored (removed) before returning.
|
|
#[test]
|
|
fn load_swarm_name_covers_missing_and_set() {
|
|
unsafe {
|
|
std::env::remove_var(NAME_ENV);
|
|
}
|
|
assert_eq!(
|
|
load_swarm_name(),
|
|
None,
|
|
"unset env var is an unnamed swarm, not a startup failure"
|
|
);
|
|
|
|
unsafe {
|
|
std::env::set_var(NAME_ENV, "constellat1on");
|
|
}
|
|
assert_eq!(load_swarm_name(), Some("constellat1on".to_string()));
|
|
|
|
unsafe {
|
|
std::env::remove_var(NAME_ENV);
|
|
}
|
|
}
|
|
|
|
// ── NotConnected maps to 503, everything else stays 500 ─────────────
|
|
|
|
/// Every inference verb refuses a name that is not an identifier with a
|
|
/// 400 and answers 503 without a queue; a PUT whose body is refused is a
|
|
/// 400 before the queue is asked.
|
|
#[tokio::test]
|
|
async fn the_inference_routes_check_the_name_and_body_before_the_queue() {
|
|
use crate::inference::{delete_agent_inference, get_agent_inference, put_agent_inference};
|
|
use axum::extract::{Path, State};
|
|
use axum::http::StatusCode;
|
|
|
|
fn status<T>(r: Result<T, problem_details::ProblemDetails>) -> Option<StatusCode> {
|
|
r.map(|_| ()).expect_err("must not succeed").status
|
|
}
|
|
let (state, _sched) = state_with_roster();
|
|
let body = |json: &str| axum::Json(serde_json::from_str(json).expect("decodes"));
|
|
for name in ["iris", "Not An Ident"] {
|
|
let want = Some(if name == "iris" {
|
|
StatusCode::SERVICE_UNAVAILABLE
|
|
} else {
|
|
StatusCode::BAD_REQUEST
|
|
});
|
|
let path = || Path(name.to_owned());
|
|
assert_eq!(
|
|
status(get_agent_inference(State(state.clone()), path()).await),
|
|
want,
|
|
"GET {name}"
|
|
);
|
|
assert_eq!(
|
|
status(
|
|
put_agent_inference(State(state.clone()), path(), body(r#"{"model":"opus"}"#))
|
|
.await
|
|
),
|
|
want,
|
|
"PUT {name}"
|
|
);
|
|
assert_eq!(
|
|
status(delete_agent_inference(State(state.clone()), path()).await),
|
|
want,
|
|
"DELETE {name}"
|
|
);
|
|
}
|
|
for bad in [r#"{"model":"opus 4"}"#, r#"{"model":"--print"}"#] {
|
|
assert_eq!(
|
|
status(
|
|
put_agent_inference(State(state.clone()), Path("iris".to_owned()), body(bad))
|
|
.await
|
|
),
|
|
Some(StatusCode::BAD_REQUEST),
|
|
"{bad}"
|
|
);
|
|
}
|
|
}
|
|
|
|
#[tokio::test]
|
|
async fn listing_subagents_needs_a_queue_and_an_identifier() {
|
|
let (state, _sched) = state_with_roster();
|
|
let no_queue = crate::subagent_term::list_subagents(
|
|
axum::extract::State(state.clone()),
|
|
axum::extract::Path("iris".to_owned()),
|
|
)
|
|
.await
|
|
.expect_err("no queue is configured");
|
|
assert_eq!(
|
|
no_queue.status,
|
|
Some(axum::http::StatusCode::SERVICE_UNAVAILABLE),
|
|
"{no_queue:?}"
|
|
);
|
|
let bad_name = crate::subagent_term::list_subagents(
|
|
axum::extract::State(state),
|
|
axum::extract::Path("iris.sub".to_owned()),
|
|
)
|
|
.await
|
|
.expect_err("a dot would widen the subject");
|
|
assert_eq!(
|
|
bad_name.status,
|
|
Some(axum::http::StatusCode::BAD_REQUEST),
|
|
"{bad_name:?}"
|
|
);
|
|
}
|
|
|
|
/// A subagent name is a subject token, so a wildcard or a dot in it is
|
|
/// refused before anything subscribes.
|
|
#[tokio::test]
|
|
async fn a_subagent_stream_refuses_a_name_that_is_not_one_token() {
|
|
let (state, _sched) = state_with_roster();
|
|
for subagent in ["*", ">", "a.b"] {
|
|
let Err(problem) = crate::subagent_term::stream_subagent_term(
|
|
axum::extract::State(state.clone()),
|
|
axum::extract::Path(("iris".to_owned(), subagent.to_owned())),
|
|
axum::http::HeaderMap::new(),
|
|
)
|
|
.await
|
|
else {
|
|
panic!("{subagent:?} must be refused");
|
|
};
|
|
assert_eq!(
|
|
problem.status,
|
|
Some(axum::http::StatusCode::BAD_REQUEST),
|
|
"{subagent:?}: {problem:?}"
|
|
);
|
|
}
|
|
}
|
|
|
|
// Paused time: each sleep completes at once and advances the clock.
|
|
#[tokio::test(start_paused = true)]
|
|
async fn registration_waits_for_the_secret_and_then_registers_with_it() {
|
|
let secret = Late::<Arc<str>>::default();
|
|
let registered = Arc::new(Mutex::new(Vec::<String>::new()));
|
|
let filling = Arc::clone(&secret);
|
|
tokio::spawn(async move {
|
|
tokio::time::sleep(std::time::Duration::from_secs(100)).await;
|
|
filling.get_or_init(|| Arc::from("the-secret"));
|
|
});
|
|
|
|
let seen = Arc::clone(®istered);
|
|
register_when_ready(&secret, move |secret| {
|
|
let seen = Arc::clone(&seen);
|
|
async move {
|
|
seen.lock().unwrap().push(secret.to_string());
|
|
Ok(())
|
|
}
|
|
})
|
|
.await;
|
|
|
|
assert_eq!(*registered.lock().unwrap(), ["the-secret"]);
|
|
}
|
|
|
|
#[tokio::test(start_paused = true)]
|
|
async fn a_forge_that_fails_past_the_old_bound_is_still_registered_with() {
|
|
let secret = Late::<Arc<str>>::default();
|
|
secret.get_or_init(|| Arc::from("the-secret"));
|
|
let start = tokio::time::Instant::now();
|
|
let mut calls = 0;
|
|
|
|
register_when_ready(&secret, |_| {
|
|
calls += 1;
|
|
let n = calls;
|
|
async move {
|
|
if n < 9 {
|
|
anyhow::bail!("502 Bad Gateway")
|
|
}
|
|
Ok(())
|
|
}
|
|
})
|
|
.await;
|
|
|
|
assert_eq!(calls, 9);
|
|
// 15, 30, 60, 120, 240, then the 300s cap three times.
|
|
assert_eq!(
|
|
start.elapsed().as_secs(),
|
|
15 + 30 + 60 + 120 + 240 + 3 * 300
|
|
);
|
|
}
|
|
}
|