the swarm controller streams every inter-agent message from the agents' terminal rows
This commit is contained in:
atlas
parent
87b1f94ac6
commit
c32906b893
11 changed files with 520 additions and 12 deletions
|
|
@ -205,7 +205,7 @@ fn queue(state: &AppState) -> Result<async_nats::Client, String> {
|
|||
}
|
||||
|
||||
/// The swarm roster, or the 503 `GET /api/agents` answers when it has none.
|
||||
async fn roster(state: &AppState) -> Result<Vec<String>, ProblemDetails> {
|
||||
pub(crate) async fn roster(state: &AppState) -> Result<Vec<String>, ProblemDetails> {
|
||||
let Some(bridge) = state.auth.as_ref() else {
|
||||
return Err(error_problem(
|
||||
StatusCode::SERVICE_UNAVAILABLE,
|
||||
|
|
|
|||
|
|
@ -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;
|
||||
|
|
|
|||
411
swarm-controller/src/message_stream.rs
Normal file
411
swarm-controller/src/message_stream.rs
Normal file
|
|
@ -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<u64>,
|
||||
}
|
||||
|
||||
#[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<AppState>,
|
||||
query: Result<Query<MessagesQuery>, QueryRejection>,
|
||||
) -> Result<Sse<impl Stream<Item = Result<Event, Infallible>>>, 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<HashSet<String>> = Arc::new(roster.into_iter().collect());
|
||||
let history = history(&client, &roster, minutes).await;
|
||||
let seen: HashSet<SwarmMessage> = 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<HashSet<String>>,
|
||||
minutes: u64,
|
||||
) -> Vec<SwarmMessage> {
|
||||
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<SwarmMessage> = stream::iter(roster.iter().cloned().collect::<Vec<_>>())
|
||||
.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<HashSet<String>>,
|
||||
) -> Vec<SwarmMessage> {
|
||||
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.<agent>` subjects.
|
||||
fn live(
|
||||
subscriber: async_nats::Subscriber,
|
||||
roster: Arc<HashSet<String>>,
|
||||
) -> impl Stream<Item = SwarmMessage> {
|
||||
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<String>,
|
||||
#[serde(default)]
|
||||
summary: String,
|
||||
body: Option<String>,
|
||||
body_format: Option<String>,
|
||||
}
|
||||
|
||||
/// 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<SwarmMessage> {
|
||||
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<HashSet<String>> {
|
||||
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<Item = SwarmMessage> + 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<Item = SwarmMessage> + 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::<Vec<_>>(),
|
||||
["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);
|
||||
}
|
||||
}
|
||||
|
|
@ -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))
|
||||
|
|
|
|||
|
|
@ -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<u64>,
|
||||
payload: bytes::Bytes,
|
||||
pub(crate) seq: Option<u64>,
|
||||
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,
|
||||
|
|
|
|||
Loading…
Reference in a new issue