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.

swarm-controller/src/term_stream.rs

@@ -117,7 +117,7 @@ pub(crate) async fn stream_agent_term(
     })?;
 
     let stream = match replay(&client, &agent, last_event_id(&headers)).await {
-        Ok(stream) => stream,
+        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 =
@@ -130,13 +130,20 @@ pub(crate) async fn stream_agent_term(
     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, Result<Event, Infallible>>, String> {
+) -> Result<BoxStream<'static, Row>, String> {
     let js = async_nats::jetstream::new(client.clone());
     let name = agent_term::stream_name(agent);
     let stream = js
@@ -164,28 +171,39 @@ async fn replay(
         .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))),
+    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: row skipped");
+                    tracing::warn!(%agent, error = %e, "term stream: consumer failed, ending the stream");
                     None
                 }
             };
-            std::future::ready(event)
+            std::future::ready(row)
         })
-        .boxed())
+        .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,
+/// 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,
     }
 }
 
@@ -222,6 +240,196 @@ pub(crate) async fn subscribe(
 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(