diff --git a/hive-agent/src/swarm_agent_state.rs b/hive-agent/src/swarm_agent_state.rs index a9a9cb16..cdfa3e69 100644 --- a/hive-agent/src/swarm_agent_state.rs +++ b/hive-agent/src/swarm_agent_state.rs @@ -36,12 +36,6 @@ use swarm_queue_client::wanted::AgentState; use crate::events::{Bus, BusEvent, LiveEvent, TurnState}; use crate::term_msg::iso8601_utc; -/// Subject family carrying agent turn-state headers, the swarm-wide -/// agreement this publisher holds up its end of. One leaf subject per -/// agent, matching `$SWARM.term`'s shape, so a subscriber can follow one -/// agent without filtering the swarm's whole header traffic. -const SUBJECT_PREFIX: &str = "$SWARM.agent-state"; - /// How much room to leave under the announced limit for everything the /// publish adds around the payload — subject, headers, protocol framing. /// Same margin and same reasoning as `swarm_term::HEADROOM`. @@ -139,11 +133,6 @@ fn snapshot(bus: &Bus) -> AgentStateMsg { } } -/// The subject this agent's header goes to. -fn subject(agent: &str) -> String { - format!("{SUBJECT_PREFIX}.{agent}") -} - /// Start the publish task, if this agent has a label to name its subject with. pub fn spawn(bus: &Bus) { let agent = crate::identity::label(); @@ -174,7 +163,7 @@ pub fn spawn(bus: &Bus) { /// changes nothing. async fn run(mut rx: broadcast::Receiver, bus: Bus, agent: String) { let client = crate::swarm_queue::client().await; - let subject = subject(&agent); + let subject = swarm_queue_client::agent_state_subject(&agent); tracing::info!(subject, "publishing agent turn state to the swarm queue"); // The last payload actually sent, so a rebuild that changed nothing is @@ -260,7 +249,7 @@ async fn publish( #[cfg(test)] mod tests { - use super::{AgentStateMsg, subject}; + use super::AgentStateMsg; use crate::events::TurnState; use swarm_queue_client::wanted::AgentState; @@ -306,15 +295,6 @@ mod tests { } } - /// The subject the swarm side subscribes to. `swarm-controller`'s relay - /// spells the same prefix in its own constant and the two never meet, so - /// this pins the half that lives here. The agent's own credential is - /// granted the subject naming the agent alone, with no hive in it. - #[test] - fn the_subject_is_the_prefix_then_the_agent() { - assert_eq!(subject("mara"), "$SWARM.agent-state.mara"); - } - /// The published contract, asserted on the JSON a subscriber parses /// rather than on the Rust struct: a renderer is written against these /// key names, and renaming a field in Rust without noticing is exactly diff --git a/hive-agent/src/swarm_term.rs b/hive-agent/src/swarm_term.rs index b6222ae8..158628a3 100644 --- a/hive-agent/src/swarm_term.rs +++ b/hive-agent/src/swarm_term.rs @@ -29,11 +29,6 @@ use crate::events::BusEvent; use crate::term_msg::{ClassifyCtx, TermMsg, classify}; use hive_sh4re::term_msg::fit; -/// Subject family carrying agent terminal rows, the swarm-wide agreement this -/// publisher holds up its end of. One leaf subject per agent, so a subscriber -/// can follow one agent without filtering the swarm's whole terminal traffic. -const SUBJECT_PREFIX: &str = "$SWARM.term"; - /// How much room to leave under the announced limit for everything the /// publish adds around the payload — subject, headers, protocol framing. /// @@ -43,11 +38,6 @@ const SUBJECT_PREFIX: &str = "$SWARM.term"; /// out of eight megabytes buys that margin for nothing. const HEADROOM: usize = 1024; -/// The subject this agent's rows go to. -fn subject(agent: &str) -> String { - format!("{SUBJECT_PREFIX}.{agent}") -} - /// Start the publish task, if this agent has a label to name its subject with. pub fn spawn(rx: broadcast::Receiver) { let agent = crate::identity::label(); @@ -60,7 +50,7 @@ pub fn spawn(rx: broadcast::Receiver) { async fn run(mut rx: broadcast::Receiver, agent: String) { let client = crate::swarm_queue::client().await; - let subject = subject(&agent); + let subject = swarm_queue_client::term_subject(&agent); tracing::info!(subject, "publishing the agent terminal to the swarm queue"); // One context for the task's whole life — see the module doc. @@ -119,7 +109,6 @@ async fn publish(client: &async_nats::Client, subject: &str, msg: TermMsg) { #[cfg(test)] mod tests { - use super::subject; use crate::events::LiveEvent; use crate::term_msg::{ClassifyCtx, classify}; use hive_sh4re::term_msg::fit; @@ -127,13 +116,6 @@ mod tests { /// The time `classify` renders for the event the test below classifies. const ROW_TS: &str = "2026-09-13T12:35:03Z"; - /// The agent's own credential is granted the subject naming the agent - /// alone, with no hive in it. - #[test] - fn the_subject_is_the_prefix_then_the_agent() { - assert_eq!(subject("mara"), "$SWARM.term.mara"); - } - /// What a subscriber actually receives: the payload this module hands /// `publish` is the bare row, so the time has to be *in* it. A `ts` that /// existed in Rust but never serialized would leave the queue exactly as diff --git a/swarm-controller/src/agent_state_stream.rs b/swarm-controller/src/agent_state_stream.rs index 4fb75aba..e00d8bcb 100644 --- a/swarm-controller/src/agent_state_stream.rs +++ b/swarm-controller/src/agent_state_stream.rs @@ -30,14 +30,6 @@ use futures_util::{Stream, StreamExt as _}; use crate::AppState; -/// Subject family carrying agent turn-state headers — must match -/// `hive-agent::swarm_agent_state::SUBJECT_PREFIX` exactly, since the two -/// ends never see the constant together. `crate::term_stream::agent_subject` -/// spells the full subject; as with the terminal relay's own copy, there is no -/// way to check the two sides agree short of this comment and the module -/// docs on both ends staying honest about it. -const SUBJECT_PREFIX: &str = "$SWARM.agent-state"; - #[utoipa::path( get, path = "/api/agents/{name}/state/stream", @@ -115,7 +107,7 @@ pub(crate) async fn stream_agent_state( let subscriber = crate::term_stream::subscribe( &client, - crate::term_stream::agent_subject(SUBJECT_PREFIX, &agent), + swarm_queue_client::agent_state_subject(&agent), "state", ) .await?; diff --git a/swarm-controller/src/term_stream.rs b/swarm-controller/src/term_stream.rs index 71ab39ee..820c59fd 100644 --- a/swarm-controller/src/term_stream.rs +++ b/swarm-controller/src/term_stream.rs @@ -31,13 +31,6 @@ use futures_util::{Stream, StreamExt as _}; use crate::AppState; -/// Subject family carrying agent terminal rows — must match -/// `hive-agent::swarm_term::SUBJECT_PREFIX` exactly, since the two ends -/// never see the constant together. [`agent_subject`] spells the full -/// subject; there is no way to check the two sides agree short of -/// this comment and the module docs on both ends staying honest about it. -const SUBJECT_PREFIX: &str = "$SWARM.term"; - #[utoipa::path( get, path = "/api/agents/{name}/term/stream", @@ -112,17 +105,12 @@ pub(crate) async fn stream_agent_term( ) })?; - let subscriber = subscribe(&client, agent_subject(SUBJECT_PREFIX, &agent), "term").await?; + let subscriber = subscribe(&client, swarm_queue_client::term_subject(&agent), "term").await?; let stream = subscriber.map(|msg| Ok(Event::default().data(String::from_utf8_lossy(&msg.payload)))); Ok(Sse::new(stream).keep_alive(KeepAlive::default())) } -/// The subject one agent publishes on under `prefix`: `.`. -pub(crate) fn agent_subject(prefix: &str, agent: &str) -> String { - format!("{prefix}.{agent}") -} - /// Subscribe to `subject`. `what` names the relay in logs and errors. pub(crate) async fn subscribe( client: &async_nats::Client, @@ -139,14 +127,3 @@ pub(crate) async fn subscribe( tracing::info!(%subject, "{what} stream: client attached"); Ok(subscriber) } - -#[cfg(test)] -mod tests { - use super::*; - - /// The subject the agent's own queue credential is granted. - #[test] - fn a_relay_subscribes_to_the_agents_own_subject() { - assert_eq!(agent_subject(SUBJECT_PREFIX, "atlas"), "$SWARM.term.atlas"); - } -} diff --git a/swarm-nats-auth/src/policy.rs b/swarm-nats-auth/src/policy.rs index b72b9dfd..8db0db2b 100644 --- a/swarm-nats-auth/src/policy.rs +++ b/swarm-nats-auth/src/policy.rs @@ -214,14 +214,14 @@ impl Policy { /// to swarm-ui, and its own reply inbox. /// /// The three families are the publishers' (`hive-agent`'s `swarm_term` - /// and `swarm_agent_state`, `swarm_queue_client::subagent_term`), which no - /// shared constant names; a rename there must change these too. Operator - /// input, deploys and knowledge are absent: the reader publishes those. + /// and `swarm_agent_state`, `swarm_queue_client::subagent_term`), spelled + /// by the same `swarm_queue_client` helpers. Operator input, deploys and + /// knowledge are absent: the reader publishes those. fn reader_subscribe(client_id: &str) -> Vec { vec![ - "$SWARM.term.*".to_owned(), - "$SWARM.term.*.sub.*".to_owned(), - "$SWARM.agent-state.*".to_owned(), + swarm_queue_client::term_subject_wildcard(), + swarm_queue_client::subagent_term::subject_wildcard(), + swarm_queue_client::agent_state_subject_wildcard(), Self::inbox(client_id), ] } @@ -1062,6 +1062,25 @@ mod tests { } } + /// The subjects the publishers and relays build from the shared helpers + /// are the ones the reader's grant covers, whatever those helpers spell. + #[test] + fn the_reader_may_subscribe_to_every_subject_the_shared_helpers_build() { + let reader = policy() + .permissions("swarm-controller") + .expect("the reader"); + for built in [ + swarm_queue_client::term_subject("atlas"), + swarm_queue_client::subagent_term::subject("atlas", "scout"), + swarm_queue_client::agent_state_subject("atlas"), + ] { + assert!( + may_subscribe(&reader, &built), + "the reader may not subscribe to {built}" + ); + } + } + #[test] fn the_prefix_alone_names_no_hive() { // `hive-` would otherwise produce `$KV.hive-status.`, a subject nobody diff --git a/swarm-queue-client/src/lib.rs b/swarm-queue-client/src/lib.rs index ae1efa3c..eb6f1ae9 100644 --- a/swarm-queue-client/src/lib.rs +++ b/swarm-queue-client/src/lib.rs @@ -272,6 +272,47 @@ pub const AGENT_INPUT_SUBJECT_WILDCARD: &str = "$SWARM.input.*"; /// Shared by [`agent_input_subject`] and [`AGENT_INPUT_SUBJECT_WILDCARD`]. const AGENT_INPUT_SUBJECT_PREFIX: &str = "$SWARM.input"; +/// The subject `agent` publishes its terminal rows on. One leaf per agent, so +/// a subscriber can follow one agent without filtering the swarm's whole +/// terminal traffic. +/// +/// Shared by the publisher (`hive-agent`), the relay that streams it to +/// swarm-ui (`swarm-controller`) and the queue's subscribe grant for readers +/// (`swarm-nats-auth`), which spells [`term_subject_wildcard`] — so a rename +/// moves all three together. +#[must_use] +pub fn term_subject(agent: &str) -> String { + format!("{TERM_SUBJECT_PREFIX}.{agent}") +} + +/// Every [`term_subject`] at once: the reader's subscribe grant, which has no +/// roster to enumerate agents from. +#[must_use] +pub fn term_subject_wildcard() -> String { + term_subject("*") +} + +/// Shared by [`term_subject`] and [`subagent_term`], whose subjects sit under +/// the same prefix. +const TERM_SUBJECT_PREFIX: &str = "$SWARM.term"; + +/// The subject `agent` publishes its turn-state header on. Same shape and +/// same three users as [`term_subject`]. +#[must_use] +pub fn agent_state_subject(agent: &str) -> String { + format!("{AGENT_STATE_SUBJECT_PREFIX}.{agent}") +} + +/// Every [`agent_state_subject`] at once, for the same reason as +/// [`term_subject_wildcard`]. +#[must_use] +pub fn agent_state_subject_wildcard() -> String { + agent_state_subject("*") +} + +/// Shared by [`agent_state_subject`] and its wildcard. +const AGENT_STATE_SUBJECT_PREFIX: &str = "$SWARM.agent-state"; + /// The reply-inbox prefix a client presenting `agent`'s own credential /// connects with. /// @@ -915,6 +956,16 @@ where mod tests { use super::*; + /// The names the publishers, the relays and the queue's grant all spell. + #[test] + fn term_and_agent_state_subjects_are_the_agreed_names() { + assert_eq!(term_subject("mara"), "$SWARM.term.mara"); + assert_eq!(term_subject_wildcard(), "$SWARM.term.*"); + assert_eq!(agent_state_subject("mara"), "$SWARM.agent-state.mara"); + assert_eq!(agent_state_subject_wildcard(), "$SWARM.agent-state.*"); + assert_eq!(subagent_term::subject_wildcard(), "$SWARM.term.*.sub.*"); + } + #[test] fn a_tls_client_config_builds_once_the_provider_is_installed() { install_crypto_provider(); diff --git a/swarm-queue-client/src/subagent_term.rs b/swarm-queue-client/src/subagent_term.rs index 965c9480..b340d0d4 100644 --- a/swarm-queue-client/src/subagent_term.rs +++ b/swarm-queue-client/src/subagent_term.rs @@ -18,8 +18,7 @@ //! queue grant in `swarm-nats.nix`, whose `{agent}` template spells the same //! subjects. -/// The subject family every agent terminal shares. -const TERM_PREFIX: &str = "$SWARM.term"; +use crate::TERM_SUBJECT_PREFIX; /// The token between the agent and the subagent name. A subagent subject has /// two more tokens than the agent's own `$SWARM.term.`, so neither @@ -47,14 +46,21 @@ pub fn stream_name(agent: &str) -> String { /// The subject `subagent`'s rows are published on. #[must_use] pub fn subject(agent: &str, subagent: &str) -> String { - format!("{TERM_PREFIX}.{agent}.{SUBAGENT_TOKEN}.{subagent}") + 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("*", "*") } /// 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_PREFIX}.{agent}.{SUBAGENT_TOKEN}.>") + format!("{TERM_SUBJECT_PREFIX}.{agent}.{SUBAGENT_TOKEN}.>") } /// The subagent a stream subject names, or `None` for a subject that is not @@ -62,7 +68,7 @@ pub fn stream_subjects(agent: &str) -> String { #[must_use] pub fn subagent_from_subject<'a>(agent: &str, subject: &'a str) -> Option<&'a str> { let name = subject - .strip_prefix(TERM_PREFIX)? + .strip_prefix(TERM_SUBJECT_PREFIX)? .strip_prefix('.')? .strip_prefix(agent)? .strip_prefix('.')?