diff --git a/docs/swarm/README.md b/docs/swarm/README.md index e7bac767..2ce2ca5b 100644 --- a/docs/swarm/README.md +++ b/docs/swarm/README.md @@ -189,9 +189,10 @@ own web UI renders also goes to `$SWARM.term.`, one subject per agent, so a swarm-level terminal can follow one agent without subscribing to the swarm's whole traffic. The queue grants that subject to the agent's own credential, and the swarm controller relays it. Publishing only: an -agent talks about itself here and reads nothing. Rows aren't retained — a -subscriber that wasn't listening missed them, the same as on the agent's own -live stream. +agent talks about itself here and reads nothing. The queue keeps the rows in +the stream `term-main-` for 24 hours or 64 MiB, the same limits and the +same creator as the subagent stream below, so a swarm terminal that attaches +later replays the newest of them. The two streams hold disjoint subjects. An agent's **subagents** publish their terminals too, from the agent's subagent daemon: each subagent's rows go to `$SWARM.term..sub.`, diff --git a/docs/swarm/ui.md b/docs/swarm/ui.md index 1fb70656..72cbad9d 100644 --- a/docs/swarm/ui.md +++ b/docs/swarm/ui.md @@ -30,8 +30,9 @@ terminal in the same preview, and **expand** opens it in its own tab, as it does for the agent. The list refreshes every 10 seconds, so a subagent the agent spawns shows up once it writes output. -A preview shows rows published while it's open. A subagent terminal has no -turn-state badges. +An agent's terminal opens with its last 100 rows from the past 24 hours, then +follows new ones; a reconnect picks up after the last row it showed. A subagent +terminal shows rows published while it's open, and has no turn-state badges. ### Scheduled prompts @@ -102,6 +103,21 @@ store identity, or no queue address, publishes no subagent terminals. +
Where an agent's terminal history comes from + +The harness publishes the agent's terminal rows on `$SWARM.term.`, into +a stream named `term-main-` that swarm-controller creates for every +declared agent, with the same 24 hours or 64 MiB as the subagent stream. That +stream holds this one subject, and the subagent stream holds only subjects +below `$SWARM.term..sub`, so they never overlap. +`GET /api/agents//term/stream` creates a short-lived consumer on it +that starts 100 rows back, so the SSE stream sends those rows and then each +new one. Each event's `id` is the row's stream sequence, and a request with +`Last-Event-ID` resumes after it. Until the stream exists, the route sends +live rows only. + +
+ ### Inference settings The agent's detail panel shows its model and effort under **inference**, each diff --git a/frontend/packages/swarm-ui/src/pages/agents/useSwarmTermStream.ts b/frontend/packages/swarm-ui/src/pages/agents/useSwarmTermStream.ts index e5a56d7f..0da6c3c4 100644 --- a/frontend/packages/swarm-ui/src/pages/agents/useSwarmTermStream.ts +++ b/frontend/packages/swarm-ui/src/pages/agents/useSwarmTermStream.ts @@ -3,9 +3,10 @@ // `term_stream.rs`). Deliberately NOT `@hive/agent`'s `useLiveStream` — // that hook's wire shape is `TermEnvelope` (`{seq, msgs: []}`) plus a // history/backfill dance; the swarm relay forwards one bare `TermMsg` -// per SSE event with no envelope and no history endpoint (a core NATS -// subject, not `JetStream` — see `term_stream.rs`'s own module doc), so -// there's nothing to buffer/dedupe/backfill against. Same row-coalescing +// per SSE event, its replayed rows first and in order on the same +// stream, so there's nothing to buffer/dedupe/backfill against. The +// browser's own `Last-Event-ID` resumes a reconnect, so rows are kept +// across one. Same row-coalescing // rule as `useLiveStream`'s `appendRow` (ported, not imported — one // small helper isn't worth a cross-package hop for). import { useEffect, useRef, useState } from "preact/hooks"; diff --git a/hive-agent/src/swarm_term.rs b/hive-agent/src/swarm_term.rs index 158628a3..f4a04f49 100644 --- a/hive-agent/src/swarm_term.rs +++ b/hive-agent/src/swarm_term.rs @@ -13,11 +13,11 @@ //! restarted its correlation state per reader would lose the `tool_use` → //! name mapping a `tool_result` needs. //! -//! **A core subject, not `JetStream`.** A terminal row has no current value a -//! late reader can ask for, and nothing here is worth the durability of the -//! notices stream: a subscriber that was not listening missed the row, the -//! same way it would have missed it on the agent's own SSE stream. That also -//! keeps the agent's grant to a plain publish. +//! **A plain core publish.** The queue keeps the rows in the stream +//! `term-main-`, which swarm-controller creates +//! (`swarm_queue_client::agent_term`), so this needs no `JetStream` call and +//! the agent's grant stays a plain publish. A row published before that +//! stream exists reaches live subscribers only. //! //! **Best-effort per row.** The turn loop and the web UI never wait on the //! queue, so the bus receiver is the only thing this task blocks on; a failed diff --git a/swarm-controller/Cargo.toml b/swarm-controller/Cargo.toml index 9a876b5a..dde94f96 100644 --- a/swarm-controller/Cargo.toml +++ b/swarm-controller/Cargo.toml @@ -92,7 +92,13 @@ swarm-authelia-bridge-sock.workspace = true # # `subagent-term`: this daemon creates every agent's subagent stream, with the # config the crate spells. -swarm-queue-client = { workspace = true, features = ["kv", "subagent-term"] } +# `agent-term`: likewise every agent's terminal stream, plus the consumer it +# replays that stream through. +swarm-queue-client = { workspace = true, features = [ + "kv", + "subagent-term", + "agent-term", +] } # `matrix_account.rs` writes the credential this daemon's route accepts. Same # crate the hive reads it back with, which is the point: the path, the field # name and the object's shape are agreements between the two ends, and a diff --git a/swarm-controller/src/main.rs b/swarm-controller/src/main.rs index 6971c207..8d6af936 100644 --- a/swarm-controller/src/main.rs +++ b/swarm-controller/src/main.rs @@ -2621,8 +2621,8 @@ fn spawn_agent_renewal( }); } -/// Start creating every live agent's subagent stream when a swarm queue is -/// wired up. Lifted out of `main` for `clippy::too_many_lines`. +/// Start creating every live agent's subagent and terminal streams when a swarm +/// queue is wired up. Lifted out of `main` for `clippy::too_many_lines`. fn spawn_subagent_streams(status: Option<&Arc>, hives: &[HiveEntry]) { let (Some(status), Some(wanted)) = (status, wanted_writer(status)) else { return; @@ -2631,7 +2631,7 @@ fn spawn_subagent_streams(status: Option<&Arc>, hives: &[H subagent_term::spawn(status.queue_client(), wanted, hives); } -/// The per-agent background passes: subagent streams and schedule keys. +/// The per-agent background passes: terminal streams and schedule keys. /// Lifted out of `main` for `clippy::too_many_lines`. fn spawn_per_agent_upkeep(status: Option<&Arc>, hives: &[HiveEntry]) { spawn_subagent_streams(status, hives); diff --git a/swarm-controller/src/subagent_term.rs b/swarm-controller/src/subagent_term.rs index 61722460..307c304b 100644 --- a/swarm-controller/src/subagent_term.rs +++ b/swarm-controller/src/subagent_term.rs @@ -5,7 +5,8 @@ //! `$SWARM.term.{agent}.sub.{subagent}` under the agent's own queue credential, //! which may publish there and nothing else of these streams. [`spawn`] //! creates the stream `term-sub-{agent}` for every live agent, with the config -//! `swarm_queue_client::subagent_term` spells. Subagents are spawned on demand +//! `swarm_queue_client::subagent_term` spells, and in the same pass the +//! agent's own terminal stream (`swarm_queue_client::agent_term`). Subagents are spawned on demand //! under caller-chosen names, so the list is the set of subjects that stream //! holds rows for, read with `STREAM.INFO`. A missing stream lists nothing. //! Each subagent carries the stream's own publish time of its newest row, so @@ -30,7 +31,7 @@ use axum::http::StatusCode; use axum::response::sse::{Event, KeepAlive, Sse}; use futures_util::{Stream, StreamExt as _, TryStreamExt as _}; use serde::Serialize; -use swarm_queue_client::subagent_term; +use swarm_queue_client::{agent_term, subagent_term}; use utoipa::ToSchema; use crate::AppState; @@ -49,8 +50,9 @@ const LOOKUP_TIMEOUT: Duration = Duration::from_secs(2); /// How many newest-row lookups [`list_subagents`] has in flight at once. const LOOKUP_CONCURRENCY: usize = 16; -/// Create the subagent stream of every live agent that has none, now and every -/// [`ENSURE_INTERVAL`] after. An existing stream is left as it is. +/// Create the subagent stream and the terminal stream (`agent_term`) of every +/// live agent that lacks one, now and every [`ENSURE_INTERVAL`] after. An +/// existing stream is left as it is. /// /// The live agents are the ones some hive's wanted-state declaration names in /// a state other than `Destroyed` (`agent_renewal::live_agents`). A pass skips @@ -63,7 +65,7 @@ pub(crate) fn spawn(client: async_nats::Client, wanted: Arc, hives loop { ticker.tick().await; if swarm_queue_client::ensure_connected(&client).is_err() { - tracing::debug!("subagent streams: queue not connected; next pass"); + tracing::debug!("terminal streams: queue not connected; next pass"); continue; } for agent in live_agents(&wanted, &hives).await { @@ -74,6 +76,13 @@ pub(crate) fn spawn(client: async_nats::Client, wanted: Arc, hives "subagent streams: creating the agent's stream failed; next pass" ); } + if let Err(e) = agent_term::open_or_create(&client, &agent).await { + tracing::warn!( + %agent, + error = %swarm_queue_client::chain(&e), + "terminal streams: creating the agent's stream failed; next pass" + ); + } } } }); diff --git a/swarm-controller/src/term_stream.rs b/swarm-controller/src/term_stream.rs index 820c59fd..47f63c06 100644 --- a/swarm-controller/src/term_stream.rs +++ b/swarm-controller/src/term_stream.rs @@ -1,4 +1,4 @@ -//! `GET /api/agents/{name}/term/stream` — a live SSE relay of one agent's +//! `GET /api/agents/{name}/term/stream` — an SSE relay of one agent's //! terminal rows, sourced straight from the swarm queue. //! //! **No `hive` in the path, on purpose.** An agent is not pinned to a @@ -6,15 +6,20 @@ //! 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}` (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 -//! never built, and this route's own job — a live tail — never needed -//! it either. +//! **Replay, then the live tail, through one consumer.** `hive-agent` +//! publishes each already-classified `TermMsg` row to the core subject +//! `$SWARM.term.{agent}`, and the stream `term-main-{agent}` +//! (`swarm_queue_client::agent_term`) keeps them. The handler creates an +//! ephemeral consumer on that stream starting at the newest +//! `agent_term::REPLAY_ROWS` rows, so the reader gets those and then every +//! new row, in stream order, with no gap and no duplicate between the two. +//! Each event's `id` is the row's stream sequence; a browser that +//! reconnects sends it back as `Last-Event-ID` and resumes right after it. +//! +//! **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. //! //! **The payload is passed through opaquely**, the same way //! `crate::status::HiveStatus::snapshot` stores a hive's status @@ -26,8 +31,11 @@ use std::convert::Infallible; use std::time::SystemTime; use axum::extract::{Path, State}; +use axum::http::HeaderMap; use axum::response::sse::{Event, KeepAlive, Sse}; +use futures_util::stream::BoxStream; use futures_util::{Stream, StreamExt as _}; +use swarm_queue_client::agent_term; use crate::AppState; @@ -38,7 +46,9 @@ use crate::AppState; responses( (status = 200, description = "server-sent event stream; each event's `data` is \ one already-classified TermMsg row, JSON as hive-agent published it \ - (opaque to this daemon) — live only, no replay", + (opaque to this daemon); the 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 name is not shaped like an identifier \ (problem+json)", body = String), @@ -55,6 +65,7 @@ use crate::AppState; pub(crate) async fn stream_agent_term( State(state): State, Path(agent): Path, + headers: HeaderMap, ) -> Result>>, problem_details::ProblemDetails> { let Some(status) = state.status.as_ref() else { return Err(crate::error_problem( @@ -105,12 +116,91 @@ pub(crate) async fn stream_agent_term( ) })?; - let subscriber = subscribe(&client, swarm_queue_client::term_subject(&agent), "term").await?; - let stream = - subscriber.map(|msg| Ok(Event::default().data(String::from_utf8_lossy(&msg.payload)))); + let stream = match replay(&client, &agent, last_event_id(&headers)).await { + Ok(stream) => stream, + 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() + } + }; Ok(Sse::new(stream).keep_alive(KeepAlive::default())) } +/// The agent's rows from its terminal stream: the replay +/// [`agent_term::replay_start`] picks, then every new row. +async fn replay( + client: &async_nats::Client, + agent: &str, + resume_after: Option, +) -> Result>, String> { + let js = async_nats::jetstream::new(client.clone()); + let name = agent_term::stream_name(agent); + let stream = js + .get_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); + 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"); + let agent = agent.to_owned(); + Ok(messages + .filter_map(move |msg| { + let event = match msg { + Ok(msg) => Some(Ok(row_event(&msg))), + Err(e) => { + tracing::warn!(%agent, error = %e, "term stream: row skipped"); + None + } + }; + std::future::ready(event) + }) + .boxed()) +} + +/// One row as an SSE event, carrying its stream sequence as the event `id` +/// when the delivery names one. +fn row_event(msg: &async_nats::jetstream::Message) -> Event { + let event = Event::default().data(String::from_utf8_lossy(&msg.payload)); + match msg.info() { + Ok(info) => event.id(info.stream_sequence.to_string()), + Err(_) => 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 { + headers + .get("last-event-id")? + .to_str() + .ok()? + .trim() + .parse() + .ok() +} + /// Subscribe to `subject`. `what` names the relay in logs and errors. pub(crate) async fn subscribe( client: &async_nats::Client, @@ -127,3 +217,33 @@ pub(crate) async fn subscribe( tracing::info!(%subject, "{what} stream: client attached"); Ok(subscriber) } + +#[cfg(test)] +mod tests { + use super::*; + + fn with_last_event_id(value: &str) -> HeaderMap { + let mut headers = HeaderMap::new(); + headers.insert( + "last-event-id", + value.parse().expect("a valid header value"), + ); + headers + } + + #[test] + fn a_reconnect_resumes_after_the_sequence_it_sends() { + assert_eq!(last_event_id(&with_last_event_id("917")), Some(917)); + assert_eq!( + agent_term::replay_start(1000, last_event_id(&with_last_event_id("917"))), + 918 + ); + } + + #[test] + fn no_or_a_foreign_last_event_id_is_a_fresh_reader() { + assert_eq!(last_event_id(&HeaderMap::new()), None); + assert_eq!(last_event_id(&with_last_event_id("r12")), None); + assert_eq!(last_event_id(&with_last_event_id("")), None); + } +} diff --git a/swarm-nats-auth/src/policy.rs b/swarm-nats-auth/src/policy.rs index 5577cf62..5615bdb3 100644 --- a/swarm-nats-auth/src/policy.rs +++ b/swarm-nats-auth/src/policy.rs @@ -538,6 +538,16 @@ impl Policy { // `MSG.GET`, the same width as `STREAM.INFO.*` above. A refused // publish reaches the client as a timeout. "$JS.API.STREAM.MSG.GET.*".to_owned(), + // Replaying an agent's terminal from `term-main-` to + // swarm-ui. Consumer create on every stream, but only in the + // extended form whose last tokens are the consumer's filter, + // which the server checks against the payload. So a consumer + // delivers only rows below `$SWARM.term`, every agent's and every + // subagent's included (the `*` also matches a literal `*` or `>`), + // which this client already subscribes to and reads by + // `MSG.GET` above. No bare `CONSUMER.CREATE.`, whose + // filter is unchecked. + swarm_queue_client::agent_term::consumer_create_grant(), "$KV.*.>".to_owned(), ]); // The per-agent status bucket, read-side only. Same five subjects as @@ -1215,6 +1225,47 @@ mod tests { assert!(p.publish.contains(&"$JS.API.STREAM.CREATE.*".to_owned())); } + #[test] + fn the_controller_may_replay_terminal_rows_and_no_other_subject() { + use swarm_queue_client::agent_term::stream_name; + use swarm_queue_client::{subagent_term, term_subject}; + + let p = policy() + .permissions("swarm-controller") + .expect("a reader is admitted"); + let may_publish = |s: &str| p.publish.iter().any(|g| covers(g, s)); + let create = |stream: &str, filter: &str| { + format!("$JS.API.CONSUMER.CREATE.{stream}.term-replay-n1.{filter}") + }; + assert!(may_publish(&create( + &stream_name("atlas"), + &term_subject("atlas") + ))); + for refused in [ + // The bare form, whose filter rides in the payload alone. + format!("$JS.API.CONSUMER.CREATE.{}", stream_name("atlas")), + create( + &subagent_term::stream_name("atlas"), + &subagent_term::subject("atlas", "scout"), + ), + create("hive-notices", "hive-notices.alpha"), + create(&stream_name("atlas"), "$SWARM.agent-state.atlas"), + create( + &stream_name("atlas"), + &format!("{}.x", term_subject("atlas")), + ), + ] { + assert!(!may_publish(&refused), "the reader may publish {refused}"); + } + // Control: no hive holds the grant. + assert!( + !policy() + .hive_subjects("alpha") + .iter() + .any(|s| covers(s, &create(&stream_name("atlas"), &term_subject("atlas")))) + ); + } + #[test] fn the_controller_may_read_a_subagents_newest_row() { let p = policy() diff --git a/swarm-queue-client/Cargo.toml b/swarm-queue-client/Cargo.toml index 4d81730c..8a549768 100644 --- a/swarm-queue-client/Cargo.toml +++ b/swarm-queue-client/Cargo.toml @@ -33,6 +33,10 @@ notices = ["async-nats/jetstream"] # `subagent_term::open_or_create`, for swarm-controller, which creates every # agent's subagent stream. Explicit for the same reason as `notices`. subagent-term = ["async-nats/jetstream"] +# `agent_term::open_or_create` and the replay consumer, for swarm-controller, +# which creates every agent's terminal stream and replays it to swarm-ui. +# Explicit for the same reason as `notices`. +agent-term = ["async-nats/jetstream"] # `agent_schedules::read_own`, for the agent harness, which reads its own # schedule key without opening the bucket. Explicit for the same reason as # `notices`. diff --git a/swarm-queue-client/src/agent_term.rs b/swarm-queue-client/src/agent_term.rs new file mode 100644 index 00000000..7699e6b6 --- /dev/null +++ b/swarm-queue-client/src/agent_term.rs @@ -0,0 +1,281 @@ +//! An agent's own terminal: the stream that keeps its rows, and the consumer +//! a reader replays them through. +//! +//! The agent publishes its classified terminal rows on +//! [`crate::term_subject`] with a plain core publish, under its own queue +//! credential, which may publish there and nothing else of these streams. +//! `swarm-controller` creates one stream per agent with the fixed config +//! [`crate::agent_term::open_or_create`] spells, so the queue keeps the rows a +//! reader who attaches later replays. +//! +//! The stream captures exactly one literal subject, `$SWARM.term.`. +//! [`crate::subagent_term`]'s stream captures `$SWARM.term..sub.>`, +//! which only matches subjects with at least two more tokens, so no subject +//! is in both and the two streams never overlap. The `JetStream` server refuses +//! to create a stream whose subjects overlap another's. +//! +//! The names live here because three crates must agree on them: the +//! publisher in `hive-agent`, `swarm-controller`, and the reader's consumer +//! grant in `swarm-nats-auth`. + +/// The token between the prefix and the agent in [`stream_name`]. It differs +/// from `term-sub-` at the same position, so no agent name makes this +/// stream's name equal another agent's subagent stream. +const STREAM_KIND: &str = "main"; + +/// How long the stream keeps a row. The same window as the subagent stream. +pub const MAX_AGE: std::time::Duration = std::time::Duration::from_hours(24); + +/// How large one agent's stream may grow before the discard policy reclaims +/// space. The same cap as the subagent stream, for the same reason: it +/// bounds the disk one agent can claim. +#[cfg(feature = "agent-term")] +const MAX_BYTES: i64 = 64 * 1024 * 1024; + +/// How many of the newest rows a reader that attaches fresh is replayed +/// before the live tail. The agent's own web UI loads the same number of +/// events as its first history page. +pub const REPLAY_ROWS: u64 = 100; + +/// The stream holding `agent`'s terminal rows: `term-main-`. +#[must_use] +pub fn stream_name(agent: &str) -> String { + format!("term-{STREAM_KIND}-{agent}") +} + +/// The one subject [`stream_name`]'s stream captures. +#[must_use] +pub fn stream_subject(agent: &str) -> String { + crate::term_subject(agent) +} + +/// The reader's grant to create a replay consumer: on any stream, under any +/// consumer name, in the extended create form whose last tokens are the +/// consumer's filter. +/// +/// The server refuses a create whose payload filter differs from those +/// tokens, so a consumer made under this grant delivers only rows on +/// subjects below `$SWARM.term`. The grant's last `*` also matches a literal +/// `*` or `>` token, so besides one agent's [`crate::term_subject`] it admits +/// `$SWARM.term.*` and `$SWARM.term.>`, which take in every agent's and every +/// subagent's terminal. The bare `CONSUMER.CREATE.` form, whose +/// filter rides only in the payload, is not covered. +#[must_use] +pub fn consumer_create_grant() -> String { + format!( + "$JS.API.CONSUMER.CREATE.*.*.{}", + crate::term_subject_wildcard() + ) +} + +/// The first stream sequence to replay from. +/// +/// `resume_after` is the last sequence the reader already holds, from a +/// reconnecting browser's `Last-Event-ID`; the replay continues right after +/// it. A fresh reader gets the newest [`REPLAY_ROWS`] below `last_sequence`, +/// the stream's newest row when it was read. A sequence the stream has since +/// discarded starts the consumer at the oldest row it still has. +#[must_use] +pub fn replay_start(last_sequence: u64, resume_after: Option) -> u64 { + match resume_after { + Some(seen) => seen.saturating_add(1), + None => last_sequence.saturating_sub(REPLAY_ROWS - 1).max(1), + } +} + +/// The config [`open_or_create`] creates `agent`'s stream with. +/// +/// `max_message_size` tracks the queue's live `max_payload`, as the subagent +/// stream's does, so the stream never rejects a row the queue accepted. +#[cfg(feature = "agent-term")] +fn stream_config(agent: &str, max_message_size: i32) -> async_nats::jetstream::stream::Config { + async_nats::jetstream::stream::Config { + name: stream_name(agent), + description: Some(format!("Terminal rows of {agent}")), + subjects: vec![stream_subject(agent)], + max_age: MAX_AGE, + max_bytes: MAX_BYTES, + discard: async_nats::jetstream::stream::DiscardPolicy::Old, + max_message_size, + ..Default::default() + } +} + +/// Open `agent`'s terminal stream, creating it if it does not exist yet. An +/// existing stream is opened as it is. +#[cfg(feature = "agent-term")] +pub async fn open_or_create( + client: &async_nats::Client, + agent: &str, +) -> Result { + let js = async_nats::jetstream::new(client.clone()); + let name = stream_name(agent); + match js.get_stream(&name).await { + Ok(stream) => Ok(stream), + Err(e) => { + tracing::info!(stream = %name, reason = %e, "agent terminal stream not available, creating it"); + let max_message_size = i32::try_from(crate::max_payload(client)).unwrap_or(i32::MAX); + js.create_stream(stream_config(agent, max_message_size)) + .await + .map_err(|source| crate::Error::CreateAgentStream { + stream: name, + source, + }) + } + } +} + +/// The consumer that replays `agent`'s rows from `start_sequence` and then +/// follows new ones, delivered to `deliver_subject` in stream order. +/// +/// Ephemeral and unacknowledged: the server drops it once nothing listens +/// on `deliver_subject`, so a reader that goes away needs no delete grant. +/// `name` must be set, because the server only takes the filter from the +/// subject [`consumer_create_grant`] covers when the create names a consumer. +#[cfg(feature = "agent-term")] +#[must_use] +pub fn replay_consumer_config( + agent: &str, + name: String, + deliver_subject: String, + start_sequence: u64, +) -> async_nats::jetstream::consumer::push::Config { + async_nats::jetstream::consumer::push::Config { + name: Some(name), + deliver_subject, + filter_subject: stream_subject(agent), + deliver_policy: async_nats::jetstream::consumer::DeliverPolicy::ByStartSequence { + start_sequence, + }, + ack_policy: async_nats::jetstream::consumer::AckPolicy::None, + ..Default::default() + } +} + +#[cfg(test)] +mod tests { + use super::*; + + /// Token-wise NATS subject overlap: whether some literal subject matches + /// both patterns. + fn overlaps(a: &str, b: &str) -> bool { + let (mut a, mut b) = (a.split('.'), b.split('.')); + loop { + match (a.next(), b.next()) { + (Some(">"), Some(_)) | (Some(_), Some(">")) | (None, None) => return true, + (None, _) | (_, None) => return false, + (Some(x), Some(y)) if x != "*" && y != "*" && x != y => return false, + _ => {} + } + } + } + + #[test] + fn names_match_the_publisher_and_the_grant() { + assert_eq!(stream_name("iris"), "term-main-iris"); + assert_eq!(stream_subject("iris"), "$SWARM.term.iris"); + assert_eq!( + consumer_create_grant(), + "$JS.API.CONSUMER.CREATE.*.*.$SWARM.term.*" + ); + } + + /// The server refuses to create a stream whose subjects overlap + /// another's, so this stream must share no subject with any agent's + /// subagent stream, its own agent's included. + #[test] + fn the_subject_is_disjoint_from_every_subagent_stream() { + for other in ["iris", "argus", "sub", "iris-2"] { + assert!( + !overlaps( + &stream_subject("iris"), + &crate::subagent_term::stream_subjects(other) + ), + "term-main-iris overlaps term-sub-{other}" + ); + } + // Control: the checker does report a real overlap. + assert!(overlaps( + "$SWARM.term.iris.>", + &crate::subagent_term::stream_subjects("iris") + )); + assert!(overlaps(&stream_subject("iris"), "$SWARM.term.*")); + } + + /// No agent name makes this stream's name another agent's subagent + /// stream: an agent may be called `sub-x`. + #[test] + fn the_name_never_collides_with_a_subagent_stream() { + for (agent, other) in [("sub-x", "x"), ("x", "sub-x"), ("main-x", "x")] { + assert_ne!(stream_name(agent), crate::subagent_term::stream_name(other)); + } + } + + /// The create the controller sends is one the grant covers, and every + /// filter the grant admits stays below `$SWARM.term`. + #[test] + fn the_grant_admits_only_filters_below_the_terminal_prefix() { + let grant = consumer_create_grant(); + let create = |filter: &str| format!("$JS.API.CONSUMER.CREATE.term-main-iris.r1.{filter}"); + assert!(overlaps(&grant, &create(&stream_subject("iris")))); + // The width the doc states: a literal wildcard as the last token. + assert!(overlaps(&grant, &create("$SWARM.term.>"))); + for other in [ + crate::subagent_term::subject("iris", "scout"), + crate::subagent_term::stream_subjects("iris"), + crate::agent_state_subject("iris"), + ] { + assert!(!overlaps(&grant, &create(&other)), "{other}"); + } + assert!(!overlaps(&grant, "$JS.API.CONSUMER.CREATE.term-main-iris")); + } + + #[test] + fn a_fresh_reader_replays_the_newest_rows() { + assert_eq!(replay_start(1000, None), 1000 - REPLAY_ROWS + 1); + assert_eq!(replay_start(5, None), 1); + assert_eq!(replay_start(0, None), 1); + } + + #[test] + fn a_reconnecting_reader_resumes_after_its_last_row() { + assert_eq!(replay_start(1000, Some(990)), 991); + assert_eq!(replay_start(1000, Some(1000)), 1001); + } + + /// The consumer starts at a stored sequence, not at the next new row, so + /// the replay reaches the reader before the live tail, in stream order, + /// through the one subscription. + #[cfg(feature = "agent-term")] + #[test] + fn the_consumer_replays_before_it_follows() { + let cfg = replay_consumer_config("iris", "r1".to_owned(), "_INBOX.x".to_owned(), 901); + assert_eq!( + cfg.deliver_policy, + async_nats::jetstream::consumer::DeliverPolicy::ByStartSequence { + start_sequence: 901 + } + ); + assert_eq!(cfg.filter_subject, "$SWARM.term.iris"); + assert_eq!(cfg.name.as_deref(), Some("r1")); + assert_eq!( + cfg.ack_policy, + async_nats::jetstream::consumer::AckPolicy::None + ); + assert!(cfg.durable_name.is_none()); + } + + #[cfg(feature = "agent-term")] + #[test] + fn stream_config_matches_the_subagent_streams_limits() { + let cfg = stream_config("iris", 8_388_608); + assert_eq!(cfg.subjects, ["$SWARM.term.iris"]); + assert_eq!(cfg.max_age, crate::subagent_term::MAX_AGE); + assert_eq!(cfg.max_bytes, MAX_BYTES); + assert_eq!( + cfg.discard, + async_nats::jetstream::stream::DiscardPolicy::Old + ); + assert_eq!(cfg.max_message_size, 8_388_608); + } +} diff --git a/swarm-queue-client/src/lib.rs b/swarm-queue-client/src/lib.rs index a76a901e..87c14990 100644 --- a/swarm-queue-client/src/lib.rs +++ b/swarm-queue-client/src/lib.rs @@ -135,10 +135,11 @@ pub enum Error { source: async_nats::jetstream::context::CreateStreamError, }, - #[cfg(feature = "subagent-term")] + #[cfg(any(feature = "subagent-term", feature = "agent-term"))] #[error("creating the {stream} stream")] CreateAgentStream { - // Owned: the name carries the agent's (`subagent_term::stream_name`). + // Owned: the name carries the agent's (`subagent_term::stream_name`, + // `agent_term::stream_name`). stream: String, #[source] source: async_nats::jetstream::context::CreateStreamError, @@ -381,6 +382,11 @@ pub struct AgentInput { #[cfg(feature = "notices")] pub mod notices; +/// The stream that keeps an agent's own terminal rows, and the consumer a +/// reader replays them through. Only the `JetStream` calls need the +/// `agent-term` feature; the names are unconditional. +pub mod agent_term; + /// The subject an agent's subagents' terminal rows go to, and the per-agent /// stream that lists them. Only opening the stream needs the `subagent-term` /// feature; the names are unconditional, like [`agent_icon`]'s.