From 839efd1642db37611d07913c86a7ff2aa258eb57 Mon Sep 17 00:00:00 2001 From: atlas Date: Sun, 18 Oct 2026 23:37:05 +0200 Subject: [PATCH] agent_permissions, agent_term and subagent_term take their wire contracts from swarm-wire --- AGENTS.md | 2 +- Cargo.lock | 1 + docs/process/conventions.md | 2 +- hive-agent/src/swarm_term.rs | 4 +- hive-c0re/src/agent_config/permissions.rs | 5 +- hive-c0re/src/workers/permissions_import.rs | 2 +- hive-c0re/src/workers/swarm_permissions.rs | 5 +- hive-subagent-mcp/Cargo.toml | 1 - hive-subagent-mcp/src/swarm_term.rs | 4 +- nix/host-modules/swarm-nats.nix | 4 +- swarm-controller/src/agent_login.rs | 5 +- swarm-controller/src/agent_permissions.rs | 11 +- swarm-controller/src/message_stream.rs | 29 ++- swarm-controller/src/routes/agents.rs | 4 +- swarm-controller/src/subagent_term.rs | 12 +- swarm-controller/src/term_stream.rs | 43 +++-- swarm-nats-auth/src/policy.rs | 36 ++-- swarm-queue-client/Cargo.toml | 2 +- swarm-queue-client/src/agent_permissions.rs | 158 ++------------- swarm-queue-client/src/agent_term.rs | 202 ++------------------ swarm-queue-client/src/lib.rs | 56 ++---- swarm-queue-client/src/subagent_term.rs | 153 +-------------- swarm-wire/Cargo.toml | 1 + swarm-wire/src/agent_permissions.rs | 143 ++++++++++++++ swarm-wire/src/agent_term.rs | 191 ++++++++++++++++++ swarm-wire/src/lib.rs | 40 ++++ swarm-wire/src/subagent_term.rs | 152 +++++++++++++++ 27 files changed, 661 insertions(+), 607 deletions(-) create mode 100644 swarm-wire/src/agent_permissions.rs create mode 100644 swarm-wire/src/agent_term.rs create mode 100644 swarm-wire/src/subagent_term.rs diff --git a/AGENTS.md b/AGENTS.md index de1786f8..7947aa98 100644 --- a/AGENTS.md +++ b/AGENTS.md @@ -139,7 +139,7 @@ Swarm-level: - **`swarm-queue-client/`** — the one authenticated connect to the swarm NATS queue. - **`swarm-wire/`** — serde-only request and reply contracts services - exchange over the queue (builder, claude login, review classifier, agent inference and login), so naming one links + exchange over the queue (builder, claude login, review classifier, agent inference, login, permissions and terminal streams), so naming one links neither the queue client nor the store client. - **`swarm-nats-auth/`** — NATS auth-callout responder; without it the queue is fail-closed and admits nobody. diff --git a/Cargo.lock b/Cargo.lock index a705ad8a..4f0e03d3 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -5090,6 +5090,7 @@ dependencies = [ name = "swarm-wire" version = "0.1.0" dependencies = [ + "hive-types", "serde", "serde_json", "utoipa", diff --git a/docs/process/conventions.md b/docs/process/conventions.md index d868c6b9..adc9ec66 100644 --- a/docs/process/conventions.md +++ b/docs/process/conventions.md @@ -297,7 +297,7 @@ or not it holds `inbox`, and an agent without it still needs to clear them. **Where they live** — in the swarm, not on the hive: the `permissions` field of the agent's key in the `agent-permissions` KV bucket -(`swarm-queue-client/src/agent_permissions.rs`), which swarm-controller alone +(`swarm-wire/src/agent_permissions.rs`), which swarm-controller alone writes. Set them with `swarmctl agent permissions get|set|clear ` (`/api/agents/{name}/permissions`), or on the swarm UI's create-agent screen (`permissions` in `POST /api/agents`). A missing key or an unset field means diff --git a/hive-agent/src/swarm_term.rs b/hive-agent/src/swarm_term.rs index f4a04f49..b496abc2 100644 --- a/hive-agent/src/swarm_term.rs +++ b/hive-agent/src/swarm_term.rs @@ -15,7 +15,7 @@ //! //! **A plain core publish.** The queue keeps the rows in the stream //! `term-main-`, which swarm-controller creates -//! (`swarm_queue_client::agent_term`), so this needs no `JetStream` call and +//! (`swarm_wire::agent_term`), so this needs no `JetStream` call and //! the agent's grant stays a plain publish. A row published before that //! stream exists reaches live subscribers only. //! @@ -50,7 +50,7 @@ pub fn spawn(rx: broadcast::Receiver) { async fn run(mut rx: broadcast::Receiver, agent: String) { let client = crate::swarm_queue::client().await; - let subject = swarm_queue_client::term_subject(&agent); + let subject = swarm_wire::term_subject(&agent); tracing::info!(subject, "publishing the agent terminal to the swarm queue"); // One context for the task's whole life — see the module doc. diff --git a/hive-c0re/src/agent_config/permissions.rs b/hive-c0re/src/agent_config/permissions.rs index f0649c5b..79dcd2ba 100644 --- a/hive-c0re/src/agent_config/permissions.rs +++ b/hive-c0re/src/agent_config/permissions.rs @@ -4,7 +4,7 @@ //! the swarm is the record, this file is the hive's last read of it. //! //! Format: agent name → the stored -//! [`swarm_queue_client::agent_permissions::AgentPermissions`]. An agent with +//! [`swarm_wire::agent_permissions::AgentPermissions`]. An agent with //! no entry, or an entry with `permissions` unset, holds //! [`Permission::DEFAULT`]; `"permissions": []` holds none. //! @@ -18,7 +18,8 @@ use std::collections::BTreeMap; use std::path::{Path, PathBuf}; use hive_types::Permission; -use swarm_queue_client::agent_permissions::{AgentPermissions, HIVE_COPY_FILE, HiveCopy}; +use swarm_queue_client::agent_permissions::{HIVE_COPY_FILE, HiveCopy}; +use swarm_wire::agent_permissions::AgentPermissions; const FILE: &str = "swarm-agent-permissions.json"; diff --git a/hive-c0re/src/workers/permissions_import.rs b/hive-c0re/src/workers/permissions_import.rs index e139d2a1..815c1b47 100644 --- a/hive-c0re/src/workers/permissions_import.rs +++ b/hive-c0re/src/workers/permissions_import.rs @@ -14,7 +14,7 @@ use std::path::{Path, PathBuf}; use anyhow::{Context, Result}; use hive_types::Permission; -use swarm_queue_client::agent_permissions::{ +use swarm_wire::agent_permissions::{ self, AgentPermissions, IMPORT_REPLY_DEADLINE, Import, ImportReply, }; diff --git a/hive-c0re/src/workers/swarm_permissions.rs b/hive-c0re/src/workers/swarm_permissions.rs index 9ed88432..5f0cd477 100644 --- a/hive-c0re/src/workers/swarm_permissions.rs +++ b/hive-c0re/src/workers/swarm_permissions.rs @@ -19,7 +19,8 @@ use std::time::Duration; use anyhow::{Context, Result}; use hive_types::Permission; -use swarm_queue_client::agent_permissions::{self, AgentPermissions, Read}; +use swarm_queue_client::agent_permissions::{self, Read}; +use swarm_wire::agent_permissions::AgentPermissions; use crate::coordinator::Coordinator; use crate::permissions::{self, Stored}; @@ -182,7 +183,7 @@ mod tests { use std::path::Path; use super::{Read, sync_with}; - use swarm_queue_client::agent_permissions::AgentPermissions; + use swarm_wire::agent_permissions::AgentPermissions; fn record(json: &str) -> AgentPermissions { serde_json::from_str(json).expect("decodes") diff --git a/hive-subagent-mcp/Cargo.toml b/hive-subagent-mcp/Cargo.toml index fac4e45f..566d5978 100644 --- a/hive-subagent-mcp/Cargo.toml +++ b/hive-subagent-mcp/Cargo.toml @@ -28,7 +28,6 @@ rmcp.workspace = true schemars.workspace = true serde.workspace = true serde_json.workspace = true -# `subagent_term`: the subject each subagent's rows are published on. # `agent-inference`: this agent's `subagent_models`, read with one direct get. swarm-queue-client = { workspace = true, features = [ "agent-inference", diff --git a/hive-subagent-mcp/src/swarm_term.rs b/hive-subagent-mcp/src/swarm_term.rs index 428aea3f..a4ecd14b 100644 --- a/hive-subagent-mcp/src/swarm_term.rs +++ b/hive-subagent-mcp/src/swarm_term.rs @@ -6,7 +6,7 @@ //! output reader, so `offer` only queues. One task drains the queue: it //! classifies each `stream-json` event into the rows the agent's own terminal //! carries (`hive_sh4re::stream_enrich`) and publishes them on -//! `$SWARM.term..sub.` (`swarm_queue_client::subagent_term`). +//! `$SWARM.term..sub.` (`swarm_wire::subagent_term`). //! //! **Output only.** Nothing here subscribes: a subagent takes no input from //! the swarm. @@ -33,12 +33,12 @@ use std::time::{Duration, Instant, SystemTime, UNIX_EPOCH}; use anyhow::{Context as _, anyhow}; use hive_sh4re::term_msg::{ClassifyCtx, Level, TermMsg, fit, iso8601_utc}; -use swarm_queue_client::subagent_term; use swarm_secret_client::{ SecretStore, client::{DEFAULT_CERT_MOUNT, ENV_ADDR, ENV_CACERT, Settings}, policy, queue, }; +use swarm_wire::subagent_term; use tokio::sync::mpsc; /// Names the agent. The store role, the store path, the queue token and the diff --git a/nix/host-modules/swarm-nats.nix b/nix/host-modules/swarm-nats.nix index a8fbc65e..db0d193e 100644 --- a/nix/host-modules/swarm-nats.nix +++ b/nix/host-modules/swarm-nats.nix @@ -889,7 +889,7 @@ in # harness mirrors its `AGENTS.md` there for the swarm to read. "--agent-token-publish-subject ${lib.escapeShellArg "\$\$KV.agents-md.{agent}"}" # Its subagents' terminal rows - # (`swarm_queue_client::subagent_term`), publish only and no + # (`swarm_wire::subagent_term`), publish only and no # `$JS.API.*` subject: swarm-controller creates the # `term-sub-{agent}` stream that keeps them, so its subjects # and limits are never the agent's to choose. @@ -905,7 +905,7 @@ in # (`swarm_wire::agent_inference`), on the same terms. "--agent-token-publish-subject ${lib.escapeShellArg "\$\$JS.API.DIRECT.GET.KV_agent-inference.\$\$KV.agent-inference.{agent}"}" # Its own key of the `agent-permissions` KV bucket - # (`swarm_queue_client::agent_permissions`), on the same terms: + # (`swarm_wire::agent_permissions`), on the same terms: # the harness and the subagent daemon read the tool permissions # it holds. "--agent-token-publish-subject ${lib.escapeShellArg "\$\$JS.API.DIRECT.GET.KV_agent-permissions.\$\$KV.agent-permissions.{agent}"}" diff --git a/swarm-controller/src/agent_login.rs b/swarm-controller/src/agent_login.rs index 7f7e20a1..708c752d 100644 --- a/swarm-controller/src/agent_login.rs +++ b/swarm-controller/src/agent_login.rs @@ -138,10 +138,7 @@ mod tests { &swarm_queue_client::agent_state_subject("iris"), &header(Some(true)), ); - reported.record( - &swarm_queue_client::term_subject("atlas"), - &header(Some(true)), - ); + reported.record(&swarm_wire::term_subject("atlas"), &header(Some(true))); assert_eq!(reported.get("iris"), Some(true)); assert_eq!(reported.get("atlas"), None); } diff --git a/swarm-controller/src/agent_permissions.rs b/swarm-controller/src/agent_permissions.rs index f20ace7d..b9aa520c 100644 --- a/swarm-controller/src/agent_permissions.rs +++ b/swarm-controller/src/agent_permissions.rs @@ -20,7 +20,10 @@ use futures_util::StreamExt as _; use hive_types::Permission; use problem_details::ProblemDetails; use serde::{Deserialize, Serialize}; -use swarm_queue_client::agent_permissions::{self, AgentPermissions, Import, ImportReply}; +use swarm_queue_client::agent_permissions; +use swarm_wire::agent_permissions::{ + AgentPermissions, Import, ImportReply, hive_from_import_subject, import_subject_wildcard, +}; use utoipa::ToSchema; use crate::routes::{AppState, error_problem, wanted_error_status}; @@ -377,7 +380,7 @@ pub(crate) fn spawn(client: async_nats::Client, wanted: Arc, hives .await; } }); - let subject = agent_permissions::import_subject_wildcard(); + let subject = import_subject_wildcard(); loop { match client.subscribe(subject.clone()).await { Ok(mut requests) => { @@ -413,7 +416,7 @@ async fn answer_import( tracing::warn!(subject = %msg.subject, %error, "agent permissions: import refused"); ImportReply::Failed { error } }; - let Some(hive) = agent_permissions::hive_from_import_subject(&msg.subject) else { + let Some(hive) = hive_from_import_subject(&msg.subject) else { return failed(format!("{} is not an import subject", msg.subject)); }; if !hives.iter().any(|h| h == hive) { @@ -529,7 +532,7 @@ mod tests { use axum::extract::{Path, State}; use axum::http::StatusCode; use hive_types::Permission; - use swarm_queue_client::agent_permissions::{Import, ImportReply}; + use swarm_wire::agent_permissions::{Import, ImportReply}; async fn published(client: &async_nats::Client, agent: &str) -> Option { let store = swarm_queue_client::agent_permissions::open_or_create(client) diff --git a/swarm-controller/src/message_stream.rs b/swarm-controller/src/message_stream.rs index 9f07e6fa..d5773b53 100644 --- a/swarm-controller/src/message_stream.rs +++ b/swarm-controller/src/message_stream.rs @@ -35,7 +35,7 @@ use futures_util::{Stream, StreamExt as _, stream}; use hive_types::term_msg::{SEND_ICON, TURN_ICON, send_recipient, turn_sender}; use serde::Deserialize; use swarm_controller_api::messages::SwarmMessage; -use swarm_queue_client::agent_term; +use swarm_wire::agent_term; use utoipa::IntoParams; use crate::routes::{AppState, error_problem}; @@ -104,12 +104,9 @@ pub(crate) async fn stream_messages( ) })?; - let subscriber = crate::term_stream::subscribe( - &client, - swarm_queue_client::term_subject_wildcard(), - "messages", - ) - .await?; + let subscriber = + crate::term_stream::subscribe(&client, swarm_wire::term_subject_wildcard(), "messages") + .await?; let roster: Arc> = Arc::new(roster.into_iter().collect()); let history = history(&client, &roster, minutes).await; let seen: HashSet = history.iter().cloned().collect(); @@ -170,7 +167,7 @@ async fn agent_history( let replayed = crate::term_stream::replay( &client, &agent_term::stream_name(&agent), - swarm_queue_client::term_subject(&agent), + swarm_wire::term_subject(&agent), None, ) .await; @@ -203,7 +200,7 @@ fn live( /// The agent a main terminal subject belongs to. A subagent's subject has /// more tokens and names none. fn agent_of(subject: &str) -> Option<&str> { - let agent = subject.strip_prefix(&swarm_queue_client::term_subject(""))?; + let agent = subject.strip_prefix(&swarm_wire::term_subject(""))?; (!agent.is_empty() && !agent.contains('.')).then_some(agent) } @@ -307,12 +304,12 @@ mod tests { let nats = TestNats::start(); let client = nats.client().await; let subscriber = client - .subscribe(swarm_queue_client::term_subject_wildcard()) + .subscribe(swarm_wire::term_subject_wildcard()) .await .expect("subscribe"); let mut messages = Box::pin(live(subscriber, roster())); - let atlas = swarm_queue_client::term_subject("atlas"); - let iris = swarm_queue_client::term_subject("iris"); + let atlas = swarm_wire::term_subject("atlas"); + let iris = swarm_wire::term_subject("iris"); publish( &client, @@ -350,7 +347,7 @@ mod tests { .await; publish( &client, - swarm_queue_client::subagent_term::subject("atlas", "scout"), + swarm_wire::subagent_term::subject("atlas", "scout"), row(SEND_ICON, "send* → iris", "from a subagent"), ) .await; @@ -366,7 +363,7 @@ mod tests { let nats = TestNats::start(); let client = nats.client().await; for agent in ["atlas", "iris"] { - agent_term::open_or_create(&client, agent) + swarm_queue_client::agent_term::open_or_create(&client, agent) .await .expect("the stream"); } @@ -381,8 +378,8 @@ mod tests { .as_secs(); let now = iso8601_utc(i64::try_from(secs).expect("fits")); let earlier = iso8601_utc(i64::try_from(secs - 600).expect("fits")); - let atlas = swarm_queue_client::term_subject("atlas"); - let iris = swarm_queue_client::term_subject("iris"); + let atlas = swarm_wire::term_subject("atlas"); + let iris = swarm_wire::term_subject("iris"); publish(&client, atlas, stamped(&now, "send* → iris", "newer")).await; publish( &client, diff --git a/swarm-controller/src/routes/agents.rs b/swarm-controller/src/routes/agents.rs index c70b62b0..cc4ae16d 100644 --- a/swarm-controller/src/routes/agents.rs +++ b/swarm-controller/src/routes/agents.rs @@ -711,7 +711,7 @@ pub(crate) async fn create_agent( struct CreateSettings { limits: Option, inference: Option, - permissions: Option, + permissions: Option, archetype: Option, agents_md: Option, } @@ -794,7 +794,7 @@ impl StoredSettings { agent: &str, limits: Option<&swarm_wire::agent_limits::AgentCaps>, inference: Option<&swarm_wire::agent_inference::AgentInference>, - permissions: Option<&swarm_queue_client::agent_permissions::AgentPermissions>, + permissions: Option<&swarm_wire::agent_permissions::AgentPermissions>, archetype: Option<&swarm_queue_client::agent_archetype::ArchetypeRef>, agents_md: Option<&str>, ) -> Result { diff --git a/swarm-controller/src/subagent_term.rs b/swarm-controller/src/subagent_term.rs index f93f27e1..5392de5c 100644 --- a/swarm-controller/src/subagent_term.rs +++ b/swarm-controller/src/subagent_term.rs @@ -34,7 +34,7 @@ use axum::http::{HeaderMap, StatusCode}; use axum::response::sse::{Event, KeepAlive, Sse}; use futures_util::{Stream, StreamExt as _, TryStreamExt as _}; use serde::Serialize; -use swarm_queue_client::{agent_term, subagent_term}; +use swarm_wire::subagent_term; use utoipa::ToSchema; use crate::routes::AppState; @@ -72,14 +72,18 @@ pub(crate) fn spawn(client: async_nats::Client, wanted: Arc, hives continue; } for agent in live_agents(&wanted, &hives).await { - if let Err(e) = subagent_term::open_or_create(&client, &agent).await { + if let Err(e) = + swarm_queue_client::subagent_term::open_or_create(&client, &agent).await + { tracing::warn!( %agent, error = %swarm_queue_client::chain(&e), "subagent streams: creating the agent's stream failed; next pass" ); } - if let Err(e) = agent_term::open_or_create(&client, &agent).await { + if let Err(e) = + swarm_queue_client::agent_term::open_or_create(&client, &agent).await + { tracing::warn!( %agent, error = %swarm_queue_client::chain(&e), @@ -287,7 +291,7 @@ fn read_failed(agent: &str, e: &dyn std::error::Error) -> problem_details::Probl #[cfg(test)] mod tests { - use swarm_queue_client::subagent_term::subagent_names; + use swarm_wire::subagent_term::subagent_names; /// The list is what `STREAM.INFO` reports for the stream's subjects, by /// name: a subagent that published once and one that published many rows diff --git a/swarm-controller/src/term_stream.rs b/swarm-controller/src/term_stream.rs index 193cb3cc..21775dc3 100644 --- a/swarm-controller/src/term_stream.rs +++ b/swarm-controller/src/term_stream.rs @@ -9,7 +9,7 @@ //! **Replay, then the live tail, through one consumer.** `hive-agent` //! publishes each already-classified `TermMsg` row to the core subject //! `$SWARM.term.{agent}`, and the stream `term-main-{agent}` -//! (`swarm_queue_client::agent_term`) keeps them. The handler creates an +//! (`swarm_wire::agent_term`) keeps them. The handler creates an //! ephemeral consumer on that stream starting at the newest //! `agent_term::REPLAY_ROWS` rows, so the reader gets those and then every //! new row, in stream order, with no gap and no duplicate between the two. @@ -35,7 +35,7 @@ use axum::http::HeaderMap; use axum::response::sse::{Event, KeepAlive, Sse}; use futures_util::stream::BoxStream; use futures_util::{Stream, StreamExt as _}; -use swarm_queue_client::agent_term; +use swarm_wire::agent_term; use crate::routes::AppState; @@ -121,7 +121,7 @@ pub(crate) async fn stream_agent_term( let stream = replay_or_live( &client, &agent_term::stream_name(&agent), - swarm_queue_client::term_subject(&agent), + swarm_wire::term_subject(&agent), last_event_id(&headers), "term", ) @@ -198,8 +198,12 @@ pub(crate) async fn replay( "term-replay-{}", deliver_subject.rsplit('.').next().unwrap_or_default() ); - let config = - agent_term::replay_consumer_config(subject.clone(), consumer_name, deliver_subject, start); + let config = swarm_queue_client::agent_term::replay_consumer_config( + subject.clone(), + consumer_name, + deliver_subject, + start, + ); let consumer = stream.create_consumer(config).await.map_err(|e| { format!( "creating a consumer on {stream_name}: {}", @@ -301,10 +305,7 @@ mod tests { async fn publish_rows(client: &async_nats::Client, rows: std::ops::RangeInclusive) { for i in rows { client - .publish( - swarm_queue_client::term_subject(AGENT), - format!("row{i}").into(), - ) + .publish(swarm_wire::term_subject(AGENT), format!("row{i}").into()) .await .expect("publish"); } @@ -335,7 +336,7 @@ mod tests { replay( client, &agent_term::stream_name(AGENT), - swarm_queue_client::term_subject(AGENT), + swarm_wire::term_subject(AGENT), resume_after, ) .await @@ -352,7 +353,7 @@ mod tests { for subagent in ["scout", "builder"] { client .publish( - swarm_queue_client::subagent_term::subject(AGENT, subagent), + swarm_wire::subagent_term::subject(AGENT, subagent), format!("{subagent}{i}").into(), ) .await @@ -368,8 +369,8 @@ mod tests { ) -> Result, String> { replay( client, - &swarm_queue_client::subagent_term::stream_name(AGENT), - swarm_queue_client::subagent_term::subject(AGENT, "scout"), + &swarm_wire::subagent_term::stream_name(AGENT), + swarm_wire::subagent_term::subject(AGENT, "scout"), resume_after, ) .await @@ -413,7 +414,7 @@ mod tests { let nats = TestNats::start(); let client = nats.client().await; - let stream = agent_term::open_or_create(&client, AGENT) + let stream = swarm_queue_client::agent_term::open_or_create(&client, AGENT) .await .expect("the stream"); publish_rows(&client, 1..=3).await; @@ -445,12 +446,12 @@ mod tests { async fn ensuring_the_terminal_stream_again_leaves_it_as_it_is() { let nats = TestNats::start(); let client = nats.client().await; - let created = agent_term::open_or_create(&client, AGENT) + let created = swarm_queue_client::agent_term::open_or_create(&client, AGENT) .await .expect("the first ensure creates the stream"); publish_rows(&client, 1..=3).await; stored(&client, 3).await; - agent_term::open_or_create(&client, AGENT) + swarm_queue_client::agent_term::open_or_create(&client, AGENT) .await .expect("a second ensure succeeds"); // Read back from the server, not from what the ensure returned. @@ -484,7 +485,7 @@ mod tests { .expect("an earlier stream"); publish_rows(&client, 1..=2).await; stored(&client, 2).await; - let opened = agent_term::open_or_create(&client, AGENT) + let opened = swarm_queue_client::agent_term::open_or_create(&client, AGENT) .await .expect("an existing stream opens"); assert_eq!(opened.cached_info().config.max_message_size, 4096); @@ -497,7 +498,7 @@ mod tests { async fn the_replay_reaches_the_reader_before_the_live_tail() { let nats = TestNats::start(); let client = nats.client().await; - agent_term::open_or_create(&client, AGENT) + swarm_queue_client::agent_term::open_or_create(&client, AGENT) .await .expect("the stream"); publish_rows(&client, 1..=150).await; @@ -514,7 +515,7 @@ mod tests { async fn a_reconnect_gets_every_later_row_once() { let nats = TestNats::start(); let client = nats.client().await; - agent_term::open_or_create(&client, AGENT) + swarm_queue_client::agent_term::open_or_create(&client, AGENT) .await .expect("the stream"); publish_rows(&client, 1..=150).await; @@ -541,7 +542,7 @@ mod tests { async fn a_subagent_replays_its_own_newest_rows_before_the_live_tail() { let nats = TestNats::start(); let client = nats.client().await; - let stream_name = swarm_queue_client::subagent_term::stream_name(AGENT); + let stream_name = swarm_wire::subagent_term::stream_name(AGENT); swarm_queue_client::subagent_term::open_or_create(&client, AGENT) .await .expect("the stream"); @@ -559,7 +560,7 @@ mod tests { async fn a_subagent_reconnect_gets_every_later_row_once() { let nats = TestNats::start(); let client = nats.client().await; - let stream_name = swarm_queue_client::subagent_term::stream_name(AGENT); + let stream_name = swarm_wire::subagent_term::stream_name(AGENT); swarm_queue_client::subagent_term::open_or_create(&client, AGENT) .await .expect("the stream"); diff --git a/swarm-nats-auth/src/policy.rs b/swarm-nats-auth/src/policy.rs index 31ea3eae..e3704e4f 100644 --- a/swarm-nats-auth/src/policy.rs +++ b/swarm-nats-auth/src/policy.rs @@ -241,7 +241,7 @@ impl Policy { subscribe: Some(Self::reader_subscribe(client_id)), // The answer to a hive's permissions import. NATS lets it // answer only a message it received, once. - responses: Some(swarm_queue_client::agent_permissions::IMPORT_REPLY_DEADLINE), + responses: Some(swarm_wire::agent_permissions::IMPORT_REPLY_DEADLINE), }); } if self.review_classifiers.iter().any(|c| c == client_id) { @@ -324,15 +324,15 @@ impl Policy { /// to swarm-ui, every hive's permissions import, and its own reply inbox. /// /// The three families are the publishers' (`hive-agent`'s `swarm_term` - /// and `swarm_agent_state`, `swarm_queue_client::subagent_term`), spelled + /// and `swarm_agent_state`, `swarm_wire::subagent_term`), spelled /// by the same `swarm_queue_client` helpers. Operator input, deploys and /// knowledge are absent: the reader publishes those. fn reader_subscribe(client_id: &str) -> Vec { vec![ - swarm_queue_client::term_subject_wildcard(), - swarm_queue_client::subagent_term::subject_wildcard(), + swarm_wire::term_subject_wildcard(), + swarm_wire::subagent_term::subject_wildcard(), swarm_queue_client::agent_state_subject_wildcard(), - swarm_queue_client::agent_permissions::import_subject_wildcard(), + swarm_wire::agent_permissions::import_subject_wildcard(), Self::inbox(client_id), ] } @@ -577,11 +577,11 @@ impl Policy { swarm_wire::agent_limits::STREAM )); // The permissions the swarm grants those agents - // (`swarm_queue_client::agent_permissions`), read the same way and + // (`swarm_wire::agent_permissions`), read the same way and // every key for the same reason. subjects.push(format!( "$JS.API.DIRECT.GET.{}.>", - swarm_queue_client::agent_permissions::STREAM + swarm_wire::agent_permissions::STREAM )); // The archetype an agent was created from // (`swarm_queue_client::agent_archetype`), read the same way and every @@ -593,7 +593,7 @@ impl Policy { // The one-shot import of the grants this hive's meta repo held, sent // as a request on this hive's own subject: the controller takes the // hive from the subject, so one hive cannot import for another. - subjects.push(swarm_queue_client::agent_permissions::import_subject(hive)); + subjects.push(swarm_wire::agent_permissions::import_subject(hive)); subjects.extend( self.extra_hive_subjects .iter() @@ -701,11 +701,11 @@ impl Policy { // which this client already subscribes to and reads by // `MSG.GET` above. No bare `CONSUMER.CREATE.`, whose // filter is unchecked. - swarm_queue_client::agent_term::consumer_create_grant(), + swarm_wire::agent_term::consumer_create_grant(), // The same for one subagent's rows in `term-sub-`: the // filter is `$SWARM.term..sub.`, two tokens // deeper than the grant above reaches. - swarm_queue_client::subagent_term::consumer_create_grant(), + swarm_wire::subagent_term::consumer_create_grant(), "$KV.*.>".to_owned(), ]); // The per-agent status bucket, read-side only. Same five subjects as @@ -925,7 +925,7 @@ mod tests { /// no other, and writes none. #[test] fn a_hive_reads_every_agents_permissions_and_an_agent_only_its_own() { - use swarm_queue_client::agent_permissions::{STREAM, key_subject}; + use swarm_wire::agent_permissions::{STREAM, key_subject}; let direct_get_subject = |agent: &str| { format!("$JS.API.DIRECT.GET.KV_agent-permissions.$KV.agent-permissions.{agent}") @@ -962,7 +962,7 @@ mod tests { /// hive's and may answer, and no other role may send one. #[test] fn a_hive_imports_only_under_its_own_name_and_the_reader_answers() { - use swarm_queue_client::agent_permissions::{ + use swarm_wire::agent_permissions::{ IMPORT_REPLY_DEADLINE, import_subject, import_subject_wildcard, }; @@ -1408,7 +1408,7 @@ mod tests { let inbox = client_inbox_prefix("swarm-controller"); for allowed in [ "$SWARM.term.atlas".to_owned(), - swarm_queue_client::subagent_term::subject("atlas", "scout"), + swarm_wire::subagent_term::subject("atlas", "scout"), "$SWARM.agent-state.atlas".to_owned(), reply(&inbox), multiplexer(&inbox), @@ -1443,8 +1443,8 @@ mod tests { .permissions("swarm-controller") .expect("the reader"); for built in [ - swarm_queue_client::term_subject("atlas"), - swarm_queue_client::subagent_term::subject("atlas", "scout"), + swarm_wire::term_subject("atlas"), + swarm_wire::subagent_term::subject("atlas", "scout"), swarm_queue_client::agent_state_subject("atlas"), ] { assert!( @@ -1778,8 +1778,8 @@ mod tests { #[test] fn the_controller_may_replay_terminal_rows_and_no_other_subject() { - use swarm_queue_client::agent_term::stream_name; - use swarm_queue_client::{subagent_term, term_subject}; + use swarm_wire::agent_term::stream_name; + use swarm_wire::{subagent_term, term_subject}; let p = policy() .permissions("swarm-controller") @@ -1983,7 +1983,7 @@ mod tests { /// the controller's to create, which its own grant covers. #[test] fn an_agents_subagent_grant_is_publish_on_its_own_family_only() { - use swarm_queue_client::subagent_term::{stream_name, stream_subjects, subject}; + use swarm_wire::subagent_term::{stream_name, stream_subjects, subject}; let p = policy() .with_agent_token_subjects(vec!["$SWARM.term.{agent}.sub.>".to_owned()]) diff --git a/swarm-queue-client/Cargo.toml b/swarm-queue-client/Cargo.toml index 75685c95..6b398996 100644 --- a/swarm-queue-client/Cargo.toml +++ b/swarm-queue-client/Cargo.toml @@ -51,7 +51,7 @@ operator-profiles = ["async-nats/jetstream"] # `agent_limits::read`, for hive-c0re, which reads the key of each agent it # runs the same way. Explicit for the same reason as `notices`. agent-limits = ["async-nats/jetstream"] -# `agent_permissions::AgentPermissions` and `read`, for hive-c0re, the agent +# `agent_permissions::read` and `HiveCopy`, for hive-c0re, the agent # harness, the subagent daemon and swarm-controller. Pulls `hive-types` for the # `Permission` names, so the crates that only need the bucket name do not link # it. diff --git a/swarm-queue-client/src/agent_permissions.rs b/swarm-queue-client/src/agent_permissions.rs index d0e86dcd..85b27ff8 100644 --- a/swarm-queue-client/src/agent_permissions.rs +++ b/swarm-queue-client/src/agent_permissions.rs @@ -1,102 +1,10 @@ -//! The per-agent permissions KV bucket: one key per agent, holding the -//! [`hive_types::Permission`] set the swarm grants it. -//! -//! **swarm-controller is the only writer**, and creates the bucket when it -//! starts. A missing key or an unset `permissions` field leaves the agent on -//! the compiled-in default (`Permission::DEFAULT`); an empty list grants -//! nothing. -//! -//! **Two readers, each enforcing its own part.** The hive running the agent -//! applies `manage_root_agent` as bind mounts at container start; its grant -//! covers every key of this bucket, read-only, for the reason -//! `swarm_wire::agent_limits` gives. The agent itself (its harness and subagent -//! daemon) applies the tool permissions; its grant covers only its own key. -//! Both read with one `DIRECT.GET` on the agent's key -//! ([`crate::agent_permissions::key_subject`]). -//! -//! Keyed by the agent alone, like `swarm_wire::agent_limits`: an agent can move -//! between hives, and its permissions go with it. -//! -//! Also carries the one-shot import a hive sends of the grants its meta repo -//! held ([`crate::agent_permissions::import_subject`]), answered by the -//! controller. +//! Opening and reading the per-agent permissions KV bucket, and the copy +//! hive-c0re keeps of a record for the agent's own processes. The names, the +//! stored type and the import messages are `swarm_wire::agent_permissions`; +//! reading needs the `agent-permissions` feature and writing needs `kv`. -/// The KV bucket permissions are published into, keyed by agent name. -pub const BUCKET: &str = "agent-permissions"; - -/// The stream backing [`BUCKET`]. `create_key_value` prefixes `KV_`. -pub const STREAM: &str = "KV_agent-permissions"; - -/// The subject `agent`'s entry is stored under. -#[must_use] -pub fn key_subject(agent: &str) -> String { - format!("$KV.{BUCKET}.{agent}") -} - -const IMPORT_SUBJECT_PREFIX: &str = "$SWARM.permissions-import"; - -/// The subject `hive` sends its import on. One per hive, so the hive's grant -/// names only its own, and the controller takes the hive from the subject -/// rather than from the payload. -#[must_use] -pub fn import_subject(hive: &str) -> String { - format!("{IMPORT_SUBJECT_PREFIX}.{hive}") -} - -/// Every [`import_subject`] at once, for the controller's subscription. -#[must_use] -pub fn import_subject_wildcard() -> String { - import_subject("*") -} - -/// The hive an [`import_subject`] names, or `None` for any other subject. -#[must_use] -pub fn hive_from_import_subject(subject: &str) -> Option<&str> { - let hive = subject - .strip_prefix(IMPORT_SUBJECT_PREFIX)? - .strip_prefix('.')?; - (!hive.is_empty() && !hive.contains('.')).then_some(hive) -} - -/// How long a hive waits for the controller's answer to an import, and the -/// queue lets the controller answer. -pub const IMPORT_REPLY_DEADLINE: std::time::Duration = std::time::Duration::from_secs(30); - -/// One agent's permissions: the value under `BUCKET`/``. `None` is -/// "not set by the swarm", which a hive reads as its default. -/// -/// No `deny_unknown_fields`: a field a newer controller adds must not cost an -/// older hive the fields it knows. #[cfg(feature = "agent-permissions")] -#[derive(Debug, Clone, Default, PartialEq, Eq, serde::Serialize, serde::Deserialize)] -pub struct AgentPermissions { - #[serde(default, skip_serializing_if = "Option::is_none")] - pub permissions: Option>, -} - -/// What a hive sends once: the permissions its meta repo held, by agent. -#[cfg(feature = "agent-permissions")] -#[derive(Debug, Clone, Default, PartialEq, Eq, serde::Serialize, serde::Deserialize)] -pub struct Import { - pub permissions: std::collections::BTreeMap>, -} - -/// The controller's answer to an [`Import`]. -#[cfg(feature = "agent-permissions")] -#[derive(Debug, Clone, PartialEq, Eq, serde::Serialize, serde::Deserialize)] -#[serde(tag = "state", rename_all = "kebab-case")] -pub enum ImportReply { - /// Every agent was handled. `written` had no key and got the hive's value; - /// `kept` already had a key, which was left alone; `skipped` is not an - /// agent the sending hive's wanted state declares. - Imported { - written: Vec, - kept: Vec, - skipped: Vec, - }, - /// Nothing is known to be written; the hive sends again later. - Failed { error: String }, -} +use swarm_wire::agent_permissions::AgentPermissions; /// Open the permissions bucket, creating it if nothing has yet. /// @@ -108,7 +16,7 @@ pub async fn open_or_create( let js = async_nats::jetstream::new(client.clone()); crate::kv::open_or_create( &js, - BUCKET, + swarm_wire::agent_permissions::BUCKET, "agent-permissions", "Permissions the swarm controller sets for each agent".to_owned(), ) @@ -209,8 +117,13 @@ pub async fn read( use async_nats::jetstream::stream::DirectGetErrorKind; let js = async_nats::jetstream::new(client.clone()); - let stream = js.get_stream_no_info(STREAM).await?; - let message = match stream.direct_get_last_for_subject(key_subject(agent)).await { + let stream = js + .get_stream_no_info(swarm_wire::agent_permissions::STREAM) + .await?; + let message = match stream + .direct_get_last_for_subject(swarm_wire::agent_permissions::key_subject(agent)) + .await + { Ok(message) => message, Err(e) if matches!(e.kind(), DirectGetErrorKind::NotFound) => { return Ok(Read::Current(AgentPermissions::default())); @@ -245,53 +158,10 @@ fn decode(operation: Option<&str>, payload: &[u8]) -> serde_json::Result`. -//! [`crate::subagent_term`]'s stream captures `$SWARM.term..sub.>`, -//! which only matches subjects with at least two more tokens, so no subject -//! is in both and the two streams never overlap. The `JetStream` server refuses -//! to create a stream whose subjects overlap another's. -//! -//! The names live here because three crates must agree on them: the -//! publisher in `hive-agent`, `swarm-controller`, and the reader's consumer -//! grant in `swarm-nats-auth`. +//! The stream captures exactly one literal subject, which +//! `swarm_wire::subagent_term`'s stream never overlaps. The `JetStream` server +//! refuses to create a stream whose subjects overlap another's. -/// The token between the prefix and the agent in [`stream_name`]. It differs -/// from `term-sub-` at the same position, so no agent name makes this -/// stream's name equal another agent's subagent stream. -const STREAM_KIND: &str = "main"; - -/// How long the stream keeps a row. The same window as the subagent stream. -pub const MAX_AGE: std::time::Duration = std::time::Duration::from_hours(24); +use swarm_wire::agent_term::{IDLE_HEARTBEAT, MAX_AGE, stream_name, stream_subject}; /// How large one agent's stream may grow before the discard policy reclaims /// space. The same cap as the subagent stream, for the same reason: it /// bounds the disk one agent can claim. -#[cfg(feature = "agent-term")] const MAX_BYTES: i64 = 64 * 1024 * 1024; -/// How many of the newest rows a reader that attaches fresh is replayed -/// before the live tail. The agent's own web UI loads the same number of -/// events as its first history page. -pub const REPLAY_ROWS: u64 = 100; - -/// How often the server sends an idle replay consumer a heartbeat. Two missed -/// in a row end the reader's row stream, so a consumer the server dropped, -/// as on a restart, ends the SSE response and the browser reconnects. -pub const IDLE_HEARTBEAT: std::time::Duration = std::time::Duration::from_secs(5); - -/// The stream holding `agent`'s terminal rows: `term-main-`. -#[must_use] -pub fn stream_name(agent: &str) -> String { - format!("term-{STREAM_KIND}-{agent}") -} - -/// The one subject [`stream_name`]'s stream captures. -#[must_use] -pub fn stream_subject(agent: &str) -> String { - crate::term_subject(agent) -} - -/// The reader's grant to create a replay consumer: on any stream, under any -/// consumer name, in the extended create form whose last tokens are the -/// consumer's filter. -/// -/// The server refuses a create whose payload filter differs from those -/// tokens, so a consumer made under this grant delivers only rows on -/// subjects below `$SWARM.term`. The grant's last `*` also matches a literal -/// `*` or `>` token, so besides one agent's [`crate::term_subject`] it admits -/// `$SWARM.term.*` and `$SWARM.term.>`, which take in every agent's and every -/// subagent's terminal. The bare `CONSUMER.CREATE.` form, whose -/// filter rides only in the payload, is not covered. -#[must_use] -pub fn consumer_create_grant() -> String { - format!( - "$JS.API.CONSUMER.CREATE.*.*.{}", - crate::term_subject_wildcard() - ) -} - -/// The first stream sequence to replay from. -/// -/// `resume_after` is the last sequence the reader already holds, from a -/// reconnecting browser's `Last-Event-ID`; the replay continues right after -/// it. A fresh reader gets the newest [`REPLAY_ROWS`] below `last_sequence`, -/// the stream's newest row when it was read. A sequence the stream has since -/// discarded starts the consumer at the oldest row it still has. -/// -/// A `resume_after` above `last_sequence` names no row this stream ever held, -/// as after the stream was created again, and is read as a fresh reader: the -/// server accepts a start past the end and delivers nothing until the stream -/// reaches it. -#[must_use] -pub fn replay_start(last_sequence: u64, resume_after: Option) -> u64 { - match resume_after { - Some(seen) if seen <= last_sequence => seen + 1, - _ => last_sequence.saturating_sub(REPLAY_ROWS - 1).max(1), - } -} - /// The config [`open_or_create`] creates `agent`'s stream with. /// /// `max_message_size` tracks the queue's live `max_payload`, as the subagent /// stream's does, so the stream never rejects a row the queue accepted. -#[cfg(feature = "agent-term")] fn stream_config(agent: &str, max_message_size: i32) -> async_nats::jetstream::stream::Config { async_nats::jetstream::stream::Config { name: stream_name(agent), @@ -113,7 +32,6 @@ fn stream_config(agent: &str, max_message_size: i32) -> async_nats::jetstream::s /// Open `agent`'s terminal stream, creating it if it does not exist yet. An /// existing stream is opened as it is. -#[cfg(feature = "agent-term")] pub async fn open_or_create( client: &async_nats::Client, agent: &str, @@ -138,15 +56,14 @@ pub async fn open_or_create( /// The consumer that replays the rows on `filter_subject` from /// `start_sequence` and then follows new ones, delivered to `deliver_subject` /// in stream order. The filter is an agent's [`stream_subject`] or one -/// subagent's [`crate::subagent_term::subject`]. +/// subagent's `swarm_wire::subagent_term::subject`. /// /// Ephemeral and unacknowledged: the server drops it once nothing listens /// on `deliver_subject`, so a reader that goes away needs no delete grant. /// `name` must be set, because the server only takes the filter from the -/// subject [`consumer_create_grant`] (or -/// [`crate::subagent_term::consumer_create_grant`]) covers when the create +/// subject `swarm_wire::agent_term::consumer_create_grant` (or +/// `swarm_wire::subagent_term::consumer_create_grant`) covers when the create /// names a consumer. -#[cfg(feature = "agent-term")] #[must_use] pub fn replay_consumer_config( filter_subject: String, @@ -171,106 +88,9 @@ pub fn replay_consumer_config( mod tests { use super::*; - /// Token-wise NATS subject overlap: whether some literal subject matches - /// both patterns. - fn overlaps(a: &str, b: &str) -> bool { - let (mut a, mut b) = (a.split('.'), b.split('.')); - loop { - match (a.next(), b.next()) { - (Some(">"), Some(_)) | (Some(_), Some(">")) | (None, None) => return true, - (None, _) | (_, None) => return false, - (Some(x), Some(y)) if x != "*" && y != "*" && x != y => return false, - _ => {} - } - } - } - - #[test] - fn names_match_the_publisher_and_the_grant() { - assert_eq!(stream_name("iris"), "term-main-iris"); - assert_eq!(stream_subject("iris"), "$SWARM.term.iris"); - assert_eq!( - consumer_create_grant(), - "$JS.API.CONSUMER.CREATE.*.*.$SWARM.term.*" - ); - } - - /// The server refuses to create a stream whose subjects overlap - /// another's, so this stream must share no subject with any agent's - /// subagent stream, its own agent's included. - #[test] - fn the_subject_is_disjoint_from_every_subagent_stream() { - for other in ["iris", "argus", "sub", "iris-2"] { - assert!( - !overlaps( - &stream_subject("iris"), - &crate::subagent_term::stream_subjects(other) - ), - "term-main-iris overlaps term-sub-{other}" - ); - } - // Control: the checker does report a real overlap. - assert!(overlaps( - "$SWARM.term.iris.>", - &crate::subagent_term::stream_subjects("iris") - )); - assert!(overlaps(&stream_subject("iris"), "$SWARM.term.*")); - } - - /// No agent name makes this stream's name another agent's subagent - /// stream: an agent may be called `sub-x`. - #[test] - fn the_name_never_collides_with_a_subagent_stream() { - for (agent, other) in [("sub-x", "x"), ("x", "sub-x"), ("main-x", "x")] { - assert_ne!(stream_name(agent), crate::subagent_term::stream_name(other)); - } - } - - /// The create the controller sends is one the grant covers, and every - /// filter the grant admits stays below `$SWARM.term`. - #[test] - fn the_grant_admits_only_filters_below_the_terminal_prefix() { - let grant = consumer_create_grant(); - let create = |filter: &str| format!("$JS.API.CONSUMER.CREATE.term-main-iris.r1.{filter}"); - assert!(overlaps(&grant, &create(&stream_subject("iris")))); - // The width the doc states: a literal wildcard as the last token. - assert!(overlaps(&grant, &create("$SWARM.term.>"))); - for other in [ - crate::subagent_term::subject("iris", "scout"), - crate::subagent_term::stream_subjects("iris"), - crate::agent_state_subject("iris"), - ] { - assert!(!overlaps(&grant, &create(&other)), "{other}"); - } - assert!(!overlaps(&grant, "$JS.API.CONSUMER.CREATE.term-main-iris")); - } - - #[test] - fn a_fresh_reader_replays_the_newest_rows() { - assert_eq!(replay_start(1000, None), 1000 - REPLAY_ROWS + 1); - assert_eq!(replay_start(5, None), 1); - assert_eq!(replay_start(0, None), 1); - } - - #[test] - fn a_reconnecting_reader_resumes_after_its_last_row() { - assert_eq!(replay_start(1000, Some(990)), 991); - assert_eq!(replay_start(1000, Some(1000)), 1001); - } - - /// An id past the stream's newest row, as after the stream was created - /// again, replays like a fresh reader instead of waiting for that number. - #[test] - fn an_id_past_the_newest_row_replays_as_a_fresh_reader() { - assert_eq!(replay_start(5, Some(99)), replay_start(5, None)); - assert_eq!(replay_start(1000, Some(1001)), replay_start(1000, None)); - assert_eq!(replay_start(0, Some(7)), 1); - } - /// The consumer starts at a stored sequence, not at the next new row, so /// the replay reaches the reader before the live tail, in stream order, /// through the one subscription. - #[cfg(feature = "agent-term")] #[test] fn the_consumer_replays_before_it_follows() { let cfg = replay_consumer_config( @@ -299,7 +119,7 @@ mod tests { /// Every limit is read from the subagent stream's own config, so the two /// fail this test the moment they drift apart. - #[cfg(all(feature = "agent-term", feature = "subagent-term"))] + #[cfg(feature = "subagent-term")] #[test] fn stream_config_matches_the_subagent_streams_limits() { let main = stream_config("iris", 8_388_608); diff --git a/swarm-queue-client/src/lib.rs b/swarm-queue-client/src/lib.rs index 63efa26f..be491399 100644 --- a/swarm-queue-client/src/lib.rs +++ b/swarm-queue-client/src/lib.rs @@ -232,10 +232,11 @@ pub mod agent_inference; /// and reading needs `agent-limits`. pub mod agent_limits; -/// The bucket swarm-controller sets each agent's permissions in — one key -/// per agent, read by the hive that runs it, plus the one-shot import a hive -/// sends. The names are unconditional; writing needs `kv` and the stored type -/// and reading need `agent-permissions`. +/// Opening and reading the bucket swarm-controller sets each agent's +/// permissions in — one key per agent, read by the hive that runs it. The names, +/// the stored type and the import messages are `swarm_wire::agent_permissions`; +/// writing needs `kv` and reading needs `agent-permissions`. +#[cfg(any(feature = "kv", feature = "agent-permissions"))] pub mod agent_permissions; /// The bucket swarm-controller publishes each operator's name and forge-profile @@ -348,39 +349,15 @@ pub const AGENT_CONTROL_SUBJECT_WILDCARD: &str = "$SWARM.control.*"; /// Shared by [`agent_control_subject`] and [`AGENT_CONTROL_SUBJECT_WILDCARD`]. const AGENT_CONTROL_SUBJECT_PREFIX: &str = "$SWARM.control"; -/// The subject `agent` publishes its terminal rows on. One leaf per agent, so -/// a subscriber can follow one agent without filtering the swarm's whole -/// terminal traffic. -/// -/// Shared by the publisher (`hive-agent`), the relay that streams it to -/// swarm-ui (`swarm-controller`) and the queue's subscribe grant for readers -/// (`swarm-nats-auth`), which spells [`term_subject_wildcard`] — so a rename -/// moves all three together. -#[must_use] -pub fn term_subject(agent: &str) -> String { - format!("{TERM_SUBJECT_PREFIX}.{agent}") -} - -/// Every [`term_subject`] at once: the reader's subscribe grant, which has no -/// roster to enumerate agents from. -#[must_use] -pub fn term_subject_wildcard() -> String { - term_subject("*") -} - -/// Shared by [`term_subject`] and [`subagent_term`], whose subjects sit under -/// the same prefix. -const TERM_SUBJECT_PREFIX: &str = "$SWARM.term"; - /// The subject `agent` publishes its turn-state header on. Same shape and -/// same three users as [`term_subject`]. +/// same three users as [`swarm_wire::term_subject`]. #[must_use] pub fn agent_state_subject(agent: &str) -> String { format!("{AGENT_STATE_SUBJECT_PREFIX}.{agent}") } /// Every [`agent_state_subject`] at once, for the same reason as -/// [`term_subject_wildcard`]. +/// [`swarm_wire::term_subject_wildcard`]. #[must_use] pub fn agent_state_subject_wildcard() -> String { agent_state_subject("*") @@ -475,14 +452,16 @@ pub struct AgentInput { #[cfg(feature = "notices")] pub mod notices; -/// The stream that keeps an agent's own terminal rows, and the consumer a -/// reader replays them through. Only the `JetStream` calls need the -/// `agent-term` feature; the names are unconditional. +/// Creating an agent's own terminal stream and the consumer a reader replays +/// it through. The names, subjects, grants and replay start are +/// `swarm_wire::agent_term`. Behind the `agent-term` feature. +#[cfg(feature = "agent-term")] pub mod agent_term; -/// The subject an agent's subagents' terminal rows go to, and the per-agent -/// stream that lists them. Only opening the stream needs the `subagent-term` -/// feature; the names are unconditional, like [`agent_icon`]'s. +/// Creating the per-agent stream that lists an agent's subagents. The names, +/// subjects and grants are `swarm_wire::subagent_term`. Behind the +/// `subagent-term` feature. +#[cfg(feature = "subagent-term")] pub mod subagent_term; /// Only the fields this needs; authelia returns several. @@ -1119,12 +1098,9 @@ mod tests { /// The names the publishers, the relays and the queue's grant all spell. #[test] - fn term_and_agent_state_subjects_are_the_agreed_names() { - assert_eq!(term_subject("mara"), "$SWARM.term.mara"); - assert_eq!(term_subject_wildcard(), "$SWARM.term.*"); + fn agent_state_subjects_are_the_agreed_names() { assert_eq!(agent_state_subject("mara"), "$SWARM.agent-state.mara"); assert_eq!(agent_state_subject_wildcard(), "$SWARM.agent-state.*"); - assert_eq!(subagent_term::subject_wildcard(), "$SWARM.term.*.sub.*"); } #[test] diff --git a/swarm-queue-client/src/subagent_term.rs b/swarm-queue-client/src/subagent_term.rs index 4c5d0c98..3a6dd853 100644 --- a/swarm-queue-client/src/subagent_term.rs +++ b/swarm-queue-client/src/subagent_term.rs @@ -1,108 +1,17 @@ -//! Subagent terminals: the subject each subagent's rows go to, and the -//! stream that keeps them. +//! Creating an agent's subagent stream. The names, subjects, grants and the +//! subagent list are `swarm_wire::subagent_term`. //! -//! An agent's subagent daemon publishes each subagent's classified terminal -//! rows on `$SWARM.term..sub.`, under the agent's own queue -//! credential. The queue grants that credential publish on this subject -//! family and no `JetStream` subject. -//! -//! The stream lets the swarm list an agent's subagents: subagents -//! are spawned on demand under caller-chosen names, and the set of subjects -//! the stream holds is the list. `swarm-controller` creates one stream per -//! agent with the fixed config [`crate::subagent_term::open_or_create`] -//! spells, and takes the list from its subject counts -//! ([`crate::subagent_term::subagent_names`]). A reader attaching to one -//! subagent's terminal replays that subagent's newest rows from it. -//! -//! The name and the subject live here because they must agree across the -//! publisher in `hive-subagent-mcp`, `swarm-controller`, the queue grant in -//! `swarm-nats.nix`, whose `{agent}` template spells the same subjects, and -//! the reader's consumer grant in `swarm-nats-auth`. +//! `swarm-controller` creates one stream per agent with the fixed config +//! [`crate::subagent_term::open_or_create`] spells. -use crate::TERM_SUBJECT_PREFIX; - -/// The token between the agent and the subagent name. A subagent subject has -/// two more tokens than the agent's own `$SWARM.term.`, so neither -/// family's subscriber receives the other's rows. -const SUBAGENT_TOKEN: &str = "sub"; - -/// How long the stream keeps a row. The subagent list is the subjects with a -/// row inside this window, so a subagent drops off the list this long after -/// its last output. -pub const MAX_AGE: std::time::Duration = std::time::Duration::from_hours(24); +use swarm_wire::subagent_term::{MAX_AGE, stream_name, stream_subjects}; /// How large one agent's stream may grow before the discard policy reclaims /// space. 64 MiB is enough rows from a day of ordinary subagent chatter to /// stay well inside it while still bounding the disk one compromised or /// runaway agent can claim for 24h. -#[cfg(feature = "subagent-term")] const MAX_BYTES: i64 = 64 * 1024 * 1024; -/// The stream holding `agent`'s subagent rows: `term-sub-`. -#[must_use] -pub fn stream_name(agent: &str) -> String { - format!("term-sub-{agent}") -} - -/// The subject `subagent`'s rows are published on. -#[must_use] -pub fn subject(agent: &str, subagent: &str) -> String { - format!("{TERM_SUBJECT_PREFIX}.{agent}.{SUBAGENT_TOKEN}.{subagent}") -} - -/// Every [`subject`] at once: the reader's subscribe grant, which has no -/// roster to enumerate agents or subagents from. -#[must_use] -pub fn subject_wildcard() -> String { - subject("*", "*") -} - -/// The reader's grant to create a replay consumer on one subagent's rows: -/// the extended create form of [`crate::agent_term::consumer_create_grant`], -/// on any stream and under any consumer name, with a [`subject`] as the -/// filter tokens the server checks the payload against. The last `*` also -/// matches a literal `*` or `>`, so it admits `$SWARM.term..sub.>` -/// too, and nothing outside [`subject_wildcard`]'s family. -#[must_use] -pub fn consumer_create_grant() -> String { - format!("$JS.API.CONSUMER.CREATE.*.*.{}", subject_wildcard()) -} - -/// Every subject [`stream_name`]'s stream captures, which is also the publish -/// grant's shape. -#[must_use] -pub fn stream_subjects(agent: &str) -> String { - format!("{TERM_SUBJECT_PREFIX}.{agent}.{SUBAGENT_TOKEN}.>") -} - -/// The subagent a stream subject names, or `None` for a subject that is not -/// exactly one subagent token under `agent`'s family. -#[must_use] -pub fn subagent_from_subject<'a>(agent: &str, subject: &'a str) -> Option<&'a str> { - let name = subject - .strip_prefix(TERM_SUBJECT_PREFIX)? - .strip_prefix('.')? - .strip_prefix(agent)? - .strip_prefix('.')? - .strip_prefix(SUBAGENT_TOKEN)? - .strip_prefix('.')?; - (!name.is_empty() && !name.contains('.')).then_some(name) -} - -/// The subagents named by a stream's subjects, sorted and deduplicated. -/// Subjects outside `agent`'s family are skipped. -#[must_use] -pub fn subagent_names<'a>(agent: &str, subjects: impl IntoIterator) -> Vec { - let mut names: Vec = subjects - .into_iter() - .filter_map(|s| subagent_from_subject(agent, s)) - .map(str::to_owned) - .collect(); - names.sort_unstable(); - names.dedup(); - names -} - /// The config [`open_or_create`] creates `agent`'s stream with. /// /// `max_message_size` is taken as a parameter rather than read from a @@ -110,7 +19,6 @@ pub fn subagent_names<'a>(agent: &str, subjects: impl IntoIterator"); - assert_eq!( - consumer_create_grant(), - "$JS.API.CONSUMER.CREATE.*.*.$SWARM.term.*.sub.*" - ); - } - - #[test] - fn a_subject_reads_back_as_its_subagent() { - assert_eq!( - subagent_from_subject("iris", &subject("iris", "scout")), - Some("scout") - ); - } - - /// The agent's own terminal, another agent's subagent, and anything - /// deeper than one token are not subagents of `iris`. - #[test] - fn other_subjects_name_no_subagent() { - for s in [ - "$SWARM.term.iris", - "$SWARM.term.h1.iris", - "$SWARM.term.iris.sub", - "$SWARM.term.iris.sub.", - "$SWARM.term.iris.sub.a.b", - "$SWARM.term.iris-2.sub.scout", - "$SWARM.term.argus.sub.scout", - "$SWARM.agent-state.iris.sub.scout", - ] { - assert_eq!(subagent_from_subject("iris", s), None, "{s}"); - } - } - - #[test] - fn the_list_is_sorted_unique_and_scoped_to_the_agent() { - let subjects = [ - "$SWARM.term.iris.sub.zed", - "$SWARM.term.iris.sub.alpha", - "$SWARM.term.argus.sub.other", - "$SWARM.term.iris.sub.alpha", - ]; - assert_eq!(subagent_names("iris", subjects), ["alpha", "zed"]); - } - /// The config the controller creates caps disk per agent and never /// fails a publish because of that cap — `discard: Old` reclaims the /// oldest rows instead of rejecting the newest. - #[cfg(feature = "subagent-term")] #[test] fn stream_config_caps_bytes_and_discards_the_oldest() { let cfg = stream_config("iris", 8_388_608); diff --git a/swarm-wire/Cargo.toml b/swarm-wire/Cargo.toml index 18f6cd21..cfb3a5a4 100644 --- a/swarm-wire/Cargo.toml +++ b/swarm-wire/Cargo.toml @@ -9,6 +9,7 @@ edition.workspace = true schema = ["dep:utoipa"] [dependencies] +hive-types.workspace = true serde.workspace = true serde_json.workspace = true utoipa = { workspace = true, optional = true } diff --git a/swarm-wire/src/agent_permissions.rs b/swarm-wire/src/agent_permissions.rs new file mode 100644 index 00000000..acafb0d7 --- /dev/null +++ b/swarm-wire/src/agent_permissions.rs @@ -0,0 +1,143 @@ +//! The per-agent permissions KV bucket: one key per agent, holding the +//! [`hive_types::Permission`] set the swarm grants it. +//! +//! **swarm-controller is the only writer**, and creates the bucket when it +//! starts. A missing key or an unset `permissions` field leaves the agent on +//! the compiled-in default (`Permission::DEFAULT`); an empty list grants +//! nothing. +//! +//! **Two readers, each enforcing its own part.** The hive running the agent +//! applies `manage_root_agent` as bind mounts at container start; its grant +//! covers every key of this bucket, read-only, for the reason +//! [`crate::agent_limits`] gives. The agent itself (its harness and subagent +//! daemon) applies the tool permissions; its grant covers only its own key. +//! Both read with one `DIRECT.GET` on the agent's key +//! ([`crate::agent_permissions::key_subject`]). +//! +//! Keyed by the agent alone, like [`crate::agent_limits`]: an agent can move +//! between hives, and its permissions go with it. +//! +//! Also carries the one-shot import a hive sends of the grants its meta repo +//! held ([`crate::agent_permissions::import_subject`]), answered by the +//! controller. + +/// The KV bucket permissions are published into, keyed by agent name. +pub const BUCKET: &str = "agent-permissions"; + +/// The stream backing [`BUCKET`]. `create_key_value` prefixes `KV_`. +pub const STREAM: &str = "KV_agent-permissions"; + +/// The subject `agent`'s entry is stored under. +#[must_use] +pub fn key_subject(agent: &str) -> String { + format!("$KV.{BUCKET}.{agent}") +} + +const IMPORT_SUBJECT_PREFIX: &str = "$SWARM.permissions-import"; + +/// The subject `hive` sends its import on. One per hive, so the hive's grant +/// names only its own, and the controller takes the hive from the subject +/// rather than from the payload. +#[must_use] +pub fn import_subject(hive: &str) -> String { + format!("{IMPORT_SUBJECT_PREFIX}.{hive}") +} + +/// Every [`import_subject`] at once, for the controller's subscription. +#[must_use] +pub fn import_subject_wildcard() -> String { + import_subject("*") +} + +/// The hive an [`import_subject`] names, or `None` for any other subject. +#[must_use] +pub fn hive_from_import_subject(subject: &str) -> Option<&str> { + let hive = subject + .strip_prefix(IMPORT_SUBJECT_PREFIX)? + .strip_prefix('.')?; + (!hive.is_empty() && !hive.contains('.')).then_some(hive) +} + +/// How long a hive waits for the controller's answer to an import, and the +/// queue lets the controller answer. +pub const IMPORT_REPLY_DEADLINE: std::time::Duration = std::time::Duration::from_secs(30); + +/// One agent's permissions: the value under `BUCKET`/``. `None` is +/// "not set by the swarm", which a hive reads as its default. +/// +/// No `deny_unknown_fields`: a field a newer controller adds must not cost an +/// older hive the fields it knows. +#[derive(Debug, Clone, Default, PartialEq, Eq, serde::Serialize, serde::Deserialize)] +pub struct AgentPermissions { + #[serde(default, skip_serializing_if = "Option::is_none")] + pub permissions: Option>, +} + +/// What a hive sends once: the permissions its meta repo held, by agent. +#[derive(Debug, Clone, Default, PartialEq, Eq, serde::Serialize, serde::Deserialize)] +pub struct Import { + pub permissions: std::collections::BTreeMap>, +} + +/// The controller's answer to an [`Import`]. +#[derive(Debug, Clone, PartialEq, Eq, serde::Serialize, serde::Deserialize)] +#[serde(tag = "state", rename_all = "kebab-case")] +pub enum ImportReply { + /// Every agent was handled. `written` had no key and got the hive's value; + /// `kept` already had a key, which was left alone; `skipped` is not an + /// agent the sending hive's wanted state declares. + Imported { + written: Vec, + kept: Vec, + skipped: Vec, + }, + /// Nothing is known to be written; the hive sends again later. + Failed { error: String }, +} + +#[cfg(test)] +mod tests { + use super::*; + + /// The names the grants in `swarm-nats-auth` are built from. + #[test] + fn names_match_the_queue_grant() { + assert_eq!(STREAM, format!("KV_{BUCKET}")); + assert_eq!(key_subject("iris"), "$KV.agent-permissions.iris"); + assert_eq!(import_subject("pr1ma"), "$SWARM.permissions-import.pr1ma"); + assert_eq!(import_subject_wildcard(), "$SWARM.permissions-import.*"); + } + + #[test] + fn the_hive_is_read_back_from_its_import_subject_only() { + assert_eq!( + hive_from_import_subject(&import_subject("pr1ma")), + Some("pr1ma") + ); + for other in [ + "$SWARM.permissions-import", + "$SWARM.permissions-import.", + "$SWARM.permissions-import.a.b", + "$SWARM.permissions-importx.pr1ma", + "$SWARM.deploy.pr1ma", + ] { + assert_eq!(hive_from_import_subject(other), None, "{other}"); + } + } + + /// Unset is left out on the wire, `{}` is unset, and an empty list stays + /// an empty list: "grants nothing" is not "not set". + #[test] + fn unset_and_empty_are_different_values() { + use hive_types::Permission; + assert_eq!( + serde_json::to_string(&AgentPermissions::default()).unwrap(), + "{}" + ); + let none: AgentPermissions = serde_json::from_str(r#"{"permissions":[]}"#).unwrap(); + assert_eq!(none.permissions, Some(Vec::new())); + let ruth: AgentPermissions = + serde_json::from_str(r#"{"permissions":["manage_root_agent"]}"#).unwrap(); + assert_eq!(ruth.permissions, Some(vec![Permission::ManageRootAgent])); + } +} diff --git a/swarm-wire/src/agent_term.rs b/swarm-wire/src/agent_term.rs new file mode 100644 index 00000000..53380676 --- /dev/null +++ b/swarm-wire/src/agent_term.rs @@ -0,0 +1,191 @@ +//! An agent's own terminal: the stream that keeps its rows, and the consumer +//! a reader replays them through. +//! +//! The agent publishes its classified terminal rows on +//! [`crate::term_subject`] with a plain core publish, under its own queue +//! credential, which may publish there and nothing else of these streams. +//! `swarm-controller` creates one stream per agent with the fixed config +//! `swarm_queue_client::agent_term::open_or_create` spells, so the queue keeps +//! the rows a reader who attaches later replays. +//! +//! The stream captures exactly one literal subject, `$SWARM.term.`. +//! [`crate::subagent_term`]'s stream captures `$SWARM.term..sub.>`, +//! which only matches subjects with at least two more tokens, so no subject +//! is in both and the two streams never overlap. The `JetStream` server refuses +//! to create a stream whose subjects overlap another's. +//! +//! The names live here because three crates must agree on them: the +//! publisher in `hive-agent`, `swarm-controller`, and the reader's consumer +//! grant in `swarm-nats-auth`. + +/// The token between the prefix and the agent in [`stream_name`]. It differs +/// from `term-sub-` at the same position, so no agent name makes this +/// stream's name equal another agent's subagent stream. +const STREAM_KIND: &str = "main"; + +/// How long the stream keeps a row. The same window as the subagent stream. +pub const MAX_AGE: std::time::Duration = std::time::Duration::from_hours(24); + +/// How many of the newest rows a reader that attaches fresh is replayed +/// before the live tail. The agent's own web UI loads the same number of +/// events as its first history page. +pub const REPLAY_ROWS: u64 = 100; + +/// How often the server sends an idle replay consumer a heartbeat. Two missed +/// in a row end the reader's row stream, so a consumer the server dropped, +/// as on a restart, ends the SSE response and the browser reconnects. +pub const IDLE_HEARTBEAT: std::time::Duration = std::time::Duration::from_secs(5); + +/// The stream holding `agent`'s terminal rows: `term-main-`. +#[must_use] +pub fn stream_name(agent: &str) -> String { + format!("term-{STREAM_KIND}-{agent}") +} + +/// The one subject [`stream_name`]'s stream captures. +#[must_use] +pub fn stream_subject(agent: &str) -> String { + crate::term_subject(agent) +} + +/// The reader's grant to create a replay consumer: on any stream, under any +/// consumer name, in the extended create form whose last tokens are the +/// consumer's filter. +/// +/// The server refuses a create whose payload filter differs from those +/// tokens, so a consumer made under this grant delivers only rows on +/// subjects below `$SWARM.term`. The grant's last `*` also matches a literal +/// `*` or `>` token, so besides one agent's [`crate::term_subject`] it admits +/// `$SWARM.term.*` and `$SWARM.term.>`, which take in every agent's and every +/// subagent's terminal. The bare `CONSUMER.CREATE.` form, whose +/// filter rides only in the payload, is not covered. +#[must_use] +pub fn consumer_create_grant() -> String { + format!( + "$JS.API.CONSUMER.CREATE.*.*.{}", + crate::term_subject_wildcard() + ) +} + +/// The first stream sequence to replay from. +/// +/// `resume_after` is the last sequence the reader already holds, from a +/// reconnecting browser's `Last-Event-ID`; the replay continues right after +/// it. A fresh reader gets the newest [`REPLAY_ROWS`] below `last_sequence`, +/// the stream's newest row when it was read. A sequence the stream has since +/// discarded starts the consumer at the oldest row it still has. +/// +/// A `resume_after` above `last_sequence` names no row this stream ever held, +/// as after the stream was created again, and is read as a fresh reader: the +/// server accepts a start past the end and delivers nothing until the stream +/// reaches it. +#[must_use] +pub fn replay_start(last_sequence: u64, resume_after: Option) -> u64 { + match resume_after { + Some(seen) if seen <= last_sequence => seen + 1, + _ => last_sequence.saturating_sub(REPLAY_ROWS - 1).max(1), + } +} + +#[cfg(test)] +mod tests { + use super::*; + + /// Token-wise NATS subject overlap: whether some literal subject matches + /// both patterns. + fn overlaps(a: &str, b: &str) -> bool { + let (mut a, mut b) = (a.split('.'), b.split('.')); + loop { + match (a.next(), b.next()) { + (Some(">"), Some(_)) | (Some(_), Some(">")) | (None, None) => return true, + (None, _) | (_, None) => return false, + (Some(x), Some(y)) if x != "*" && y != "*" && x != y => return false, + _ => {} + } + } + } + + #[test] + fn names_match_the_publisher_and_the_grant() { + assert_eq!(stream_name("iris"), "term-main-iris"); + assert_eq!(stream_subject("iris"), "$SWARM.term.iris"); + assert_eq!( + consumer_create_grant(), + "$JS.API.CONSUMER.CREATE.*.*.$SWARM.term.*" + ); + } + + /// The server refuses to create a stream whose subjects overlap + /// another's, so this stream must share no subject with any agent's + /// subagent stream, its own agent's included. + #[test] + fn the_subject_is_disjoint_from_every_subagent_stream() { + for other in ["iris", "argus", "sub", "iris-2"] { + assert!( + !overlaps( + &stream_subject("iris"), + &crate::subagent_term::stream_subjects(other) + ), + "term-main-iris overlaps term-sub-{other}" + ); + } + // Control: the checker does report a real overlap. + assert!(overlaps( + "$SWARM.term.iris.>", + &crate::subagent_term::stream_subjects("iris") + )); + assert!(overlaps(&stream_subject("iris"), "$SWARM.term.*")); + } + + /// No agent name makes this stream's name another agent's subagent + /// stream: an agent may be called `sub-x`. + #[test] + fn the_name_never_collides_with_a_subagent_stream() { + for (agent, other) in [("sub-x", "x"), ("x", "sub-x"), ("main-x", "x")] { + assert_ne!(stream_name(agent), crate::subagent_term::stream_name(other)); + } + } + + /// The create the controller sends is one the grant covers, and every + /// filter the grant admits stays below `$SWARM.term`. + #[test] + fn the_grant_admits_only_filters_below_the_terminal_prefix() { + let grant = consumer_create_grant(); + let create = |filter: &str| format!("$JS.API.CONSUMER.CREATE.term-main-iris.r1.{filter}"); + assert!(overlaps(&grant, &create(&stream_subject("iris")))); + // The width the doc states: a literal wildcard as the last token. + assert!(overlaps(&grant, &create("$SWARM.term.>"))); + for other in [ + crate::subagent_term::subject("iris", "scout"), + crate::subagent_term::stream_subjects("iris"), + // `swarm_queue_client::agent_state_subject("iris")`, whose value + // that crate's tests pin. + "$SWARM.agent-state.iris".to_owned(), + ] { + assert!(!overlaps(&grant, &create(&other)), "{other}"); + } + assert!(!overlaps(&grant, "$JS.API.CONSUMER.CREATE.term-main-iris")); + } + + #[test] + fn a_fresh_reader_replays_the_newest_rows() { + assert_eq!(replay_start(1000, None), 1000 - REPLAY_ROWS + 1); + assert_eq!(replay_start(5, None), 1); + assert_eq!(replay_start(0, None), 1); + } + + #[test] + fn a_reconnecting_reader_resumes_after_its_last_row() { + assert_eq!(replay_start(1000, Some(990)), 991); + assert_eq!(replay_start(1000, Some(1000)), 1001); + } + + /// An id past the stream's newest row, as after the stream was created + /// again, replays like a fresh reader instead of waiting for that number. + #[test] + fn an_id_past_the_newest_row_replays_as_a_fresh_reader() { + assert_eq!(replay_start(5, Some(99)), replay_start(5, None)); + assert_eq!(replay_start(1000, Some(1001)), replay_start(1000, None)); + assert_eq!(replay_start(0, Some(7)), 1); + } +} diff --git a/swarm-wire/src/lib.rs b/swarm-wire/src/lib.rs index e9b1bdf6..fbdf8fd3 100644 --- a/swarm-wire/src/lib.rs +++ b/swarm-wire/src/lib.rs @@ -9,10 +9,50 @@ pub mod agent_icon; pub mod agent_inference; pub mod agent_limits; pub mod agent_login; +pub mod agent_permissions; pub mod agent_schedules; pub mod agent_status; +pub mod agent_term; pub mod builder; pub mod claude_login; pub mod notices; pub mod review_classifier; +pub mod subagent_term; pub mod wanted; + +/// The subject `agent` publishes its terminal rows on. One leaf per agent, so +/// a subscriber can follow one agent without filtering the swarm's whole +/// terminal traffic. +/// +/// Shared by the publisher (`hive-agent`), the relay that streams it to +/// swarm-ui (`swarm-controller`) and the queue's subscribe grant for readers +/// (`swarm-nats-auth`), which spells [`term_subject_wildcard`] — so a rename +/// moves all three together. +#[must_use] +pub fn term_subject(agent: &str) -> String { + format!("{TERM_SUBJECT_PREFIX}.{agent}") +} + +/// Every [`term_subject`] at once: the reader's subscribe grant, which has no +/// roster to enumerate agents from. +#[must_use] +pub fn term_subject_wildcard() -> String { + term_subject("*") +} + +/// Shared by [`term_subject`] and [`subagent_term`], whose subjects sit under +/// the same prefix. +pub const TERM_SUBJECT_PREFIX: &str = "$SWARM.term"; + +#[cfg(test)] +mod tests { + use super::*; + + /// The names the publishers, the relays and the queue's grant all spell. + #[test] + fn term_subjects_are_the_agreed_names() { + assert_eq!(term_subject("mara"), "$SWARM.term.mara"); + assert_eq!(term_subject_wildcard(), "$SWARM.term.*"); + assert_eq!(subagent_term::subject_wildcard(), "$SWARM.term.*.sub.*"); + } +} diff --git a/swarm-wire/src/subagent_term.rs b/swarm-wire/src/subagent_term.rs new file mode 100644 index 00000000..8204d90e --- /dev/null +++ b/swarm-wire/src/subagent_term.rs @@ -0,0 +1,152 @@ +//! Subagent terminals: the subject each subagent's rows go to, and the +//! stream that keeps them. +//! +//! An agent's subagent daemon publishes each subagent's classified terminal +//! rows on `$SWARM.term..sub.`, under the agent's own queue +//! credential. The queue grants that credential publish on this subject +//! family and no `JetStream` subject. +//! +//! The stream lets the swarm list an agent's subagents: subagents +//! are spawned on demand under caller-chosen names, and the set of subjects +//! the stream holds is the list. `swarm-controller` creates one stream per +//! agent with the fixed config +//! `swarm_queue_client::subagent_term::open_or_create` spells, and takes the +//! list from its subject counts ([`crate::subagent_term::subagent_names`]). A +//! reader attaching to one subagent's terminal replays that subagent's newest +//! rows from it. +//! +//! The name and the subject live here because they must agree across the +//! publisher in `hive-subagent-mcp`, `swarm-controller`, the queue grant in +//! `swarm-nats.nix`, whose `{agent}` template spells the same subjects, and +//! the reader's consumer grant in `swarm-nats-auth`. + +use crate::TERM_SUBJECT_PREFIX; + +/// The token between the agent and the subagent name. A subagent subject has +/// two more tokens than the agent's own `$SWARM.term.`, so neither +/// family's subscriber receives the other's rows. +const SUBAGENT_TOKEN: &str = "sub"; + +/// How long the stream keeps a row. The subagent list is the subjects with a +/// row inside this window, so a subagent drops off the list this long after +/// its last output. +pub const MAX_AGE: std::time::Duration = std::time::Duration::from_hours(24); + +/// The stream holding `agent`'s subagent rows: `term-sub-`. +#[must_use] +pub fn stream_name(agent: &str) -> String { + format!("term-sub-{agent}") +} + +/// The subject `subagent`'s rows are published on. +#[must_use] +pub fn subject(agent: &str, subagent: &str) -> String { + format!("{TERM_SUBJECT_PREFIX}.{agent}.{SUBAGENT_TOKEN}.{subagent}") +} + +/// Every [`subject`] at once: the reader's subscribe grant, which has no +/// roster to enumerate agents or subagents from. +#[must_use] +pub fn subject_wildcard() -> String { + subject("*", "*") +} + +/// The reader's grant to create a replay consumer on one subagent's rows: +/// the extended create form of [`crate::agent_term::consumer_create_grant`], +/// on any stream and under any consumer name, with a [`subject`] as the +/// filter tokens the server checks the payload against. The last `*` also +/// matches a literal `*` or `>`, so it admits `$SWARM.term..sub.>` +/// too, and nothing outside [`subject_wildcard`]'s family. +#[must_use] +pub fn consumer_create_grant() -> String { + format!("$JS.API.CONSUMER.CREATE.*.*.{}", subject_wildcard()) +} + +/// Every subject [`stream_name`]'s stream captures, which is also the publish +/// grant's shape. +#[must_use] +pub fn stream_subjects(agent: &str) -> String { + format!("{TERM_SUBJECT_PREFIX}.{agent}.{SUBAGENT_TOKEN}.>") +} + +/// The subagent a stream subject names, or `None` for a subject that is not +/// exactly one subagent token under `agent`'s family. +#[must_use] +pub fn subagent_from_subject<'a>(agent: &str, subject: &'a str) -> Option<&'a str> { + let name = subject + .strip_prefix(TERM_SUBJECT_PREFIX)? + .strip_prefix('.')? + .strip_prefix(agent)? + .strip_prefix('.')? + .strip_prefix(SUBAGENT_TOKEN)? + .strip_prefix('.')?; + (!name.is_empty() && !name.contains('.')).then_some(name) +} + +/// The subagents named by a stream's subjects, sorted and deduplicated. +/// Subjects outside `agent`'s family are skipped. +#[must_use] +pub fn subagent_names<'a>(agent: &str, subjects: impl IntoIterator) -> Vec { + let mut names: Vec = subjects + .into_iter() + .filter_map(|s| subagent_from_subject(agent, s)) + .map(str::to_owned) + .collect(); + names.sort_unstable(); + names.dedup(); + names +} + +#[cfg(test)] +mod tests { + use super::*; + + /// The names the grant in `swarm-nats.nix` spells with `{agent}`. + #[test] + fn names_match_the_queue_grant() { + assert_eq!(stream_name("iris"), "term-sub-iris"); + assert_eq!(subject("iris", "scout"), "$SWARM.term.iris.sub.scout"); + assert_eq!(stream_subjects("iris"), "$SWARM.term.iris.sub.>"); + assert_eq!( + consumer_create_grant(), + "$JS.API.CONSUMER.CREATE.*.*.$SWARM.term.*.sub.*" + ); + } + + #[test] + fn a_subject_reads_back_as_its_subagent() { + assert_eq!( + subagent_from_subject("iris", &subject("iris", "scout")), + Some("scout") + ); + } + + /// The agent's own terminal, another agent's subagent, and anything + /// deeper than one token are not subagents of `iris`. + #[test] + fn other_subjects_name_no_subagent() { + for s in [ + "$SWARM.term.iris", + "$SWARM.term.h1.iris", + "$SWARM.term.iris.sub", + "$SWARM.term.iris.sub.", + "$SWARM.term.iris.sub.a.b", + "$SWARM.term.iris-2.sub.scout", + "$SWARM.term.argus.sub.scout", + "$SWARM.agent-state.iris.sub.scout", + ] { + assert_eq!(subagent_from_subject("iris", s), None, "{s}"); + } + } + + #[test] + fn the_list_is_sorted_unique_and_scoped_to_the_agent() { + let subjects = [ + "$SWARM.term.iris.sub.zed", + "$SWARM.term.iris.sub.alpha", + "$SWARM.term.argus.sub.other", + "$SWARM.term.iris.sub.alpha", + ]; + assert_eq!(subagent_names("iris", subjects), ["alpha", "zed"]); + } +}