From 1ca758daa7c07096df44ea6d498cdc627145652a Mon Sep 17 00:00:00 2001 From: atlas Date: Wed, 7 Oct 2026 21:02:45 +0200 Subject: [PATCH] swarm-controller grants internal/docs read to agents; hives stop doing it Agent creation gains a GrantDocsRead node (after CreateForgeUser) that adds the agent as a read collaborator on internal/docs, and forge::docs_grant runs a pass over the store's hive-agent-* roster at start and every five minutes that queues the same job for any agent that is not one. The pass only adds; an unreadable roster or forge read changes nothing. hive-c0re drops shared_docs_access, its call in sync_agent, and the SHARED_ORG / SHARED_DOCS_REPO constants only it used. --- docs/swarm/README.md | 9 + hive-c0re/src/forge/mod.rs | 33 +- hive-c0re/src/forge/repos.rs | 29 +- swarm-controller/src/forge.rs | 21 +- swarm-controller/src/forge/docs_grant.rs | 427 +++++++++++++++++++++++ swarm-controller/src/forge/objects.rs | 10 +- swarm-controller/src/main.rs | 173 ++++++++- 7 files changed, 643 insertions(+), 59 deletions(-) create mode 100644 swarm-controller/src/forge/docs_grant.rs diff --git a/docs/swarm/README.md b/docs/swarm/README.md index 1123f741..1b45c78f 100644 --- a/docs/swarm/README.md +++ b/docs/swarm/README.md @@ -294,6 +294,15 @@ A pass that can't finish logs a `journalctl -u swarm-controller`, and retries on the next tick. While the controller is down the objects stay as they are. +The controller gives each agent **read** on `internal/docs`. A `GrantDocsRead` +node in the agent's create job adds it as a collaborator, unless it already holds read or more. A pass over the +store's `hive-agent-*` roster (`forge/docs_grant.rs`, at start and every five +minutes) adds any agent that isn't one, so an agent whose create-time grant failed +gets it within five minutes. The pass only adds: a roster it can't read, a +forge read that fails, or an `internal/docs` that doesn't exist (checked once per +pass, logged at warn) changes nothing. The grant doesn't depend on the hive +the agent runs on, or on a forge container there. + ### Swarm-wide forge webhooks At startup the controller registers two Forgejo hooks pointing at diff --git a/hive-c0re/src/forge/mod.rs b/hive-c0re/src/forge/mod.rs index ab2c5666..676aaf1d 100644 --- a/hive-c0re/src/forge/mod.rs +++ b/hive-c0re/src/forge/mod.rs @@ -1,9 +1,9 @@ -//! Optional Forgejo wiring — the `core/meta` repo, meta read-access -//! grants, and each agent's read-only grant on `internal/docs` (the -//! swarm-controller creates that repo, with the other swarm-wide orgs, -//! teams and repos). Those writes, as `core`, run only beside a local -//! `hive-forge` container; reads of the forge go through [`reader_token`] on -//! every hive. Full design: `docs/integrations/forge.md`. +//! Optional Forgejo wiring — the `core/meta` repo and meta read-access +//! grants. Those writes, as `core`, run only beside a local `hive-forge` +//! container; reads of the forge go through [`reader_token`] on every hive. +//! The swarm-wide orgs, teams and repos, and each agent's read grant on +//! `internal/docs`, are swarm-controller's. Full design: +//! `docs/integrations/forge.md`. mod pr_comment; mod reader; @@ -16,7 +16,7 @@ pub use reader::reader_token; pub use reconcile::{fetch_forge_main, reconcile_config_apply, reconcile_config_status}; pub use repos::{ clone_config_into_proposed, ensure_meta_remote, ensure_repo, fast_forward_applied_main, - fetch_config_main_into_applied, meta_read_access, push_meta, shared_docs_access, + fetch_config_main_into_applied, meta_read_access, push_meta, }; pub use users::core_token; @@ -100,16 +100,6 @@ fn basic_auth_header(login: &str, token: &str) -> String { /// deploys through swarm-controller. The repos remain private, so an agent /// still can't reach *another* agent's config. hive-c0re only reads from it. pub(crate) const CONFIG_ORG: &str = "agent-configs"; -/// Forgejo org hosting the operator-curated shared docs/skills repo -/// that every agent gets read-only access to. Agents use it as a -/// common reference without the operator having to bake content into -/// the system prompt or rely on `/shared`. Only the manager + operator -/// (i.e. `core` user) can push. The swarm-controller creates the org -/// and the repo; this hive only grants its agents read access. -const SHARED_ORG: &str = "internal"; -/// The shared docs repo inside `SHARED_ORG`. Cloneable by every agent -/// at `{forge_http_base()}/internal/docs.git`. -const SHARED_DOCS_REPO: &str = "docs"; /// Leak `s` to get a `&'static str` warning `kind` for the small, bounded /// set of boot warnings in [`ensure_all`] keyed by a runtime name (at @@ -219,15 +209,6 @@ pub async fn sync_agent(name: &str, core_token: Option<&str>) -> bool { tracing::warn!(%name, error = ?e, "forge: ensure_meta_remote failed"); ok = false; } - // Grant read-only access to internal/docs so the agent can clone - // the operator-curated shared skills/runbook repo. Best-effort. - if let Some(token) = core_token - && let Err(e) = shared_docs_access(name, token).await - { - tracing::warn!(%name, error = ?e, "forge: shared_docs_access failed"); - ok = false; - } - // internal/knowledge is public — no per-agent collaborator grant needed. ok } diff --git a/hive-c0re/src/forge/repos.rs b/hive-c0re/src/forge/repos.rs index 726bbc0f..6718d2ae 100644 --- a/hive-c0re/src/forge/repos.rs +++ b/hive-c0re/src/forge/repos.rs @@ -1,7 +1,7 @@ -//! Repo + org plumbing on the local Forgejo: the meta + shared-docs + -//! knowledge repos, meta mirroring (`push_meta`), collaborator grants, -//! pull-mirrors, and branch-protection rules. The typed API-client -//! constructor + org-name constants live in the module root (`super`). +//! Repo + org plumbing on the local Forgejo: the meta + knowledge repos, +//! meta mirroring (`push_meta`), collaborator grants, pull-mirrors, and +//! branch-protection rules. The typed API-client constructor + org-name +//! constants live in the module root (`super`). use std::path::Path; use std::sync::Mutex; @@ -16,8 +16,8 @@ use reqwest::StatusCode; use crate::coordinator::Coordinator; use super::{ - CONFIG_ORG, SHARED_DOCS_REPO, SHARED_ORG, api, core_auth_header, core_token, forge_git_url, - forge_http_base, is_present, reader_token, + CONFIG_ORG, api, core_auth_header, core_token, forge_git_url, forge_http_base, is_present, + reader_token, }; /// Creation options for an empty repo defaulting to `main`. @@ -144,23 +144,6 @@ pub async fn push_meta(dir: &Path) -> Result<()> { Ok(()) } -/// Grant agent `name` read-only collaborator access to `internal/docs`. -/// Idempotent: re-adding an existing collaborator succeeds (Forgejo -/// answers 204 either way). Mirrors `meta_read_access` so agents can -/// clone the shared docs repo without authentication hassle. -pub async fn shared_docs_access(name: &str, core_token: &str) -> Result<()> { - add_collaborator( - SHARED_ORG, - SHARED_DOCS_REPO, - name, - AddCollaboratorOptionPermission::Read, - core_token, - ) - .await?; - tracing::info!(%name, "forge: granted shared-docs read access"); - Ok(()) -} - /// Grant agent `name` read-only collaborator access to `core/meta` on /// the forge so the agent can clone/fetch the meta flake. Idempotent: /// re-adding an existing collaborator succeeds (Forgejo answers 204 diff --git a/swarm-controller/src/forge.rs b/swarm-controller/src/forge.rs index a4f3a10b..4d2c488d 100644 --- a/swarm-controller/src/forge.rs +++ b/swarm-controller/src/forge.rs @@ -34,6 +34,7 @@ use utoipa::ToSchema; use crate::webhook::DeliveryKind; pub mod agent_token; +pub mod docs_grant; pub mod hive_token; pub mod legacy_tokens; pub mod objects; @@ -233,11 +234,23 @@ impl Client { } /// Add `user` as a collaborator on `CONFIG_ORG/repo` at `permission`. + async fn add_collaborator( + &self, + repo: &str, + user: &str, + permission: AddCollaboratorOptionPermission, + ) -> Result<()> { + self.add_collaborator_to(CONFIG_ORG, repo, user, permission) + .await + } + + /// Add `user` as a collaborator on `owner/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( + async fn add_collaborator_to( &self, + owner: &str, repo: &str, user: &str, permission: AddCollaboratorOptionPermission, @@ -245,7 +258,7 @@ impl Client { let res = self .api .repo_add_collaborator( - CONFIG_ORG, + owner, repo, user, AddCollaboratorOption { @@ -258,10 +271,10 @@ impl Client { Err(ForgejoError::UnexpectedStatusCode(s)) if s == StatusCode::CREATED => {} Err(e) => { return Err(e) - .with_context(|| format!("add collaborator {user} to {CONFIG_ORG}/{repo}")); + .with_context(|| format!("add collaborator {user} to {owner}/{repo}")); } } - tracing::debug!(%repo, %user, ?permission, "swarm forge: collaborator set"); + tracing::debug!(%owner, %repo, %user, ?permission, "swarm forge: collaborator set"); Ok(()) } diff --git a/swarm-controller/src/forge/docs_grant.rs b/swarm-controller/src/forge/docs_grant.rs new file mode 100644 index 00000000..6595cb1d --- /dev/null +++ b/swarm-controller/src/forge/docs_grant.rs @@ -0,0 +1,427 @@ +//! Each agent's read access to `internal/docs`: a collaborator grant at +//! **read**, made here with the admin API so it does not depend on which hive +//! the agent runs on or whether that hive has a forge container. +//! +//! Two callers insert the same `GrantDocsRead` job node: agent creation, and +//! [`spawn`]'s pass at start and every [`RECONCILE_INTERVAL`] over every +//! agent that holds a store identity. The pass reaches agents created before +//! this node existed and agents whose creation-time grant failed. It only ever +//! adds: nobody is removed, whatever the roster or the forge says. The +//! decision is pure ([`plan`]), so the tests pin it. + +use std::sync::Arc; + +use anyhow::{Context, Result}; +use forgejo_api::structs::{AddCollaboratorOptionPermission, RepoCollaboratorPermission}; +use swarm_secret_client::{client::DEFAULT_CERT_MOUNT, policy}; + +use super::Client; +use super::agent_token::is_not_found; +use super::objects::{SHARED_DOCS_REPO, SHARED_ORG}; + +/// How often [`spawn`] re-checks every agent. +const RECONCILE_INTERVAL: std::time::Duration = std::time::Duration::from_mins(5); + +/// What one pass found for one agent. +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub enum Observed { + /// The agent is already a collaborator on `internal/docs`, at any level. + Granted, + /// The forge says the agent is not a collaborator, or has no such user — + /// the queued job creates the user before it grants. + Missing, + /// The read failed. Nothing is known, so nothing is done. + Unknown, +} + +/// The agents a pass queues a grant for: those the forge says are not +/// collaborators. [`Observed::Unknown`] is never queued. +pub fn plan(observed: &[(String, Observed)]) -> Vec { + observed + .iter() + .filter(|(_, o)| *o == Observed::Missing) + .map(|(agent, _)| agent.clone()) + .collect() +} + +/// Whether a permission read says the user already has read access or more. +fn holds_read(p: &RepoCollaboratorPermission) -> bool { + matches!( + p.permission.as_deref(), + Some("read" | "write" | "admin" | "owner") + ) +} + +impl Client { + /// Make `agent` a **read** collaborator on `internal/docs` — the whole job + /// of the `GrantDocsRead` node. Idempotent, and never lowers: an agent that + /// already holds read or more (write, admin, owner) is left as it is, since + /// the add endpoint would overwrite the permission. Needs the forge user to + /// exist. + pub async fn grant_docs_read(&self, agent: &str) -> Result<()> { + match self + .api + .repo_get_repo_permissions(SHARED_ORG, SHARED_DOCS_REPO, agent) + .await + { + Ok(p) if holds_read(&p) => return Ok(()), + Ok(_) => {} + Err(e) if is_not_found(&e) => {} + Err(e) => { + return Err(e).with_context(|| { + format!("read {agent}'s permission on {SHARED_ORG}/{SHARED_DOCS_REPO}") + }); + } + } + self.add_collaborator_to( + SHARED_ORG, + SHARED_DOCS_REPO, + agent, + AddCollaboratorOptionPermission::Read, + ) + .await + } + + /// Whether `agent` is a collaborator on `internal/docs` now. + async fn observe_grant(&self, agent: &str) -> Observed { + match self + .api + .repo_check_collaborator(SHARED_ORG, SHARED_DOCS_REPO, agent) + .await + { + Ok(()) => Observed::Granted, + Err(e) if is_not_found(&e) => Observed::Missing, + Err(e) => { + tracing::warn!(agent, error = %e, "docs grant: forge read failed"); + Observed::Unknown + } + } + } + + /// Whether `internal/docs` exists now: `None` when the forge could not + /// say. A missing repo makes every collaborator check answer 404 too, so + /// it has to be told apart from "not a collaborator" before they run. + async fn docs_repo_exists(&self) -> Option { + match self.api.repo_get(SHARED_ORG, SHARED_DOCS_REPO).await { + Ok(_) => Some(true), + Err(e) if is_not_found(&e) => Some(false), + Err(e) => { + tracing::warn!(error = %e, "docs grants: forge read of the repo failed"); + None + } + } + } + + /// One pass over `roster`. + async fn observe_grants(&self, roster: &[String]) -> Vec<(String, Observed)> { + let mut observed = Vec::with_capacity(roster.len()); + for agent in roster { + observed.push((agent.clone(), self.observe_grant(agent).await)); + } + observed + } +} + +/// The store's `hive-agent-*` cert-auth roles: exactly the agents that hold a +/// store identity. +async fn read_roster() -> Result> { + let store = crate::store::connect() + .await + .context("logging in to the swarm secret store")?; + let roles = store + .list_cert_roles(DEFAULT_CERT_MOUNT) + .await + .context("listing the store's cert-auth roles")?; + Ok(policy::agents_from_role_names(&roles)) +} + +/// One pass: observe every agent of `roster` and return the ones without a +/// grant. A roster that could not be read yields none and touches nothing; so +/// does a pass whose `internal/docs` is missing or unreadable, which is checked +/// once before any agent. +async fn run_pass(client: &Client, roster: Result>) -> Vec { + let roster = match roster { + Ok(roster) => roster, + Err(e) => { + tracing::warn!( + error = %format!("{e:#}"), + retry_in_s = RECONCILE_INTERVAL.as_secs(), + "docs grants: no roster; retrying next tick" + ); + return Vec::new(); + } + }; + match client.docs_repo_exists().await { + Some(true) => {} + Some(false) => { + tracing::warn!( + repo = format!("{SHARED_ORG}/{SHARED_DOCS_REPO}"), + "docs grants: repo missing; no grants queued" + ); + return Vec::new(); + } + None => return Vec::new(), + } + let observed = client.observe_grants(&roster).await; + let agents = plan(&observed); + if agents.is_empty() { + tracing::debug!(checked = observed.len(), "docs grants: all current"); + } else { + tracing::info!( + checked = observed.len(), + queueing = agents.len(), + "docs grants: queueing grants" + ); + } + agents +} + +/// Check every agent now and every [`RECONCILE_INTERVAL`] after, and hand the +/// agents without a grant to `enqueue`, which inserts a `GrantDocsRead` node +/// for each. A pass that fails is logged and retried on the next tick; it +/// never stops the daemon. +pub fn spawn(client: Arc, enqueue: impl Fn(Vec) + Send + 'static) { + tokio::spawn(async move { + let mut ticker = tokio::time::interval(RECONCILE_INTERVAL); + loop { + ticker.tick().await; + let agents = run_pass(&client, read_roster().await).await; + if !agents.is_empty() { + enqueue(agents); + } + } + }); +} + +#[cfg(test)] +mod tests { + use std::collections::BTreeMap; + use std::sync::Mutex; + + use axum::extract::{Path, State}; + use axum::http::{StatusCode, header::CONTENT_TYPE}; + use forgejo_api::{Auth, Forgejo}; + + use super::*; + + /// What a stand-in `internal/docs` holds: who is a collaborator, and the + /// request log. + #[derive(Default)] + struct Docs { + /// The repo does not exist: its own lookup and every collaborator + /// check answer 404. + repo_missing: bool, + collaborators: BTreeMap, + requests: Vec<(String, String)>, + /// Answer every collaborator check with a 500. + broken: bool, + } + + type Shared = Arc>; + + /// A stand-in forge on a loopback port — never the real one — serving + /// the collaborator check (`GET`) and add (`PUT`) on `internal/docs`. + async fn stub_forge(collaborators: &[&str], broken: bool) -> (Client, Shared) { + serve(Docs { + collaborators: collaborators + .iter() + .map(|c| ((*c).to_owned(), "read".to_owned())) + .collect(), + broken, + ..Docs::default() + }) + .await + } + + async fn serve(docs: Docs) -> (Client, Shared) { + let docs: Shared = Arc::new(Mutex::new(docs)); + let app = axum::Router::new() + .route("/api/v1/repos/internal/docs", axum::routing::get(repo)) + .route( + "/api/v1/repos/internal/docs/collaborators/{user}", + axum::routing::get(check).put(add), + ) + .route( + "/api/v1/repos/internal/docs/collaborators/{user}/permission", + axum::routing::get(permission), + ) + .with_state(Arc::clone(&docs)); + let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap(); + let url = url::Url::parse(&format!("http://{}/", listener.local_addr().unwrap())).unwrap(); + tokio::spawn(async move { axum::serve(listener, app).await }); + let api = Forgejo::new(Auth::Token("stub-token"), url.clone()).unwrap(); + (Client { api, url }, docs) + } + + async fn check( + State(docs): State, + Path(user): Path, + ) -> ( + StatusCode, + [(axum::http::HeaderName, &'static str); 1], + &'static str, + ) { + let mut docs = docs.lock().unwrap(); + docs.requests.push(("GET".to_owned(), user.clone())); + let status = if docs.repo_missing { + StatusCode::NOT_FOUND + } else if docs.broken { + StatusCode::INTERNAL_SERVER_ERROR + } else if docs.collaborators.contains_key(&user) { + StatusCode::NO_CONTENT + } else { + StatusCode::NOT_FOUND + }; + (status, [(CONTENT_TYPE, "application/json")], "{}") + } + + async fn repo( + State(docs): State, + ) -> ( + StatusCode, + [(axum::http::HeaderName, &'static str); 1], + &'static str, + ) { + let mut docs = docs.lock().unwrap(); + docs.requests.push(("GET".to_owned(), "repo".to_owned())); + let status = if docs.repo_missing { + StatusCode::NOT_FOUND + } else { + StatusCode::OK + }; + (status, [(CONTENT_TYPE, "application/json")], "{}") + } + + async fn permission(State(docs): State, Path(user): Path) -> String { + let mut docs = docs.lock().unwrap(); + docs.requests + .push(("GET".to_owned(), format!("{user} permission"))); + let level = docs.collaborators.get(&user).map_or("none", String::as_str); + format!(r#"{{"permission":"{level}"}}"#) + } + + async fn add(State(docs): State, Path(user): Path, body: String) -> StatusCode { + let mut docs = docs.lock().unwrap(); + docs.requests + .push(("PUT".to_owned(), format!("{user} {body}"))); + docs.collaborators.insert(user, "read".to_owned()); + StatusCode::NO_CONTENT + } + + fn roster(agents: &[&str]) -> Vec { + agents.iter().map(|a| (*a).to_owned()).collect() + } + + fn puts(docs: &Shared) -> Vec { + docs.lock() + .unwrap() + .requests + .iter() + .filter(|(m, _)| m == "PUT") + .map(|(_, what)| what.clone()) + .collect() + } + + #[test] + fn only_a_missing_grant_is_planned() { + let observed = [ + ("a".to_owned(), Observed::Granted), + ("b".to_owned(), Observed::Missing), + ("c".to_owned(), Observed::Unknown), + ]; + assert_eq!(plan(&observed), ["b"]); + } + + /// The `GrantDocsRead` node's work: a **read** collaborator on + /// `internal/docs`, not write. + #[tokio::test] + async fn the_grant_adds_the_agent_as_a_read_collaborator() { + let (client, docs) = stub_forge(&[], false).await; + + client.grant_docs_read("alice").await.unwrap(); + + let sent = puts(&docs); + assert_eq!(sent.len(), 1, "{sent:?}"); + let (user, body) = sent[0].split_once(' ').unwrap(); + assert_eq!(user, "alice"); + let body: serde_json::Value = serde_json::from_str(body).unwrap(); + assert_eq!(body["permission"], "read"); + } + + /// The backfill queues the agents that lack the grant, and only those. + #[tokio::test] + async fn the_pass_queues_the_agents_missing_the_grant() { + let (client, docs) = stub_forge(&["bob"], false).await; + let queued = run_pass(&client, Ok(roster(&["alice", "bob", "carol"]))).await; + + assert_eq!(queued, ["alice", "carol"]); + assert!(puts(&docs).is_empty(), "the pass itself only reads"); + } + + /// Once every agent holds the grant a pass queues nothing, and writes + /// nothing — in particular it does not re-add a collaborator that holds a + /// higher permission. + #[tokio::test] + async fn the_pass_is_a_no_op_for_agents_already_granted() { + let (client, docs) = stub_forge(&["alice", "bob"], false).await; + let queued = run_pass(&client, Ok(roster(&["alice", "bob"]))).await; + + assert!(queued.is_empty()); + assert!(puts(&docs).is_empty()); + } + + /// A roster that could not be read changes nothing: no grant queued, no + /// request to the forge at all. + #[tokio::test] + async fn an_unreadable_roster_changes_nothing() { + let (client, docs) = stub_forge(&["alice"], false).await; + let queued = run_pass(&client, Err(anyhow::anyhow!("store is down"))).await; + + assert!(queued.is_empty()); + assert!(docs.lock().unwrap().requests.is_empty()); + } + + /// A forge that cannot answer the check is not a forge that says "not a + /// collaborator": nothing is queued. + #[tokio::test] + async fn an_unreadable_forge_queues_nothing() { + let (client, docs) = stub_forge(&[], true).await; + let queued = run_pass(&client, Ok(roster(&["alice"]))).await; + + assert!(queued.is_empty()); + assert!(puts(&docs).is_empty()); + } + + /// A missing `internal/docs` answers 404 to every collaborator check, which + /// would read as "not a collaborator" for every agent: nothing is queued, + /// and the agents are not asked about one by one. + #[tokio::test] + async fn a_missing_repo_queues_nothing() { + let (client, docs) = serve(Docs { + repo_missing: true, + ..Docs::default() + }) + .await; + let queued = run_pass(&client, Ok(roster(&["alice", "bob"]))).await; + + assert!(queued.is_empty()); + let requests = docs.lock().unwrap().requests.clone(); + assert_eq!(requests, [("GET".to_owned(), "repo".to_owned())]); + } + + /// An agent an operator already raised above read keeps it: the node reads + /// the permission first and sends no write. + #[tokio::test] + async fn the_grant_leaves_a_higher_permission_alone() { + let (client, docs) = serve(Docs { + collaborators: BTreeMap::from([("alice".to_owned(), "write".to_owned())]), + ..Docs::default() + }) + .await; + + client.grant_docs_read("alice").await.unwrap(); + + assert!(puts(&docs).is_empty(), "{:?}", puts(&docs)); + assert_eq!(docs.lock().unwrap().collaborators["alice"], "write"); + } +} diff --git a/swarm-controller/src/forge/objects.rs b/swarm-controller/src/forge/objects.rs index 1b10ec8b..52b3348a 100644 --- a/swarm-controller/src/forge/objects.rs +++ b/swarm-controller/src/forge/objects.rs @@ -48,13 +48,13 @@ const AGENTS_ORG: &str = "agents"; /// Org hosting the operator-curated shared repos. Same org as the knowledge /// repo's, named separately for the docs repo it also holds. -const SHARED_ORG: &str = KNOWLEDGE_ORG; +pub(super) const SHARED_ORG: &str = KNOWLEDGE_ORG; /// The shared docs repo inside [`SHARED_ORG`]. **Private**: every agent gets -/// read-only collaborator access, granted per agent by its hive. Agents use it -/// as a common reference without the operator having to bake content into the -/// system prompt. -const SHARED_DOCS_REPO: &str = "docs"; +/// read-only collaborator access, granted per agent by [`super::docs_grant`]. +/// Agents use it as a common reference without the operator having to bake +/// content into the system prompt. +pub(super) const SHARED_DOCS_REPO: &str = "docs"; /// The orgs ensured on every pass. The meta repo lives at `core/meta`, the /// `core` user's own namespace, so no org is needed for it (and it is not a diff --git a/swarm-controller/src/main.rs b/swarm-controller/src/main.rs index 17691787..f0aefd1c 100644 --- a/swarm-controller/src/main.rs +++ b/swarm-controller/src/main.rs @@ -96,6 +96,11 @@ enum SwarmNodeKind { /// Add `agent` as a write collaborator on its own repo. See /// `forge::Client::add_repo_member`. AddRepoMember { agent: String }, + /// Add `agent` as a read collaborator on `internal/docs`. See + /// `forge::docs_grant`. + /// + /// Carries no hive: the grant is on a swarm-wide repo. + GrantDocsRead { agent: String }, /// Seed the agent's repo with `agent.nix` + `flake.nix`. See /// `forge::Client::seed_agent_config`. /// @@ -171,6 +176,7 @@ impl hive_jobq_wire::WireNode for SwarmNodeKind { SwarmNodeKind::CreateRepo { .. } => "create_repo".to_owned(), SwarmNodeKind::CreateForgeUser { .. } => "create_forge_user".to_owned(), SwarmNodeKind::AddRepoMember { .. } => "add_repo_member".to_owned(), + SwarmNodeKind::GrantDocsRead { .. } => "grant_docs_read".to_owned(), SwarmNodeKind::InitAgentConfigRepo { .. } => "init_agent_config_repo".to_owned(), SwarmNodeKind::MintAgentIdentity { .. } => "mint_agent_identity".to_owned(), SwarmNodeKind::MintAgentForgeToken { .. } => "mint_agent_forge_token".to_owned(), @@ -197,6 +203,7 @@ impl hive_jobq_wire::WireNode for SwarmNodeKind { | SwarmNodeKind::CreateRepo { agent } | SwarmNodeKind::CreateForgeUser { agent } | SwarmNodeKind::AddRepoMember { agent } + | SwarmNodeKind::GrantDocsRead { agent } | SwarmNodeKind::InitAgentConfigRepo { agent } | SwarmNodeKind::MintAgentIdentity { agent } | SwarmNodeKind::MintAgentForgeToken { agent } @@ -340,6 +347,7 @@ async fn run_swarm_node( Err(e) => Outcome::Failed(format!("{e:#}")), }, }, + SwarmNodeKind::GrantDocsRead { agent } => grant_docs_read(deps.forge, &agent).await, SwarmNodeKind::InitAgentConfigRepo { agent } => match deps.forge { None => Outcome::Failed( "no forge configured on this host (SWARM_CONTROLLER_FORGE_URL / \ @@ -414,6 +422,26 @@ async fn mint_forge_token( } } +/// The `GrantDocsRead` arm, lifted out for the same reason. +async fn grant_docs_read( + forge: Option>, + agent: &str, +) -> hive_jobq::scheduler::Outcome { + use hive_jobq::scheduler::Outcome; + + let Some(client) = forge else { + return Outcome::Failed( + "no forge configured on this host (SWARM_CONTROLLER_FORGE_URL / \ + SWARM_CONTROLLER_FORGE_TOKEN_FILE unset), so no docs grant can be made" + .to_owned(), + ); + }; + match client.grant_docs_read(agent).await { + Ok(()) => Outcome::Done, + Err(e) => Outcome::Failed(format!("{e:#}")), + } +} + /// The `RenewAgentQueueCredential` arm, lifted out for the same reason. async fn renew_queue_credential(agent: &str) -> hive_jobq::scheduler::Outcome { use hive_jobq::scheduler::Outcome; @@ -1980,6 +2008,14 @@ fn declare_agent_job( }) .after_ok(create_repo) .after_ok(create_forge_user); + // Read on `internal/docs`, a repo that exists whatever this agent is, so + // it needs only the user. Not a parent of the deploy: the agent reads docs + // once running, and a failed grant is retried by `forge::docs_grant::spawn`. + let _grant_docs = b + .node(SwarmNodeKind::GrantDocsRead { + agent: agent.to_owned(), + }) + .after_ok(create_forge_user); let init_config = b .node(SwarmNodeKind::InitAgentConfigRepo { agent: agent.to_owned(), @@ -2266,6 +2302,37 @@ fn queue_forge_token_mints( Ok(ids) } +/// Insert one job per agent: `CreateForgeUser`, then `GrantDocsRead` `after_ok` +/// it — the same edge [`declare_agent_job`] uses. Returns the grant nodes' ids. +/// `forge::docs_grant::spawn`'s periodic pass comes through here, so a +/// backfilled grant is the same node a new agent gets; the user node is a +/// no-op for an agent that has an account. +fn queue_docs_grants( + sched: &Mutex>, + agents: Vec, +) -> Result> { + let mut sched = sched + .lock() + .unwrap_or_else(std::sync::PoisonError::into_inner); + let mut ids = Vec::with_capacity(agents.len()); + for agent in agents { + let queued = sched + .insert_job(None, |b| { + let create_forge_user = b.node(SwarmNodeKind::CreateForgeUser { + agent: agent.clone(), + }); + vec![ + b.node(SwarmNodeKind::GrantDocsRead { agent }) + .after_ok(create_forge_user) + .guid(), + ] + }) + .map_err(|e| anyhow::anyhow!("{e}"))?; + ids.extend(queued); + } + Ok(ids) +} + /// Insert one `MintAgentMatrixAccount` job per agent and return the nodes' /// ids. `matrix_account::agent_token::spawn`'s periodic pass comes through /// here, so a backfilled mint is the same node a new agent gets. @@ -2373,7 +2440,8 @@ fn spawn_matrix_account_backfill( } /// Start the forge's periodic passes — the swarm-wide objects, the runner -/// registration token, hives' read tokens, and agents' tokens (which also +/// registration token, hives' read tokens, agents' `internal/docs` grants, and +/// agents' tokens (which also /// sweeps legacy tokens; see `forge::legacy_tokens`) — /// when a forge is configured. Lifted out of `main` for /// `clippy::too_many_lines`. @@ -2395,6 +2463,12 @@ fn spawn_forge_workers( } }); 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"); @@ -4529,6 +4603,103 @@ mod tests { ); } + /// Agent creation grants `internal/docs` read once the forge user exists, + /// and a failed grant does not hold up the deploy (the pass retries it). + #[tokio::test] + async fn the_docs_grant_follows_the_user_and_is_not_a_deploy_parent() { + use hive_jobq_wire::WireNode as _; + + let (state, sched) = state_with_roster(); + let _queued = super::create_agent( + axum::extract::State(state), + axum::Json(super::CreateAgentRequest { + name: "atlas".to_owned(), + hive: "pr1ma".to_owned(), + }), + ) + .await + .expect("a hive in the roster must be accepted"); + + let guard = sched + .lock() + .unwrap_or_else(std::sync::PoisonError::into_inner); + let graph = guard.graph(); + let id_of = |label: &str| { + graph + .nodes() + .find(|n| n.payload.label() == label) + .unwrap_or_else(|| panic!("the graph holds a {label} node")) + .id + }; + let grant = graph.node(id_of("grant_docs_read")).expect("just found"); + assert_eq!(grant.payload.data(grant.id.get())["agent"], "atlas"); + let when = grant + .deps + .iter() + .find_map(|d| match d { + hive_jobq::Dep::Node { id, when } if *id == id_of("create_forge_user") => { + Some(*when) + } + _ => None, + }) + .expect("the grant waits for the forge user"); + assert!( + !when.accepts(hive_jobq::TerminalState::Failed), + "adding a user that was never created as a collaborator is a forge error; \ + this edge has to be `after_ok`" + ); + let deploy = graph.node(id_of("trigger_deploy")).expect("just found"); + assert!( + !deploy.deps.iter().any(|d| matches!( + d, + hive_jobq::Dep::Node { id, .. } if *id == grant.id + )), + "the deploy must not wait on the docs grant" + ); + } + + /// The periodic pass queues the same pair of nodes agent creation does. + #[test] + fn a_queued_docs_grant_creates_the_forge_user_first() { + use hive_jobq_wire::WireNode as _; + + let sched = std::sync::Mutex::new(hive_jobq::scheduler::Scheduler::new( + hive_jobq::Graph::new(), + hive_jobq::resources::ResourceTable::new(), + )); + let ids = super::queue_docs_grants(&sched, vec!["a".to_owned(), "b".to_owned()]) + .expect("two jobs insert"); + assert_eq!(ids.len(), 2, "one returned id per agent: its grant node"); + let guard = sched + .lock() + .unwrap_or_else(std::sync::PoisonError::into_inner); + let graph = guard.graph(); + let mut labels: Vec = graph.nodes().map(|n| n.payload.label()).collect(); + labels.sort(); + assert_eq!( + labels, + [ + "create_forge_user", + "create_forge_user", + "grant_docs_read", + "grant_docs_read" + ] + ); + for grant in graph + .nodes() + .filter(|n| n.payload.label() == "grant_docs_read") + { + assert!(ids.contains(&grant.id)); + assert!( + grant + .deps + .iter() + .any(|d| matches!(d, hive_jobq::Dep::Node { when, .. } if !when.accepts(hive_jobq::TerminalState::Failed))), + "each grant waits `after_ok` on its user" + ); + } + } + #[test] fn a_forge_token_node_renders_the_agent() { use hive_jobq_wire::WireNode as _;