Watch
0
0
Fork
You've already forked hyperhive
0

swarm-nats-auth: take the subscribe subjects from the shared definitions

The reader's subscribe allow-list spelled `$SWARM.term.*`,
`$SWARM.term.*.sub.*` and `$SWARM.agent-state.*` as literals, copies of
private prefixes in hive-agent and swarm-controller. A rename on either side
left the policy behind and the stream stopped with only a connection-level
server error.

swarm-queue-client now owns `term_subject`, `agent_state_subject` and their
`*_wildcard` forms, and `subagent_term::subject_wildcard`; hive-agent's
publishers, swarm-controller's relays and the policy all build from them.
Subjects are byte-identical.
This commit is contained in:
atlas 2026-10-07 16:51:53 +02:00
commit 1b061fc9d5
7 changed files with 92 additions and 85 deletions

View file

@ -36,12 +36,6 @@ use swarm_queue_client::wanted::AgentState;
use crate::events::{Bus, BusEvent, LiveEvent, TurnState}; use crate::events::{Bus, BusEvent, LiveEvent, TurnState};
use crate::term_msg::iso8601_utc; 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 /// How much room to leave under the announced limit for everything the
/// publish adds around the payload — subject, headers, protocol framing. /// publish adds around the payload — subject, headers, protocol framing.
/// Same margin and same reasoning as `swarm_term::HEADROOM`. /// 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. /// Start the publish task, if this agent has a label to name its subject with.
pub fn spawn(bus: &Bus) { pub fn spawn(bus: &Bus) {
let agent = crate::identity::label(); let agent = crate::identity::label();
@ -174,7 +163,7 @@ pub fn spawn(bus: &Bus) {
/// changes nothing. /// changes nothing.
async fn run(mut rx: broadcast::Receiver<BusEvent>, bus: Bus, agent: String) { async fn run(mut rx: broadcast::Receiver<BusEvent>, bus: Bus, agent: String) {
let client = crate::swarm_queue::client().await; 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"); tracing::info!(subject, "publishing agent turn state to the swarm queue");
// The last payload actually sent, so a rebuild that changed nothing is // The last payload actually sent, so a rebuild that changed nothing is
@ -260,7 +249,7 @@ async fn publish(
#[cfg(test)] #[cfg(test)]
mod tests { mod tests {
use super::{AgentStateMsg, subject}; use super::AgentStateMsg;
use crate::events::TurnState; use crate::events::TurnState;
use swarm_queue_client::wanted::AgentState; 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 /// The published contract, asserted on the JSON a subscriber parses
/// rather than on the Rust struct: a renderer is written against these /// rather than on the Rust struct: a renderer is written against these
/// key names, and renaming a field in Rust without noticing is exactly /// key names, and renaming a field in Rust without noticing is exactly

View file

@ -29,11 +29,6 @@ use crate::events::BusEvent;
use crate::term_msg::{ClassifyCtx, TermMsg, classify}; use crate::term_msg::{ClassifyCtx, TermMsg, classify};
use hive_sh4re::term_msg::fit; 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 /// How much room to leave under the announced limit for everything the
/// publish adds around the payload — subject, headers, protocol framing. /// 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. /// out of eight megabytes buys that margin for nothing.
const HEADROOM: usize = 1024; 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. /// Start the publish task, if this agent has a label to name its subject with.
pub fn spawn(rx: broadcast::Receiver<BusEvent>) { pub fn spawn(rx: broadcast::Receiver<BusEvent>) {
let agent = crate::identity::label(); let agent = crate::identity::label();
@ -60,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 = subject(&agent); let subject = swarm_queue_client::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.
@ -119,7 +109,6 @@ async fn publish(client: &async_nats::Client, subject: &str, msg: TermMsg) {
#[cfg(test)] #[cfg(test)]
mod tests { mod tests {
use super::subject;
use crate::events::LiveEvent; use crate::events::LiveEvent;
use crate::term_msg::{ClassifyCtx, classify}; use crate::term_msg::{ClassifyCtx, classify};
use hive_sh4re::term_msg::fit; use hive_sh4re::term_msg::fit;
@ -127,13 +116,6 @@ mod tests {
/// The time `classify` renders for the event the test below classifies. /// The time `classify` renders for the event the test below classifies.
const ROW_TS: &str = "2026-09-13T12:35:03Z"; 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 /// 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 /// `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 /// existed in Rust but never serialized would leave the queue exactly as

View file

@ -30,14 +30,6 @@ use futures_util::{Stream, StreamExt as _};
use crate::AppState; 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( #[utoipa::path(
get, get,
path = "/api/agents/{name}/state/stream", path = "/api/agents/{name}/state/stream",
@ -115,7 +107,7 @@ pub(crate) async fn stream_agent_state(
let subscriber = crate::term_stream::subscribe( let subscriber = crate::term_stream::subscribe(
&client, &client,
crate::term_stream::agent_subject(SUBJECT_PREFIX, &agent), swarm_queue_client::agent_state_subject(&agent),
"state", "state",
) )
.await?; .await?;

View file

@ -31,13 +31,6 @@ use futures_util::{Stream, StreamExt as _};
use crate::AppState; 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( #[utoipa::path(
get, get,
path = "/api/agents/{name}/term/stream", 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 = let stream =
subscriber.map(|msg| Ok(Event::default().data(String::from_utf8_lossy(&msg.payload)))); subscriber.map(|msg| Ok(Event::default().data(String::from_utf8_lossy(&msg.payload))));
Ok(Sse::new(stream).keep_alive(KeepAlive::default())) Ok(Sse::new(stream).keep_alive(KeepAlive::default()))
} }
/// The subject one agent publishes on under `prefix`: `<prefix>.<agent>`.
pub(crate) fn agent_subject(prefix: &str, agent: &str) -> String {
format!("{prefix}.{agent}")
}
/// Subscribe to `subject`. `what` names the relay in logs and errors. /// Subscribe to `subject`. `what` names the relay in logs and errors.
pub(crate) async fn subscribe( pub(crate) async fn subscribe(
client: &async_nats::Client, client: &async_nats::Client,
@ -139,14 +127,3 @@ pub(crate) async fn subscribe(
tracing::info!(%subject, "{what} stream: client attached"); tracing::info!(%subject, "{what} stream: client attached");
Ok(subscriber) 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");
}
}

View file

@ -214,14 +214,14 @@ impl Policy {
/// to swarm-ui, and its own reply inbox. /// to swarm-ui, 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`), which no /// and `swarm_agent_state`, `swarm_queue_client::subagent_term`), spelled
/// shared constant names; a rename there must change these too. Operator /// by the same `swarm_queue_client` helpers. Operator input, deploys and
/// 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.term.*".to_owned(), swarm_queue_client::term_subject_wildcard(),
"$SWARM.term.*.sub.*".to_owned(), swarm_queue_client::subagent_term::subject_wildcard(),
"$SWARM.agent-state.*".to_owned(), swarm_queue_client::agent_state_subject_wildcard(),
Self::inbox(client_id), 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] #[test]
fn the_prefix_alone_names_no_hive() { fn the_prefix_alone_names_no_hive() {
// `hive-` would otherwise produce `$KV.hive-status.`, a subject nobody // `hive-` would otherwise produce `$KV.hive-status.`, a subject nobody

View file

@ -272,6 +272,47 @@ pub const AGENT_INPUT_SUBJECT_WILDCARD: &str = "$SWARM.input.*";
/// Shared by [`agent_input_subject`] and [`AGENT_INPUT_SUBJECT_WILDCARD`]. /// Shared by [`agent_input_subject`] and [`AGENT_INPUT_SUBJECT_WILDCARD`].
const AGENT_INPUT_SUBJECT_PREFIX: &str = "$SWARM.input"; 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 /// The reply-inbox prefix a client presenting `agent`'s own credential
/// connects with. /// connects with.
/// ///
@ -915,6 +956,16 @@ where
mod tests { mod tests {
use super::*; 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] #[test]
fn a_tls_client_config_builds_once_the_provider_is_installed() { fn a_tls_client_config_builds_once_the_provider_is_installed() {
install_crypto_provider(); install_crypto_provider();

View file

@ -18,8 +18,7 @@
//! queue grant in `swarm-nats.nix`, whose `{agent}` template spells the same //! queue grant in `swarm-nats.nix`, whose `{agent}` template spells the same
//! subjects. //! subjects.
/// The subject family every agent terminal shares. use crate::TERM_SUBJECT_PREFIX;
const TERM_PREFIX: &str = "$SWARM.term";
/// The token between the agent and the subagent name. A subagent subject has /// 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 /// two more tokens than the agent's own `$SWARM.term.<agent>`, so neither
@ -47,14 +46,21 @@ pub fn stream_name(agent: &str) -> String {
/// The subject `subagent`'s rows are published on. /// The subject `subagent`'s rows are published on.
#[must_use] #[must_use]
pub fn subject(agent: &str, subagent: &str) -> String { 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 /// Every subject [`stream_name`]'s stream captures, which is also the publish
/// grant's shape. /// grant's shape.
#[must_use] #[must_use]
pub fn stream_subjects(agent: &str) -> String { 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 /// 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] #[must_use]
pub fn subagent_from_subject<'a>(agent: &str, subject: &'a str) -> Option<&'a str> { pub fn subagent_from_subject<'a>(agent: &str, subject: &'a str) -> Option<&'a str> {
let name = subject let name = subject
.strip_prefix(TERM_PREFIX)? .strip_prefix(TERM_SUBJECT_PREFIX)?
.strip_prefix('.')? .strip_prefix('.')?
.strip_prefix(agent)? .strip_prefix(agent)?
.strip_prefix('.')? .strip_prefix('.')?