diff --git a/docs/swarm/README.md b/docs/swarm/README.md index 60c72a18..808431d7 100644 --- a/docs/swarm/README.md +++ b/docs/swarm/README.md @@ -221,7 +221,7 @@ identity, exactly as the harness does, and publishes nothing without it. The queue keeps these rows in the stream `term-sub-` for 24 hours or 64 MiB, whichever comes first, dropping the oldest rows once either limit kicks in so a publish never fails because of it, and the swarm lists an agent's subagents -from that stream's subjects. The swarm controller creates that stream for +from that stream's subjects and replays a subagent's newest rows from it. The swarm controller creates that stream for every agent a hive's wanted state names, within a minute of the agent appearing there. The agent's grant is publish on its own `$SWARM.term..sub.>` and no `JetStream` subject, so the stream's diff --git a/docs/swarm/ui.md b/docs/swarm/ui.md index 33eec06f..43188320 100644 --- a/docs/swarm/ui.md +++ b/docs/swarm/ui.md @@ -51,7 +51,7 @@ dimmed, as **not running**. 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. +terminal does the same with that subagent's rows, and has no turn-state badges. The settings menu in the header has a **hide debug output** toggle. It drops the muted debug rows (thinking, coalesced ticks, harness chatter) from every @@ -256,7 +256,12 @@ or `null` when that row isn't read within two seconds. The lookups run concurrently, sixteen at a time, so the list waits at most two seconds for each sixteen subagents. swarm-controller also relays one subagent's rows as SSE on -`GET /api/agents//subagents//term/stream`. An agent with no +`GET /api/agents//subagents//term/stream`. Like the agent's +terminal route below, it creates a short-lived consumer on that stream, here +filtered to the one subagent, which sends its newest 100 rows and then each new +one, with the same `Last-Event-ID` resume. The stream holds every subagent of +the agent, so for a fresh reader the consumer starts at the oldest row and +swarm-controller drops all but the newest 100 before sending. An agent with no store identity, or no queue address, publishes no subagent terminals. diff --git a/swarm-controller/src/main.rs b/swarm-controller/src/main.rs index 9b8eb672..43c709a3 100644 --- a/swarm-controller/src/main.rs +++ b/swarm-controller/src/main.rs @@ -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 { diff --git a/swarm-controller/src/subagent_term.rs b/swarm-controller/src/subagent_term.rs index 47dcdd88..06c9791e 100644 --- a/swarm-controller/src/subagent_term.rs +++ b/swarm-controller/src/subagent_term.rs @@ -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, Path((agent, subagent)): Path<(String, String)>, + headers: HeaderMap, ) -> Result>>, 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())) } diff --git a/swarm-controller/src/term_stream.rs b/swarm-controller/src/term_stream.rs index 6a333cdf..e3cdf710 100644 --- a/swarm-controller/src/term_stream.rs +++ b/swarm-controller/src/term_stream.rs @@ -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, + what: &str, +) -> Result>, 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, 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-` 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, ) -> Result, 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( messages: impl Stream> + Send + 'static, - agent: String, + subject: String, ) -> BoxStream<'static, Row> { messages .scan((), move |(), msg| { @@ -189,7 +239,7 @@ fn rows_until_error( 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 { +pub(crate) fn last_event_id(headers: &HeaderMap) -> Option { 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, + ) -> Result, 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, + ) { + 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, + ) -> Result, 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) -> 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 = 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( diff --git a/swarm-nats-auth/src/policy.rs b/swarm-nats-auth/src/policy.rs index 7776d3fa..c18b1153 100644 --- a/swarm-nats-auth/src/policy.rs +++ b/swarm-nats-auth/src/policy.rs @@ -702,6 +702,10 @@ impl Policy { // `MSG.GET` above. No bare `CONSUMER.CREATE.`, whose // filter is unchecked. swarm_queue_client::agent_term::consumer_create_grant(), + // The same for one subagent's rows in `term-sub-`: the + // filter is `$SWARM.term..sub.`, two tokens + // deeper than the grant above reaches. + swarm_queue_client::subagent_term::consumer_create_grant(), "$KV.*.>".to_owned(), ]); // The per-agent status bucket, read-side only. Same five subjects as @@ -1788,12 +1792,24 @@ mod tests { &stream_name("atlas"), &term_subject("atlas") ))); + assert!(may_publish(&create( + &subagent_term::stream_name("atlas"), + &subagent_term::subject("atlas", "scout"), + ))); for refused in [ // The bare form, whose filter rides in the payload alone. format!("$JS.API.CONSUMER.CREATE.{}", stream_name("atlas")), + format!( + "$JS.API.CONSUMER.CREATE.{}", + subagent_term::stream_name("atlas") + ), create( &subagent_term::stream_name("atlas"), - &subagent_term::subject("atlas", "scout"), + &format!("{}.x", subagent_term::subject("atlas", "scout")), + ), + create( + &subagent_term::stream_name("atlas"), + "$SWARM.agent-state.atlas.sub.scout", ), create("hive-notices", "hive-notices.alpha"), create(&stream_name("atlas"), "$SWARM.agent-state.atlas"), diff --git a/swarm-queue-client/src/agent_term.rs b/swarm-queue-client/src/agent_term.rs index f0bdd060..5ad2a07d 100644 --- a/swarm-queue-client/src/agent_term.rs +++ b/swarm-queue-client/src/agent_term.rs @@ -135,17 +135,21 @@ pub async fn open_or_create( } } -/// The consumer that replays `agent`'s rows from `start_sequence` and then -/// follows new ones, delivered to `deliver_subject` in stream order. +/// The consumer that replays the rows on `filter_subject` from +/// `start_sequence` and then follows new ones, delivered to `deliver_subject` +/// in stream order. The filter is an agent's [`stream_subject`] or one +/// subagent's [`crate::subagent_term::subject`]. /// /// 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. +/// subject [`consumer_create_grant`] (or +/// [`crate::subagent_term::consumer_create_grant`]) covers when the create +/// names a consumer. #[cfg(feature = "agent-term")] #[must_use] pub fn replay_consumer_config( - agent: &str, + filter_subject: String, name: String, deliver_subject: String, start_sequence: u64, @@ -153,7 +157,7 @@ pub fn replay_consumer_config( async_nats::jetstream::consumer::push::Config { name: Some(name), deliver_subject, - filter_subject: stream_subject(agent), + filter_subject, deliver_policy: async_nats::jetstream::consumer::DeliverPolicy::ByStartSequence { start_sequence, }, @@ -269,7 +273,12 @@ mod tests { #[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); + let cfg = replay_consumer_config( + stream_subject("iris"), + "r1".to_owned(), + "_INBOX.x".to_owned(), + 901, + ); assert_eq!( cfg.deliver_policy, async_nats::jetstream::consumer::DeliverPolicy::ByStartSequence { diff --git a/swarm-queue-client/src/subagent_term.rs b/swarm-queue-client/src/subagent_term.rs index aeef3d85..4c5d0c98 100644 --- a/swarm-queue-client/src/subagent_term.rs +++ b/swarm-queue-client/src/subagent_term.rs @@ -6,17 +6,18 @@ //! credential. The queue grants that credential publish on this subject //! family and no `JetStream` subject. //! -//! The stream exists so the swarm can list an agent's subagents: subagents +//! The stream lets the swarm list an agent's subagents: subagents //! are spawned on demand under caller-chosen names, and the set of subjects //! the stream holds is the list. `swarm-controller` creates one stream per //! agent with the fixed config [`crate::subagent_term::open_or_create`] //! spells, and takes the list from its subject counts -//! ([`crate::subagent_term::subagent_names`]). +//! ([`crate::subagent_term::subagent_names`]). A reader attaching to one +//! subagent's terminal replays that subagent's newest rows from it. //! -//! The name and the subject live here because three crates must agree on -//! them: the publisher in `hive-subagent-mcp`, `swarm-controller`, and the -//! queue grant in `swarm-nats.nix`, whose `{agent}` template spells the same -//! subjects. +//! The name and the subject live here because they must agree across the +//! publisher in `hive-subagent-mcp`, `swarm-controller`, the queue grant in +//! `swarm-nats.nix`, whose `{agent}` template spells the same subjects, and +//! the reader's consumer grant in `swarm-nats-auth`. use crate::TERM_SUBJECT_PREFIX; @@ -56,6 +57,17 @@ pub fn subject_wildcard() -> String { subject("*", "*") } +/// The reader's grant to create a replay consumer on one subagent's rows: +/// the extended create form of [`crate::agent_term::consumer_create_grant`], +/// on any stream and under any consumer name, with a [`subject`] as the +/// filter tokens the server checks the payload against. The last `*` also +/// matches a literal `*` or `>`, so it admits `$SWARM.term..sub.>` +/// too, and nothing outside [`subject_wildcard`]'s family. +#[must_use] +pub fn consumer_create_grant() -> String { + format!("$JS.API.CONSUMER.CREATE.*.*.{}", subject_wildcard()) +} + /// Every subject [`stream_name`]'s stream captures, which is also the publish /// grant's shape. #[must_use] @@ -149,6 +161,10 @@ mod tests { assert_eq!(stream_name("iris"), "term-sub-iris"); assert_eq!(subject("iris", "scout"), "$SWARM.term.iris.sub.scout"); assert_eq!(stream_subjects("iris"), "$SWARM.term.iris.sub.>"); + assert_eq!( + consumer_create_grant(), + "$JS.API.CONSUMER.CREATE.*.*.$SWARM.term.*.sub.*" + ); } #[test]