diff --git a/docs/swarm/README.md b/docs/swarm/README.md index c09e404e..80e9a640 100644 --- a/docs/swarm/README.md +++ b/docs/swarm/README.md @@ -403,6 +403,36 @@ take the connection down with it, so the harness drops such a row's body before sending and leaves a marker in its place; the summary, level and icon still arrive. A row that's too large even without its body is logged and skipped. +The second thing an agent publishes is its **turn-state header**, on +`$SWARM.agent-state..` — same shape of subject, same grant +mechanics, same lack of retention. It carries what a header bar wants: what the +turn loop is doing (`turn_state`, plus `turn_state_since` as an ISO 8601 UTC +stamp), which model (`model` and the resolved id the last turn actually ran on), +the context budget and the last turn's context and cost token blocks, and +`agent_state`. + +`agent_state` reuses the swarm's own wanted-state vocabulary +(`up`/`offline`/`paused`/`destroyed`) so a reader can compare what an agent _is_ +against what the swarm declared it should be without translating between two +spellings. ⚠️ From inside the container only two of those four are sayable: the +harness reports `up`, or `paused` when the pause marker is present. `offline` and +`destroyed` are hive-c0re's observations — a stopped agent publishes nothing and +a destroyed one doesn't exist — so a view that needs the full four-state picture +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. + +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. +The payload passes through opaquely — the controller never parses a header. + ### Swarm-wide forge webhooks At startup the controller registers two Forgejo hooks pointing at diff --git a/hive-agent/src/main.rs b/hive-agent/src/main.rs index 55f99144..d75ba543 100644 --- a/hive-agent/src/main.rs +++ b/hive-agent/src/main.rs @@ -28,6 +28,7 @@ mod serve_common; mod state_entry_watch; mod stats; mod stream_enrich; +mod swarm_agent_state; mod swarm_queue; mod swarm_term; mod term_msg; @@ -509,6 +510,10 @@ async fn serve_main(socket: &Path, poll_ms: u64) -> Result<()> { // Subscribes before anything is emitted, so the swarm sees this agent's // terminal from its first row rather than from whenever a reader attached. swarm_term::spawn(bus.subscribe()); + // Same reason, one subject over: the swarm's header bar wants the turn + // state from this agent's first transition, not from whenever a reader + // first asked. + swarm_agent_state::spawn(&bus); // Set by the web UI's `/api/cancel` on a successful SIGINT, read-and- // cleared by `handle_turn` before building the next wake prompt — see // `hive_sh4re::inbox::INTERRUPTED_HINT`. Shared between the web server task and diff --git a/hive-agent/src/swarm_agent_state.rs b/hive-agent/src/swarm_agent_state.rs new file mode 100644 index 00000000..4a21bfbf --- /dev/null +++ b/hive-agent/src/swarm_agent_state.rs @@ -0,0 +1,486 @@ +//! Publishing this agent's turn-state header onto the swarm queue. +//! +//! The per-agent web UI's `/api/state` already carries what a header bar +//! wants — what the turn loop is doing, since when, on which model, how much +//! context and cost the last turn spent. None of it reaches the swarm: the +//! `agent-status` KV bucket republishes once a minute, which is fine for +//! "is this agent alive" and useless for "is it thinking right now". This +//! 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 +//! 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. +//! +//! **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. +//! +//! **Best-effort in every direction.** The turn loop and the web UI must not +//! notice whether the queue exists, 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 serde::Serialize; +use tokio::sync::broadcast; + +use swarm_queue_client::wanted::AgentState; + +use crate::events::{Bus, BusEvent, LiveEvent, TurnState}; +use crate::term_msg::iso8601_utc; + +/// Subject family carrying agent turn-state headers, the swarm-wide +/// agreement this publisher holds up its end of. One leaf subject per +/// agent, matching `$SWARM.term`'s shape, so a subscriber can follow one +/// agent without filtering the swarm's whole header traffic. +const SUBJECT_PREFIX: &str = "$SWARM.agent-state"; + +/// What the queue's client id looks like either side of the hive's own name. +/// +/// These mirror the auth-callout responder's `--hive-client-prefix` and +/// `--agent-client-suffix`, which is where the authoritative pair lives: the +/// responder parses the hive back out of the presented client id and grants +/// exactly the configured subjects for *that* hive. The agent is not told +/// those flags, so it restates their defaults — a deployment that retunes +/// either one has to change them here too, and the symptom of not doing so +/// is every publish refused rather than a wrong subject being accepted. +const CLIENT_ID_PREFIX: &str = "hive-"; +const CLIENT_ID_SUFFIX: &str = "-agent"; + +/// How much room to leave under the announced limit for everything the +/// publish adds around the payload — subject, headers, protocol framing. +/// Same margin and same reasoning as [`crate::swarm_term::HEADROOM`]. +const HEADROOM: usize = 1024; + +/// One turn-state header, as a swarm-level reader receives it. +/// +/// The field names here are the published contract; two of them +/// deliberately do **not** match the per-agent web UI's `StateSnapshot`, +/// which is the other place these same values are served from: +/// +/// - `turn_state_since` is an ISO 8601 / RFC 3339 UTC string, where +/// `StateSnapshot` sends unix seconds. The sibling subject's +/// [`crate::term_msg::TermMsg::ts`] is already spelled that way, and two +/// messages on adjacent subjects disagreeing about how to write a time is +/// a cost paid by every reader of both. +/// - `agent_state` replaces `StateSnapshot`'s `paused: bool` with the +/// swarm's own [`AgentState`] — the same enum the *wanted* state is +/// declared in, so a reader can compare actual against wanted directly +/// instead of translating one into the other's vocabulary first. +/// +/// `turn_state` and `agent_state` are two axes, not one: [`AgentState`] has +/// no notion of `thinking` and [`TurnState`] has no notion of `offline`, so +/// a single field would have to drop one of them. +#[derive(Debug, Clone, PartialEq, Eq, Serialize)] +struct AgentStateMsg { + /// What the turn loop is doing right now. + turn_state: TurnState, + /// When it entered that state, ISO 8601 / RFC 3339 UTC — see the struct + /// doc. A reader ticks the elapsed time off this rather than being sent + /// an age that is wrong the moment it is serialized. + turn_state_since: String, + /// What this agent can honestly say it *is* — see [`observed_state`], + /// which documents why only two of the four variants can ever appear + /// here. + agent_state: AgentState, + /// The alias `claude --model` was invoked with (e.g. `"sonnet"`). + model: String, + /// The concrete model id the most recent completed turn actually ran + /// on, resolved from that turn's assistant events. `None` before any + /// turn has completed. + resolved_model: Option, + /// Effective context-window token budget for the current model. + context_window_tokens: u64, + /// Last-inference token usage from the most recent completed turn — the + /// current context-window occupancy. `None` until the first turn ends. + ctx_usage: Option, + /// Cumulative token usage across the most recent turn's inferences, the + /// cost signal. `None` until the first turn ends. + cost_usage: Option, +} + +/// The hive a queue client id names. +/// +/// The hive in the subject has to be the string the **responder** parses out +/// of this same client id, because that is what its grant is built from. The +/// harness knows a hive display name too, from a different source and with +/// no rule tying the two together — deriving the subject from that one +/// instead produces a publish the broker refuses, which surfaces as a header +/// that is simply never populated. +/// +/// `None` for an id that does not have the expected shape: a publisher that +/// guessed at a subject would be asking for a grant it cannot have, so the +/// caller disables itself instead. +fn hive_from_client_id(client_id: &str) -> Option<&str> { + let hive = client_id + .strip_prefix(CLIENT_ID_PREFIX)? + .strip_suffix(CLIENT_ID_SUFFIX)?; + (!hive.is_empty()).then_some(hive) +} + +/// What this process can honestly report about the agent-state axis. +/// +/// Only two of [`AgentState`]'s four variants are knowable from inside the +/// container, and the split is not arbitrary: +/// +/// - [`AgentState::Up`] and [`AgentState::Paused`] are the agent's own +/// facts. Running is self-evident — this code is executing — and the +/// pause marker is a file in the harness dir that this process reads on +/// every turn-loop iteration anyway (`crate::paths::paused_marker`). +/// - [`AgentState::Offline`] and [`AgentState::Destroyed`] are **not +/// reportable from here, ever**. A stopped agent publishes nothing and a +/// destroyed one does not exist, so a message claiming either would be a +/// message from a process contradicting itself. Those two remain +/// hive-c0re's to observe and the `agent-status` bucket's to carry; a +/// reader wanting the full four-state picture needs both sources, and +/// this one's silence is the only evidence it can offer for the other +/// two. +/// +/// Read fresh per publish rather than cached: the marker is written by +/// hive-c0re (through hive-priv) and by this process's own auto-pause, so +/// nothing in here observes every write. +fn observed_state() -> AgentState { + if crate::paths::paused_marker().exists() { + AgentState::Paused + } else { + AgentState::Up + } +} + +/// Build the current header from the bus plus the pause marker. +fn snapshot(bus: &Bus) -> AgentStateMsg { + let (turn_state, since_unix) = bus.state_snapshot(); + let model = bus.model(); + AgentStateMsg { + turn_state, + turn_state_since: iso8601_utc(since_unix), + agent_state: observed_state(), + context_window_tokens: bus.effective_context_window(&model), + model, + resolved_model: bus.last_resolved_model(), + ctx_usage: bus.last_ctx_usage(), + cost_usage: bus.last_cost_usage(), + } +} + +/// Start the publish task, if this agent has both queue coordinates and an +/// identity the subject can be derived from. +/// +/// Returns without spawning in every other case — no queue, an unparseable +/// client id, no label — each of which is a legal state for an agent rather +/// than an error, and each logged once here rather than per transition. +pub fn spawn(bus: &Bus) { + let Some(cfg) = crate::swarm_queue::config() else { + // `swarm_queue::init` already said why at boot; repeating it here + // would be the same fact logged twice. + return; + }; + let Some(hive) = hive_from_client_id(&cfg.client_id) else { + tracing::warn!( + client_id = %cfg.client_id, + expected = format!("{CLIENT_ID_PREFIX}{CLIENT_ID_SUFFIX}"), + "queue client id does not name a hive; not publishing turn state upward" + ); + return; + }; + let agent = crate::identity::label(); + if agent.is_empty() { + tracing::warn!("this agent has no label; not publishing turn state upward"); + return; + } + let subject = format!("{SUBJECT_PREFIX}.{hive}.{agent}"); + tokio::spawn(run(bus.subscribe(), bus.clone(), subject)); +} + +/// Watch the bus and publish whenever the header actually changed. +/// +/// **Why the change test is on the serialized bytes rather than on which +/// event arrived.** The header is assembled from six independent pieces of +/// `Bus` state, only some of which announce themselves with an event of +/// their own — `set_resolved_model`, for one, emits nothing and lands +/// alongside the `TokenUsageChanged` of the same turn. An allow-list of +/// interesting variants would therefore have to encode which event happens +/// to be adjacent to each field's write, and would go quietly stale the +/// first time a field moved. Rebuilding on (almost) any event and comparing +/// the result is the same publish traffic with none of that coupling: a +/// rebuild is a handful of mutex reads and one `stat`. +/// +/// `Stream` is the exception, skipped before the rebuild: it is the only +/// high-rate variant — one per `stream-json` line, so hundreds per turn — +/// 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. +async fn run(mut rx: broadcast::Receiver, bus: Bus, subject: String) { + let Some(client) = crate::swarm_queue::client().await else { + return; + }; + 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. + 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, + } + publish(&client, &subject, &snapshot(&bus), &mut last).await; + } +} + +/// Offer one header, if it differs from the last one sent. Every failure is +/// terminal for that publish and for nothing else. +/// +/// `last` is only updated on a publish the client accepted, so a transition +/// lost to a transport failure is re-sent by the next one rather than +/// deduped away against a value the swarm never saw. +async fn publish( + client: &async_nats::Client, + subject: &str, + msg: &AgentStateMsg, + last: &mut Option>, +) { + let payload = match serde_json::to_vec(msg) { + Ok(payload) => payload, + Err(e) => { + tracing::warn!(error = %e, "swarm agent state: serialising failed, header dropped"); + return; + } + }; + if last.as_ref() == Some(&payload) { + return; + } + // An unconnected client does not fail a publish, it buffers it — and the + // limit read below is the library's pre-connect default until the server + // has announced its own, which is smaller than any deployment sets. + if let Err(e) = swarm_queue_client::ensure_connected(client) { + tracing::warn!(error = %swarm_queue_client::chain(&e), "swarm agent state: publish skipped"); + return; + } + // No degrade path, unlike `swarm_term`'s: every field here is a number, + // an enum, or a model name, so there is no arbitrary-length body worth + // spending to get under the limit. The check stays because being one + // byte over does not truncate the message, it gets it refused and the + // connection closed — which would cost every header racing behind it + // through the reconnect, not just this one. + let limit = swarm_queue_client::max_payload(client).saturating_sub(HEADROOM); + if payload.len() > limit { + tracing::warn!( + len = payload.len(), + limit, + "swarm agent state: header over the payload limit, dropped" + ); + return; + } + if let Err(e) = client + .publish(subject.to_owned(), payload.clone().into()) + .await + { + tracing::warn!(error = %e, "swarm agent state: publish failed, header dropped"); + return; + } + *last = Some(payload); +} + +#[cfg(test)] +mod tests { + use super::{AgentStateMsg, SUBJECT_PREFIX, hive_from_client_id}; + use crate::events::TurnState; + use swarm_queue_client::wanted::AgentState; + + /// `TokenUsage` is `#[non_exhaustive]`, so a downstream crate cannot + /// write one as a struct expression at all — see `turn_stats`'s note on + /// the same restriction. Deserializing one is the shortest way to a + /// populated fixture here, and it keeps the numbers next to the field + /// names they belong to. + /// + /// ⚠️ All four fields, every time: none of them carries a serde default, + /// so a block written with only the field a given test reads fails to + /// deserialize at all rather than zero-filling the rest. + fn usage(json: &serde_json::Value) -> hive_claude::TokenUsage { + serde_json::from_value(json.clone()).expect("a usage block") + } + + /// A header with every optional field populated, so a test asserting on + /// the serialized shape sees the widest form a reader can receive. + fn msg() -> AgentStateMsg { + AgentStateMsg { + turn_state: TurnState::Thinking, + // As `iso8601_utc` renders a transition time. + turn_state_since: "2026-09-13T12:35:03Z".to_owned(), + agent_state: AgentState::Paused, + model: "sonnet".to_owned(), + resolved_model: Some("claude-sonnet-4-5-20260805".to_owned()), + context_window_tokens: 200_000, + ctx_usage: Some(usage(&serde_json::json!({ + "input_tokens": 11, + "output_tokens": 22, + "cache_read_input_tokens": 33, + "cache_creation_input_tokens": 44, + }))), + // Distinct from `ctx_usage`'s numbers in every field, so a + // renderer served one block where it asked for the other shows + // up as a wrong value rather than as a coincidence. + cost_usage: Some(usage(&serde_json::json!({ + "input_tokens": 55, + "output_tokens": 66, + "cache_read_input_tokens": 77, + "cache_creation_input_tokens": 88, + }))), + } + } + + #[test] + fn a_client_id_names_the_hive_between_the_prefix_and_the_suffix() { + assert_eq!(hive_from_client_id("hive-alpha-agent"), Some("alpha")); + // A hive whose own name contains the suffix still resolves: only the + // trailing one is stripped, so the responder and this agree. + assert_eq!( + hive_from_client_id("hive-alpha-agent-agent"), + Some("alpha-agent") + ); + } + + /// Each way an id can fail to name a hive. A guessed subject would be + /// refused by the broker, and a refusal reaches an operator as a header + /// that never fills in rather than as an error, so none of these may + /// fall back. + #[test] + fn an_id_of_another_shape_names_no_hive() { + for id in [ + // The hive's own id, not an agent's. + "hive-alpha", + // Missing the prefix the responder keys on. + "alpha-agent", + // Prefix and suffix but nothing between them: the subject would + // be `$SWARM.agent-state..`, whose empty token matches no + // grant. + "hive--agent", + // Prefix and suffix overlapping with no hive at all. + "hive-agent", + "", + ] { + assert_eq!(hive_from_client_id(id), None, "{id} must name no hive"); + } + } + + /// The subject the swarm side subscribes to, built end to end from the + /// two things that identify this publisher. `swarm-controller`'s relay + /// spells the same prefix in its own constant and the two never meet, so + /// this pins the half that lives here. + #[test] + fn the_subject_is_the_prefix_then_the_hive_then_the_agent() { + let hive = hive_from_client_id("hive-alpha-agent").expect("names a hive"); + assert_eq!( + format!("{SUBJECT_PREFIX}.{hive}.mara"), + "$SWARM.agent-state.alpha.mara" + ); + } + + /// The published contract, asserted on the JSON a subscriber parses + /// rather than on the Rust struct: a renderer is written against these + /// key names, and renaming a field in Rust without noticing is exactly + /// the failure this catches. + #[test] + fn the_payload_carries_the_contracts_field_names() { + let value: serde_json::Value = + serde_json::from_slice(&serde_json::to_vec(&msg()).expect("serialises")) + .expect("valid json"); + let obj = value.as_object().expect("an object"); + let mut keys: Vec<&str> = obj.keys().map(String::as_str).collect(); + keys.sort_unstable(); + assert_eq!( + keys, + [ + "agent_state", + "context_window_tokens", + "cost_usage", + "ctx_usage", + "model", + "resolved_model", + "turn_state", + "turn_state_since", + ] + ); + } + + /// Both enums cross the wire as the snake-case strings their `serde` + /// attributes promise, not as Rust variant names — and the two stay + /// separate fields, since neither vocabulary contains the other's values. + #[test] + fn the_two_state_axes_serialise_as_their_own_snake_case_strings() { + let value = serde_json::to_value(msg()).expect("serialises"); + assert_eq!(value["turn_state"], serde_json::json!("thinking")); + assert_eq!(value["agent_state"], serde_json::json!("paused")); + } + + /// The departure from `StateSnapshot` that a reader is most likely to + /// get wrong: this field is a string, and a number here would parse as a + /// valid — and completely wrong — date on the other side. + #[test] + fn turn_state_since_is_an_iso_8601_string_not_unix_seconds() { + let value = serde_json::to_value(msg()).expect("serialises"); + assert_eq!( + value["turn_state_since"], + serde_json::json!("2026-09-13T12:35:03Z") + ); + } + + /// The token blocks ship whole rather than pre-summed: a header renderer + /// wanting the context percentage needs the same three fields + /// `TokenUsage::context_tokens` adds up, and one of them alone reads as + /// near-zero once prompt caching is on. + #[test] + fn the_usage_blocks_ship_their_own_fields() { + let value = serde_json::to_value(msg()).expect("serialises"); + assert_eq!(value["ctx_usage"]["cache_read_input_tokens"], 33); + assert_eq!(value["cost_usage"]["input_tokens"], 55); + assert_eq!(value["context_window_tokens"], 200_000); + } + + /// A pre-first-turn header, which is what a freshly booted agent + /// actually publishes. The nullable fields have to be present and null + /// rather than absent — a renderer that reads `ctx_usage` off a header + /// missing the key gets `undefined`, which is not the same thing as + /// "this agent has not finished a turn yet". + #[test] + fn a_header_from_before_the_first_turn_still_carries_every_key() { + let fresh = AgentStateMsg { + turn_state: TurnState::Idle, + agent_state: AgentState::Up, + resolved_model: None, + ctx_usage: None, + cost_usage: None, + ..msg() + }; + let value = serde_json::to_value(fresh).expect("serialises"); + assert_eq!(value["resolved_model"], serde_json::Value::Null); + assert_eq!(value["ctx_usage"], serde_json::Value::Null); + assert_eq!(value["cost_usage"], serde_json::Value::Null); + assert_eq!(value["turn_state"], serde_json::json!("idle")); + assert_eq!(value["agent_state"], serde_json::json!("up")); + } +} diff --git a/nix/host-modules/swarm-nats.nix b/nix/host-modules/swarm-nats.nix index 4541fef2..1e7145da 100644 --- a/nix/host-modules/swarm-nats.nix +++ b/nix/host-modules/swarm-nats.nix @@ -793,6 +793,14 @@ in # responder as the empty expansion of an unset `SWARM` and the # grant silently becomes `.term.{hive}.>`. "--agent-publish-subject ${lib.escapeShellArg "\$\$SWARM.term.{hive}.>"}" + # The turn-state header the swarm's agent view renders from + # (`hive-agent::swarm_agent_state`). Its own subject family + # rather than a leaf under `.term.`: a subscriber following + # one agent's terminal should not also be handed every + # header, and the two have opposite shapes — a terminal is + # an append-only row stream, a header is one current value + # republished on change. + "--agent-publish-subject ${lib.escapeShellArg "\$\$SWARM.agent-state.{hive}.>"}" ]; # Every credential arrives by `LoadCredential` and is named # on the command line only as a **path** — `argv` is diff --git a/nix/module-eval.nix b/nix/module-eval.nix index e764d29f..a2682c6f 100644 --- a/nix/module-eval.nix +++ b/nix/module-eval.nix @@ -910,6 +910,25 @@ let in lib.hasInfix "--agent-publish-subject " exec && lib.hasInfix "$$SWARM.term.{hive}.>" exec; } + { + # Second grant, same escaping trap, asserted separately: the two + # subject families are independent features (terminal rows and the + # turn-state header) and dropping either should fail as its own arm + # rather than being masked by the other still being present. + # + # Flag and argument are matched as one infix rather than as two + # independent `hasInfix` calls: the responder takes the flag + # repeatedly, so the thing worth pinning is that THIS subject is the + # argument of one of them, which two separate presence checks would + # both pass on while the subject sat under some other flag entirely. + name = "the responder grants agents their hive's agent-state subject too"; + ok = + let + exec = + natsOldPath.containers.swarm-nats.config.systemd.services.swarm-nats-auth.serviceConfig.ExecStart; + in + lib.hasInfix "--agent-publish-subject '$$SWARM.agent-state.{hive}.>'" exec; + } { # Reads the host's tmpfiles rules, not the options: the socket directory # nginx and the container share is created there, so a rename that diff --git a/swarm-controller/src/agent_state_stream.rs b/swarm-controller/src/agent_state_stream.rs new file mode 100644 index 00000000..a9adc67c --- /dev/null +++ b/swarm-controller/src/agent_state_stream.rs @@ -0,0 +1,133 @@ +//! `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 resolving the hive at request time rather than +//! naming it in 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. +//! +//! **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 +//! alternative would be a KV bucket written on every turn-state flip, which +//! is the write rate this feature exists to avoid. +//! +//! **The payload is passed through opaquely**, same as the terminal relay: +//! this daemon 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 +//! `hive-agent::swarm_agent_state::publish` sent. + +use std::convert::Infallible; +use std::time::SystemTime; + +use axum::extract::{Path, State}; +use axum::response::sse::{Event, KeepAlive, Sse}; +use futures_util::{Stream, StreamExt as _}; + +use crate::AppState; + +/// Subject family carrying agent turn-state headers — must match +/// `hive-agent::swarm_agent_state::SUBJECT_PREFIX` exactly, since the two +/// ends never see the constant together. `$SWARM.agent-state..` +/// is the full subject; as with the terminal relay's own copy, there is no +/// way to check the two sides agree short of this comment and the module +/// docs on both ends staying honest about it. +const SUBJECT_PREFIX: &str = "$SWARM.agent-state"; + +#[utoipa::path( + get, + path = "/api/agents/{name}/state/stream", + params(("name" = String, Path, description = "agent whose turn state to stream")), + responses( + (status = 200, description = "server-sent event stream; each event's `data` is one \ + turn-state header, JSON as hive-agent published it (opaque to this daemon) — \ + `turn_state`, `turn_state_since` (ISO 8601 UTC), `agent_state`, `model`, \ + `resolved_model`, `context_window_tokens`, `ctx_usage`, `cost_usage`. Sent on \ + transition only, live, no replay: an idle agent emits nothing until it changes", + body = String, content_type = "text/event-stream"), + (status = 400, description = "the agent name is not shaped like an identifier \ + (problem+json)", body = String), + (status = 404, description = "the agent has never reported to the swarm, so no \ + hive is on record to stream its state from (problem+json)", body = String), + (status = 503, description = "no swarm queue is configured on this host, no \ + agent-status reader is wired up, or the queue could not be reached \ + (problem+json)", body = String), + (status = 500, description = "reading the agent's last-known hive failed, or the \ + subscribe itself failed (problem+json)", body = String), + ), + tag = "agents" +)] +pub(crate) async fn stream_agent_state( + State(state): State, + Path(agent): Path, +) -> Result>>, problem_details::ProblemDetails> { + let Some(status) = state.status.as_ref() else { + return Err(crate::error_problem( + axum::http::StatusCode::SERVICE_UNAVAILABLE, + "no swarm queue is configured on this host", + )); + }; + let Some(agent_status) = state.agent_status.as_ref() else { + return Err(crate::error_problem( + axum::http::StatusCode::SERVICE_UNAVAILABLE, + "no agent-status reader is configured on this host", + )); + }; + let agent = hive_types::Ident::parse(&agent) + .map_err(|reason| crate::error_problem(axum::http::StatusCode::BAD_REQUEST, reason))? + .into_string(); + + // One-agent "roster": reuses `AgentStatusReader::view`'s already-tested + // row logic rather than a second, narrower lookup that could disagree + // with it about what "never reported" means. + let hive = agent_status + .view(std::slice::from_ref(&agent), SystemTime::now()) + .await + .map_err(|e| { + let detail = format!("{e:#}"); + tracing::warn!(%agent, error = %detail, "state stream: reading agent status failed"); + crate::error_problem(axum::http::StatusCode::INTERNAL_SERVER_ERROR, &detail) + })? + .into_iter() + .find_map(|row| row.hive) + .ok_or_else(|| { + crate::error_problem( + axum::http::StatusCode::NOT_FOUND, + &format!( + "{agent:?} has never reported to the swarm; no hive is on record to \ + stream its state from" + ), + ) + })?; + + let client = status.queue_client(); + // Before subscribing: an unconnected client does not fail a subscribe + // request outright, but there is no point opening a subscription + // against a queue that is not there — the honest answer is 503, the + // same shape every other queue-backed route here already uses. + swarm_queue_client::ensure_connected(&client).map_err(|e| { + crate::error_problem( + axum::http::StatusCode::SERVICE_UNAVAILABLE, + &swarm_queue_client::chain(&e), + ) + })?; + + let subject = format!("{SUBJECT_PREFIX}.{hive}.{agent}"); + let subscriber = client.subscribe(subject.clone()).await.map_err(|e| { + tracing::warn!(%subject, error = %e, "state stream: subscribe failed"); + crate::error_problem( + axum::http::StatusCode::INTERNAL_SERVER_ERROR, + &format!("subscribing to {subject} failed: {e}"), + ) + })?; + + tracing::info!(%subject, "state stream: client attached"); + let stream = + subscriber.map(|msg| Ok(Event::default().data(String::from_utf8_lossy(&msg.payload)))); + Ok(Sse::new(stream).keep_alive(KeepAlive::default())) +} diff --git a/swarm-controller/src/main.rs b/swarm-controller/src/main.rs index bd5c674c..0f254506 100644 --- a/swarm-controller/src/main.rs +++ b/swarm-controller/src/main.rs @@ -40,6 +40,7 @@ use swarm_authelia_bridge_sock::BridgeResponse; use utoipa::{OpenApi, ToSchema}; use utoipa_axum::{router::OpenApiRouter, routes}; +mod agent_state_stream; mod agent_status; mod auth; mod config_pr; @@ -1676,6 +1677,7 @@ fn build_app(state: AppState) -> axum::Router { .routes(routes!(matrix_account::put_matrix_account)) .routes(routes!(get_hive_wanted)) .routes(routes!(term_stream::stream_agent_term)) + .routes(routes!(agent_state_stream::stream_agent_state)) .routes(routes!(issue_report::get_repos)) .routes(routes!(issue_report::get_issue_report_all)) .routes(routes!(issue_report::get_issue_report))