hyperhive/swarm-controller/src/main.rs

//! Swarm-level controller daemon. Runs as the unprivileged
//! `swarm-controller` user on whichever host the operator flips
//! `services.hyperhive.deploy.swarm-controller.enable` on, and serves HTTP over a
//! unix socket that the hive-gateway's nginx proxies to.
//!
//! Configuration is read-only and loaded once at startup from env vars the
//! NixOS module sets (`services.hyperhive.swarm.controller`) — see
//! `startup::load_hives`. A config change means a redeploy, same as every other
//! option this process reads.
//!
//! It keeps the webhook HMAC secret in the swarm secret store (see `webhook`),
//! because that key is handed to Forgejo at registration and so cannot be
//! regenerated per boot, and the operator admin sync's granted set under its
//! `StateDirectory` (see `matrix_account::operator_admin`).
//!
//! Distinct from `hive-c0re`, which is per-hive: c0re owns the agents on
//! one host, this owns what is true across hives.

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

use anyhow::{Context, Result};

mod agent_archetype;
mod agent_control;
mod agent_custom;
mod agent_entity;
mod agent_icon;
mod agent_identity;
mod agent_input;
mod agent_limits;
mod agent_login;
mod agent_permissions;
mod agent_renewal;
mod agent_role_ttl;
mod agent_state_stream;
mod agent_status;
mod agents_md;
mod agents_md_initial;
mod alerts;
mod archetypes;
mod attic_copy;
mod auth;
mod backend_egress;
mod backend_env;
mod builder;
mod cache_miss;
mod claude_login;
mod config_drift;
mod config_pr;
mod egress_allow;
mod egress_deny;
mod forge;
mod forge_account;
mod forge_events;
mod github_account;
mod gitlab_account;
mod held_deploys;
mod inference;
mod issue_report;
mod jobq;
mod link_egress;
mod linked_accounts;
mod matrix_account;
mod message_stream;
mod meta_build;
mod meta_fanout;
mod meta_flake;
mod notify_flows;
mod notify_router_api;
mod observability_grant;
mod otel_http_client;
mod proxy_credential;
mod pulls;
mod queue_identity;
mod read_policy;
mod request_reviewer;
mod review_classifier;
mod routes;
mod schedules;
mod startup;
mod status;
mod store;
mod subagent_term;
mod term_stream;
#[cfg(test)]
mod test_nats;
mod vcs_metrics;
mod wanted;
mod webhook;

use jobq::{Late, WorkerDeps, spawn_jobq_worker};
use routes::AppState;
use startup::{
    agent_icon_reader, agent_status_reader, agents_md_reader, build_app,
    configured_matrix_homeserver, connect_status_reader, load_hives, load_links, load_swarm_name,
    reported_inference, socket_path, spawn_agent_renewal, spawn_matrix_account_backfill,
    spawn_per_agent_upkeep, start_forge, wanted_writer,
};

#[tokio::main]
async fn main() -> Result<()> {
    hive_log::init();
    swarm_secret_client::install_crypto_provider();

    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.**
    // The unit makes the directory `0750`, group `swarm-controller-clients`,
    // so only nginx and root reach the socket. The directory holds one
    // socket, which 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");

    // Before every user of the identity: the queue, the bridge and both OTLP
    // exporters read it from `queue_identity::get`.
    queue_identity::init().await?;
    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 access still serves
    // everything else, and the forge-shaped node kinds give an honest
    // per-job failure rather than this fn refusing to start. Both slots stay
    // empty until `start_forge` below fills them.
    let forge_client: Late<Arc<forge::Client>> = Arc::default();
    let config_prs: Late<Arc<config_pr::ConfigPrCache>> = Arc::default();
    let deps = WorkerDeps {
        auth: auth.clone(),
        forge: Arc::clone(&forge_client),
        queue: status.as_ref().map(|s| s.queue_client()),
        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(),
    )));
    let hives = load_hives();
    spawn_matrix_account_backfill(&jobq, deps.matrix_homeserver.clone(), &hives, auth.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<dyn opentelemetry_http::HttpClient>);
    let jobq_metrics_provider =
        hive_jobq_metrics::spawn_exporter(Arc::clone(&jobq), "swarm-controller", jobq_http_client);
    register_cert_gauges(jobq_metrics_provider.as_ref());
    held_deploys::register(jobq_metrics_provider.as_ref(), &jobq);

    let webhook_secret = webhook::load().await;

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

    start_forge(
        Arc::clone(&forge_client),
        Arc::clone(&config_prs),
        Arc::clone(&jobq),
        hives.clone(),
        Arc::clone(&webhook_secret),
        auth.clone(),
        status.as_ref().map(|s| s.queue_client()),
    )
    .await;

    spawn_agent_renewal(&jobq, wanted_writer(status.as_ref()), &hives);
    spawn_per_agent_upkeep(status.as_ref(), &hives);
    // Before serving, because a hive whose role does not exist cannot log in,
    // and one whose policy does not exist logs in able to read nothing —
    // either way it cannot collect what this daemon writes for it. A store
    // that is not up yet is retried in the background instead of being
    // abandoned, since the two of them boot together.
    let hive_names: Vec<String> = hives.iter().map(|h| h.name.clone()).collect();
    read_policy::ensure_hive_access(hive_names).await;
    // The one policy every agent's role attaches, and the entity each agent's
    // login is read under, are checked here at every start, and so is the
    // token lifetime of the roles that exist.
    read_policy::ensure_agent_policies().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()),
        agent_icons: agent_icon_reader(status.as_ref()),
        agents_md: agents_md_reader(status.as_ref()),
        reported_inference: reported_inference(status.as_ref()),
        agent_login: startup::agent_login(status.as_ref()),
        runs: review_classifier::runs::Runs::Live(status.as_ref().map(|s| s.queue_client())),
        builds: builder::runs::Builds::Live(status.as_ref().map(|s| s.queue_client())),
        status,
        jobq,
        webhook_secret,
        config_prs,
        swarm_name: load_swarm_name().map(Arc::from),
        auth,
        forge: forge_client,
        create_gate: Arc::default(),
        accounts: linked_accounts::Accounts::Live,
        try_outs: review_classifier::TryOuts::default(),
        reviews: Arc::default(),
        alerts: alerts::from_env(),
    };
    serve(listener, state).await
}

/// Registers the `bao.client_cert.*` gauges for the controller's own store
/// certificate (`controller.pem`) on the jobq exporter, the one provider built
/// at startup rather than on first use. Nothing when no exporter is configured.
fn register_cert_gauges(provider: Option<&opentelemetry_sdk::metrics::SdkMeterProvider>) {
    use opentelemetry::metrics::MeterProvider as _;
    if let Some(provider) = provider {
        swarm_secret_client::cert_metrics::register(
            &provider.meter("hyperhive.bao_client_cert"),
            "controller",
        );
    }
}

/// Start taking the queued review requests and answering the notification
/// router's, then serve `state` on `listener`.
async fn serve(listener: tokio::net::UnixListener, state: AppState) -> Result<()> {
    review_classifier::review::spawn_queue(&state);
    request_reviewer::spawn(&state);
    axum::serve(listener, build_app(state))
        .await
        .context("serving swarm-controller")
}