diff --git a/docs/swarm/README.md b/docs/swarm/README.md index 1144d0f7..8c7e92ed 100644 --- a/docs/swarm/README.md +++ b/docs/swarm/README.md @@ -274,7 +274,8 @@ That's the whole point of the subject — the `agent-status` bucket already republishes once a minute, which is far too slow for "is this agent thinking right now." A core subject keeps nothing for a late subscriber, so the harness also re-sends the current header once a minute: a subscriber attaching -mid-idle sees it within that minute. A renderer opens with the bucket's +mid-idle sees it within that minute. The swarm's stream sends a new client the +latest header it holds first, when that's under 150 seconds old. A renderer opens with the bucket's snapshot and lets this stream refine it. Swarm-side, `GET /api/agents//state/stream` relays that subject as SSE, diff --git a/docs/swarm/ui.md b/docs/swarm/ui.md index 7b89ea64..ccc7e733 100644 --- a/docs/swarm/ui.md +++ b/docs/swarm/ui.md @@ -93,8 +93,10 @@ The preview carries only the connection badge and **expand**, so it stays clear of the text. What the agent is doing shows in the details table above it, in the **state** row: the turn state (idle, thinking or compacting), a **paused** badge for a paused agent, and the tokens in its context and billed. The row -reads "no report yet" until the first report arrives, and for an agent that's -offline. The agent sends a report when its state changes and once a minute. If +shows the latest report the controller holds, if it's under 150 seconds old, as +soon as you look. It reads "no report yet" when there is none: for an agent +that's offline, and until the first report after the controller restarts. The +agent sends a report when its state changes and once a minute. If the browser gives up on the stream, a **reconnect** badge opens it again. The terminal's own tab keeps these badges, with the model and the effort, beside the connection badge; the turn-state badge says how long the agent has been in that diff --git a/swarm-controller/src/agent_state_stream.rs b/swarm-controller/src/agent_state_stream.rs index dc990428..7189dcf3 100644 --- a/swarm-controller/src/agent_state_stream.rs +++ b/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())) } diff --git a/swarm-controller/src/inference.rs b/swarm-controller/src/inference.rs index 02437dd1..3097befb 100644 --- a/swarm-controller/src/inference.rs +++ b/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>, + /// Each agent's newest header exactly as published, with when it came. + raw: Mutex)>>, } 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> { + 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::(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();