diff --git a/docs/swarm/README.md b/docs/swarm/README.md index 3d085e0f..c09e404e 100644 --- a/docs/swarm/README.md +++ b/docs/swarm/README.md @@ -403,36 +403,6 @@ 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 d75ba543..55f99144 100644 --- a/hive-agent/src/main.rs +++ b/hive-agent/src/main.rs @@ -28,7 +28,6 @@ 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; @@ -510,10 +509,6 @@ 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 deleted file mode 100644 index 4c7a988e..00000000 --- a/hive-agent/src/swarm_agent_state.rs +++ /dev/null @@ -1,486 +0,0 @@ -//! 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 `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 1e7145da..4541fef2 100644 --- a/nix/host-modules/swarm-nats.nix +++ b/nix/host-modules/swarm-nats.nix @@ -793,14 +793,6 @@ 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 a2682c6f..e764d29f 100644 --- a/nix/module-eval.nix +++ b/nix/module-eval.nix @@ -910,25 +910,6 @@ 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 deleted file mode 100644 index a9adc67c..00000000 --- a/swarm-controller/src/agent_state_stream.rs +++ /dev/null @@ -1,133 +0,0 @@ -//! `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 0f254506..bd5c674c 100644 --- a/swarm-controller/src/main.rs +++ b/swarm-controller/src/main.rs @@ -40,7 +40,6 @@ 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; @@ -1677,7 +1676,6 @@ 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))