From 8a7733b3f5687791ac96705517195639e425da69 Mon Sep 17 00:00:00 2001 From: atlas Date: Sun, 4 Oct 2026 21:43:33 +0200 Subject: [PATCH] swarm-controller: drop the relay subscription to the refused hive-scoped agent subject --- swarm-controller/src/agent_state_stream.rs | 13 ++-- swarm-controller/src/subagent_term.rs | 9 ++- swarm-controller/src/term_stream.rs | 78 ++++++++-------------- 3 files changed, 37 insertions(+), 63 deletions(-) diff --git a/swarm-controller/src/agent_state_stream.rs b/swarm-controller/src/agent_state_stream.rs index 7eca9f93..8594d724 100644 --- a/swarm-controller/src/agent_state_stream.rs +++ b/swarm-controller/src/agent_state_stream.rs @@ -2,8 +2,7 @@ //! turn-state header, sourced straight from the swarm queue. //! //! Sibling of [`crate::term_stream`] in every structural respect — the -//! reasoning there for resolving the hive at request time rather than -//! naming it in the path, and for a live tail with no replay, applies +//! reasoning there for keeping the hive out of the path, and for a live tail with no replay, applies //! unchanged here and is not repeated. What differs is only what the two //! subjects carry: a terminal row is an event that happened, a header is //! the agent's current state. @@ -33,8 +32,8 @@ 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_subjects` -/// spells the full subjects; as with the terminal relay's own copy, there is no +/// 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"; @@ -85,7 +84,7 @@ pub(crate) async fn stream_agent_state( // One-agent "roster": reuses `AgentStatusReader::view`'s already-tested // row logic rather than a second, narrower lookup that could disagree // with it about what "never reported" means. - let hive = agent_status + agent_status .view(std::slice::from_ref(&agent), SystemTime::now()) .await .map_err(|e| { @@ -117,9 +116,9 @@ pub(crate) async fn stream_agent_state( ) })?; - let subscriber = crate::term_stream::subscribe_all( + let subscriber = crate::term_stream::subscribe( &client, - &crate::term_stream::agent_subjects(SUBJECT_PREFIX, &hive, &agent), + crate::term_stream::agent_subject(SUBJECT_PREFIX, &agent), "state", ) .await?; diff --git a/swarm-controller/src/subagent_term.rs b/swarm-controller/src/subagent_term.rs index 3372d636..61722460 100644 --- a/swarm-controller/src/subagent_term.rs +++ b/swarm-controller/src/subagent_term.rs @@ -11,9 +11,8 @@ //! Each subagent carries the stream's own publish time of its newest row, so //! a reader can tell a subagent that is running from one that has gone quiet. //! -//! **No hive lookup.** Unlike [`crate::term_stream`], there is no -//! hive-scoped subject to follow: only the agent's own credential is granted -//! these subjects. +//! **No agent-status lookup.** Unlike [`crate::term_stream`], an agent that +//! has never reported gets an empty stream, not a 404. //! //! **Live tail only, output only.** The relay is a core subscription, the //! same as the agent's own terminal route, so a reader sees rows published @@ -221,9 +220,9 @@ pub(crate) async fn stream_subagent_term( let agent = ident(&agent).map_err(bad_request)?; let subagent = ident(&subagent).map_err(bad_request)?; let client = connected_client(&state).map_err(|e| unavailable(&e))?; - let subscriber = crate::term_stream::subscribe_all( + let subscriber = crate::term_stream::subscribe( &client, - &[subagent_term::subject(&agent, &subagent)], + subagent_term::subject(&agent, &subagent), "subagent term", ) .await?; diff --git a/swarm-controller/src/term_stream.rs b/swarm-controller/src/term_stream.rs index 3cd6a878..9c72de1d 100644 --- a/swarm-controller/src/term_stream.rs +++ b/swarm-controller/src/term_stream.rs @@ -3,15 +3,13 @@ //! //! **No `hive` in the path, on purpose.** An agent is not pinned to a //! hive forever — it can move — so a URL naming one would go stale the -//! moment it did. The hive is resolved at request time from -//! `crate::agent_status`'s bucket instead, the same "an agent's hive -//! comes from its own last report" rule that module already documents. +//! moment it did. `crate::agent_status`'s bucket is read only to answer +//! 404 for an agent that has never reported. //! //! **Live tail only, on purpose.** `hive-agent` publishes each //! already-classified `TermMsg` row to the core subject -//! `$SWARM.term.{agent}` (`$SWARM.term.{hive}.{agent}` from an agent on its -//! hive's shared credential; see `hive-agent::swarm_term`'s module -//! doc) — a core subject, not `JetStream`, so a subscriber who was not +//! `$SWARM.term.{agent}` (see `hive-agent::swarm_term`'s module doc) — +//! a core subject, not `JetStream`, so a subscriber who was not //! listening missed the row, same as on the agent's own local SSE //! stream. This handler relays exactly that: no replay, no last-N //! history. The publish side's own retention half was deliberately @@ -35,8 +33,8 @@ 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_subjects`] spells the full -/// subjects; there is no way to check the two sides agree short of +/// 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"; @@ -84,7 +82,7 @@ pub(crate) async fn stream_agent_term( // One-agent "roster": reuses `AgentStatusReader::view`'s already-tested // row logic rather than a second, narrower lookup that could disagree // with it about what "never reported" means. - let hive = agent_status + agent_status .view(std::slice::from_ref(&agent), SystemTime::now()) .await .map_err(|e| { @@ -117,63 +115,41 @@ pub(crate) async fn stream_agent_term( ) })?; - let subscriber = subscribe_all( - &client, - &agent_subjects(SUBJECT_PREFIX, &hive, &agent), - "term", - ) - .await?; + let subscriber = subscribe(&client, agent_subject(SUBJECT_PREFIX, &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 subjects one agent publishes on under `prefix`: its own -/// `.`, and `..`, which an agent still on -/// its hive's shared queue credential publishes to. -pub(crate) fn agent_subjects(prefix: &str, hive: &str, agent: &str) -> [String; 2] { - [ - format!("{prefix}.{agent}"), - format!("{prefix}.{hive}.{agent}"), - ] +/// The subject one agent publishes on under `prefix`: `.`. +pub(crate) fn agent_subject(prefix: &str, agent: &str) -> String { + format!("{prefix}.{agent}") } -/// Subscribe to every one of `subjects`, merged into one stream. `what` names -/// the relay in logs and errors. -pub(crate) async fn subscribe_all( +/// Subscribe to `subject`. `what` names the relay in logs and errors. +pub(crate) async fn subscribe( client: &async_nats::Client, - subjects: &[String], + subject: String, what: &str, -) -> Result, problem_details::ProblemDetails> -{ - let mut subscribers = Vec::with_capacity(subjects.len()); - for subject in subjects { - let subscriber = client.subscribe(subject.clone()).await.map_err(|e| { - tracing::warn!(%subject, error = %e, "{what} stream: subscribe failed"); - crate::error_problem( - axum::http::StatusCode::INTERNAL_SERVER_ERROR, - &format!("subscribing to {subject} failed: {e}"), - ) - })?; - subscribers.push(subscriber); - } - tracing::info!(?subjects, "{what} stream: client attached"); - Ok(futures_util::stream::select_all(subscribers)) +) -> Result { + let subscriber = client.subscribe(subject.clone()).await.map_err(|e| { + tracing::warn!(%subject, error = %e, "{what} stream: subscribe failed"); + crate::error_problem( + axum::http::StatusCode::INTERNAL_SERVER_ERROR, + &format!("subscribing to {subject} failed: {e}"), + ) + })?; + tracing::info!(%subject, "{what} stream: client attached"); + Ok(subscriber) } #[cfg(test)] mod tests { use super::*; - /// Both shapes, so an agent on either credential is streamed. + /// The subject the agent's own queue credential is granted. #[test] - fn a_relay_subscribes_to_the_agents_own_subject_and_its_hive_scoped_one() { - assert_eq!( - agent_subjects(SUBJECT_PREFIX, "h1", "atlas"), - [ - "$SWARM.term.atlas".to_owned(), - "$SWARM.term.h1.atlas".to_owned(), - ] - ); + fn a_relay_subscribes_to_the_agents_own_subject() { + assert_eq!(agent_subject(SUBJECT_PREFIX, "atlas"), "$SWARM.term.atlas"); } }