From c32906b8936274ef5a05e9c3336515b32919f219 Mon Sep 17 00:00:00 2001 From: atlas Date: Sun, 18 Oct 2026 17:27:32 +0200 Subject: [PATCH] the swarm controller streams every inter-agent message from the agents' terminal rows --- docs/swarm/ui.md | 12 + hive-agent/src/term_msg.rs | 4 +- hive-sh4re/src/stream_enrich.rs | 33 +- hive-sh4re/src/term_msg.rs | 35 +++ swarm-controller-api/src/lib.rs | 1 + swarm-controller-api/src/messages.rs | 20 ++ swarm-controller/src/agent_input.rs | 2 +- swarm-controller/src/main.rs | 1 + swarm-controller/src/message_stream.rs | 411 +++++++++++++++++++++++++ swarm-controller/src/startup.rs | 5 +- swarm-controller/src/term_stream.rs | 8 +- 11 files changed, 520 insertions(+), 12 deletions(-) create mode 100644 swarm-controller-api/src/messages.rs create mode 100644 swarm-controller/src/message_stream.rs diff --git a/docs/swarm/ui.md b/docs/swarm/ui.md index bce82c8d..8cdacf6e 100644 --- a/docs/swarm/ui.md +++ b/docs/swarm/ui.md @@ -233,6 +233,18 @@ whole request and sends nothing when `body` or `agents` is empty, when `agents` names an agent twice, names more than 64, or names one that `GET /api/agents` doesn't list. +`GET /api/messages/stream` is an SSE feed of the messages between agents, and +from senders that are no agent (the operator, `reminder`, `compact`) to an +agent. It reads the terminal rows on `$SWARM.term.` and has no hive in +the path. Each event's `data` is `{ts, from, to, body, body_format?}`. An +agent's `send` row gives one event from that agent to the recipient in the +row's summary. A turn started by a non-agent sender gives one event to the +agent whose terminal holds the row. A turn started by an agent gives none, +since that agent's `send` row already did. The query `since_minutes` (default +60, at most 1440, 0 for live only) sends recent history first, oldest first, +read from each agent's newest 100 rows. A message an agent drained with an +in-turn `recv` starts no turn and so is missing. + A command has its own subject, so the harness never mistakes typed text for one. `POST /api/agents//cancel` sends `"cancel"` as a NATS request on `$SWARM.control.`, and only swarm-controller may publish there. The diff --git a/hive-agent/src/term_msg.rs b/hive-agent/src/term_msg.rs index caeddcc8..55553c95 100644 --- a/hive-agent/src/term_msg.rs +++ b/hive-agent/src/term_msg.rs @@ -11,6 +11,7 @@ //! both a text block and a `tool_use` block produces two rows. pub use hive_sh4re::term_msg::{ClassifyCtx, Level, TermMsg, iso8601_utc}; +use hive_sh4re::term_msg::{TURN_ICON, TURN_SUMMARY_PREFIX}; use crate::events::LiveEvent; @@ -33,7 +34,8 @@ fn classify_rows(ev: &LiveEvent, ctx: &mut ClassifyCtx) -> Vec { LiveEvent::TurnStart { from, body, .. } => { // `unread` (the third field) was dropped โ€” mara: "i can see // the todo/inbox count in toolbar and dont need the N unread". - let mut m = TermMsg::new(Level::Info, format!("TURN โ† {from}")).icon("โ—†"); + let mut m = + TermMsg::new(Level::Info, format!("{TURN_SUMMARY_PREFIX}{from}")).icon(TURN_ICON); if !body.trim().is_empty() { m = m.body(body.clone(), None); } diff --git a/hive-sh4re/src/stream_enrich.rs b/hive-sh4re/src/stream_enrich.rs index 0cb709e3..67ac545d 100644 --- a/hive-sh4re/src/stream_enrich.rs +++ b/hive-sh4re/src/stream_enrich.rs @@ -15,7 +15,7 @@ //! `mcp__hyperhive__send`'s row say, which tools get an expandable body) //! didn't change; only the shape it gets packed into did. -use crate::term_msg::{BodyFormat, ClassifyCtx, Level, TermMsg}; +use crate::term_msg::{BodyFormat, ClassifyCtx, Level, SEND_ICON, SEND_SUMMARY_PREFIX, TermMsg}; use serde_json::{Value, json}; /// Classify one raw claude stream-json line into zero or more terminal rows. @@ -490,7 +490,7 @@ fn rich_tool_body(name: &str, input: &Value) -> Option<(String, &'static str)> { fn tool_icon(name: &str) -> &'static str { // Exact-match table first, then prefix/contains fallbacks. match name { - "mcp__hyperhive__send" => "๐Ÿ“ค", + "mcp__hyperhive__send" => SEND_ICON, "mcp__hyperhive__recv" => "๐Ÿ“ฅ", "mcp__hyperhive__remind" => "โฐ", "mcp__hyperhive__set_status" => "๐Ÿท๏ธ", @@ -610,9 +610,9 @@ fn fmt_hyperhive_message_tool(name: &str, short: &str, input: &Value) -> String let to = sv(input, "to"); let lines = sv(input, "body").lines().count(); if lines > 1 { - format!("{short} โ†’ {to} ยท {lines}L") + format!("{SEND_SUMMARY_PREFIX}{to} ยท {lines}L") } else { - format!("{short} โ†’ {to}") + format!("{SEND_SUMMARY_PREFIX}{to}") } } _ => fmt_args_generic(short, input), @@ -1331,4 +1331,29 @@ mod tests { assert_eq!(fmt_room("!ร„ร–รœ:server"), "!ร„ร–รœ"); assert_eq!(fmt_room("!ร„ร–รœ"), "!ร„ร–รœ"); } + + /// The send row the classifier builds is the shape `send_recipient` + /// reads back, with and without the line-count suffix. + #[test] + fn a_send_row_names_its_recipient_through_the_shared_parser() { + for (body, lines) in [("hi", false), ("one\ntwo\nthree", true)] { + let row = one(&json!({ + "type": "assistant", + "message": { "content": [{ + "type": "tool_use", "id": "s1", "name": "mcp__hyperhive__send", + "input": { "to": "iris", "body": body }, + }]}, + })); + assert_eq!(row.icon.as_deref(), Some(SEND_ICON)); + assert_eq!(row.summary.contains(" ยท 3L"), lines, "{}", row.summary); + assert_eq!(crate::term_msg::send_recipient(&row.summary), Some("iris")); + } + assert_eq!(crate::term_msg::send_recipient("send* โ†’"), None); + assert_eq!(crate::term_msg::send_recipient("Read(x)"), None); + // A suffix lookalike in the name is not stripped. + assert_eq!( + crate::term_msg::send_recipient("send* โ†’ a ยท xL"), + Some("a ยท xL") + ); + } } diff --git a/hive-sh4re/src/term_msg.rs b/hive-sh4re/src/term_msg.rs index d896934c..429e45be 100644 --- a/hive-sh4re/src/term_msg.rs +++ b/hive-sh4re/src/term_msg.rs @@ -128,6 +128,41 @@ pub fn iso8601_utc(unix_seconds: i64) -> String { .to_rfc3339_opts(SecondsFormat::Secs, true) } +/// Icon of the row an agent's `send` tool call makes. +pub const SEND_ICON: &str = "๐Ÿ“ค"; + +/// What a `send` row's summary starts with; the recipient follows. A +/// multi-line body appends ` ยท L`, which [`send_recipient`] drops. +pub const SEND_SUMMARY_PREFIX: &str = "send* โ†’ "; + +/// Icon of the row a turn started by a message makes. +pub const TURN_ICON: &str = "โ—†"; + +/// What a turn-start row's summary starts with; the sender follows. +pub const TURN_SUMMARY_PREFIX: &str = "TURN โ† "; + +/// The recipient a `send` row's summary names, without the line-count suffix. +#[must_use] +pub fn send_recipient(summary: &str) -> Option<&str> { + let rest = summary.strip_prefix(SEND_SUMMARY_PREFIX)?; + Some(match rest.rsplit_once(" ยท ") { + Some((to, lines)) + if lines + .strip_suffix('L') + .is_some_and(|n| !n.is_empty() && n.bytes().all(|b| b.is_ascii_digit())) => + { + to + } + _ => rest, + }) +} + +/// The sender a turn-start row's summary names. +#[must_use] +pub fn turn_sender(summary: &str) -> Option<&str> { + summary.strip_prefix(TURN_SUMMARY_PREFIX) +} + /// Per-connection/per-request classification state. A live SSE stream keeps /// one of these alive for the connection's lifetime โ€” `tool_use` id โ†’ name /// correlation, so a `tool_result` can tell it's answering a `recv` call and diff --git a/swarm-controller-api/src/lib.rs b/swarm-controller-api/src/lib.rs index fc9c47d1..5c515656 100644 --- a/swarm-controller-api/src/lib.rs +++ b/swarm-controller-api/src/lib.rs @@ -7,4 +7,5 @@ pub mod agents; pub mod agents_md; pub mod forge; +pub mod messages; pub mod schedules; diff --git a/swarm-controller-api/src/messages.rs b/swarm-controller-api/src/messages.rs new file mode 100644 index 00000000..56661b37 --- /dev/null +++ b/swarm-controller-api/src/messages.rs @@ -0,0 +1,20 @@ +//! `GET /api/messages/stream`: the swarm-wide message feed. + +use serde::{Deserialize, Serialize}; + +/// One message between two parties, as one event of `GET /api/messages/stream`. +#[derive(Debug, Clone, PartialEq, Eq, Hash, Serialize, Deserialize)] +#[cfg_attr(feature = "schema", derive(utoipa::ToSchema))] +pub struct SwarmMessage { + /// When the sender's terminal row was stamped, ISO 8601 / RFC 3339 UTC. + pub ts: String, + /// An agent, or a non-agent sender such as `reminder`, `compact` or the + /// operator's name. + pub from: String, + /// The receiving agent. + pub to: String, + pub body: String, + /// `markdown` or `diff`; absent for plain text. + #[serde(default, skip_serializing_if = "Option::is_none")] + pub body_format: Option, +} diff --git a/swarm-controller/src/agent_input.rs b/swarm-controller/src/agent_input.rs index aeaf8f7c..b371c428 100644 --- a/swarm-controller/src/agent_input.rs +++ b/swarm-controller/src/agent_input.rs @@ -205,7 +205,7 @@ fn queue(state: &AppState) -> Result { } /// The swarm roster, or the 503 `GET /api/agents` answers when it has none. -async fn roster(state: &AppState) -> Result, ProblemDetails> { +pub(crate) async fn roster(state: &AppState) -> Result, ProblemDetails> { let Some(bridge) = state.auth.as_ref() else { return Err(error_problem( StatusCode::SERVICE_UNAVAILABLE, diff --git a/swarm-controller/src/main.rs b/swarm-controller/src/main.rs index 1e927f79..545bf774 100644 --- a/swarm-controller/src/main.rs +++ b/swarm-controller/src/main.rs @@ -48,6 +48,7 @@ mod issue_report; mod jobq; mod linked_accounts; mod matrix_account; +mod message_stream; mod otel_http_client; mod pulls; mod queue_identity; diff --git a/swarm-controller/src/message_stream.rs b/swarm-controller/src/message_stream.rs new file mode 100644 index 00000000..c130686b --- /dev/null +++ b/swarm-controller/src/message_stream.rs @@ -0,0 +1,411 @@ +//! `GET /api/messages/stream` โ€” an SSE feed of every message between agents, +//! and from non-agent senders to agents, read from the agents' terminal rows +//! on the swarm queue. No hive is involved. +//! +//! **Two kinds of row make an event.** An agent's `send` call leaves a +//! [`SEND_ICON`] row on its own terminal; the sender is that terminal's +//! agent, the recipient is in the summary. A turn started by a message leaves +//! a [`TURN_ICON`] row on the receiver's terminal; it makes an event only when +//! the sender is not in the roster (`reminder`, `compact`, the operator), since +//! an agent sender's `send` row already made one. +//! +//! **Not every message appears.** Messages an agent drains with an in-turn +//! `recv` start no turn and leave no `TURN โ†` row, and a `send` row exists +//! only while the sender's terminal publisher was up. A sender added to the +//! roster after the connection opened counts as a non-agent until the next +//! connection, so its messages show twice. +//! +//! **History, then live, with no gap.** The live subscription on +//! `$SWARM.term.*` (one token, so subagent subjects are excluded) opens first. +//! Then each roster agent's stream gives its newest +//! [`agent_term::REPLAY_ROWS`] rows, which are filtered by `since_minutes`, +//! merged by `ts` and sent; a live event equal to a history event is dropped. +//! History is bounded by those 100 rows per agent, not only by `since_minutes`. + +use std::collections::HashSet; +use std::convert::Infallible; +use std::sync::Arc; +use std::time::{Duration, SystemTime, UNIX_EPOCH}; + +use axum::extract::rejection::QueryRejection; +use axum::extract::{Query, State}; +use axum::http::StatusCode; +use axum::response::sse::{Event, KeepAlive, Sse}; +use futures_util::{Stream, StreamExt as _, stream}; +use hive_sh4re::term_msg::{SEND_ICON, TURN_ICON, iso8601_utc, send_recipient, turn_sender}; +use serde::Deserialize; +use swarm_controller_api::messages::SwarmMessage; +use swarm_queue_client::agent_term; +use utoipa::IntoParams; + +use crate::routes::{AppState, error_problem}; + +/// `since_minutes` when the query names none. +const DEFAULT_SINCE_MINUTES: u64 = 60; + +/// The largest `since_minutes`: the stream's own retention. +const MAX_SINCE_MINUTES: u64 = 24 * 60; + +/// How long a stream that has delivered its rows stays quiet before the +/// replay counts as caught up. +const HISTORY_IDLE: Duration = Duration::from_millis(500); + +/// How many agents' histories are read at once. +const HISTORY_CONCURRENCY: usize = 16; + +/// `stream_messages`'s query. +#[derive(Deserialize, IntoParams)] +pub struct MessagesQuery { + /// How many minutes of history to send before the live tail. Default + /// 60, at most 1440; 0 sends live messages only. + since_minutes: Option, +} + +#[utoipa::path( + get, + path = "/api/messages/stream", + params(MessagesQuery), + responses( + (status = 200, description = "server-sent event stream; each event's `data` is \ + one `SwarmMessage` as JSON: first the recent history, oldest first, then \ + live messages. Built from agents' terminal rows, so a message an agent \ + drained with an in-turn `recv` is missing", + body = SwarmMessage, content_type = "text/event-stream"), + (status = 400, description = "`since_minutes` is not a number (problem+json)", + body = String), + (status = 503, description = "no swarm queue or identity bridge is configured on \ + this host, the roster could not be read, or the queue could not be reached \ + (problem+json)", body = String), + (status = 500, description = "the subscribe failed (problem+json)", body = String), + ), + tag = "agents" +)] +pub(crate) async fn stream_messages( + State(state): State, + query: Result, QueryRejection>, +) -> Result>>, problem_details::ProblemDetails> { + let Query(query) = query.map_err(|e| error_problem(StatusCode::BAD_REQUEST, &e.body_text()))?; + let minutes = query + .since_minutes + .unwrap_or(DEFAULT_SINCE_MINUTES) + .min(MAX_SINCE_MINUTES); + let Some(status) = state.status.as_ref() else { + return Err(error_problem( + StatusCode::SERVICE_UNAVAILABLE, + "no swarm queue is configured on this host", + )); + }; + let roster = crate::agent_input::roster(&state).await?; + let client = status.queue_client(); + swarm_queue_client::ensure_connected(&client).map_err(|e| { + error_problem( + StatusCode::SERVICE_UNAVAILABLE, + &swarm_queue_client::chain(&e), + ) + })?; + + let subscriber = crate::term_stream::subscribe( + &client, + swarm_queue_client::term_subject_wildcard(), + "messages", + ) + .await?; + let roster: Arc> = Arc::new(roster.into_iter().collect()); + let history = history(&client, &roster, minutes).await; + let seen: HashSet = history.iter().cloned().collect(); + let live = live(subscriber, roster).filter(move |m| std::future::ready(!seen.contains(m))); + let events = stream::iter(history) + .chain(live) + .map(|m| Ok(message_event(&m))); + Ok(Sse::new(events).keep_alive(KeepAlive::default())) +} + +fn message_event(message: &SwarmMessage) -> Event { + // A struct of strings always serializes. + Event::default().data(serde_json::to_string(message).unwrap_or_default()) +} + +/// The messages in the newest rows of each of `roster`'s streams from the last +/// `minutes`, oldest first. An agent whose stream cannot be read adds none. +async fn history( + client: &async_nats::Client, + roster: &Arc>, + minutes: u64, +) -> Vec { + if minutes == 0 { + return Vec::new(); + } + let now = SystemTime::now() + .duration_since(UNIX_EPOCH) + .map_or(0, |d| i64::try_from(d.as_secs()).unwrap_or(i64::MAX)); + let cutoff = iso8601_utc(now.saturating_sub(i64::try_from(minutes * 60).unwrap_or(i64::MAX))); + let mut messages: Vec = stream::iter(roster.iter().cloned().collect::>()) + .map(|agent| agent_history(client.clone(), agent, roster.clone())) + .buffer_unordered(HISTORY_CONCURRENCY) + .flat_map(stream::iter) + .filter(|m| std::future::ready(m.ts >= cutoff)) + .collect() + .await; + messages.sort_by(|a, b| a.ts.cmp(&b.ts)); + messages +} + +/// The messages in `agent`'s replayed rows, read until the stream goes quiet. +async fn agent_history( + client: async_nats::Client, + agent: String, + roster: Arc>, +) -> Vec { + let replayed = crate::term_stream::replay( + &client, + &agent_term::stream_name(&agent), + swarm_queue_client::term_subject(&agent), + None, + ) + .await; + let mut rows = match replayed { + Ok(rows) => rows, + Err(error) => { + tracing::info!(%agent, %error, "messages stream: no history for this agent"); + return Vec::new(); + } + }; + let mut messages = Vec::new(); + while let Ok(Some(row)) = tokio::time::timeout(HISTORY_IDLE, rows.next()).await { + messages.extend(message_of(&agent, &row.payload, |a| roster.contains(a))); + } + messages +} + +/// The messages in the rows arriving on `$SWARM.term.` subjects. +fn live( + subscriber: async_nats::Subscriber, + roster: Arc>, +) -> impl Stream { + subscriber.filter_map(move |msg| { + let message = agent_of(msg.subject.as_str()) + .and_then(|agent| message_of(agent, &msg.payload, |a| roster.contains(a))); + std::future::ready(message) + }) +} + +/// The agent a main terminal subject belongs to. A subagent's subject has +/// more tokens and names none. +fn agent_of(subject: &str) -> Option<&str> { + let agent = subject.strip_prefix(&swarm_queue_client::term_subject(""))?; + (!agent.is_empty() && !agent.contains('.')).then_some(agent) +} + +/// The fields of a terminal row this feed reads. +#[derive(Deserialize)] +struct RowView { + #[serde(default)] + ts: String, + icon: Option, + #[serde(default)] + summary: String, + body: Option, + body_format: Option, +} + +/// The message in `agent`'s terminal row `payload`, if it holds one. +/// `is_agent` says whether a turn's sender is an agent. +fn message_of( + agent: &str, + payload: &[u8], + is_agent: impl Fn(&str) -> bool, +) -> Option { + let row: RowView = serde_json::from_slice(payload).ok()?; + let (from, to) = match row.icon.as_deref()? { + SEND_ICON => ( + agent, + send_recipient(&row.summary).filter(|to| !to.is_empty())?, + ), + TURN_ICON => { + let sender = turn_sender(&row.summary).filter(|s| !s.is_empty() && !is_agent(s))?; + (sender, agent) + } + _ => return None, + }; + Some(SwarmMessage { + ts: row.ts, + from: from.to_owned(), + to: to.to_owned(), + body: row.body.unwrap_or_default(), + body_format: row.body_format, + }) +} + +#[cfg(test)] +mod tests { + use super::*; + + use serde_json::json; + + use crate::test_nats::TestNats; + + fn row(icon: &str, summary: &str, body: &str) -> bytes::Bytes { + json!({ + "ts": "2026-10-18T12:00:00Z", "icon": icon, "level": "info", + "summary": summary, "body": body, "body_format": "markdown", + }) + .to_string() + .into() + } + + fn roster() -> Arc> { + Arc::new(["atlas", "iris"].map(String::from).into()) + } + + async fn publish(client: &async_nats::Client, subject: String, payload: bytes::Bytes) { + client.publish(subject, payload).await.expect("publish"); + client.flush().await.expect("flush"); + } + + async fn next(messages: &mut (impl Stream + Unpin)) -> SwarmMessage { + tokio::time::timeout(Duration::from_secs(5), messages.next()) + .await + .expect("a message arrived") + .expect("the feed ended") + } + + async fn nothing_more(messages: &mut (impl Stream + Unpin)) { + if let Ok(Some(m)) = tokio::time::timeout(Duration::from_millis(500), messages.next()).await + { + panic!("an extra message: {m:?}"); + } + } + + /// The last row published is a `send`, so reaching it proves the rows + /// before it were read and dropped rather than not yet delivered. + #[tokio::test] + async fn live_rows_become_messages_and_the_rest_are_dropped() { + let nats = TestNats::start(); + let client = nats.client().await; + let subscriber = client + .subscribe(swarm_queue_client::term_subject_wildcard()) + .await + .expect("subscribe"); + let mut messages = Box::pin(live(subscriber, roster())); + let atlas = swarm_queue_client::term_subject("atlas"); + let iris = swarm_queue_client::term_subject("iris"); + + publish( + &client, + atlas.clone(), + row(SEND_ICON, "send* โ†’ iris ยท 3L", "hello"), + ) + .await; + let m = next(&mut messages).await; + assert_eq!( + (m.from.as_str(), m.to.as_str(), m.body.as_str()), + ("atlas", "iris", "hello") + ); + assert_eq!(m.ts, "2026-10-18T12:00:00Z"); + assert_eq!(m.body_format.as_deref(), Some("markdown")); + + // A reminder's turn is a message from `reminder` to the agent. + publish( + &client, + iris.clone(), + row(TURN_ICON, "TURN โ† reminder", "stand up"), + ) + .await; + let m = next(&mut messages).await; + assert_eq!( + (m.from.as_str(), m.to.as_str(), m.body.as_str()), + ("reminder", "iris", "stand up") + ); + + // A turn from an agent, a subagent subject, and other rows: no event. + publish( + &client, + iris.clone(), + row(TURN_ICON, "TURN โ† atlas", "hello"), + ) + .await; + publish( + &client, + swarm_queue_client::subagent_term::subject("atlas", "scout"), + row(SEND_ICON, "send* โ†’ iris", "from a subagent"), + ) + .await; + publish(&client, iris.clone(), row("๐Ÿ”ง", "Bash", "ls")).await; + publish(&client, iris, bytes::Bytes::from_static(b"not json")).await; + publish(&client, atlas, row(SEND_ICON, "send* โ†’ iris", "last")).await; + assert_eq!(next(&mut messages).await.body, "last"); + nothing_more(&mut messages).await; + } + + #[tokio::test] + async fn history_holds_only_rows_since_the_cutoff_oldest_first() { + let nats = TestNats::start(); + let client = nats.client().await; + for agent in ["atlas", "iris"] { + agent_term::open_or_create(&client, agent) + .await + .expect("the stream"); + } + let stamped = |ts: &str, summary: &str, body: &str| -> bytes::Bytes { + json!({"ts": ts, "icon": SEND_ICON, "level": "info", "summary": summary, "body": body}) + .to_string() + .into() + }; + let secs = SystemTime::now() + .duration_since(UNIX_EPOCH) + .expect("after the epoch") + .as_secs(); + let now = iso8601_utc(i64::try_from(secs).expect("fits")); + let earlier = iso8601_utc(i64::try_from(secs - 600).expect("fits")); + let atlas = swarm_queue_client::term_subject("atlas"); + let iris = swarm_queue_client::term_subject("iris"); + publish(&client, atlas, stamped(&now, "send* โ†’ iris", "newer")).await; + publish( + &client, + iris.clone(), + stamped(&earlier, "send* โ†’ atlas", "older"), + ) + .await; + publish( + &client, + iris, + stamped("2000-01-01T00:00:00Z", "send* โ†’ atlas", "ancient"), + ) + .await; + let js = async_nats::jetstream::new(client.clone()); + for (agent, want) in [("atlas", 1), ("iris", 2)] { + for _ in 0..100 { + let mut s = js + .get_stream(agent_term::stream_name(agent)) + .await + .expect("the stream"); + if s.info().await.expect("info").state.messages >= want { + break; + } + tokio::time::sleep(Duration::from_millis(50)).await; + } + } + let got = history(&client, &roster(), 60).await; + assert_eq!( + got.iter().map(|m| m.body.as_str()).collect::>(), + ["older", "newer"] + ); + // No window reads no history. + assert!(history(&client, &roster(), 0).await.is_empty()); + } + + #[test] + fn a_row_without_a_party_makes_no_message() { + let none = |_: &str| false; + assert!(message_of("atlas", &row(SEND_ICON, "send* โ†’ ", "x"), none).is_none()); + assert!(message_of("atlas", &row(TURN_ICON, "TURN โ† ", "x"), none).is_none()); + assert!(message_of("atlas", &row(SEND_ICON, "send* โ†’ iris", "x"), none).is_some()); + } + + #[test] + fn only_main_terminal_subjects_name_an_agent() { + assert_eq!(agent_of("$SWARM.term.atlas"), Some("atlas")); + assert_eq!(agent_of("$SWARM.term.atlas.sub.scout"), None); + assert_eq!(agent_of("$SWARM.term."), None); + } +} diff --git a/swarm-controller/src/startup.rs b/swarm-controller/src/startup.rs index ec4d5e11..29a0b418 100644 --- a/swarm-controller/src/startup.rs +++ b/swarm-controller/src/startup.rs @@ -31,8 +31,8 @@ use crate::{ agent_control, agent_icon, agent_input, agent_limits, agent_login, agent_permissions, agent_renewal, agent_state_stream, agent_status, agents_md, alerts, archetypes, auth, builder, claude_login, config_pr, forge, forge_account, github_account, inference, issue_report, - linked_accounts, matrix_account, pulls, queue_identity, review_classifier, schedules, status, - subagent_term, term_stream, wanted, webhook, + linked_accounts, matrix_account, message_stream, pulls, queue_identity, review_classifier, + schedules, status, subagent_term, term_stream, wanted, webhook, }; /// Where the daemon binds, overridable via `SWARM_CONTROLLER_SOCKET`. @@ -602,6 +602,7 @@ pub(crate) fn build_app(state: AppState) -> axum::Router { .routes(routes!(linked_accounts::get_linked_accounts)) .routes(routes!(agents::get_hive_wanted)) .routes(routes!(term_stream::stream_agent_term)) + .routes(routes!(message_stream::stream_messages)) .routes(routes!(subagent_term::list_subagents)) .routes(routes!(subagent_term::stream_subagent_term)) .routes(routes!(agent_state_stream::stream_agent_state)) diff --git a/swarm-controller/src/term_stream.rs b/swarm-controller/src/term_stream.rs index 8a40aa9d..193cb3cc 100644 --- a/swarm-controller/src/term_stream.rs +++ b/swarm-controller/src/term_stream.rs @@ -153,10 +153,10 @@ pub(crate) async fn replay_or_live( } /// One row read from a terminal stream. -struct Row { +pub(crate) struct Row { /// Its stream sequence, when the delivery names one. - seq: Option, - payload: bytes::Bytes, + pub(crate) seq: Option, + pub(crate) payload: bytes::Bytes, } /// The rows on `subject` in the stream `stream_name`: every row after @@ -170,7 +170,7 @@ struct Row { /// subject's rows, so a fresh reader's consumer starts at the oldest row and /// the reader drops all but the newest [`agent_term::REPLAY_ROWS`] it counts /// as pending when the consumer is created. -async fn replay( +pub(crate) async fn replay( client: &async_nats::Client, stream_name: &str, subject: String,