the agent state stream sends a new client the latest header first

The stream was a live tail with no replay, and the agent re-sends its header
once a minute, so a card opened mid-idle read "no report yet" for up to a
minute. The controller already follows every agent's header for the inference
route; it now keeps the newest one as published, and the stream sends it
first when it is under 150 seconds old. An agent that stopped keeps no state
older than that.

swarm-controller/src/agent_state_stream.rs

@@ -16,6 +16,13 @@
 //! alternative would be a KV bucket written on every turn-state flip, which
 //! is the write rate this feature exists to avoid.
 //!
+//! **A client that attaches mid-idle gets the latest header at once.**
+//! `crate::inference::Reported` keeps it, and the stream sends it first when
+//! it is under [`REPLAY_MAX_AGE`] old: two re-send intervals, so an agent that
+//! stopped does not keep a state it no longer has. After that it is the live
+//! tail. A controller that has just started holds none and, as before, waits
+//! for the agent's next header.
+//!
 //! **The payload is passed through opaquely**, same as the terminal relay:
 //! this relay never inspects a header, so it does not depend on
 //! `hive-agent`'s type to re-serialise one. The bytes on the wire are what
@@ -23,11 +30,15 @@
 //! reads its own fields out of the same subject separately.
 
 use std::convert::Infallible;
-use std::time::SystemTime;
+use std::time::{Duration, SystemTime};
 
 use axum::extract::{Path, State};
 use axum::response::sse::{Event, KeepAlive, Sse};
-use futures_util::{Stream, StreamExt as _};
+use futures_util::{Stream, StreamExt as _, stream};
+
+/// How old a header the stream still hands a new client first: two of the
+/// agent's one-minute re-sends, so one missed re-send does not hide it.
+const REPLAY_MAX_AGE: Duration = Duration::from_secs(150);
 
 use crate::routes::AppState;
 
@@ -41,7 +52,8 @@ use crate::routes::AppState;
             `turn_state`, `turn_state_since` (ISO 8601 UTC), `agent_state`, `model`, `effort`, \
             `model_source`, `effort_source`, `resolved_model`, `context_window_tokens`, \
             `ctx_usage`, `cost_usage`, `needs_login`. Sent on transition and re-sent once a \
-            minute, live, no replay",
+            minute, live. A new client is first sent the latest one the controller holds, \
+            if it is under 150 seconds old",
             body = String, content_type = "text/event-stream"),
         (status = 400, description = "the agent name is not shaped like an identifier \
             (problem+json)", body = String),
@@ -115,7 +127,17 @@ pub(crate) async fn stream_agent_state(
         "state",
     )
     .await?;
-    let stream =
+    // Read after subscribing, so a header published in between is in the tail
+    // too; the client may see one twice, never miss one.
+    let latest = state
+        .reported_inference
+        .as_ref()
+        .and_then(|reported| reported.header_within(&agent, REPLAY_MAX_AGE));
+    let first = stream::iter(latest.map(|payload| {
+        Ok::<_, Infallible>(Event::default().data(String::from_utf8_lossy(&payload)))
+    }));
+    let live =
         subscriber.map(|msg| Ok(Event::default().data(String::from_utf8_lossy(&msg.payload))));
+    let stream = first.chain(live);
     Ok(Sse::new(stream).keep_alive(KeepAlive::default()))
 }

swarm-controller/src/inference.rs

@@ -12,10 +12,15 @@
 //! Agents publish a header when it changes and re-send it once a minute, and
 //! the queue keeps none, so an agent has no entry for up to a minute after
 //! this daemon starts.
+//!
+//! [`Reported`] also keeps each agent's newest header as it arrived, so the
+//! state stream (`crate::agent_state_stream`) can hand a client that attaches
+//! mid-idle the last header at once, rather than leaving it waiting for the
+//! next re-send.
 
 use std::collections::{BTreeSet, HashMap};
 use std::sync::{Arc, Mutex, PoisonError};
-use std::time::Duration;
+use std::time::{Duration, Instant};
 
 use axum::Json;
 use axum::extract::{Path, State};
@@ -108,6 +113,8 @@ fn effective_view(v: Effective) -> EffectiveInference {
 #[derive(Debug, Default)]
 pub(crate) struct Reported {
     latest: Mutex<HashMap<String, Effective>>,
+    /// Each agent's newest header exactly as published, with when it came.
+    raw: Mutex<HashMap<String, (Instant, Vec<u8>)>>,
 }
 
 impl Reported {
@@ -139,12 +146,24 @@ impl Reported {
         }
     }
 
+    /// `agent`'s newest header, as published, when it arrived within `max_age`.
+    /// An older one describes an agent that may have stopped since.
+    pub(crate) fn header_within(&self, agent: &str, max_age: Duration) -> Option<Vec<u8>> {
+        let raw = self.raw.lock().unwrap_or_else(PoisonError::into_inner);
+        let (at, payload) = raw.get(agent)?;
+        (at.elapsed() <= max_age).then(|| payload.clone())
+    }
+
     /// Keep one header's inference fields. A header without them, from an
     /// agent that does not report them, removes what was kept.
     fn record(&self, subject: &str, payload: &[u8]) {
         let Some(agent) = swarm_queue_client::agent_from_state_subject(subject) else {
             return;
         };
+        self.raw
+            .lock()
+            .unwrap_or_else(PoisonError::into_inner)
+            .insert(agent.to_owned(), (Instant::now(), payload.to_vec()));
         let mut latest = self.latest.lock().unwrap_or_else(PoisonError::into_inner);
         match serde_json::from_slice::<Effective>(payload) {
             Ok(effective) => {
@@ -436,6 +455,38 @@ mod tests {
         assert_eq!(reported.get("atlas"), None);
     }
 
+    #[test]
+    fn the_newest_header_is_kept_as_published_while_it_is_fresh() {
+        let reported = Reported::default();
+        reported.record(&iris(), OLD_HEADER);
+        reported.record(&iris(), NEW_HEADER);
+        assert_eq!(
+            reported.header_within("iris", Duration::from_mins(1)),
+            Some(NEW_HEADER.to_vec())
+        );
+        assert_eq!(
+            reported.header_within("atlas", Duration::from_mins(1)),
+            None
+        );
+        // A header with no inference fields still counts as the newest one.
+        reported.record(&iris(), OLD_HEADER);
+        assert_eq!(
+            reported.header_within("iris", Duration::from_mins(1)),
+            Some(OLD_HEADER.to_vec())
+        );
+    }
+
+    #[test]
+    fn a_header_older_than_the_limit_is_not_handed_out() {
+        let reported = Reported::default();
+        reported.record(&iris(), NEW_HEADER);
+        std::thread::sleep(Duration::from_millis(20));
+        assert_eq!(
+            reported.header_within("iris", Duration::from_millis(1)),
+            None
+        );
+    }
+
     #[test]
     fn a_header_without_the_inference_fields_leaves_no_entry() {
         let reported = Reported::default();