swarm-controller: end a dead replay, clamp stale resume ids, test on a real server

A replay consumer now has a 5s idle heartbeat, and the row stream ends at
the first consumer error instead of skipping it. When the server drops the
consumer, as on a restart, two missed heartbeats end the SSE response and
the browser reconnects with Last-Event-ID. Before, the response stayed
open and delivered nothing.

A Last-Event-ID above the stream's newest sequence now replays as a fresh
reader. The server accepts a start past the end and delivers nothing
until the stream reaches it, so a stale id stalled the replay.

swarm-controller's stream tests run against a real nats-server on
loopback (`test_nats`, binary from HIVE_NATS_SERVER, exported by
checks.nix and devshell.nix). They cover: ensuring the stream twice, an
existing stream with other limits being opened as it is, the replay
arriving before live rows, a reconnect getting every later row exactly
once, and a dropped consumer ending the row stream.

The limits test now compares against the subagent stream's own config.
Two comments that described the agent's terminal route as live-only are
corrected.

docs/swarm/README.md

@@ -189,9 +189,10 @@ own web UI renders also goes to `$SWARM.term.<agent>`, 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-<agent>` 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.<agent>.sub.<subagent>`,

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,24 @@ store identity, or no queue address, publishes no subagent terminals.
 
 </details>
 
+<details><summary>Where an agent's terminal history comes from</summary>
+
+The harness publishes the agent's terminal rows on `$SWARM.term.<agent>`, into
+a stream named `term-main-<agent>` 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.<agent>.sub`, so they never overlap.
+`GET /api/agents/<name>/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. An id past the newest row of the stream, as
+after swarm-controller recreates the stream, starts over as a fresh reader. When the
+consumer stops sending its heartbeat, as after a queue restart, the route ends
+the response and the browser reconnects. Until the stream exists, the route sends
+live rows only.
+
+</details>
+
 ### Inference settings
 
 The agent's detail panel shows its model and effort under **inference**, each

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";

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-<agent>`, 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

nix/checks.nix

@@ -222,6 +222,11 @@ in
     # quietly does nothing is worse than no drift test, because it still
     # shows up green.
     HIVE_RESERVED_NAMES = pkgs.lib.concatStringsSep " " (import ./reserved-names.nix);
+    # swarm-controller's stream and replay tests run against a real
+    # `nats-server` on loopback (`swarm-controller/src/test_nats.rs`), since
+    # what they check is the server's own create and delivery behaviour.
+    # Exported here and in `devshell.nix`; the tests panic when it is unset.
+    HIVE_NATS_SERVER = "${pkgs.nats-server}/bin/nats-server";
   };
 
   # Rustdoc gate. Builds the workspace's docs and turns rustdoc's own

nix/devshell.nix

@@ -11,6 +11,9 @@
     # developer's local run and CI would disagree about a test that exists
     # precisely to stop two spellings of one fact from drifting.
     HIVE_RESERVED_NAMES = pkgs.lib.concatStringsSep " " (import ./reserved-names.nix);
+    # The `nats-server` swarm-controller's stream tests start, the same one
+    # `checks.nix` gives the test derivation. Those tests panic without it.
+    HIVE_NATS_SERVER = "${pkgs.nats-server}/bin/nats-server";
     packages =
       rust.nativeBuildInputs
       ++ (with pkgs; [

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

swarm-controller/src/agent_state_stream.rs

@@ -1,11 +1,11 @@
 //! `GET /api/agents/{name}/state/stream` — a live SSE relay of one agent's
 //! turn-state header, sourced straight from the swarm queue.
 //!
-//! Sibling of [`crate::term_stream`] in every structural respect — the
-//! reasoning there for keeping the hive out of the path, and for a live tail with no replay, applies
-//! unchanged here and is not repeated. What differs is only what the two
-//! subjects carry: a terminal row is an event that happened, a header is
-//! the agent's current state.
+//! Sibling of [`crate::term_stream`]: the reasoning there for keeping the
+//! hive out of the path applies unchanged here and is not repeated. Unlike
+//! that route this one is a live tail with no replay, because no stream keeps
+//! headers. The two subjects also differ in what they carry: a terminal row
+//! is an event that happened, a header is the agent's current state.
 //!
 //! **That difference is the one thing worth reading twice.** `hive-agent`
 //! publishes a header on transition and re-sends it once a minute, to a core

swarm-controller/src/main.rs

@@ -64,6 +64,8 @@ mod status;
 mod store;
 mod subagent_term;
 mod term_stream;
+#[cfg(test)]
+mod test_nats;
 mod vcs_metrics;
 mod wanted;
 mod webhook;
@@ -2621,8 +2623,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<status::StatusReader>>, hives: &[HiveEntry]) {
     let (Some(status), Some(wanted)) = (status, wanted_writer(status)) else {
         return;
@@ -2631,7 +2633,8 @@ fn spawn_subagent_streams(status: Option<&Arc<status::StatusReader>>, 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: subagent and terminal streams, and
+/// schedule keys.
 /// Lifted out of `main` for `clippy::too_many_lines`.
 fn spawn_per_agent_upkeep(status: Option<&Arc<status::StatusReader>>, hives: &[HiveEntry]) {
     spawn_subagent_streams(status, hives);

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
@@ -14,9 +15,9 @@
 //! **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, the
-//! same as the agent's own terminal route, so a reader sees rows published
-//! while it is attached. Nothing is sent toward a subagent: subagents take no
+//! **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
 //! input from the swarm.
 
 use std::collections::BTreeSet;
@@ -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<WantedWriter>, 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<WantedWriter>, 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"
+                    );
+                }
             }
         }
     });

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<AppState>,
     Path(agent): Path<String>,
+    headers: HeaderMap,
 ) -> Result<Sse<impl Stream<Item = Result<Event, Infallible>>>, problem_details::ProblemDetails> {
     let Some(status) = state.status.as_ref() else {
         return Err(crate::error_problem(
@@ -105,12 +116,109 @@ 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(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()
+        }
+    };
     Ok(Sse::new(stream).keep_alive(KeepAlive::default()))
 }
 
+/// One row read from the agent's 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.
+async fn replay(
+    client: &async_nats::Client,
+    agent: &str,
+    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)
+        .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");
+    Ok(rows_until_error(messages, agent.to_owned()))
+}
+
+/// The rows of a consumer's deliveries, ending at the first error. A missed
+/// heartbeat is such an error, and ending there closes the SSE response, so
+/// 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,
+) -> BoxStream<'static, Row> {
+    messages
+        .scan((), move |(), msg| {
+            let row = match msg {
+                Ok(msg) => Some(Row {
+                    seq: msg.info().ok().map(|info| info.stream_sequence),
+                    payload: msg.message.payload,
+                }),
+                Err(e) => {
+                    tracing::warn!(%agent, error = %e, "term stream: consumer failed, ending the stream");
+                    None
+                }
+            };
+            std::future::ready(row)
+        })
+        .boxed()
+}
+
+/// One row as an SSE event, carrying its stream sequence as the event `id`.
+fn row_event(row: &Row) -> Event {
+    let event = Event::default().data(String::from_utf8_lossy(&row.payload));
+    match row.seq {
+        Some(seq) => event.id(seq.to_string()),
+        None => 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> {
+    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 +235,223 @@ pub(crate) async fn subscribe(
     tracing::info!(%subject, "{what} stream: client attached");
     Ok(subscriber)
 }
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    use std::time::Duration;
+
+    use crate::test_nats::TestNats;
+
+    const AGENT: &str = "iris";
+
+    async fn publish_rows(client: &async_nats::Client, rows: std::ops::RangeInclusive<u64>) {
+        for i in rows {
+            client
+                .publish(
+                    swarm_queue_client::term_subject(AGENT),
+                    format!("row{i}").into(),
+                )
+                .await
+                .expect("publish");
+        }
+        client.flush().await.expect("flush");
+    }
+
+    /// Wait until the stream holds row `last`, so the next step sees it.
+    async fn stored(client: &async_nats::Client, 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");
+            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}");
+    }
+
+    async fn take(rows: &mut BoxStream<'static, Row>, n: usize) -> Vec<(u64, String)> {
+        let mut out = Vec::with_capacity(n);
+        for _ in 0..n {
+            let row = tokio::time::timeout(Duration::from_secs(5), rows.next())
+                .await
+                .unwrap_or_else(|_| panic!("row {} of {n} never came; got {out:?}", out.len() + 1))
+                .expect("the row stream ended");
+            out.push((
+                row.seq.expect("a delivered row names its sequence"),
+                String::from_utf8_lossy(&row.payload).into_owned(),
+            ));
+        }
+        out
+    }
+
+    fn rows(range: std::ops::RangeInclusive<u64>) -> Vec<(u64, String)> {
+        range.map(|i| (i, format!("row{i}"))).collect()
+    }
+
+    async fn nothing_more(rows: &mut BoxStream<'static, Row>) {
+        if let Ok(Some(row)) = tokio::time::timeout(Duration::from_millis(500), rows.next()).await {
+            panic!("an extra row: seq {:?}", row.seq);
+        }
+    }
+
+    /// A consumer the server dropped, as on a restart, ends the row stream
+    /// once its heartbeats stop, so the SSE response closes and the browser
+    /// reconnects instead of keeping a tail that never delivers.
+    #[tokio::test]
+    async fn a_dropped_consumer_ends_the_row_stream() {
+        use futures_util::TryStreamExt as _;
+
+        let nats = TestNats::start();
+        let client = nats.client().await;
+        let stream = agent_term::open_or_create(&client, AGENT)
+            .await
+            .expect("the stream");
+        publish_rows(&client, 1..=3).await;
+        stored(&client, 3).await;
+        let mut replayed = replay(&client, AGENT, None).await.expect("a replay");
+        assert_eq!(take(&mut replayed, 3).await, rows(1..=3));
+        let consumers: Vec<String> = stream
+            .consumer_names()
+            .try_collect()
+            .await
+            .expect("consumer names");
+        let [consumer] = consumers.as_slice() else {
+            panic!("expected the one replay consumer, got {consumers:?}");
+        };
+        stream
+            .delete_consumer(consumer)
+            .await
+            .expect("deleting the consumer");
+        let end = tokio::time::timeout(agent_term::IDLE_HEARTBEAT * 4, replayed.next())
+            .await
+            .expect("the row stream ends once heartbeats stop");
+        assert!(
+            end.is_none(),
+            "a row arrived after the consumer was deleted"
+        );
+    }
+
+    #[tokio::test]
+    async fn ensuring_the_terminal_stream_again_leaves_it_as_it_is() {
+        let nats = TestNats::start();
+        let client = nats.client().await;
+        let created = agent_term::open_or_create(&client, AGENT)
+            .await
+            .expect("the first ensure creates the stream");
+        publish_rows(&client, 1..=3).await;
+        stored(&client, 3).await;
+        agent_term::open_or_create(&client, AGENT)
+            .await
+            .expect("a second ensure succeeds");
+        // Read back from the server, not from what the ensure returned.
+        let after = async_nats::jetstream::new(client.clone())
+            .get_stream(agent_term::stream_name(AGENT))
+            .await
+            .expect("the stream")
+            .info()
+            .await
+            .expect("its info")
+            .clone();
+        assert_eq!(after.config, created.cached_info().config);
+        assert_eq!(after.state.messages, 3);
+    }
+
+    /// An existing stream is opened, not created again: one made when the
+    /// queue announced a different `max_payload` keeps its own config and
+    /// rows, where a second create would be refused for the mismatch.
+    #[tokio::test]
+    async fn an_existing_terminal_stream_with_other_limits_is_opened_as_it_is() {
+        let nats = TestNats::start();
+        let client = nats.client().await;
+        let js = async_nats::jetstream::new(client.clone());
+        js.create_stream(async_nats::jetstream::stream::Config {
+            name: agent_term::stream_name(AGENT),
+            subjects: vec![agent_term::stream_subject(AGENT)],
+            max_message_size: 4096,
+            ..Default::default()
+        })
+        .await
+        .expect("an earlier stream");
+        publish_rows(&client, 1..=2).await;
+        stored(&client, 2).await;
+        let opened = agent_term::open_or_create(&client, AGENT)
+            .await
+            .expect("an existing stream opens");
+        assert_eq!(opened.cached_info().config.max_message_size, 4096);
+        assert_eq!(opened.cached_info().state.messages, 2);
+    }
+
+    /// Live rows published after the reader attached, before it read any
+    /// replayed row, still arrive after the whole replay.
+    #[tokio::test]
+    async fn the_replay_reaches_the_reader_before_the_live_tail() {
+        let nats = TestNats::start();
+        let client = nats.client().await;
+        agent_term::open_or_create(&client, AGENT)
+            .await
+            .expect("the stream");
+        publish_rows(&client, 1..=150).await;
+        stored(&client, 150).await;
+        let mut replayed = replay(&client, AGENT, 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;
+    }
+
+    /// A reader that reconnects with the last id it was sent gets every row
+    /// after it exactly once, including those published while it was away.
+    #[tokio::test]
+    async fn a_reconnect_gets_every_later_row_once() {
+        let nats = TestNats::start();
+        let client = nats.client().await;
+        agent_term::open_or_create(&client, AGENT)
+            .await
+            .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 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)
+            .await
+            .expect("a resumed replay");
+        publish_rows(&client, 161..=163).await;
+        assert_eq!(take(&mut resumed, 13).await, rows(151..=163));
+        nothing_more(&mut resumed).await;
+    }
+
+    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);
+    }
+}

swarm-controller/src/test_nats.rs

@@ -0,0 +1,66 @@
+//! A throwaway `nats-server` with `JetStream`, for tests whose subject is the
+//! server's own behaviour: stream creation, consumer delivery order.
+//!
+//! The binary comes from [`NATS_SERVER_ENV`], which `nix/checks.nix` and
+//! `nix/devshell.nix` export. A test that needs it panics when it is unset
+//! rather than skipping, so it never shows green without having run.
+
+use std::path::PathBuf;
+use std::process::{Child, Command, Stdio};
+use std::time::Duration;
+
+/// The variable naming the `nats-server` binary.
+pub(crate) const NATS_SERVER_ENV: &str = "HIVE_NATS_SERVER";
+
+/// A running server on a loopback port, with its store in a fresh directory.
+/// Dropping it kills the server and removes the store.
+pub(crate) struct TestNats {
+    child: Child,
+    store: PathBuf,
+    url: String,
+}
+
+impl TestNats {
+    pub(crate) fn start() -> Self {
+        let bin = std::env::var(NATS_SERVER_ENV).unwrap_or_else(|_| {
+            panic!("{NATS_SERVER_ENV} is exported by nix/checks.nix and nix/devshell.nix")
+        });
+        // Bound and released to learn a free port; the server binds it next.
+        let port = std::net::TcpListener::bind("127.0.0.1:0")
+            .and_then(|l| l.local_addr())
+            .expect("a free loopback port")
+            .port();
+        let store = std::env::temp_dir().join(format!("test-nats-{}-{port}", std::process::id()));
+        let child = Command::new(&bin)
+            .args(["-a", "127.0.0.1", "-p", &port.to_string(), "-js", "-sd"])
+            .arg(&store)
+            .stdout(Stdio::null())
+            .stderr(Stdio::null())
+            .spawn()
+            .unwrap_or_else(|e| panic!("starting {bin}: {e}"));
+        Self {
+            child,
+            store,
+            url: format!("nats://127.0.0.1:{port}"),
+        }
+    }
+
+    /// A client connected to the server, once it accepts connections.
+    pub(crate) async fn client(&self) -> async_nats::Client {
+        for _ in 0..100 {
+            if let Ok(client) = async_nats::connect(&self.url).await {
+                return client;
+            }
+            tokio::time::sleep(Duration::from_millis(100)).await;
+        }
+        panic!("nats-server at {} never accepted a connection", self.url);
+    }
+}
+
+impl Drop for TestNats {
+    fn drop(&mut self) {
+        let _ = self.child.kill();
+        let _ = self.child.wait();
+        let _ = std::fs::remove_dir_all(&self.store);
+    }
+}

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-<agent>` 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.<stream>`, 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()

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`.

swarm-queue-client/src/agent_term.rs

@@ -0,0 +1,307 @@
+//! 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.<agent>`.
+//! [`crate::subagent_term`]'s stream captures `$SWARM.term.<agent>.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;
+
+/// How often the server sends an idle replay consumer a heartbeat. Two missed
+/// in a row end the reader's row stream, so a consumer the server dropped,
+/// as on a restart, ends the SSE response and the browser reconnects.
+pub const IDLE_HEARTBEAT: std::time::Duration = std::time::Duration::from_secs(5);
+
+/// The stream holding `agent`'s terminal rows: `term-main-<agent>`.
+#[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.<stream>` 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.
+///
+/// A `resume_after` above `last_sequence` names no row this stream ever held,
+/// as after the stream was created again, and is read as a fresh reader: the
+/// server accepts a start past the end and delivers nothing until the stream
+/// reaches it.
+#[must_use]
+pub fn replay_start(last_sequence: u64, resume_after: Option<u64>) -> u64 {
+    match resume_after {
+        Some(seen) if seen <= last_sequence => seen + 1,
+        _ => 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<async_nats::jetstream::stream::Stream, crate::Error> {
+    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,
+        idle_heartbeat: IDLE_HEARTBEAT,
+        ..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);
+    }
+
+    /// An id past the stream's newest row, as after the stream was created
+    /// again, replays like a fresh reader instead of waiting for that number.
+    #[test]
+    fn an_id_past_the_newest_row_replays_as_a_fresh_reader() {
+        assert_eq!(replay_start(5, Some(99)), replay_start(5, None));
+        assert_eq!(replay_start(1000, Some(1001)), replay_start(1000, None));
+        assert_eq!(replay_start(0, Some(7)), 1);
+    }
+
+    /// 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());
+        // Without it a dropped consumer is never noticed.
+        assert_eq!(cfg.idle_heartbeat, IDLE_HEARTBEAT);
+        assert!(!cfg.idle_heartbeat.is_zero());
+    }
+
+    /// Every limit is read from the subagent stream's own config, so the two
+    /// fail this test the moment they drift apart.
+    #[cfg(all(feature = "agent-term", feature = "subagent-term"))]
+    #[test]
+    fn stream_config_matches_the_subagent_streams_limits() {
+        let main = stream_config("iris", 8_388_608);
+        let sub = crate::subagent_term::stream_config("iris", 8_388_608);
+        assert_eq!(main.subjects, ["$SWARM.term.iris"]);
+        assert_eq!(main.max_age, sub.max_age);
+        assert_eq!(main.max_bytes, sub.max_bytes);
+        assert_eq!(main.discard, sub.discard);
+        assert_eq!(main.max_message_size, sub.max_message_size);
+        assert_eq!(main.retention, sub.retention);
+        assert_eq!(main.storage, sub.storage);
+        assert_eq!(main.num_replicas, sub.num_replicas);
+    }
+}

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.

swarm-queue-client/src/subagent_term.rs

@@ -99,7 +99,10 @@ pub fn subagent_names<'a>(agent: &str, subjects: impl IntoIterator<Item = &'a st
 /// rows against, so the stream never rejects a row the queue itself
 /// accepted.
 #[cfg(feature = "subagent-term")]
-fn stream_config(agent: &str, max_message_size: i32) -> async_nats::jetstream::stream::Config {
+pub(crate) 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}'s subagents")),