1b061fc9d5
The reader's subscribe allow-list spelled `$SWARM.term.*`, `$SWARM.term.*.sub.*` and `$SWARM.agent-state.*` as literals, copies of private prefixes in hive-agent and swarm-controller. A rename on either side left the policy behind and the stream stopped with only a connection-level server error. swarm-queue-client now owns `term_subject`, `agent_state_subject` and their `*_wildcard` forms, and `subagent_term::subject_wildcard`; hive-agent's publishers, swarm-controller's relays and the policy all build from them. Subjects are byte-identical.
381 lines
17 KiB
Rust
381 lines
17 KiB
Rust
//! 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 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 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;
|
|
|
|
/// 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<String>,
|
|
/// 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<hive_claude::TokenUsage>,
|
|
/// Cumulative token usage across the most recent turn's inferences, the
|
|
/// cost signal. `None` until the first turn ends.
|
|
cost_usage: Option<hive_claude::TokenUsage>,
|
|
}
|
|
|
|
/// 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 a label to name its subject with.
|
|
pub fn spawn(bus: &Bus) {
|
|
let agent = crate::identity::label();
|
|
if agent.is_empty() {
|
|
tracing::warn!("this agent has no label; not publishing turn state upward");
|
|
return;
|
|
}
|
|
tokio::spawn(run(bus.subscribe(), bus.clone(), agent));
|
|
}
|
|
|
|
/// 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<BusEvent>, 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.
|
|
let mut last: Option<Vec<u8>> = 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<Vec<u8>>,
|
|
) {
|
|
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;
|
|
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,
|
|
}))),
|
|
}
|
|
}
|
|
|
|
/// 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"));
|
|
}
|
|
}
|