vinzenz/hyperhive
vinzenz/hyperhive
Watch
0
0
Fork
You've already forked hyperhive
 0
Code Activity

agent_permissions, agent_term and subagent_term take their wire contracts from swarm-wire

Browse source
This commit is contained in:
atlas 2026-10-18 23:37:05 +02:00 • committed by mara
commit 839efd1642
27 changed files with 661 additions and 607 deletions
Download patch file Download diff file Expand all files Collapse all files

2
AGENTS.md
View file

@ -139,7 +139,7 @@ Swarm-level:
- **`swarm-queue-client/`** — the one authenticated connect to the swarm - **`swarm-queue-client/`** — the one authenticated connect to the swarm
NATS queue. NATS queue.
- **`swarm-wire/`** — serde-only request and reply contracts services - **`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. neither the queue client nor the store client.
- **`swarm-nats-auth/`** — NATS auth-callout responder; without it the - **`swarm-nats-auth/`** — NATS auth-callout responder; without it the
queue is fail-closed and admits nobody. queue is fail-closed and admits nobody.

1
Cargo.lock generated
View file

@ -5090,6 +5090,7 @@ dependencies = [
name = "swarm-wire" name = "swarm-wire"
version = "0.1.0" version = "0.1.0"
dependencies = [ dependencies = [
"hive-types",
"serde", "serde",
"serde_json", "serde_json",
"utoipa", "utoipa",

2
docs/process/conventions.md
View file

@ -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 **Where they live** — in the swarm, not on the hive: the `permissions` field of
the agent's key in the `agent-permissions` KV bucket 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 <agent>` writes. Set them with `swarmctl agent permissions get|set|clear <agent>`
(`/api/agents/{name}/permissions`), or on the swarm UI's create-agent screen (`/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 (`permissions` in `POST /api/agents`). A missing key or an unset field means

4
hive-agent/src/swarm_term.rs
View file

@ -15,7 +15,7 @@
//! //!
//! **A plain core publish.** The queue keeps the rows in the stream //! **A plain core publish.** The queue keeps the rows in the stream
//! `term-main-<agent>`, which swarm-controller creates //! `term-main-<agent>`, 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 //! the agent's grant stays a plain publish. A row published before that
//! stream exists reaches live subscribers only. //! stream exists reaches live subscribers only.
//! //!
@ -50,7 +50,7 @@ pub fn spawn(rx: broadcast::Receiver<BusEvent>) {
async fn run(mut rx: broadcast::Receiver<BusEvent>, agent: String) { async fn run(mut rx: broadcast::Receiver<BusEvent>, agent: String) {
let client = crate::swarm_queue::client().await; 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"); tracing::info!(subject, "publishing the agent terminal to the swarm queue");
// One context for the task's whole life — see the module doc. // One context for the task's whole life — see the module doc.

5
hive-c0re/src/agent_config/permissions.rs
View file

@ -4,7 +4,7 @@
//! the swarm is the record, this file is the hive's last read of it. //! the swarm is the record, this file is the hive's last read of it.
//! //!
//! Format: agent name → the stored //! 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 //! no entry, or an entry with `permissions` unset, holds
//! [`Permission::DEFAULT`]; `"permissions": []` holds none. //! [`Permission::DEFAULT`]; `"permissions": []` holds none.
//! //!
@ -18,7 +18,8 @@ use std::collections::BTreeMap;
use std::path::{Path, PathBuf}; use std::path::{Path, PathBuf};
use hive_types::Permission; 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"; const FILE: &str = "swarm-agent-permissions.json";

2
hive-c0re/src/workers/permissions_import.rs
View file

@ -14,7 +14,7 @@ use std::path::{Path, PathBuf};
use anyhow::{Context, Result}; use anyhow::{Context, Result};
use hive_types::Permission; use hive_types::Permission;
use swarm_queue_client::agent_permissions::{ use swarm_wire::agent_permissions::{
self, AgentPermissions, IMPORT_REPLY_DEADLINE, Import, ImportReply, self, AgentPermissions, IMPORT_REPLY_DEADLINE, Import, ImportReply,
}; };

5
hive-c0re/src/workers/swarm_permissions.rs
View file

@ -19,7 +19,8 @@ use std::time::Duration;
use anyhow::{Context, Result}; use anyhow::{Context, Result};
use hive_types::Permission; 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::coordinator::Coordinator;
use crate::permissions::{self, Stored}; use crate::permissions::{self, Stored};
@ -182,7 +183,7 @@ mod tests {
use std::path::Path; use std::path::Path;
use super::{Read, sync_with}; use super::{Read, sync_with};
use swarm_queue_client::agent_permissions::AgentPermissions; use swarm_wire::agent_permissions::AgentPermissions;
fn record(json: &str) -> AgentPermissions { fn record(json: &str) -> AgentPermissions {
serde_json::from_str(json).expect("decodes") serde_json::from_str(json).expect("decodes")

1
hive-subagent-mcp/Cargo.toml
View file

@ -28,7 +28,6 @@ rmcp.workspace = true
schemars.workspace = true schemars.workspace = true
serde.workspace = true serde.workspace = true
serde_json.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. # `agent-inference`: this agent's `subagent_models`, read with one direct get.
swarm-queue-client = { workspace = true, features = [ swarm-queue-client = { workspace = true, features = [
"agent-inference", "agent-inference",

4
hive-subagent-mcp/src/swarm_term.rs
View file

@ -6,7 +6,7 @@
//! output reader, so `offer` only queues. One task drains the queue: it //! 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 //! classifies each `stream-json` event into the rows the agent's own terminal
//! carries (`hive_sh4re::stream_enrich`) and publishes them on //! carries (`hive_sh4re::stream_enrich`) and publishes them on
//! `$SWARM.term.<agent>.sub.<subagent>` (`swarm_queue_client::subagent_term`). //! `$SWARM.term.<agent>.sub.<subagent>` (`swarm_wire::subagent_term`).
//! //!
//! **Output only.** Nothing here subscribes: a subagent takes no input from //! **Output only.** Nothing here subscribes: a subagent takes no input from
//! the swarm. //! the swarm.
@ -33,12 +33,12 @@ use std::time::{Duration, Instant, SystemTime, UNIX_EPOCH};
use anyhow::{Context as _, anyhow}; use anyhow::{Context as _, anyhow};
use hive_sh4re::term_msg::{ClassifyCtx, Level, TermMsg, fit, iso8601_utc}; use hive_sh4re::term_msg::{ClassifyCtx, Level, TermMsg, fit, iso8601_utc};
use swarm_queue_client::subagent_term;
use swarm_secret_client::{ use swarm_secret_client::{
SecretStore, SecretStore,
client::{DEFAULT_CERT_MOUNT, ENV_ADDR, ENV_CACERT, Settings}, client::{DEFAULT_CERT_MOUNT, ENV_ADDR, ENV_CACERT, Settings},
policy, queue, policy, queue,
}; };
use swarm_wire::subagent_term;
use tokio::sync::mpsc; use tokio::sync::mpsc;
/// Names the agent. The store role, the store path, the queue token and the /// Names the agent. The store role, the store path, the queue token and the

4
nix/host-modules/swarm-nats.nix
View file

@ -889,7 +889,7 @@ in
# harness mirrors its `AGENTS.md` there for the swarm to read. # harness mirrors its `AGENTS.md` there for the swarm to read.
"--agent-token-publish-subject ${lib.escapeShellArg "\$\$KV.agents-md.{agent}"}" "--agent-token-publish-subject ${lib.escapeShellArg "\$\$KV.agents-md.{agent}"}"
# Its subagents' terminal rows # 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 # `$JS.API.*` subject: swarm-controller creates the
# `term-sub-{agent}` stream that keeps them, so its subjects # `term-sub-{agent}` stream that keeps them, so its subjects
# and limits are never the agent's to choose. # and limits are never the agent's to choose.
@ -905,7 +905,7 @@ in
# (`swarm_wire::agent_inference`), on the same terms. # (`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}"}" "--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 # 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 # the harness and the subagent daemon read the tool permissions
# it holds. # it holds.
"--agent-token-publish-subject ${lib.escapeShellArg "\$\$JS.API.DIRECT.GET.KV_agent-permissions.\$\$KV.agent-permissions.{agent}"}" "--agent-token-publish-subject ${lib.escapeShellArg "\$\$JS.API.DIRECT.GET.KV_agent-permissions.\$\$KV.agent-permissions.{agent}"}"

5
swarm-controller/src/agent_login.rs
View file

@ -138,10 +138,7 @@ mod tests {
&swarm_queue_client::agent_state_subject("iris"), &swarm_queue_client::agent_state_subject("iris"),
&header(Some(true)), &header(Some(true)),
); );
reported.record( reported.record(&swarm_wire::term_subject("atlas"), &header(Some(true)));
&swarm_queue_client::term_subject("atlas"),
&header(Some(true)),
);
assert_eq!(reported.get("iris"), Some(true)); assert_eq!(reported.get("iris"), Some(true));
assert_eq!(reported.get("atlas"), None); assert_eq!(reported.get("atlas"), None);
} }

11
swarm-controller/src/agent_permissions.rs
View file

@ -20,7 +20,10 @@ use futures_util::StreamExt as _;
use hive_types::Permission; use hive_types::Permission;
use problem_details::ProblemDetails; use problem_details::ProblemDetails;
use serde::{Deserialize, Serialize}; 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 utoipa::ToSchema;
use crate::routes::{AppState, error_problem, wanted_error_status}; use crate::routes::{AppState, error_problem, wanted_error_status};
@ -377,7 +380,7 @@ pub(crate) fn spawn(client: async_nats::Client, wanted: Arc<WantedWriter>, hives
.await; .await;
} }
}); });
let subject = agent_permissions::import_subject_wildcard(); let subject = import_subject_wildcard();
loop { loop {
match client.subscribe(subject.clone()).await { match client.subscribe(subject.clone()).await {
Ok(mut requests) => { Ok(mut requests) => {
@ -413,7 +416,7 @@ async fn answer_import(
tracing::warn!(subject = %msg.subject, %error, "agent permissions: import refused"); tracing::warn!(subject = %msg.subject, %error, "agent permissions: import refused");
ImportReply::Failed { error } 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)); return failed(format!("{} is not an import subject", msg.subject));
}; };
if !hives.iter().any(|h| h == hive) { if !hives.iter().any(|h| h == hive) {
@ -529,7 +532,7 @@ mod tests {
use axum::extract::{Path, State}; use axum::extract::{Path, State};
use axum::http::StatusCode; use axum::http::StatusCode;
use hive_types::Permission; 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<String> { async fn published(client: &async_nats::Client, agent: &str) -> Option<String> {
let store = swarm_queue_client::agent_permissions::open_or_create(client) let store = swarm_queue_client::agent_permissions::open_or_create(client)

29
swarm-controller/src/message_stream.rs
View file

@ -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 hive_types::term_msg::{SEND_ICON, TURN_ICON, send_recipient, turn_sender};
use serde::Deserialize; use serde::Deserialize;
use swarm_controller_api::messages::SwarmMessage; use swarm_controller_api::messages::SwarmMessage;
use swarm_queue_client::agent_term; use swarm_wire::agent_term;
use utoipa::IntoParams; use utoipa::IntoParams;
use crate::routes::{AppState, error_problem}; use crate::routes::{AppState, error_problem};
@ -104,12 +104,9 @@ pub(crate) async fn stream_messages(
) )
})?; })?;
let subscriber = crate::term_stream::subscribe( let subscriber =
&client, crate::term_stream::subscribe(&client, swarm_wire::term_subject_wildcard(), "messages")
swarm_queue_client::term_subject_wildcard(), .await?;
"messages",
)
.await?;
let roster: Arc<HashSet<String>> = Arc::new(roster.into_iter().collect()); let roster: Arc<HashSet<String>> = Arc::new(roster.into_iter().collect());
let history = history(&client, &roster, minutes).await; let history = history(&client, &roster, minutes).await;
let seen: HashSet<SwarmMessage> = history.iter().cloned().collect(); let seen: HashSet<SwarmMessage> = history.iter().cloned().collect();
@ -170,7 +167,7 @@ async fn agent_history(
let replayed = crate::term_stream::replay( let replayed = crate::term_stream::replay(
&client, &client,
&agent_term::stream_name(&agent), &agent_term::stream_name(&agent),
swarm_queue_client::term_subject(&agent), swarm_wire::term_subject(&agent),
None, None,
) )
.await; .await;
@ -203,7 +200,7 @@ fn live(
/// The agent a main terminal subject belongs to. A subagent's subject has /// The agent a main terminal subject belongs to. A subagent's subject has
/// more tokens and names none. /// more tokens and names none.
fn agent_of(subject: &str) -> Option<&str> { 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) (!agent.is_empty() && !agent.contains('.')).then_some(agent)
} }
@ -307,12 +304,12 @@ mod tests {
let nats = TestNats::start(); let nats = TestNats::start();
let client = nats.client().await; let client = nats.client().await;
let subscriber = client let subscriber = client
.subscribe(swarm_queue_client::term_subject_wildcard()) .subscribe(swarm_wire::term_subject_wildcard())
.await .await
.expect("subscribe"); .expect("subscribe");
let mut messages = Box::pin(live(subscriber, roster())); let mut messages = Box::pin(live(subscriber, roster()));
let atlas = swarm_queue_client::term_subject("atlas"); let atlas = swarm_wire::term_subject("atlas");
let iris = swarm_queue_client::term_subject("iris"); let iris = swarm_wire::term_subject("iris");
publish( publish(
&client, &client,
@ -350,7 +347,7 @@ mod tests {
.await; .await;
publish( publish(
&client, &client,
swarm_queue_client::subagent_term::subject("atlas", "scout"), swarm_wire::subagent_term::subject("atlas", "scout"),
row(SEND_ICON, "send* → iris", "from a subagent"), row(SEND_ICON, "send* → iris", "from a subagent"),
) )
.await; .await;
@ -366,7 +363,7 @@ mod tests {
let nats = TestNats::start(); let nats = TestNats::start();
let client = nats.client().await; let client = nats.client().await;
for agent in ["atlas", "iris"] { for agent in ["atlas", "iris"] {
agent_term::open_or_create(&client, agent) swarm_queue_client::agent_term::open_or_create(&client, agent)
.await .await
.expect("the stream"); .expect("the stream");
} }
@ -381,8 +378,8 @@ mod tests {
.as_secs(); .as_secs();
let now = iso8601_utc(i64::try_from(secs).expect("fits")); let now = iso8601_utc(i64::try_from(secs).expect("fits"));
let earlier = iso8601_utc(i64::try_from(secs - 600).expect("fits")); let earlier = iso8601_utc(i64::try_from(secs - 600).expect("fits"));
let atlas = swarm_queue_client::term_subject("atlas"); let atlas = swarm_wire::term_subject("atlas");
let iris = swarm_queue_client::term_subject("iris"); let iris = swarm_wire::term_subject("iris");
publish(&client, atlas, stamped(&now, "send* → iris", "newer")).await; publish(&client, atlas, stamped(&now, "send* → iris", "newer")).await;
publish( publish(
&client, &client,

4
swarm-controller/src/routes/agents.rs
View file

@ -711,7 +711,7 @@ pub(crate) async fn create_agent(
struct CreateSettings { struct CreateSettings {
limits: Option<swarm_wire::agent_limits::AgentCaps>, limits: Option<swarm_wire::agent_limits::AgentCaps>,
inference: Option<swarm_wire::agent_inference::AgentInference>, 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>, archetype: Option<swarm_queue_client::agent_archetype::ArchetypeRef>,
agents_md: Option<String>, agents_md: Option<String>,
} }
@ -794,7 +794,7 @@ impl StoredSettings {
agent: &str, agent: &str,
limits: Option<&swarm_wire::agent_limits::AgentCaps>, limits: Option<&swarm_wire::agent_limits::AgentCaps>,
inference: Option<&swarm_wire::agent_inference::AgentInference>, 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>, archetype: Option<&swarm_queue_client::agent_archetype::ArchetypeRef>,
agents_md: Option<&str>, agents_md: Option<&str>,
) -> Result<Self, problem_details::ProblemDetails> { ) -> Result<Self, problem_details::ProblemDetails> {

12
swarm-controller/src/subagent_term.rs
View file

@ -34,7 +34,7 @@ use axum::http::{HeaderMap, StatusCode};
use axum::response::sse::{Event, KeepAlive, Sse}; use axum::response::sse::{Event, KeepAlive, Sse};
use futures_util::{Stream, StreamExt as _, TryStreamExt as _}; use futures_util::{Stream, StreamExt as _, TryStreamExt as _};
use serde::Serialize; use serde::Serialize;
use swarm_queue_client::{agent_term, subagent_term}; use swarm_wire::subagent_term;
use utoipa::ToSchema; use utoipa::ToSchema;
use crate::routes::AppState; use crate::routes::AppState;
@ -72,14 +72,18 @@ pub(crate) fn spawn(client: async_nats::Client, wanted: Arc<WantedWriter>, hives
continue; continue;
} }
for agent in live_agents(&wanted, &hives).await { 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!( tracing::warn!(
%agent, %agent,
error = %swarm_queue_client::chain(&e), error = %swarm_queue_client::chain(&e),
"subagent streams: creating the agent's stream failed; next pass" "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!( tracing::warn!(
%agent, %agent,
error = %swarm_queue_client::chain(&e), 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)] #[cfg(test)]
mod tests { 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 /// 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 /// name: a subagent that published once and one that published many rows

43
swarm-controller/src/term_stream.rs
View file

@ -9,7 +9,7 @@
//! **Replay, then the live tail, through one consumer.** `hive-agent` //! **Replay, then the live tail, through one consumer.** `hive-agent`
//! publishes each already-classified `TermMsg` row to the core subject //! publishes each already-classified `TermMsg` row to the core subject
//! `$SWARM.term.{agent}`, and the stream `term-main-{agent}` //! `$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 //! ephemeral consumer on that stream starting at the newest
//! `agent_term::REPLAY_ROWS` rows, so the reader gets those and then every //! `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. //! 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 axum::response::sse::{Event, KeepAlive, Sse};
use futures_util::stream::BoxStream; use futures_util::stream::BoxStream;
use futures_util::{Stream, StreamExt as _}; use futures_util::{Stream, StreamExt as _};
use swarm_queue_client::agent_term; use swarm_wire::agent_term;
use crate::routes::AppState; use crate::routes::AppState;
@ -121,7 +121,7 @@ pub(crate) async fn stream_agent_term(
let stream = replay_or_live( let stream = replay_or_live(
&client, &client,
&agent_term::stream_name(&agent), &agent_term::stream_name(&agent),
swarm_queue_client::term_subject(&agent), swarm_wire::term_subject(&agent),
last_event_id(&headers), last_event_id(&headers),
"term", "term",
) )
@ -198,8 +198,12 @@ pub(crate) async fn replay(
"term-replay-{}", "term-replay-{}",
deliver_subject.rsplit('.').next().unwrap_or_default() deliver_subject.rsplit('.').next().unwrap_or_default()
); );
let config = let config = swarm_queue_client::agent_term::replay_consumer_config(
agent_term::replay_consumer_config(subject.clone(), consumer_name, deliver_subject, start); subject.clone(),
consumer_name,
deliver_subject,
start,
);
let consumer = stream.create_consumer(config).await.map_err(|e| { let consumer = stream.create_consumer(config).await.map_err(|e| {
format!( format!(
"creating a consumer on {stream_name}: {}", "creating a consumer on {stream_name}: {}",
@ -301,10 +305,7 @@ mod tests {
async fn publish_rows(client: &async_nats::Client, rows: std::ops::RangeInclusive<u64>) { async fn publish_rows(client: &async_nats::Client, rows: std::ops::RangeInclusive<u64>) {
for i in rows { for i in rows {
client client
.publish( .publish(swarm_wire::term_subject(AGENT), format!("row{i}").into())
swarm_queue_client::term_subject(AGENT),
format!("row{i}").into(),
)
.await .await
.expect("publish"); .expect("publish");
} }
@ -335,7 +336,7 @@ mod tests {
replay( replay(
client, client,
&agent_term::stream_name(AGENT), &agent_term::stream_name(AGENT),
swarm_queue_client::term_subject(AGENT), swarm_wire::term_subject(AGENT),
resume_after, resume_after,
) )
.await .await
@ -352,7 +353,7 @@ mod tests {
for subagent in ["scout", "builder"] { for subagent in ["scout", "builder"] {
client client
.publish( .publish(
swarm_queue_client::subagent_term::subject(AGENT, subagent), swarm_wire::subagent_term::subject(AGENT, subagent),
format!("{subagent}{i}").into(), format!("{subagent}{i}").into(),
) )
.await .await
@ -368,8 +369,8 @@ mod tests {
) -> Result<BoxStream<'static, Row>, String> { ) -> Result<BoxStream<'static, Row>, String> {
replay( replay(
client, client,
&swarm_queue_client::subagent_term::stream_name(AGENT), &swarm_wire::subagent_term::stream_name(AGENT),
swarm_queue_client::subagent_term::subject(AGENT, "scout"), swarm_wire::subagent_term::subject(AGENT, "scout"),
resume_after, resume_after,
) )
.await .await
@ -413,7 +414,7 @@ mod tests {
let nats = TestNats::start(); let nats = TestNats::start();
let client = nats.client().await; 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 .await
.expect("the stream"); .expect("the stream");
publish_rows(&client, 1..=3).await; publish_rows(&client, 1..=3).await;
@ -445,12 +446,12 @@ mod tests {
async fn ensuring_the_terminal_stream_again_leaves_it_as_it_is() { async fn ensuring_the_terminal_stream_again_leaves_it_as_it_is() {
let nats = TestNats::start(); let nats = TestNats::start();
let client = nats.client().await; 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 .await
.expect("the first ensure creates the stream"); .expect("the first ensure creates the stream");
publish_rows(&client, 1..=3).await; publish_rows(&client, 1..=3).await;
stored(&client, 3).await; stored(&client, 3).await;
agent_term::open_or_create(&client, AGENT) swarm_queue_client::agent_term::open_or_create(&client, AGENT)
.await .await
.expect("a second ensure succeeds"); .expect("a second ensure succeeds");
// Read back from the server, not from what the ensure returned. // Read back from the server, not from what the ensure returned.
@ -484,7 +485,7 @@ mod tests {
.expect("an earlier stream"); .expect("an earlier stream");
publish_rows(&client, 1..=2).await; publish_rows(&client, 1..=2).await;
stored(&client, 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 .await
.expect("an existing stream opens"); .expect("an existing stream opens");
assert_eq!(opened.cached_info().config.max_message_size, 4096); 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() { async fn the_replay_reaches_the_reader_before_the_live_tail() {
let nats = TestNats::start(); let nats = TestNats::start();
let client = nats.client().await; let client = nats.client().await;
agent_term::open_or_create(&client, AGENT) swarm_queue_client::agent_term::open_or_create(&client, AGENT)
.await .await
.expect("the stream"); .expect("the stream");
publish_rows(&client, 1..=150).await; publish_rows(&client, 1..=150).await;
@ -514,7 +515,7 @@ mod tests {
async fn a_reconnect_gets_every_later_row_once() { async fn a_reconnect_gets_every_later_row_once() {
let nats = TestNats::start(); let nats = TestNats::start();
let client = nats.client().await; let client = nats.client().await;
agent_term::open_or_create(&client, AGENT) swarm_queue_client::agent_term::open_or_create(&client, AGENT)
.await .await
.expect("the stream"); .expect("the stream");
publish_rows(&client, 1..=150).await; 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() { async fn a_subagent_replays_its_own_newest_rows_before_the_live_tail() {
let nats = TestNats::start(); let nats = TestNats::start();
let client = nats.client().await; 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) swarm_queue_client::subagent_term::open_or_create(&client, AGENT)
.await .await
.expect("the stream"); .expect("the stream");
@ -559,7 +560,7 @@ mod tests {
async fn a_subagent_reconnect_gets_every_later_row_once() { async fn a_subagent_reconnect_gets_every_later_row_once() {
let nats = TestNats::start(); let nats = TestNats::start();
let client = nats.client().await; 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) swarm_queue_client::subagent_term::open_or_create(&client, AGENT)
.await .await
.expect("the stream"); .expect("the stream");

36
swarm-nats-auth/src/policy.rs
View file

@ -241,7 +241,7 @@ impl Policy {
subscribe: Some(Self::reader_subscribe(client_id)), subscribe: Some(Self::reader_subscribe(client_id)),
// The answer to a hive's permissions import. NATS lets it // The answer to a hive's permissions import. NATS lets it
// answer only a message it received, once. // 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) { 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. /// to swarm-ui, every hive's permissions import, and its own reply inbox.
/// ///
/// The three families are the publishers' (`hive-agent`'s `swarm_term` /// 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 /// by the same `swarm_queue_client` helpers. Operator input, deploys and
/// knowledge are absent: the reader publishes those. /// knowledge are absent: the reader publishes those.
fn reader_subscribe(client_id: &str) -> Vec<String> { fn reader_subscribe(client_id: &str) -> Vec<String> {
vec![ vec![
swarm_queue_client::term_subject_wildcard(), swarm_wire::term_subject_wildcard(),
swarm_queue_client::subagent_term::subject_wildcard(), swarm_wire::subagent_term::subject_wildcard(),
swarm_queue_client::agent_state_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), Self::inbox(client_id),
] ]
} }
@ -577,11 +577,11 @@ impl Policy {
swarm_wire::agent_limits::STREAM swarm_wire::agent_limits::STREAM
)); ));
// The permissions the swarm grants those agents // 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. // every key for the same reason.
subjects.push(format!( subjects.push(format!(
"$JS.API.DIRECT.GET.{}.>", "$JS.API.DIRECT.GET.{}.>",
swarm_queue_client::agent_permissions::STREAM swarm_wire::agent_permissions::STREAM
)); ));
// The archetype an agent was created from // The archetype an agent was created from
// (`swarm_queue_client::agent_archetype`), read the same way and every // (`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 // 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 // as a request on this hive's own subject: the controller takes the
// hive from the subject, so one hive cannot import for another. // 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( subjects.extend(
self.extra_hive_subjects self.extra_hive_subjects
.iter() .iter()
@ -701,11 +701,11 @@ impl Policy {
// which this client already subscribes to and reads by // which this client already subscribes to and reads by
// `MSG.GET` above. No bare `CONSUMER.CREATE.<stream>`, whose // `MSG.GET` above. No bare `CONSUMER.CREATE.<stream>`, whose
// filter is unchecked. // 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-<agent>`: the // The same for one subagent's rows in `term-sub-<agent>`: the
// filter is `$SWARM.term.<agent>.sub.<subagent>`, two tokens // filter is `$SWARM.term.<agent>.sub.<subagent>`, two tokens
// deeper than the grant above reaches. // deeper than the grant above reaches.
swarm_queue_client::subagent_term::consumer_create_grant(), swarm_wire::subagent_term::consumer_create_grant(),
"$KV.*.>".to_owned(), "$KV.*.>".to_owned(),
]); ]);
// The per-agent status bucket, read-side only. Same five subjects as // The per-agent status bucket, read-side only. Same five subjects as
@ -925,7 +925,7 @@ mod tests {
/// no other, and writes none. /// no other, and writes none.
#[test] #[test]
fn a_hive_reads_every_agents_permissions_and_an_agent_only_its_own() { 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| { let direct_get_subject = |agent: &str| {
format!("$JS.API.DIRECT.GET.KV_agent-permissions.$KV.agent-permissions.{agent}") 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. /// hive's and may answer, and no other role may send one.
#[test] #[test]
fn a_hive_imports_only_under_its_own_name_and_the_reader_answers() { 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, IMPORT_REPLY_DEADLINE, import_subject, import_subject_wildcard,
}; };
@ -1408,7 +1408,7 @@ mod tests {
let inbox = client_inbox_prefix("swarm-controller"); let inbox = client_inbox_prefix("swarm-controller");
for allowed in [ for allowed in [
"$SWARM.term.atlas".to_owned(), "$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(), "$SWARM.agent-state.atlas".to_owned(),
reply(&inbox), reply(&inbox),
multiplexer(&inbox), multiplexer(&inbox),
@ -1443,8 +1443,8 @@ mod tests {
.permissions("swarm-controller") .permissions("swarm-controller")
.expect("the reader"); .expect("the reader");
for built in [ for built in [
swarm_queue_client::term_subject("atlas"), swarm_wire::term_subject("atlas"),
swarm_queue_client::subagent_term::subject("atlas", "scout"), swarm_wire::subagent_term::subject("atlas", "scout"),
swarm_queue_client::agent_state_subject("atlas"), swarm_queue_client::agent_state_subject("atlas"),
] { ] {
assert!( assert!(
@ -1778,8 +1778,8 @@ mod tests {
#[test] #[test]
fn the_controller_may_replay_terminal_rows_and_no_other_subject() { fn the_controller_may_replay_terminal_rows_and_no_other_subject() {
use swarm_queue_client::agent_term::stream_name; use swarm_wire::agent_term::stream_name;
use swarm_queue_client::{subagent_term, term_subject}; use swarm_wire::{subagent_term, term_subject};
let p = policy() let p = policy()
.permissions("swarm-controller") .permissions("swarm-controller")
@ -1983,7 +1983,7 @@ mod tests {
/// the controller's to create, which its own grant covers. /// the controller's to create, which its own grant covers.
#[test] #[test]
fn an_agents_subagent_grant_is_publish_on_its_own_family_only() { 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() let p = policy()
.with_agent_token_subjects(vec!["$SWARM.term.{agent}.sub.>".to_owned()]) .with_agent_token_subjects(vec!["$SWARM.term.{agent}.sub.>".to_owned()])

2
swarm-queue-client/Cargo.toml
View file

@ -51,7 +51,7 @@ operator-profiles = ["async-nats/jetstream"]
# `agent_limits::read`, for hive-c0re, which reads the key of each agent it # `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`. # runs the same way. Explicit for the same reason as `notices`.
agent-limits = ["async-nats/jetstream"] 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 # 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 # `Permission` names, so the crates that only need the bucket name do not link
# it. # it.

158
swarm-queue-client/src/agent_permissions.rs
View file

@ -1,102 +1,10 @@
//! The per-agent permissions KV bucket: one key per agent, holding the //! Opening and reading the per-agent permissions KV bucket, and the copy
//! [`hive_types::Permission`] set the swarm grants it. //! 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`;
//! **swarm-controller is the only writer**, and creates the bucket when it //! reading needs the `agent-permissions` feature and writing needs `kv`.
//! 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.
/// 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`/`<agent>`. `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")] #[cfg(feature = "agent-permissions")]
#[derive(Debug, Clone, Default, PartialEq, Eq, serde::Serialize, serde::Deserialize)] use swarm_wire::agent_permissions::AgentPermissions;
pub struct AgentPermissions {
#[serde(default, skip_serializing_if = "Option::is_none")]
pub permissions: Option<Vec<hive_types::Permission>>,
}
/// 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<String, Vec<hive_types::Permission>>,
}
/// 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<String>,
kept: Vec<String>,
skipped: Vec<String>,
},
/// Nothing is known to be written; the hive sends again later.
Failed { error: String },
}
/// Open the permissions bucket, creating it if nothing has yet. /// 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()); let js = async_nats::jetstream::new(client.clone());
crate::kv::open_or_create( crate::kv::open_or_create(
&js, &js,
BUCKET, swarm_wire::agent_permissions::BUCKET,
"agent-permissions", "agent-permissions",
"Permissions the swarm controller sets for each agent".to_owned(), "Permissions the swarm controller sets for each agent".to_owned(),
) )
@ -209,8 +117,13 @@ pub async fn read(
use async_nats::jetstream::stream::DirectGetErrorKind; use async_nats::jetstream::stream::DirectGetErrorKind;
let js = async_nats::jetstream::new(client.clone()); let js = async_nats::jetstream::new(client.clone());
let stream = js.get_stream_no_info(STREAM).await?; let stream = js
let message = match stream.direct_get_last_for_subject(key_subject(agent)).await { .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, Ok(message) => message,
Err(e) if matches!(e.kind(), DirectGetErrorKind::NotFound) => { Err(e) if matches!(e.kind(), DirectGetErrorKind::NotFound) => {
return Ok(Read::Current(AgentPermissions::default())); return Ok(Read::Current(AgentPermissions::default()));
@ -245,53 +158,10 @@ fn decode(operation: Option<&str>, payload: &[u8]) -> serde_json::Result<AgentPe
serde_json::from_slice(payload) serde_json::from_slice(payload)
} }
#[cfg(test)] #[cfg(all(test, feature = "agent-permissions"))]
mod tests { mod tests {
use super::*; 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".
#[cfg(feature = "agent-permissions")]
#[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]));
}
/// Before a read the stored list stands; after one, a set key wins and an /// Before a read the stored list stands; after one, a set key wins and an
/// unset key is the default unless the import is pending. /// unset key is the default unless the import is pending.
#[cfg(feature = "agent-permissions")] #[cfg(feature = "agent-permissions")]

202
swarm-queue-client/src/agent_term.rs
View file

@ -1,103 +1,22 @@
//! An agent's own terminal: the stream that keeps its rows, and the consumer //! An agent's own terminal stream: its creation and the consumer a reader
//! a reader replays them through. //! replays it through. The names, subjects, grants and replay start are
//! `swarm_wire::agent_term`.
//! //!
//! The agent publishes its classified terminal rows on //! The stream captures exactly one literal subject, which
//! [`crate::term_subject`] with a plain core publish, under its own queue //! `swarm_wire::subagent_term`'s stream never overlaps. The `JetStream` server
//! credential, which may publish there and nothing else of these streams. //! refuses to create a stream whose subjects overlap another's.
//! `swarm-controller` creates one stream per agent with the fixed config
//! [`crate::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.<agent>`.
//! [`crate::subagent_term`]'s stream captures `$SWARM.term.<agent>.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 use swarm_wire::agent_term::{IDLE_HEARTBEAT, MAX_AGE, stream_name, stream_subject};
/// 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 large one agent's stream may grow before the discard policy reclaims /// 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 /// space. The same cap as the subagent stream, for the same reason: it
/// bounds the disk one agent can claim. /// bounds the disk one agent can claim.
#[cfg(feature = "agent-term")]
const MAX_BYTES: i64 = 64 * 1024 * 1024; 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-<agent>`.
#[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.<stream>` 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>) -> 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. /// The config [`open_or_create`] creates `agent`'s stream with.
/// ///
/// `max_message_size` tracks the queue's live `max_payload`, as the subagent /// `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. /// 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 { fn stream_config(agent: &str, max_message_size: i32) -> async_nats::jetstream::stream::Config {
async_nats::jetstream::stream::Config { async_nats::jetstream::stream::Config {
name: stream_name(agent), 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 /// Open `agent`'s terminal stream, creating it if it does not exist yet. An
/// existing stream is opened as it is. /// existing stream is opened as it is.
#[cfg(feature = "agent-term")]
pub async fn open_or_create( pub async fn open_or_create(
client: &async_nats::Client, client: &async_nats::Client,
agent: &str, agent: &str,
@ -138,15 +56,14 @@ pub async fn open_or_create(
/// The consumer that replays the rows on `filter_subject` from /// The consumer that replays the rows on `filter_subject` from
/// `start_sequence` and then follows new ones, delivered to `deliver_subject` /// `start_sequence` and then follows new ones, delivered to `deliver_subject`
/// in stream order. The filter is an agent's [`stream_subject`] or one /// 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 /// Ephemeral and unacknowledged: the server drops it once nothing listens
/// on `deliver_subject`, so a reader that goes away needs no delete grant. /// 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 /// `name` must be set, because the server only takes the filter from the
/// subject [`consumer_create_grant`] (or /// subject `swarm_wire::agent_term::consumer_create_grant` (or
/// [`crate::subagent_term::consumer_create_grant`]) covers when the create /// `swarm_wire::subagent_term::consumer_create_grant`) covers when the create
/// names a consumer. /// names a consumer.
#[cfg(feature = "agent-term")]
#[must_use] #[must_use]
pub fn replay_consumer_config( pub fn replay_consumer_config(
filter_subject: String, filter_subject: String,
@ -171,106 +88,9 @@ pub fn replay_consumer_config(
mod tests { mod tests {
use super::*; 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 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, /// the replay reaches the reader before the live tail, in stream order,
/// through the one subscription. /// through the one subscription.
#[cfg(feature = "agent-term")]
#[test] #[test]
fn the_consumer_replays_before_it_follows() { fn the_consumer_replays_before_it_follows() {
let cfg = replay_consumer_config( 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 /// Every limit is read from the subagent stream's own config, so the two
/// fail this test the moment they drift apart. /// fail this test the moment they drift apart.
#[cfg(all(feature = "agent-term", feature = "subagent-term"))] #[cfg(feature = "subagent-term")]
#[test] #[test]
fn stream_config_matches_the_subagent_streams_limits() { fn stream_config_matches_the_subagent_streams_limits() {
let main = stream_config("iris", 8_388_608); let main = stream_config("iris", 8_388_608);

56
swarm-queue-client/src/lib.rs
View file

@ -232,10 +232,11 @@ pub mod agent_inference;
/// and reading needs `agent-limits`. /// and reading needs `agent-limits`.
pub mod agent_limits; pub mod agent_limits;
/// The bucket swarm-controller sets each agent's permissions in — one key /// Opening and reading the bucket swarm-controller sets each agent's
/// per agent, read by the hive that runs it, plus the one-shot import a hive /// permissions in — one key per agent, read by the hive that runs it. The names,
/// sends. The names are unconditional; writing needs `kv` and the stored type /// the stored type and the import messages are `swarm_wire::agent_permissions`;
/// and reading need `agent-permissions`. /// writing needs `kv` and reading needs `agent-permissions`.
#[cfg(any(feature = "kv", feature = "agent-permissions"))]
pub mod agent_permissions; pub mod agent_permissions;
/// The bucket swarm-controller publishes each operator's name and forge-profile /// 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`]. /// Shared by [`agent_control_subject`] and [`AGENT_CONTROL_SUBJECT_WILDCARD`].
const AGENT_CONTROL_SUBJECT_PREFIX: &str = "$SWARM.control"; 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 /// 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] #[must_use]
pub fn agent_state_subject(agent: &str) -> String { pub fn agent_state_subject(agent: &str) -> String {
format!("{AGENT_STATE_SUBJECT_PREFIX}.{agent}") format!("{AGENT_STATE_SUBJECT_PREFIX}.{agent}")
} }
/// Every [`agent_state_subject`] at once, for the same reason as /// Every [`agent_state_subject`] at once, for the same reason as
/// [`term_subject_wildcard`]. /// [`swarm_wire::term_subject_wildcard`].
#[must_use] #[must_use]
pub fn agent_state_subject_wildcard() -> String { pub fn agent_state_subject_wildcard() -> String {
agent_state_subject("*") agent_state_subject("*")
@ -475,14 +452,16 @@ pub struct AgentInput {
#[cfg(feature = "notices")] #[cfg(feature = "notices")]
pub mod notices; pub mod notices;
/// The stream that keeps an agent's own terminal rows, and the consumer a /// Creating an agent's own terminal stream and the consumer a reader replays
/// reader replays them through. Only the `JetStream` calls need the /// it through. The names, subjects, grants and replay start are
/// `agent-term` feature; the names are unconditional. /// `swarm_wire::agent_term`. Behind the `agent-term` feature.
#[cfg(feature = "agent-term")]
pub mod agent_term; pub mod agent_term;
/// The subject an agent's subagents' terminal rows go to, and the per-agent /// Creating the per-agent stream that lists an agent's subagents. The names,
/// stream that lists them. Only opening the stream needs the `subagent-term` /// subjects and grants are `swarm_wire::subagent_term`. Behind the
/// feature; the names are unconditional, like [`agent_icon`]'s. /// `subagent-term` feature.
#[cfg(feature = "subagent-term")]
pub mod subagent_term; pub mod subagent_term;
/// Only the fields this needs; authelia returns several. /// 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. /// The names the publishers, the relays and the queue's grant all spell.
#[test] #[test]
fn term_and_agent_state_subjects_are_the_agreed_names() { fn agent_state_subjects_are_the_agreed_names() {
assert_eq!(term_subject("mara"), "$SWARM.term.mara");
assert_eq!(term_subject_wildcard(), "$SWARM.term.*");
assert_eq!(agent_state_subject("mara"), "$SWARM.agent-state.mara"); assert_eq!(agent_state_subject("mara"), "$SWARM.agent-state.mara");
assert_eq!(agent_state_subject_wildcard(), "$SWARM.agent-state.*"); assert_eq!(agent_state_subject_wildcard(), "$SWARM.agent-state.*");
assert_eq!(subagent_term::subject_wildcard(), "$SWARM.term.*.sub.*");
} }
#[test] #[test]

153
swarm-queue-client/src/subagent_term.rs
View file

@ -1,108 +1,17 @@
//! Subagent terminals: the subject each subagent's rows go to, and the //! Creating an agent's subagent stream. The names, subjects, grants and the
//! stream that keeps them. //! subagent list are `swarm_wire::subagent_term`.
//! //!
//! An agent's subagent daemon publishes each subagent's classified terminal //! `swarm-controller` creates one stream per agent with the fixed config
//! rows on `$SWARM.term.<agent>.sub.<subagent>`, under the agent's own queue //! [`crate::subagent_term::open_or_create`] spells.
//! 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`.
use crate::TERM_SUBJECT_PREFIX; use swarm_wire::subagent_term::{MAX_AGE, stream_name, stream_subjects};
/// The token between the agent and the subagent name. A subagent subject has
/// two more tokens than the agent's own `$SWARM.term.<agent>`, 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);
/// How large one agent's stream may grow before the discard policy reclaims /// 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 /// 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 /// stay well inside it while still bounding the disk one compromised or
/// runaway agent can claim for 24h. /// runaway agent can claim for 24h.
#[cfg(feature = "subagent-term")]
const MAX_BYTES: i64 = 64 * 1024 * 1024; const MAX_BYTES: i64 = 64 * 1024 * 1024;
/// The stream holding `agent`'s subagent rows: `term-sub-<agent>`.
#[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.<agent>.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<Item = &'a str>) -> Vec<String> {
let mut names: Vec<String> = 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. /// The config [`open_or_create`] creates `agent`'s stream with.
/// ///
/// `max_message_size` is taken as a parameter rather than read from a /// `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<Item = &'a st
/// ([`crate::max_payload`]), the same value the publisher already sizes its /// ([`crate::max_payload`]), the same value the publisher already sizes its
/// rows against, so the stream never rejects a row the queue itself /// rows against, so the stream never rejects a row the queue itself
/// accepted. /// accepted.
#[cfg(feature = "subagent-term")]
pub(crate) fn stream_config( pub(crate) fn stream_config(
agent: &str, agent: &str,
max_message_size: i32, max_message_size: i32,
@ -129,7 +37,6 @@ pub(crate) fn stream_config(
/// Open `agent`'s subagent stream, creating it if it does not exist yet. An /// Open `agent`'s subagent stream, creating it if it does not exist yet. An
/// existing stream is opened as it is. /// existing stream is opened as it is.
#[cfg(feature = "subagent-term")]
pub async fn open_or_create( pub async fn open_or_create(
client: &async_nats::Client, client: &async_nats::Client,
agent: &str, agent: &str,
@ -155,59 +62,9 @@ pub async fn open_or_create(
mod tests { mod tests {
use super::*; 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"]);
}
/// The config the controller creates caps disk per agent and never /// The config the controller creates caps disk per agent and never
/// fails a publish because of that cap — `discard: Old` reclaims the /// fails a publish because of that cap — `discard: Old` reclaims the
/// oldest rows instead of rejecting the newest. /// oldest rows instead of rejecting the newest.
#[cfg(feature = "subagent-term")]
#[test] #[test]
fn stream_config_caps_bytes_and_discards_the_oldest() { fn stream_config_caps_bytes_and_discards_the_oldest() {
let cfg = stream_config("iris", 8_388_608); let cfg = stream_config("iris", 8_388_608);

1
swarm-wire/Cargo.toml
View file

@ -9,6 +9,7 @@ edition.workspace = true
schema = ["dep:utoipa"] schema = ["dep:utoipa"]
[dependencies] [dependencies]
hive-types.workspace = true
serde.workspace = true serde.workspace = true
serde_json.workspace = true serde_json.workspace = true
utoipa = { workspace = true, optional = true } utoipa = { workspace = true, optional = true }

143
swarm-wire/src/agent_permissions.rs Normal file
View file

@ -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`/`<agent>`. `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<Vec<hive_types::Permission>>,
}
/// 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<String, Vec<hive_types::Permission>>,
}
/// 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<String>,
kept: Vec<String>,
skipped: Vec<String>,
},
/// 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]));
}
}

191
swarm-wire/src/agent_term.rs Normal file
View file

@ -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.<agent>`.
//! [`crate::subagent_term`]'s stream captures `$SWARM.term.<agent>.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-<agent>`.
#[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.<stream>` 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>) -> 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);
}
}

40
swarm-wire/src/lib.rs
View file

@ -9,10 +9,50 @@ pub mod agent_icon;
pub mod agent_inference; pub mod agent_inference;
pub mod agent_limits; pub mod agent_limits;
pub mod agent_login; pub mod agent_login;
pub mod agent_permissions;
pub mod agent_schedules; pub mod agent_schedules;
pub mod agent_status; pub mod agent_status;
pub mod agent_term;
pub mod builder; pub mod builder;
pub mod claude_login; pub mod claude_login;
pub mod notices; pub mod notices;
pub mod review_classifier; pub mod review_classifier;
pub mod subagent_term;
pub mod wanted; 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.*");
}
}

152
swarm-wire/src/subagent_term.rs Normal file
View file

@ -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.<agent>.sub.<subagent>`, 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.<agent>`, 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-<agent>`.
#[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.<agent>.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<Item = &'a str>) -> Vec<String> {
let mut names: Vec<String> = 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"]);
}
}