From 1d31bb6e80de7b91fb17b04f2e7ab79e38efcc41 Mon Sep 17 00:00:00 2001 From: damocles Date: Sun, 16 Aug 2026 23:05:13 +0200 Subject: [PATCH] swarm-controller: CreateRepo/AddRepoMember/InitAgentConfigRepo forge nodes --- Cargo.lock | 3 + swarm-controller/Cargo.toml | 18 +- swarm-controller/src/forge.rs | 410 ++++++++++++++++++++++++++++++++++ swarm-controller/src/main.rs | 262 ++++++++++++++++++---- 4 files changed, 647 insertions(+), 46 deletions(-) create mode 100644 swarm-controller/src/forge.rs diff --git a/Cargo.lock b/Cargo.lock index c4c5b8ed..3e1d5ad1 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -4585,6 +4585,8 @@ dependencies = [ "anyhow", "async-nats", "axum", + "base64", + "forgejo-api", "futures-util", "hive-jobq", "hive-jobq-wire", @@ -4596,6 +4598,7 @@ dependencies = [ "tokio", "tracing", "tracing-subscriber", + "url", "utoipa", "utoipa-axum", ] diff --git a/swarm-controller/Cargo.toml b/swarm-controller/Cargo.toml index b94ebe9f..69e94710 100644 --- a/swarm-controller/Cargo.toml +++ b/swarm-controller/Cargo.toml @@ -17,17 +17,26 @@ anyhow.workspace = true # neither, does not claim to need them. async-nats = { workspace = true, features = ["kv"] } axum.workspace = true +# swarm-controller's own forge client (`forge.rs`) — self-contained, +# deliberately not sharing code with `hive-c0re::forge` across the crate +# boundary (see #3306's design discussion: forcing that split now, over a +# few idempotent CRUD-ish calls, is premature plumbing). +forgejo-api.workspace = true +# Only for base64-encoding file content for `forge.rs`'s +# `repo_change_files` calls — forgejo's content API takes base64, never +# raw bytes. +base64.workspace = true futures-util.workspace = true # The graph itself, held directly rather than behind a c0re-style wrapper # module — that layering (`hive-c0re::job_queue`) is partially legacy (predates # `hive-jobq`'s extraction into its own crate) and this daemon does not need it -# repeated. No `Scheduler` yet either: nothing here submits or executes a job, -# so there is nothing to schedule — just a `Graph` for the read-only endpoints -# to serve. +# repeated. Driven by `hive_jobq::scheduler::Scheduler` (`spawn_jobq_worker`), +# same shape `hive-c0re/src/job_queue/scheduler.rs` uses over its own graph. hive-jobq.workspace = true hive-jobq-wire.workspace = true # `auth`'s bridge client — same crate the bridge itself uses to define the -# request/response shape, so the two ends cannot drift. +# request/response shape, so the two ends cannot drift. `forge.rs` also +# uses this directly for `StatusCode` in its error-classification helpers. reqwest.workspace = true serde.workspace = true serde_json.workspace = true @@ -44,6 +53,7 @@ swarm-queue-client = { workspace = true, features = ["kv"] } tokio.workspace = true tracing.workspace = true tracing-subscriber.workspace = true +url.workspace = true utoipa.workspace = true utoipa-axum.workspace = true diff --git a/swarm-controller/src/forge.rs b/swarm-controller/src/forge.rs new file mode 100644 index 00000000..af4a2089 --- /dev/null +++ b/swarm-controller/src/forge.rs @@ -0,0 +1,410 @@ +//! Swarm-controller's own forgejo client. Self-contained — deliberately +//! not sharing code with `hive-c0re::forge` across the crate boundary +//! (forcing that split now, over a handful of idempotent CRUD-ish +//! calls, is premature plumbing; factor out a shared crate later if the +//! duplication actually starts to hurt). +//! +//! Reads its identity from the credentials `swarm-controller.nix`'s +//! `forgeEnv` wires in (`SWARM_CONTROLLER_FORGE_URL` + +//! `SWARM_CONTROLLER_FORGE_TOKEN_FILE`, the latter pointing at a +//! systemd `LoadCredential`-delivered file) — see that module and +//! `hive-forge/default.nix`'s `forgejo-swarm-controller-account` / +//! `hive-forge-swarm-controller-token` units for how the token itself +//! gets minted and delivered. Absent env (forge not configured on this +//! host) means [`Client::from_env`] returns `None` — every caller +//! treats that as "no forge access here" and continues, the same +//! graceful-absence shape the queue coordinates already use. + +use anyhow::{Context, Result}; +use forgejo_api::structs::{ + AddCollaboratorOption, AddCollaboratorOptionPermission, ChangeFileOperation, + ChangeFileOperationOperation, ChangeFilesOptions, CreateBranchProtectionOption, + CreateRepoOption, RepoGetContentsQuery, +}; +use forgejo_api::{ApiErrorKind, Auth, Forgejo, ForgejoError}; +use reqwest::StatusCode; + +/// The forge org that owns agent repos. Same org `hive-c0re::forge` +/// already uses for its own single-hive `CreateRepo` path — this is +/// the same forge instance, not a separate one, so the same org. +pub const AGENTS_ORG: &str = "agents"; + +/// 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 +/// assumption ever breaks (this becomes reachable before any hive's +/// `hive-c0re` has run its startup sweep), branch-protection creation +/// below will fail loudly rather than silently no-op — see +/// [`apply_operator_branch_protection`]'s doc comment. +const OPERATORS_TEAM: &str = "operators"; + +/// The env vars `swarm-controller.nix`'s `forgeEnv` sets. Named here +/// once so [`Client::from_env`] and the nix module can't drift silently +/// — a rename on one side without the other fails loudly (env var +/// absent) rather than silently losing forge access. +const URL_ENV: &str = "SWARM_CONTROLLER_FORGE_URL"; +const TOKEN_FILE_ENV: &str = "SWARM_CONTROLLER_FORGE_TOKEN_FILE"; + +/// Typed forgejo client, built once at startup from env. +pub struct Client { + api: Forgejo, +} + +impl Client { + /// Build from env. `Ok(None)` (not an error) when neither env var is + /// set — forge isn't configured on this host, which is a normal, + /// expected deployment shape (see the module doc comment), not a + /// failure. `Err` only for a genuine misconfiguration: one of the + /// two env vars present without the other, the token file unreadable, + /// or the URL/client failing to build — those ARE failures, because + /// they mean forge was *meant* to be configured here and isn't + /// usable, which should be loud rather than silently downgrading to + /// "no forge access." + pub fn from_env() -> Result> { + let url = std::env::var(URL_ENV).ok(); + let token_file = std::env::var(TOKEN_FILE_ENV).ok(); + let (url, token_file) = match (url, token_file) { + (None, None) => return Ok(None), + (Some(url), Some(token_file)) => (url, token_file), + (Some(_), None) => { + anyhow::bail!("{URL_ENV} is set but {TOKEN_FILE_ENV} is not — partial forge config") + } + (None, Some(_)) => { + anyhow::bail!("{TOKEN_FILE_ENV} is set but {URL_ENV} is not — partial forge config") + } + }; + let token = std::fs::read_to_string(&token_file) + .with_context(|| format!("read forge token from {token_file}"))?; + let token = token.trim(); + if token.is_empty() { + anyhow::bail!("forge token file {token_file} is empty"); + } + let parsed_url = + url::Url::parse(&url).with_context(|| format!("parse {URL_ENV} ({url}) as a URL"))?; + let api = Forgejo::new(Auth::Token(token), parsed_url).context("build forgejo client")?; + Ok(Some(Self { api })) + } + + /// Creation options for an empty repo defaulting to `main`. + fn repo_option(name: &str, private: bool) -> CreateRepoOption { + CreateRepoOption { + auto_init: Some(false), + default_branch: Some("main".to_owned()), + description: None, + gitignores: None, + issue_labels: None, + license: None, + name: name.to_owned(), + object_format_name: None, + private: Some(private), + readme: None, + template: None, + trust_model: None, + } + } + + /// Create `name` inside [`AGENTS_ORG`]. Idempotent: an existing repo + /// (409/422) is folded into success. + async fn ensure_org_repo(&self, name: &str) -> Result<()> { + match self + .api + .create_org_repo(AGENTS_ORG, Self::repo_option(name, true)) + .await + { + Ok(_) => { + tracing::info!(%name, "swarm forge: created repo in {AGENTS_ORG}"); + Ok(()) + } + Err(e) if is_already_exists(&e) => { + tracing::debug!(%name, "swarm forge: repo already exists"); + Ok(()) + } + Err(e) => Err(e).with_context(|| format!("create repo {AGENTS_ORG}/{name}")), + } + } + + /// Add `user` as a collaborator on `AGENTS_ORG/repo` at `permission`. + /// Idempotent: re-adding an existing collaborator just updates its + /// permission (forgejo answers 204 either way; a 201 from older + /// server versions is tolerated defensively). + async fn add_collaborator( + &self, + repo: &str, + user: &str, + permission: AddCollaboratorOptionPermission, + ) -> Result<()> { + let res = self + .api + .repo_add_collaborator( + AGENTS_ORG, + repo, + user, + AddCollaboratorOption { + permission: Some(permission), + }, + ) + .await; + match res { + Ok(()) => {} + Err(ForgejoError::UnexpectedStatusCode(s)) if s == StatusCode::CREATED => {} + Err(e) => { + return Err(e) + .with_context(|| format!("add collaborator {user} to {AGENTS_ORG}/{repo}")); + } + } + tracing::debug!(%repo, %user, ?permission, "swarm forge: collaborator set"); + Ok(()) + } + + /// Apply the operator merge-gate branch protection to `repo`'s + /// default branch — only [`OPERATORS_TEAM`] members can merge, and + /// an approving review from that team is required, so the agent (a + /// write-level collaborator, not in the team) cannot merge its own + /// PR. Mirrors `hive-c0re::forge::repos::apply_operator_branch_protection` + /// exactly (same policy, same verify-don't-trust shape): a create + /// failure is ambiguous (already-exists vs. a silent reject that + /// created no rule), so on any error this GETs the `main` rule and + /// only treats it as success if the rule is actually present — a + /// fail-open merge gate is a security bug, not a shrug. + /// + /// Assumes [`OPERATORS_TEAM`] already exists in [`AGENTS_ORG`] + /// (provisioned by `hive-c0re::forge::repos::ensure_operators_team` + /// on its own startup sweep, not re-provisioned here) — if it + /// doesn't yet, this fails loudly rather than silently leaving the + /// repo unprotected, which is the correct failure mode for a + /// prerequisite that's supposed to already be there. + async fn apply_operator_branch_protection(&self, repo: &str) -> Result<()> { + let rule = CreateBranchProtectionOption { + apply_to_admins: None, + approvals_whitelist_teams: Some(vec![OPERATORS_TEAM.to_owned()]), + approvals_whitelist_username: None, + block_on_official_review_requests: Some(true), + block_on_outdated_branch: None, + block_on_rejected_reviews: None, + branch_name: Some("main".to_owned()), + dismiss_stale_approvals: None, + enable_approvals_whitelist: Some(true), + enable_merge_whitelist: Some(true), + enable_push: None, + enable_push_whitelist: None, + enable_status_check: None, + ignore_stale_approvals: None, + merge_whitelist_teams: Some(vec![OPERATORS_TEAM.to_owned()]), + merge_whitelist_usernames: None, + protected_file_patterns: None, + push_whitelist_deploy_keys: None, + push_whitelist_teams: None, + push_whitelist_usernames: None, + require_signed_commits: None, + required_approvals: Some(1), + rule_name: None, + status_check_contexts: None, + unprotected_file_patterns: None, + }; + let Err(create_err) = self + .api + .repo_create_branch_protection(AGENTS_ORG, repo, rule) + .await + else { + tracing::info!(%repo, "swarm forge: applied operator branch protection"); + return Ok(()); + }; + match self + .api + .repo_get_branch_protection(AGENTS_ORG, repo, "main") + .await + { + Ok(_) => { + tracing::debug!( + %repo, create_error = %create_err, + "swarm forge: operator branch protection already present" + ); + Ok(()) + } + Err(check_err) => anyhow::bail!( + "branch protection for {AGENTS_ORG}/{repo} not applied: create failed \ + ({create_err}); GET main rule failed ({check_err}), no `main` rule present" + ), + } + } + + /// Create `repo` in [`AGENTS_ORG`] and apply the operator merge gate to + /// its default branch — the whole job of the `CreateRepo` node. + /// Branch protection is folded in here rather than a separate node: it + /// has no independent retry value apart from the repo existing (there + /// is nothing useful to retry "branch protection on a repo that + /// doesn't exist yet"), unlike collaborator-add and config-seed, which + /// are genuinely separate operations against an already-existing repo. + /// Idempotent — safe to call again. + pub async fn create_repo(&self, repo: &str) -> Result { + self.ensure_org_repo(repo).await?; + self.apply_operator_branch_protection(repo).await?; + tracing::info!(%repo, "swarm forge: created repo in {AGENTS_ORG} with operator merge gate"); + Ok(format!("{AGENTS_ORG}/{repo}")) + } + + /// Add `agent` as a **write** collaborator on `repo` (can push + open + /// PRs, cannot bypass branch protection) — the whole job of the + /// `AddRepoMember` node. Idempotent. + pub async fn add_repo_member(&self, repo: &str, agent: &str) -> Result<()> { + self.add_collaborator(repo, agent, AddCollaboratorOptionPermission::Write) + .await + } + + /// Seed `repo` with the two files every agent config repo needs: + /// `agent.nix` (the agent's own module) and `flake.nix` (the + /// boilerplate that lets the meta flake import this repo as a flake + /// input) — same content `hive-c0re::lifecycle::setup::setup_proposed` + /// writes at the per-hive level, committed here in one atomic + /// `repo_change_files` call instead of a local `git commit` (this + /// process has no working tree to commit from — it only ever talks to + /// the forge over HTTP). The whole job of the `InitAgentConfigRepo` + /// node. + /// + /// Idempotent by construction rather than by catching a conflict: an + /// unconditional `repo_change_files` against an already-seeded repo + /// would overwrite `agent.nix` with the seed content again, clobbering + /// any edits merged since — so this checks for `agent.nix`'s presence + /// first and treats it as "already seeded, nothing to do" rather than + /// re-committing over it. + pub async fn seed_agent_config(&self, repo: &str, agent: &str) -> Result<()> { + if self.file_exists(repo, "agent.nix").await? { + tracing::debug!(%repo, "swarm forge: config already seeded"); + return Ok(()); + } + let files = vec![ + ChangeFileOperation { + content: Some(base64_encode(&initial_agent_nix(agent))), + from_path: None, + operation: ChangeFileOperationOperation::Create, + path: "agent.nix".to_owned(), + sha: None, + }, + ChangeFileOperation { + content: Some(base64_encode(initial_flake_nix())), + from_path: None, + operation: ChangeFileOperationOperation::Create, + path: "flake.nix".to_owned(), + sha: None, + }, + ]; + self.api + .repo_change_files( + AGENTS_ORG, + repo, + ChangeFilesOptions { + author: None, + branch: None, + committer: None, + dates: None, + files, + force_overwrite_new_branch: None, + message: Some("swarm-controller: seed agent config".to_owned()), + new_branch: None, + signoff: None, + }, + ) + .await + .with_context(|| format!("seed config files in {AGENTS_ORG}/{repo}"))?; + tracing::info!(%repo, %agent, "swarm forge: seeded agent.nix + flake.nix"); + Ok(()) + } + + /// Whether `path` exists on `repo`'s default branch — the idempotency + /// check [`Self::seed_agent_config`] uses instead of catching a + /// create-conflict, since `repo_change_files` has no + /// already-exists-is-fine status the way `create_org_repo` does. + async fn file_exists(&self, repo: &str, path: &str) -> Result { + match self + .api + .repo_get_contents(AGENTS_ORG, repo, path, RepoGetContentsQuery::default()) + .await + { + Ok(_) => Ok(true), + Err(ForgejoError::UnexpectedStatusCode(s)) if s == StatusCode::NOT_FOUND => Ok(false), + Err(e) => Err(e).with_context(|| format!("check for {path} in {AGENTS_ORG}/{repo}")), + } + } +} + +/// Base64-encode `content` for a forgejo content-API call — every file +/// write on this API takes base64, never raw bytes. +fn base64_encode(content: &str) -> String { + use base64::Engine as _; + base64::engine::general_purpose::STANDARD.encode(content) +} + +/// Per-agent NixOS module seed — same content +/// `hive-c0re::lifecycle::setup::initial_agent_nix` writes at the per-hive +/// level (this process has no access to that function across the crate +/// boundary, and it's three lines — not worth a shared crate for). +fn initial_agent_nix(name: &str) -> String { + format!( + "{{ config, pkgs, lib, ... }}:\n{{\n # Per-agent overrides for {name}. This is a regular NixOS module\n # — add packages, services, modules, imports as needed.\n #\n # imports = [ ./extra-module.nix ];\n # environment.systemPackages = with pkgs; [ ];\n}}\n", + ) +} + +/// Module-only flake seed, byte-identical to +/// `hive-c0re::lifecycle::setup::initial_flake_nix` — see that function's +/// doc comment for why the shape looks the way it does (`flakeInputs` +/// forwarding, meta-flake consumption). Not shared across the crate +/// boundary for the same reason [`initial_agent_nix`] isn't. +fn initial_flake_nix() -> &'static str { + "{\n description = \"hyperhive agent\";\n inputs = { };\n outputs =\n { self, ... }@inputs:\n {\n nixosModules.default = {\n imports = [ ./agent.nix ];\n _module.args.flakeInputs = builtins.removeAttrs inputs [ \"self\" ];\n };\n };\n}\n" +} + +/// Whether a create-style call failed because the object already +/// exists. Forgejo signals this as HTTP 409 (conflict) or 422 +/// (validation) — match both defensively, same as +/// `hive-c0re::forge::repos::is_already_exists`. +fn is_already_exists(e: &ForgejoError) -> bool { + match e { + ForgejoError::ApiError(api) => match api.error_kind() { + ApiErrorKind::ValidationFailed => true, + ApiErrorKind::Other(s) => *s == StatusCode::CONFLICT, + _ => false, + }, + ForgejoError::UnexpectedStatusCode(s) => { + *s == StatusCode::CONFLICT || *s == StatusCode::UNPROCESSABLE_ENTITY + } + _ => false, + } +} + +#[cfg(test)] +mod tests { + use super::*; + + // Both scenarios live in ONE test, not two, deliberately: `cargo + // test` runs tests in parallel by default, and two tests mutating + // the same process-global env vars would race each other. A single + // test's body is one thread, so sequential set/clear within it is + // race-free without needing a cross-test lock. + #[test] + fn from_env_reads_both_vars_or_neither() { + // SAFETY: test-only mutation of env vars this module itself + // owns (`URL_ENV`/`TOKEN_FILE_ENV`), scoped to this one test + // function so there's no cross-test race (see comment above). + unsafe { + std::env::remove_var(URL_ENV); + std::env::remove_var(TOKEN_FILE_ENV); + } + assert!(Client::from_env().unwrap().is_none()); + + unsafe { + std::env::set_var(URL_ENV, "https://forge.example.invalid"); + } + // Not `.unwrap_err()`: that needs `Client: Debug` for its panic + // message, and the wrapped `forgejo_api::Forgejo` doesn't derive + // it. Match directly instead of adding a `Debug` impl nothing + // else needs. + match Client::from_env() { + Err(e) => assert!(e.to_string().contains(TOKEN_FILE_ENV)), + Ok(_) => panic!("expected an error for a partial forge config"), + } + + unsafe { + std::env::remove_var(URL_ENV); + } + } +} diff --git a/swarm-controller/src/main.rs b/swarm-controller/src/main.rs index d65c1e03..5903da83 100644 --- a/swarm-controller/src/main.rs +++ b/swarm-controller/src/main.rs @@ -33,28 +33,43 @@ use utoipa::{OpenApi, ToSchema}; use utoipa_axum::{router::OpenApiRouter, routes}; mod auth; +mod forge; mod status; /// 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` is the first real variant — "the minimal shape for +/// `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 this landed before the forge-node work a -/// standalone `CreateRepo` variant would have started with). Its only -/// effect is calling `swarm-controller::auth`, which calls -/// `swarm-authelia-bridge`; nothing forge- or deploy-shaped happens yet. +/// 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 `repo` in `forge::AGENTS_ORG` with the operator merge gate + /// on its default branch. See `forge::Client::create_repo`. + CreateRepo { repo: String }, + /// Add `agent` as a write collaborator on `repo`. See + /// `forge::Client::add_repo_member`. + AddRepoMember { repo: String, agent: String }, + /// Seed `repo` with `agent.nix` + `flake.nix`. See + /// `forge::Client::seed_agent_config`. + InitAgentConfigRepo { repo: 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::AddRepoMember { .. } => "add_repo_member".to_owned(), + SwarmNodeKind::InitAgentConfigRepo { .. } => "init_agent_config_repo".to_owned(), } } @@ -63,6 +78,13 @@ impl hive_jobq_wire::WireNode for SwarmNodeKind { SwarmNodeKind::CreateIdentity { agent } => { serde_json::json!({ "agent": agent }) } + SwarmNodeKind::CreateRepo { repo } => { + serde_json::json!({ "repo": repo }) + } + SwarmNodeKind::AddRepoMember { repo, agent } + | SwarmNodeKind::InitAgentConfigRepo { repo, agent } => { + serde_json::json!({ "repo": repo, "agent": agent }) + } } } } @@ -78,31 +100,78 @@ impl hive_jobq_wire::WireResource for SwarmResourceKind { } } +/// 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>, +} + /// 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. -/// -/// `auth` is `None` on a host that runs a controller split from -/// `swarm-authelia` (no bridge configured there) — `CreateIdentity` fails -/// explicitly in that case rather than this fn papering over a -/// misconfigured deployment. async fn run_swarm_node( _id: hive_jobq::NodeId, kind: SwarmNodeKind, builder: hive_jobq::builder::JobBuilder, - auth: Option>, + deps: WorkerDeps, ) -> ( hive_jobq::builder::JobBuilder, hive_jobq::scheduler::Outcome, ) { + use hive_jobq::scheduler::Outcome; + let outcome = match kind { - SwarmNodeKind::CreateIdentity { agent } => match auth { - None => hive_jobq::scheduler::Outcome::Failed( - "no swarm-authelia-bridge is configured on this host".to_owned(), - ), + 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 { - Ok(_) => hive_jobq::scheduler::Outcome::Done, - Err(e) => hive_jobq::scheduler::Outcome::Failed(format!("{e:#}")), + Ok(_) => Outcome::Done, + Err(e) => Outcome::Failed(format!("{e:#}")), + }, + }, + SwarmNodeKind::CreateRepo { repo } => 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(&repo).await { + Ok(full_name) => { + tracing::info!(%full_name, "swarm jobq: create_repo done"); + Outcome::Done + } + Err(e) => Outcome::Failed(format!("{e:#}")), + }, + }, + SwarmNodeKind::AddRepoMember { repo, 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(&repo, &agent).await { + Ok(()) => Outcome::Done, + Err(e) => Outcome::Failed(format!("{e:#}")), + }, + }, + SwarmNodeKind::InitAgentConfigRepo { repo, 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(&repo, &agent).await { + Ok(()) => Outcome::Done, + Err(e) => Outcome::Failed(format!("{e:#}")), }, }, }; @@ -124,20 +193,20 @@ async fn run_swarm_node( /// Cheap to run with an empty graph: `claim_next` on a graph nothing was /// ever inserted into just returns `None` every poll. /// -/// `auth` is cloned per iteration (an `Arc` clone, not a reconnect) 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. +/// `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>>, - auth: Option>, + deps: WorkerDeps, ) { tokio::spawn(async move { loop { - let auth = auth.clone(); + let deps = deps.clone(); let runner = hive_jobq::scheduler::Scheduler::claim_next(&sched, move |id, kind, builder| { - run_swarm_node(id, kind, builder, auth) + run_swarm_node(id, kind, builder, deps) }); match runner { Some(runner) => { @@ -403,10 +472,10 @@ async fn get_hives_status( } } -/// Body of `POST /api/agents` — the agent name to create a swarm-level -/// identity for. No other fields: this endpoint is deliberately narrow — -/// "identity in authelia only, no forge or deploy yet" — so it asks for -/// nothing a `CreateIdentity` node doesn't use. +/// Body of `POST /api/agents` — the agent name to create. The repo name +/// inside `forge::AGENTS_ORG` is the same string: one repo per agent, +/// named after it, same convention `hive-c0re::forge` already uses for its +/// own single-hive `CreateRepo` path. #[derive(Clone, Debug, Deserialize, ToSchema)] struct CreateAgentRequest { name: String, @@ -420,20 +489,32 @@ struct CreateAgentResponse { node_id: u64, } -/// Queue a `CreateIdentity` job for `name`. Returns as soon as the node is -/// inserted — **not** once the identity exists; `run_swarm_node` does that +/// Queue the whole agent-creation job graph for `name` — `CreateIdentity` +/// then, once that succeeds, `CreateRepo`; once THAT succeeds, +/// `AddRepoMember` and `InitAgentConfigRepo` both run off it — a fan-out, +/// not a chain, since adding a collaborator and seeding config files are +/// independent operations against the same already-created, already- +/// protected repo and have no ordering requirement on each other (mara, +/// design review: `InitAgentConfigRepo` does not depend on +/// `AddRepoMember` — 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 /// work asynchronously off the scheduler loop already running -/// (`spawn_jobq_worker`), same as every other node kind. This is also the -/// first genuine non-test caller `SwarmNodeKind::CreateIdentity` has: the -/// node kind's `dead_code` bound was the whole reason this endpoint had to -/// land in the same change as the variant, not as a follow-up. +/// (`spawn_jobq_worker`), same as every other node kind. The response +/// reports `CreateIdentity`'s id, the graph's entry point — a caller +/// watches the whole thing settle via `/api/jobq/graph`, which serves +/// every root, not just this one. +/// +/// This endpoint is also the first genuine non-test caller all four +/// `SwarmNodeKind` variants have: each node kind's `dead_code` bound is +/// the whole reason the executor arm and this endpoint had to land in the +/// same change as the variant, not as a follow-up. #[utoipa::path( post, path = "/api/agents", request_body = CreateAgentRequest, responses( - (status = 200, description = "job queued", body = CreateAgentResponse), - (status = 500, description = "the job could not be queued", body = String), + (status = 200, description = "job chain queued", body = CreateAgentResponse), + (status = 500, description = "the job chain could not be queued", body = String), ), tag = "agents" )] @@ -445,12 +526,29 @@ async fn create_agent( .jobq .lock() .unwrap_or_else(std::sync::PoisonError::into_inner); + let agent = req.name; + let repo = agent.clone(); let ids = sched .insert_job(None, |b| { - vec![ - b.node(SwarmNodeKind::CreateIdentity { agent: req.name }) - .guid(), - ] + let create_identity = b.node(SwarmNodeKind::CreateIdentity { + agent: agent.clone(), + }); + let create_repo = b + .node(SwarmNodeKind::CreateRepo { repo: repo.clone() }) + .after_ok(create_identity); + // Both fan out from `create_repo` directly — independent + // operations on the same repo, no ordering requirement on + // each other (see the doc comment above). + let _add_repo_member = b + .node(SwarmNodeKind::AddRepoMember { + repo: repo.clone(), + agent: agent.clone(), + }) + .after_ok(create_repo); + let _init_config = b + .node(SwarmNodeKind::InitAgentConfigRepo { repo, agent }) + .after_ok(create_repo); + vec![create_identity.guid()] }) .map_err(|e| (axum::http::StatusCode::INTERNAL_SERVER_ERROR, e.to_string()))?; let [id] = ids[..] else { @@ -605,12 +703,26 @@ async fn main() -> Result<()> { 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, + forge: forge_client, + }; let jobq = Arc::new(Mutex::new(hive_jobq::scheduler::Scheduler::new( hive_jobq::Graph::new(), hive_jobq::resources::ResourceTable::new(), ))); - spawn_jobq_worker(Arc::clone(&jobq), auth.clone()); + spawn_jobq_worker(Arc::clone(&jobq), deps); let state = AppState { hives: Arc::new(load_hives()), @@ -646,10 +758,76 @@ async fn main() -> Result<()> { #[cfg(test)] mod tests { use super::{ - DEFAULT_SOCKET, HIVES_ENV, HiveEntry, LINKS_ENV, ServiceLink, load_hives, load_links, + DEFAULT_SOCKET, HIVES_ENV, HiveEntry, LINKS_ENV, ServiceLink, SwarmNodeKind, WorkerDeps, + load_hives, load_links, run_swarm_node, }; use std::path::Path; + /// 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. + #[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 { + repo: "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, + }; + let runner = + hive_jobq::scheduler::Scheduler::claim_next(&sched, move |id, kind, builder| { + run_swarm_node(id, kind, builder, deps) + }) + .expect("the node just inserted is runnable"); + runner + .await + .1 + .expect("no growth declared, nothing to reject"); + + let guard = sched.lock().unwrap(); + let node = guard.graph().node(id).expect("node still present"); + assert_eq!(node.state, hive_jobq::State::Failed); + assert!( + node.error.as_deref().unwrap_or_default().contains("forge"), + "expected a forge-not-configured error, got {:?}", + node.error + ); + } + /// The socket must not share a directory with anything else, because /// the socket is `0666` and the directory is therefore the only access /// control it has. `/run/hyperhive` in particular holds hive-c0re's