vinzenz/hyperhive
vinzenz/hyperhive
Watch
0
0
Fork
You've already forked hyperhive
 0
Code Activity

swarm-controller: replay a subagent's newest rows before its live tail

Browse source 
A subagent's terminal opened directly showed nothing until the subagent
published again: the relay was a core subscription, though its rows sit
in term-sub-<agent> for 24 hours. It now goes through the agent route's
replay-then-tail, so it opens with the subagent's newest 100 rows and
resumes after Last-Event-ID.

- term_stream.rs: the replay takes a stream name and a filter subject
  instead of an agent, and `replay_or_live` (replay, else the live core
  subscription) is the one fn both routes call. When the filter is one of
  several subjects in the stream, a fresh reader's consumer starts at the
  oldest row and drops all but the newest REPLAY_ROWS it counts as
  pending, since the stream's newest sequences may belong to sibling
  subagents. A single-subject stream (term-main-<agent>) picks its start
  with replay_start exactly as before. Tests for both subagent cases.
- subagent_term.rs: the relay calls replay_or_live on the subagent's
  subject in term-sub-<agent> and reads Last-Event-ID; module doc and
  OpenAPI text no longer say "live only".
- main.rs: the handler test passes the new headers argument.
- swarm-queue-client agent_term.rs: replay_consumer_config takes the
  filter subject instead of the agent.
- swarm-queue-client subagent_term.rs: consumer_create_grant, the
  extended CONSUMER.CREATE form with a subagent subject as filter.
- swarm-nats-auth policy.rs: the controller holds that grant; the agent
  grant's `$SWARM.term.*` filter is two tokens short of a subagent
  subject, and the test that refused it now admits it.
- docs/swarm/ui.md, README.md: a subagent terminal opens with its newest
  rows.

Fixes #5527
This commit is contained in:
atlas 2026-10-18 02:11:02 +02:00 • committed by mara
commit d9c586ef94
8 changed files with 272 additions and 71 deletions
Download patch file Download diff file Expand all files Collapse all files

1
swarm-controller/src/main.rs
View file

@ -6560,6 +6560,7 @@ mod tests {
let Err(problem) = super::subagent_term::stream_subagent_term(
axum::extract::State(state.clone()),
axum::extract::Path(("iris".to_owned(), subagent.to_owned())),
axum::http::HeaderMap::new(),
)
.await
else {

24
swarm-controller/src/subagent_term.rs
View file

@ -1,5 +1,5 @@
//! An agent's subagents: the stream that keeps their rows, which ones the
//! swarm has rows for, and a live SSE relay of one subagent's terminal.
//! swarm has rows for, and an SSE relay of one subagent's terminal.
//!
//! The agent's subagent daemon publishes each subagent's classified rows on
//! `$SWARM.term.{agent}.sub.{subagent}` under the agent's own queue credential,
@ -15,9 +15,12 @@
//! **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, so a
//! reader sees rows published while it is attached; unlike the agent's own
//! terminal route, it replays nothing from the stream. Nothing is sent toward a subagent: subagents take no
//! **Replay, then the live tail, output only.** The relay is
//! [`crate::term_stream::replay_or_live`], the agent's own terminal route's,
//! on the subagent's subject in `term-sub-{agent}`: the subagent's newest
//! `agent_term::REPLAY_ROWS` rows, then each new one, with the same
//! `Last-Event-ID` resume and the same live-only fallback while the stream
//! cannot be read. Nothing is sent toward a subagent: subagents take no
//! input from the swarm.
use std::collections::BTreeSet;
@ -27,7 +30,7 @@ use std::time::Duration;
use axum::Json;
use axum::extract::{Path, State};
use axum::http::StatusCode;
use axum::http::{HeaderMap, StatusCode};
use axum::response::sse::{Event, KeepAlive, Sse};
use futures_util::{Stream, StreamExt as _, TryStreamExt as _};
use serde::Serialize;
@ -211,7 +214,9 @@ async fn last_row_unix(
responses(
(status = 200, description = "server-sent event stream; each event's `data` is \
one already-classified TermMsg row, JSON as the agent's subagent daemon \
published it (opaque to this daemon) — live only, no replay",
published it (opaque to this daemon); the subagent's newest retained rows \
first, then live rows. Each event's `id` is the row's stream sequence, and a \
`Last-Event-ID` request header resumes right after that row",
body = String, content_type = "text/event-stream"),
(status = 400, description = "the agent or subagent name is not shaped like an \
identifier (problem+json)", body = String),
@ -225,18 +230,19 @@ async fn last_row_unix(
pub(crate) async fn stream_subagent_term(
State(state): State<AppState>,
Path((agent, subagent)): Path<(String, String)>,
headers: HeaderMap,
) -> Result<Sse<impl Stream<Item = Result<Event, Infallible>>>, problem_details::ProblemDetails> {
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(
let stream = crate::term_stream::replay_or_live(
&client,
&subagent_term::stream_name(&agent),
subagent_term::subject(&agent, &subagent),
crate::term_stream::last_event_id(&headers),
"subagent 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()))
}

244
swarm-controller/src/term_stream.rs
View file

@ -19,7 +19,7 @@
//! **No stream, no replay.** While the stream does not exist yet (up to a
//! minute after the agent is declared) or cannot be read, the handler
//! falls back to a core subscription: rows published while the reader is
//! attached, and nothing before.
//! attached, and nothing before. `crate::subagent_term` shares both: [`replay_or_live`].
//!
//! **The payload is passed through opaquely**, the same way
//! `crate::status::HiveStatus::snapshot` stores a hive's status
@ -116,62 +116,112 @@ pub(crate) async fn stream_agent_term(
)
})?;
let stream = match replay(&client, &agent, last_event_id(&headers)).await {
Ok(rows) => rows.map(|row| Ok(row_event(&row))).boxed(),
Err(error) => {
tracing::warn!(%agent, %error, "term stream: no replay, live rows only");
let subscriber =
subscribe(&client, swarm_queue_client::term_subject(&agent), "term").await?;
subscriber
.map(|msg| Ok(Event::default().data(String::from_utf8_lossy(&msg.payload))))
.boxed()
}
};
let stream = replay_or_live(
&client,
&agent_term::stream_name(&agent),
swarm_queue_client::term_subject(&agent),
last_event_id(&headers),
"term",
)
.await?;
Ok(Sse::new(stream).keep_alive(KeepAlive::default()))
}
/// One row read from the agent's terminal stream.
/// The SSE events of a terminal relay: the rows [`replay`] reads from
/// `stream_name`, or, when the stream cannot be read, a core subscription
/// to `subject` that sends live rows only. `what` names the relay in logs
/// and errors.
pub(crate) async fn replay_or_live(
client: &async_nats::Client,
stream_name: &str,
subject: String,
resume_after: Option<u64>,
what: &str,
) -> Result<BoxStream<'static, Result<Event, Infallible>>, problem_details::ProblemDetails> {
match replay(client, stream_name, subject.clone(), resume_after).await {
Ok(rows) => Ok(rows.map(|row| Ok(row_event(&row))).boxed()),
Err(error) => {
tracing::warn!(%subject, %error, "{what} stream: no replay, live rows only");
let subscriber = subscribe(client, subject, what).await?;
Ok(subscriber
.map(|msg| Ok(Event::default().data(String::from_utf8_lossy(&msg.payload))))
.boxed())
}
}
}
/// One row read from a terminal stream.
struct Row {
/// Its stream sequence, when the delivery names one.
seq: Option<u64>,
payload: bytes::Bytes,
}
/// The agent's rows from its terminal stream: the replay
/// [`agent_term::replay_start`] picks, then every new row.
/// The rows on `subject` in the stream `stream_name`: every row after
/// `resume_after`, or for a fresh reader the newest
/// [`agent_term::REPLAY_ROWS`] of them, then every new row.
///
/// When `subject` is the stream's only subject, its sequences are the
/// subject's rows and [`agent_term::replay_start`] alone picks the start.
/// In a stream that holds several subjects, as a `term-sub-<agent>` holds
/// every subagent of the agent, the newest sequences may hold none of this
/// 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(
client: &async_nats::Client,
agent: &str,
stream_name: &str,
subject: String,
resume_after: Option<u64>,
) -> Result<BoxStream<'static, Row>, String> {
let js = async_nats::jetstream::new(client.clone());
let name = agent_term::stream_name(agent);
let stream = js
.get_stream(&name)
.get_stream(stream_name)
.await
.map_err(|e| format!("opening {name}: {}", swarm_queue_client::chain(&e)))?;
let start = agent_term::replay_start(stream.cached_info().state.last_sequence, resume_after);
.map_err(|e| format!("opening {stream_name}: {}", swarm_queue_client::chain(&e)))?;
let info = stream.cached_info();
let last_sequence = info.state.last_sequence;
let shared = info.config.subjects != [subject.as_str()];
// The case `replay_start` resumes in: a sequence the stream has held.
let resumed = resume_after.is_some_and(|seen| seen <= last_sequence);
let trim = shared && !resumed;
let start = if trim {
1
} else {
agent_term::replay_start(last_sequence, resume_after)
};
let deliver_subject = client.new_inbox();
// The inbox's last token is a fresh nuid, which is also a valid name.
let consumer_name = format!(
"term-replay-{}",
deliver_subject.rsplit('.').next().unwrap_or_default()
);
let config = agent_term::replay_consumer_config(agent, consumer_name, deliver_subject, start);
let messages = stream
.create_consumer(config)
.await
.map_err(|e| {
format!(
"creating a consumer on {name}: {}",
swarm_queue_client::chain(&e)
)
})?
.messages()
.await
.map_err(|e| format!("subscribing to {name}: {}", swarm_queue_client::chain(&e)))?;
tracing::info!(stream = %name, start, "term stream: client attached");
Ok(rows_until_error(messages, agent.to_owned()))
let config =
agent_term::replay_consumer_config(subject.clone(), consumer_name, deliver_subject, start);
let consumer = stream.create_consumer(config).await.map_err(|e| {
format!(
"creating a consumer on {stream_name}: {}",
swarm_queue_client::chain(&e)
)
})?;
let skip = if trim {
consumer
.cached_info()
.num_pending
.saturating_sub(agent_term::REPLAY_ROWS)
} else {
0
};
let messages = consumer.messages().await.map_err(|e| {
format!(
"subscribing to {stream_name}: {}",
swarm_queue_client::chain(&e)
)
})?;
tracing::info!(stream = %stream_name, %subject, start, skip, "term stream: client attached");
Ok(rows_until_error(messages, subject)
.skip(usize::try_from(skip).unwrap_or(usize::MAX))
.boxed())
}
/// The rows of a consumer's deliveries, ending at the first error. A missed
@ -179,7 +229,7 @@ async fn replay(
/// the browser reconnects with `Last-Event-ID` and gets a fresh consumer.
fn rows_until_error<E: std::fmt::Display>(
messages: impl Stream<Item = Result<async_nats::jetstream::Message, E>> + Send + 'static,
agent: String,
subject: String,
) -> BoxStream<'static, Row> {
messages
.scan((), move |(), msg| {
@ -189,7 +239,7 @@ fn rows_until_error<E: std::fmt::Display>(
payload: msg.message.payload,
}),
Err(e) => {
tracing::warn!(%agent, error = %e, "term stream: consumer failed, ending the stream");
tracing::warn!(%subject, error = %e, "term stream: consumer failed, ending the stream");
None
}
};
@ -209,7 +259,7 @@ fn row_event(row: &Row) -> Event {
/// The stream sequence a reconnecting browser last received, from its
/// `Last-Event-ID` header. Absent or unparseable means a fresh reader.
fn last_event_id(headers: &HeaderMap) -> Option<u64> {
pub(crate) fn last_event_id(headers: &HeaderMap) -> Option<u64> {
headers
.get("last-event-id")?
.to_str()
@ -259,20 +309,72 @@ mod tests {
client.flush().await.expect("flush");
}
/// Wait until the stream holds row `last`, so the next step sees it.
/// Wait until the agent's stream holds row `last`, so the next step sees it.
async fn stored(client: &async_nats::Client, last: u64) {
stored_in(client, &agent_term::stream_name(AGENT), last).await;
}
async fn stored_in(client: &async_nats::Client, stream_name: &str, last: u64) {
let js = async_nats::jetstream::new(client.clone());
for _ in 0..100 {
let mut stream = js
.get_stream(agent_term::stream_name(AGENT))
.await
.expect("the stream");
let mut stream = js.get_stream(stream_name).await.expect("the stream");
if stream.info().await.expect("info").state.last_sequence >= last {
return;
}
tokio::time::sleep(Duration::from_millis(50)).await;
}
panic!("the stream never stored row {last}");
panic!("{stream_name} never stored row {last}");
}
async fn replay_agent(
client: &async_nats::Client,
resume_after: Option<u64>,
) -> Result<BoxStream<'static, Row>, String> {
replay(
client,
&agent_term::stream_name(AGENT),
swarm_queue_client::term_subject(AGENT),
resume_after,
)
.await
}
/// Row `i` of subagents `scout` and `builder` in turn, so `scout`'s row
/// `i` is stream sequence `2i - 1` and the newest sequences hold only
/// half as many of its rows.
async fn publish_subagent_rows(
client: &async_nats::Client,
rows: std::ops::RangeInclusive<u64>,
) {
for i in rows {
for subagent in ["scout", "builder"] {
client
.publish(
swarm_queue_client::subagent_term::subject(AGENT, subagent),
format!("{subagent}{i}").into(),
)
.await
.expect("publish");
}
}
client.flush().await.expect("flush");
}
async fn replay_scout(
client: &async_nats::Client,
resume_after: Option<u64>,
) -> Result<BoxStream<'static, Row>, String> {
replay(
client,
&swarm_queue_client::subagent_term::stream_name(AGENT),
swarm_queue_client::subagent_term::subject(AGENT, "scout"),
resume_after,
)
.await
}
fn scout_rows(range: std::ops::RangeInclusive<u64>) -> Vec<(u64, String)> {
range.map(|i| (2 * i - 1, format!("scout{i}"))).collect()
}
async fn take(rows: &mut BoxStream<'static, Row>, n: usize) -> Vec<(u64, String)> {
@ -314,7 +416,7 @@ mod tests {
.expect("the stream");
publish_rows(&client, 1..=3).await;
stored(&client, 3).await;
let mut replayed = replay(&client, AGENT, None).await.expect("a replay");
let mut replayed = replay_agent(&client, None).await.expect("a replay");
assert_eq!(take(&mut replayed, 3).await, rows(1..=3));
let consumers: Vec<String> = stream
.consumer_names()
@ -398,7 +500,7 @@ mod tests {
.expect("the stream");
publish_rows(&client, 1..=150).await;
stored(&client, 150).await;
let mut replayed = replay(&client, AGENT, None).await.expect("a replay");
let mut replayed = replay_agent(&client, None).await.expect("a replay");
publish_rows(&client, 151..=153).await;
assert_eq!(take(&mut replayed, 103).await, rows(51..=153));
nothing_more(&mut replayed).await;
@ -415,14 +517,14 @@ mod tests {
.expect("the stream");
publish_rows(&client, 1..=150).await;
stored(&client, 150).await;
let mut first = replay(&client, AGENT, None).await.expect("a replay");
let mut first = replay_agent(&client, None).await.expect("a replay");
let seen = take(&mut first, 100).await;
drop(first);
publish_rows(&client, 151..=160).await;
stored(&client, 160).await;
let last_id = seen.last().expect("rows were seen").0.to_string();
let resume_after = last_event_id(&with_last_event_id(&last_id));
let mut resumed = replay(&client, AGENT, resume_after)
let mut resumed = replay_agent(&client, resume_after)
.await
.expect("a resumed replay");
publish_rows(&client, 161..=163).await;
@ -430,6 +532,52 @@ mod tests {
nothing_more(&mut resumed).await;
}
/// A subagent's fresh reader gets its own newest rows, in order, then its
/// live rows, though its sibling's rows fill half the stream's newest
/// sequences.
#[tokio::test]
async fn a_subagent_replays_its_own_newest_rows_before_the_live_tail() {
let nats = TestNats::start();
let client = nats.client().await;
let stream_name = swarm_queue_client::subagent_term::stream_name(AGENT);
swarm_queue_client::subagent_term::open_or_create(&client, AGENT)
.await
.expect("the stream");
publish_subagent_rows(&client, 1..=150).await;
stored_in(&client, &stream_name, 300).await;
let mut replayed = replay_scout(&client, None).await.expect("a replay");
publish_subagent_rows(&client, 151..=153).await;
assert_eq!(take(&mut replayed, 103).await, scout_rows(51..=153));
nothing_more(&mut replayed).await;
}
/// A subagent's reader that reconnects with the last id it was sent gets
/// every later row of that subagent once.
#[tokio::test]
async fn a_subagent_reconnect_gets_every_later_row_once() {
let nats = TestNats::start();
let client = nats.client().await;
let stream_name = swarm_queue_client::subagent_term::stream_name(AGENT);
swarm_queue_client::subagent_term::open_or_create(&client, AGENT)
.await
.expect("the stream");
publish_subagent_rows(&client, 1..=150).await;
stored_in(&client, &stream_name, 300).await;
let mut first = replay_scout(&client, None).await.expect("a replay");
let seen = take(&mut first, 100).await;
drop(first);
publish_subagent_rows(&client, 151..=160).await;
stored_in(&client, &stream_name, 320).await;
let last_id = seen.last().expect("rows were seen").0.to_string();
let resume_after = last_event_id(&with_last_event_id(&last_id));
let mut resumed = replay_scout(&client, resume_after)
.await
.expect("a resumed replay");
publish_subagent_rows(&client, 161..=163).await;
assert_eq!(take(&mut resumed, 13).await, scout_rows(151..=163));
nothing_more(&mut resumed).await;
}
fn with_last_event_id(value: &str) -> HeaderMap {
let mut headers = HeaderMap::new();
headers.insert(