From 66c494697f3d90bcf23dbfbfdb589e5f51a318db Mon Sep 17 00:00:00 2001 From: damocles Date: Wed, 19 Aug 2026 20:11:00 +0200 Subject: [PATCH] swarm-controller: track agent config-PR status independently of hives --- swarm-controller/src/config_pr.rs | 91 +++++++++++++++++++++++++++++++ swarm-controller/src/forge.rs | 79 ++++++++++++++++++++++++++- swarm-controller/src/main.rs | 49 ++++++++++++++++- 3 files changed, 217 insertions(+), 2 deletions(-) create mode 100644 swarm-controller/src/config_pr.rs diff --git a/swarm-controller/src/config_pr.rs b/swarm-controller/src/config_pr.rs new file mode 100644 index 00000000..4eef46e7 --- /dev/null +++ b/swarm-controller/src/config_pr.rs @@ -0,0 +1,91 @@ +//! Swarm-level config-PR status, polled independently of any one hive. +//! +//! `hive-c0re::forge::config_pr_poll` already scans `agent-configs/*` for +//! open PRs — but it does that once per hive, to queue that hive's own +//! `MergeConfigPr` approval, and nothing at swarm level reads the result. +//! swarm-ui's config-PR panel needs a *swarm*-level answer to "does agent X +//! have an open config PR" that does not depend on which hive currently +//! hosts X being reachable. +//! +//! This mirrors `hive-c0re`'s poll shape (period, idempotent full rescan) +//! rather than consuming the swarm-wide `pull_request` webhook +//! (`crate::webhook::DeliveryKind::ConfigPr`) that already lands here: that +//! delivery is unparsed today, and building a reliable event path means +//! building this poll as a missed-delivery backstop anyway (the same reason +//! `hive-c0re`'s own poller exists). Shipping the backstop alone first +//! avoids paying for both at once; wiring the webhook as a low-latency nudge +//! on top is a cheap follow-up once this cache is the source of truth. + +use std::collections::HashMap; +use std::sync::{Arc, Mutex}; + +use crate::forge::{Client, ConfigPrStatus}; + +/// How often to rescan `agent-configs/*`. Matches the interval named in +/// `hive-c0re::forge::config_pr_poll`'s own doc comment — same org, same +/// staleness tolerance, no reason for the two to disagree. +const POLL_INTERVAL: std::time::Duration = std::time::Duration::from_mins(5); + +/// The latest full scan, replaced atomically each cycle. +/// +/// A full replace rather than an incremental merge: `Client::list_open_config_prs` +/// already returns the complete current set (an agent with no open PR is +/// simply absent), so merging would need its own stale-entry eviction to +/// avoid an agent's long-closed PR lingering forever — the same reconcile +/// problem `hive-c0re`'s poller solves for its approvals. A full replace +/// sidesteps needing that logic twice: the map at any moment is nothing more +/// than "the last successful scan's answer." +pub struct ConfigPrCache(Mutex>); + +impl ConfigPrCache { + fn new() -> Self { + Self(Mutex::new(HashMap::new())) + } + + /// `agent`'s open PR, if the last successful scan found one. + /// + /// Returns `None` both when the agent has no open PR and when no scan + /// has completed yet — the caller (`GET /api/agents//config-pr`) + /// treats both as "nothing to show," which is the honest answer for a + /// value that's a best-effort cache, not a live read. + pub fn get(&self, agent: &str) -> Option { + self.0 + .lock() + .unwrap_or_else(std::sync::PoisonError::into_inner) + .get(agent) + .cloned() + } + + fn replace(&self, scan: HashMap) { + *self + .0 + .lock() + .unwrap_or_else(std::sync::PoisonError::into_inner) = scan; + } +} + +/// Build an empty cache and spawn the periodic scan that keeps it current. +/// +/// The scan runs immediately on the first tick (`tokio::time::interval`'s +/// default), so the cache is populated on startup rather than staying empty +/// for a full [`POLL_INTERVAL`] after boot. +pub fn spawn(client: Arc) -> Arc { + let cache = Arc::new(ConfigPrCache::new()); + let cache_for_task = Arc::clone(&cache); + tokio::spawn(async move { + let mut ticker = tokio::time::interval(POLL_INTERVAL); + loop { + ticker.tick().await; + match client.list_open_config_prs().await { + Ok(scan) => cache_for_task.replace(scan), + Err(e) => { + tracing::warn!( + error = %format!("{e:#}"), + "config-pr poll: scan failed, cache keeps its last value" + ); + } + } + } + }); + cache +} diff --git a/swarm-controller/src/forge.rs b/swarm-controller/src/forge.rs index d524ccae..897d7640 100644 --- a/swarm-controller/src/forge.rs +++ b/swarm-controller/src/forge.rs @@ -20,14 +20,26 @@ use forgejo_api::structs::{ AddCollaboratorOption, AddCollaboratorOptionPermission, ChangeFileOperation, ChangeFileOperationOperation, ChangeFilesOptions, CreateBranchProtectionOption, CreateHookOption, CreateHookOptionConfig, CreateHookOptionType, CreateRepoOption, - RepoGetContentsQuery, + RepoGetContentsQuery, RepoListPullRequestsQuery, RepoListPullRequestsQueryState, }; use forgejo_api::{ApiErrorKind, Auth, Forgejo, ForgejoError}; use reqwest::StatusCode; +use serde::Serialize; use std::collections::BTreeMap; +use utoipa::ToSchema; use crate::webhook::DeliveryKind; +/// An agent's open config-PR, as [`Client::list_open_config_prs`] reports it +/// and [`crate::get_agent_config_pr`] serves it. +#[derive(Clone, Debug, PartialEq, Serialize, ToSchema)] +pub struct ConfigPrStatus { + pub pr_number: u64, + /// Absent only if Forgejo itself omitted the field — every real PR has + /// one; not worth failing the whole scan over. + pub html_url: Option, +} + /// The `operators` team, whitelisted for the merge gate on every repo /// this client protects — provisioned by `hive-c0re::forge::repos` /// already (`ensure_operators_team`), not re-provisioned here. If that @@ -351,6 +363,71 @@ impl Client { } } + /// Every agent in [`CONFIG_ORG`] with an open config PR, keyed by agent + /// name. Mirrors `hive-c0re::forge::config_pr_poll::poll_open_config_prs`'s + /// scan shape (list repos in the org, list open PRs per repo) but returns + /// data instead of side-effecting an approval queue — this daemon has no + /// approval system of its own; it exists so [`crate::get_agent_config_pr`] + /// has something to answer from, independent of any one hive being up. + /// + /// A single repo's list failing does not fail the whole scan — logged and + /// skipped, so one flaky repo can't blank out every other agent's status. + pub async fn list_open_config_prs( + &self, + ) -> Result> { + let repos = self + .api + .org_list_repos(CONFIG_ORG) + .all() + .await + .with_context(|| format!("list repos in {CONFIG_ORG}"))?; + + let mut out = std::collections::HashMap::new(); + for repo in repos { + let Some(agent) = repo.name.as_deref() else { + continue; + }; + let query = RepoListPullRequestsQuery { + state: Some(RepoListPullRequestsQueryState::Open), + sort: None, + milestone: None, + labels: None, + poster: None, + base: None, + head: None, + }; + let prs = match self + .api + .repo_list_pull_requests(CONFIG_ORG, agent, query) + .all() + .await + { + Ok(prs) => prs, + Err(e) => { + tracing::debug!(%agent, error = %e, "swarm forge: listing config PRs failed, skipping repo"); + continue; + } + }; + // Only the first open PR matters for the panel — a config repo + // is meant to carry at most one live proposal at a time (the + // same assumption `hive-c0re`'s poller and the `MergeConfigPr` + // approval flow both make). + if let Some(pr) = prs.into_iter().next() { + let Some(pr_number) = pr.number.and_then(|n| u64::try_from(n).ok()) else { + continue; + }; + out.insert( + agent.to_owned(), + ConfigPrStatus { + pr_number, + html_url: pr.html_url.map(|u| u.to_string()), + }, + ); + } + } + Ok(out) + } + /// Register the swarm-wide hooks against this controller, so a real /// forge event reaches [`crate::webhook`] instead of the endpoint only /// being reachable by hand. diff --git a/swarm-controller/src/main.rs b/swarm-controller/src/main.rs index 703faf65..e5280a7a 100644 --- a/swarm-controller/src/main.rs +++ b/swarm-controller/src/main.rs @@ -29,13 +29,18 @@ use std::path::PathBuf; use std::sync::{Arc, Mutex}; use anyhow::{Context, Result}; -use axum::{Json, extract::State, routing::get}; +use axum::{ + Json, + extract::{Path, State}, + routing::get, +}; use hive_jobq_wire::GraphWire as _; use serde::{Deserialize, Serialize}; use utoipa::{OpenApi, ToSchema}; use utoipa_axum::{router::OpenApiRouter, routes}; mod auth; +mod config_pr; mod forge; mod status; mod webhook; @@ -352,6 +357,11 @@ struct AppState { /// 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 @@ -760,6 +770,38 @@ async fn get_jobq_rollup(State(state): State) -> Json), + (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))) +} + /// 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 @@ -938,6 +980,8 @@ async fn main() -> Result<()> { } }; + let config_prs = forge_client.clone().map(config_pr::spawn); + register_swarm_webhooks(forge_client, webhook_secret.clone()); let state = AppState { @@ -946,6 +990,7 @@ async fn main() -> Result<()> { status, jobq, webhook_secret, + config_prs, swarm_name: load_swarm_name().map(Arc::from), }; @@ -957,6 +1002,7 @@ async fn main() -> Result<()> { .routes(routes!(get_swarm_info)) .routes(routes!(get_jobq_graph)) .routes(routes!(get_jobq_rollup)) + .routes(routes!(get_agent_config_pr)) .routes(routes!(create_agent)) .routes(routes!(webhook::post_webhook_forge)) .split_for_parts(); @@ -1063,6 +1109,7 @@ mod tests { status: None, jobq: std::sync::Arc::clone(&sched), webhook_secret: None, + config_prs: None, swarm_name: None, }; (state, sched)