diff --git a/docs/swarm/README.md b/docs/swarm/README.md index 57ef34b7..03fb6f1d 100644 --- a/docs/swarm/README.md +++ b/docs/swarm/README.md @@ -230,13 +230,14 @@ a destroyed one doesn't exist — so a view that needs the full four-state pictu takes them from the `agent-status` bucket and uses this subject to sharpen the rest. -Headers go out **on transition, not on a timer**: the harness rebuilds the -header whenever its event bus moves and publishes only when the result differs -from what it last sent. 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." The cost of a core subject is that a -subscriber attaching mid-idle sees nothing until the next change, so a renderer -opens with the bucket's snapshot and lets this stream refine it. +Headers go out **on transition**: the harness rebuilds the header whenever its +event bus moves and publishes when the result differs from what it last sent. +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 +snapshot and lets this stream refine it. Swarm-side, `GET /api/agents//state/stream` relays that subject as SSE, resolving the agent's hive at request time exactly as the terminal stream does. diff --git a/docs/swarm/inference.md b/docs/swarm/inference.md index 5de2d546..eeb3ef30 100644 --- a/docs/swarm/inference.md +++ b/docs/swarm/inference.md @@ -51,10 +51,9 @@ Each running value names where it comes from: When the agent starts without a swarm value, its agent.nix model wins over its `/model` pick, and its `/effort` pick wins over its agent.nix effort. -The agent reports these whenever they change. swarm-controller keeps only -what it has received since it started, so after a controller restart `get` -shows `running: not reported…` for an agent until its next change (its next -turn, or a setting changing). +The agent reports these whenever they change, and again once a minute. +swarm-controller keeps only what it has received since it started, so for up +to a minute after a controller restart `get` shows `running: not reported…`. ## When an agent picks a change up diff --git a/frontend/packages/swarm-ui/src/pages/agents/useSwarmAgentStateStream.ts b/frontend/packages/swarm-ui/src/pages/agents/useSwarmAgentStateStream.ts index c4ebb6ca..a8f8d026 100644 --- a/frontend/packages/swarm-ui/src/pages/agents/useSwarmAgentStateStream.ts +++ b/frontend/packages/swarm-ui/src/pages/agents/useSwarmAgentStateStream.ts @@ -7,13 +7,14 @@ // coalesce: each event is just the latest value. // // **No seed on attach.** `agent_state_stream.rs`'s own module doc names -// this limitation: the subject is published on transition only, so an -// agent that has sat idle since before this hook mounted announces -// nothing until its next change. `header` stays `null` until the first -// event lands — callers should render no badges in that gap rather than -// a stale or guessed one (there is no swarm-level endpoint today that -// can seed turn_state/model/ctx/cost synchronously — `GET -// /api/agents/status`'s snapshot carries only `status_text`/`running`). +// this limitation: the subject is published on transition and re-sent +// once a minute, so an agent that has sat idle since before this hook +// mounted announces nothing for up to a minute. `header` stays `null` +// until the first event lands — callers should render no badges in that +// gap rather than a stale or guessed one (there is no swarm-level +// endpoint today that can seed turn_state/model/ctx/cost synchronously — +// `GET /api/agents/status`'s snapshot carries only +// `status_text`/`running`). import { useEffect, useState } from "preact/hooks"; export type ConnectionState = "connecting" | "open" | "reconnecting" | "closed"; diff --git a/hive-agent/Cargo.toml b/hive-agent/Cargo.toml index 3ef9f6da..d4bc292a 100644 --- a/hive-agent/Cargo.toml +++ b/hive-agent/Cargo.toml @@ -52,8 +52,11 @@ tokio-stream.workspace = true tower-http.workspace = true tracing.workspace = true +# `test-util` for `#[tokio::test(start_paused = true)]`: `swarm_agent_state`'s +# refresh test waits out its one-minute interval on paused time. [dev-dependencies] tempfile = "3" +tokio = { workspace = true, features = ["test-util"] } # Single harness serve-loop binary: `hive-agent` (from `src/main.rs`). # The sibling MCP server is its own bin crate now (`hive-agent-mcp`). diff --git a/hive-agent/src/swarm_agent_state.rs b/hive-agent/src/swarm_agent_state.rs index 5c498ea3..5794b38b 100644 --- a/hive-agent/src/swarm_agent_state.rs +++ b/hive-agent/src/swarm_agent_state.rs @@ -8,9 +8,8 @@ //! offers the same values upward on their own subject so a swarm-level //! header can render an agent without reaching into its hive. //! -//! **Published on transition, not on a timer.** The whole reason this -//! exists is that the once-a-minute bucket is too stale, so a second -//! periodic publisher would reproduce the problem it is here to fix. This +//! **Published on transition.** The whole reason this exists is that the +//! once-a-minute bucket is too stale for "is it thinking right now". This //! subscribes to the event bus and republishes whenever the snapshot it //! builds differs from the one it last sent — see [`run`] for why the //! comparison is on the serialized bytes rather than on an event allow-list. @@ -18,18 +17,20 @@ //! **A core subject, not `JetStream`**, for the reason //! [`crate::swarm_term`] gives at more length: nothing here is worth the //! durability of the notices stream, and it keeps the agent's grant to a -//! plain publish. It does cost this subject the one thing a header would -//! like and a terminal would not — a late subscriber sees nothing until the -//! next transition, rather than the current value. Accepted, because the -//! alternative is a KV bucket written on every turn-state flip, and the -//! swarm already has one of those at the resolution it can afford. +//! plain publish. The queue keeps no header for a late subscriber, so the +//! current header is also re-sent every [`REFRESH_INTERVAL`]: a reader that +//! attaches while the agent is idle (swarm-controller after a restart) has +//! it within that interval. Transitions still go out as they happen. //! //! **Best-effort per transition.** 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 publish is a log line and the next transition is still attempted. +use std::time::Duration; + use serde::Serialize; use tokio::sync::broadcast; +use tokio::time::{Interval, MissedTickBehavior}; use swarm_queue_client::agent_inference::Effective; use swarm_queue_client::wanted::AgentState; @@ -42,6 +43,9 @@ use crate::term_msg::iso8601_utc; /// Same margin and same reasoning as `swarm_term::HEADROOM`. const HEADROOM: usize = 1024; +/// How often the current header is re-sent when nothing changed it. +const REFRESH_INTERVAL: Duration = Duration::from_mins(1); + /// One turn-state header, as a swarm-level reader receives it. /// /// The field names here are the published contract; two of them @@ -170,38 +174,75 @@ pub fn spawn(bus: &Bus) { /// and it carries nothing this header reads. Every other variant, including /// any added later, funnels into the comparison and costs nothing when it /// changes nothing. +/// +/// A [`Trigger::Refresh`] re-sends the header even when it is unchanged. async fn run(mut rx: broadcast::Receiver, bus: Bus, agent: String) { let client = crate::swarm_queue::client().await; let subject = swarm_queue_client::agent_state_subject(&agent); tracing::info!(subject, "publishing agent turn state to the swarm queue"); // The last payload actually sent, so a rebuild that changed nothing is - // dropped here instead of on the wire. `None` until the first publish, - // which is what makes the boot-time header go out at all: an agent that - // sat idle from boot would otherwise never announce itself. + // dropped here instead of on the wire. `None` until the first publish. let mut last: Option> = None; - loop { - match rx.recv().await { - // Skipped before the rebuild — see this function's doc. - Ok(BusEvent { - event: LiveEvent::Stream(_), - .. - }) => continue, - Ok(_) => {} - // The bus drops events for a subscriber that falls behind. A - // header is a current value rather than a log, so a missed - // event costs nothing here: the rebuild below reads the state - // that the missed events led to, not the events themselves. - Err(broadcast::error::RecvError::Lagged(missed)) => { - tracing::debug!(missed, "swarm agent state: lagged, rebuilding anyway"); - } - // Every sender is gone, so the harness is shutting down. - Err(broadcast::error::RecvError::Closed) => return, + let mut refresh = refresh_timer(); + while let Some(trigger) = next_trigger(&mut rx, &mut refresh).await { + if trigger == Trigger::Refresh { + last = None; } publish(&client, &subject, &snapshot(&bus), &mut last).await; } } +/// Why a header is offered. +#[derive(Debug, PartialEq, Eq)] +enum Trigger { + /// A bus event that may have changed it. + Event, + /// [`REFRESH_INTERVAL`] passed. + Refresh, +} + +/// Ticks every [`REFRESH_INTERVAL`], the first one an interval from now, +/// since the harness's boot events already send the first header. +fn refresh_timer() -> Interval { + let mut timer = tokio::time::interval_at( + tokio::time::Instant::now() + REFRESH_INTERVAL, + REFRESH_INTERVAL, + ); + timer.set_missed_tick_behavior(MissedTickBehavior::Delay); + timer +} + +/// The next reason to offer a header; `None` once the bus is closed. +async fn next_trigger( + rx: &mut broadcast::Receiver, + refresh: &mut Interval, +) -> Option { + loop { + tokio::select! { + _ = refresh.tick() => return Some(Trigger::Refresh), + event = rx.recv() => match event { + // Skipped before the rebuild — see [`run`]'s doc. + Ok(BusEvent { + event: LiveEvent::Stream(_), + .. + }) => {} + Ok(_) => return Some(Trigger::Event), + // The bus drops events for a subscriber that falls behind. A + // header is a current value rather than a log, so a missed + // event costs nothing here: the rebuild reads the state that + // the missed events led to, not the events themselves. + Err(broadcast::error::RecvError::Lagged(missed)) => { + tracing::debug!(missed, "swarm agent state: lagged, rebuilding anyway"); + return Some(Trigger::Event); + } + // Every sender is gone, so the harness is shutting down. + Err(broadcast::error::RecvError::Closed) => return None, + }, + } + } +} + /// Offer one header, if it differs from the last one sent. Every failure is /// terminal for that publish and for nothing else. /// @@ -258,10 +299,12 @@ async fn publish( #[cfg(test)] mod tests { - use super::AgentStateMsg; - use crate::events::TurnState; + use super::{AgentStateMsg, REFRESH_INTERVAL, Trigger, next_trigger, refresh_timer}; + use crate::events::{BusEvent, LiveEvent, TurnState}; use swarm_queue_client::agent_inference::{Effective, Source}; use swarm_queue_client::wanted::AgentState; + use tokio::sync::broadcast; + use tokio::time::{Duration, Instant}; /// `TokenUsage` is `#[non_exhaustive]`, so a downstream crate cannot /// write one as a struct expression at all — see `turn_stats`'s note on @@ -404,4 +447,49 @@ mod tests { assert_eq!(value["turn_state"], serde_json::json!("idle")); assert_eq!(value["agent_state"], serde_json::json!("up")); } + + fn event(event: LiveEvent) -> BusEvent { + BusEvent { + seq: 0, + ts: 0, + event, + } + } + + /// An idle bus still yields a refresh once per interval; a bus event + /// yields at once, a stream line alone yields nothing, and a closed bus + /// ends the loop. + #[tokio::test(start_paused = true)] + async fn an_idle_header_is_refreshed_once_per_interval() { + let (tx, mut rx) = broadcast::channel(16); + let mut refresh = refresh_timer(); + let start = Instant::now(); + assert_eq!( + next_trigger(&mut rx, &mut refresh).await, + Some(Trigger::Refresh) + ); + assert_eq!(start.elapsed(), REFRESH_INTERVAL); + + tx.send(event(LiveEvent::Stream(serde_json::json!({})))) + .unwrap(); + tx.send(event(LiveEvent::Note { + text: "claude exited 0".to_owned(), + })) + .unwrap(); + let at = Instant::now(); + assert_eq!( + next_trigger(&mut rx, &mut refresh).await, + Some(Trigger::Event) + ); + assert_eq!(at.elapsed(), Duration::ZERO); + + assert_eq!( + next_trigger(&mut rx, &mut refresh).await, + Some(Trigger::Refresh) + ); + assert_eq!(start.elapsed(), 2 * REFRESH_INTERVAL); + + drop(tx); + assert_eq!(next_trigger(&mut rx, &mut refresh).await, None); + } } diff --git a/swarm-controller/src/agent_state_stream.rs b/swarm-controller/src/agent_state_stream.rs index c1482f1a..ebd02eb4 100644 --- a/swarm-controller/src/agent_state_stream.rs +++ b/swarm-controller/src/agent_state_stream.rs @@ -8,11 +8,11 @@ //! the agent's current state. //! //! **That difference is the one thing worth reading twice.** `hive-agent` -//! publishes a header on transition, to a core subject — so a client that -//! attaches mid-idle sees nothing until the next change, which for an agent -//! parked overnight can be hours. A renderer therefore has to open with -//! whatever `GET /api/agents/status` already gives it and let this stream -//! sharpen it, rather than treating an empty stream as an empty agent. The +//! publishes a header on transition and re-sends it once a minute, to a core +//! subject — so a client that attaches mid-idle sees nothing for up to a +//! minute. A renderer therefore has to open with whatever +//! `GET /api/agents/status` already gives it and let this stream sharpen +//! it, rather than treating an empty stream as an empty agent. The //! alternative would be a KV bucket written on every turn-state flip, which //! is the write rate this feature exists to avoid. //! @@ -40,8 +40,8 @@ use crate::AppState; turn-state header, JSON as hive-agent published it (opaque to this relay) — \ `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`. Sent on \ - transition only, live, no replay: an idle agent emits nothing until it changes", + `ctx_usage`, `cost_usage`. Sent on transition and re-sent once a minute, live, \ + no replay", body = String, content_type = "text/event-stream"), (status = 400, description = "the agent name is not shaped like an identifier \ (problem+json)", body = String), diff --git a/swarm-controller/src/inference.rs b/swarm-controller/src/inference.rs index 364f1386..343e5bb0 100644 --- a/swarm-controller/src/inference.rs +++ b/swarm-controller/src/inference.rs @@ -8,8 +8,9 @@ //! //! What the agent runs on comes from its turn-state header //! (`$SWARM.agent-state.`), which [`Reported`] keeps the latest of. -//! Agents publish a header only when it changes and the queue keeps none, so -//! an agent that has not changed since this daemon started has no entry. +//! 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. use std::collections::{BTreeSet, HashMap}; use std::sync::{Arc, Mutex, PoisonError}; diff --git a/swarmctl/src/inference.rs b/swarmctl/src/inference.rs index 9eae3b3c..6cdb8db5 100644 --- a/swarmctl/src/inference.rs +++ b/swarmctl/src/inference.rs @@ -151,8 +151,8 @@ fn render(s: &Settings) -> String { fn render_effective(e: Option<&Effective>) -> String { e.map_or_else( || { - "running: not reported since swarm-controller started; the agent reports on its \ - next change\n" + "running: not reported since swarm-controller started; the agent reports within a \ + minute\n" .to_owned() }, |e| {