e5c55257f8
the harness writes the reason into the pause marker when it pauses itself and reports it as paused_by_hive in its turn-state header. swarm-controller declares such an agent paused (only if declared up) and shows the reason on the agents status row and the swarm ui card.
679 lines
28 KiB
Rust
679 lines
28 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.** 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.
|
|
//!
|
|
//! **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. 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::sync::{Arc, Mutex, PoisonError};
|
|
use std::time::Duration;
|
|
|
|
use serde::Serialize;
|
|
use tokio::sync::broadcast;
|
|
use tokio::time::{Interval, MissedTickBehavior};
|
|
|
|
use swarm_wire::agent_inference::Effective;
|
|
use swarm_wire::agent_login::Login;
|
|
use swarm_wire::agent_pause::HivePause;
|
|
use swarm_wire::hive_status::{Capabilities, TOLERANT_PERMISSIONS};
|
|
use swarm_wire::wanted::AgentState;
|
|
|
|
use crate::events::{Bus, BusEvent, LiveEvent, TurnState};
|
|
use crate::login::LoginState;
|
|
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;
|
|
|
|
/// 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
|
|
/// 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,
|
|
/// `model`, `effort`, `model_source` and `effort_source`: what the next
|
|
/// turn runs on and where each comes from. swarm-controller decodes
|
|
/// these four out of the header with the same type.
|
|
#[serde(flatten)]
|
|
inference: Effective,
|
|
/// 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>,
|
|
/// `needs_login`, from this process's own [`LoginState`], so always set.
|
|
#[serde(flatten)]
|
|
login: Login,
|
|
/// `paused_by_hive`, the reason in the pause marker when this harness wrote
|
|
/// it itself; absent otherwise.
|
|
#[serde(flatten)]
|
|
hive_pause: HivePause,
|
|
/// What this harness can do, so swarm-controller grants an agent a
|
|
/// permission only when the harness running it can decode it.
|
|
#[serde(flatten)]
|
|
capabilities: Capabilities,
|
|
}
|
|
|
|
/// 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
|
|
}
|
|
}
|
|
|
|
/// Longest reason reported; the marker is a file the agent can write to.
|
|
const PAUSE_REASON_MAX_CHARS: usize = 200;
|
|
|
|
/// The header's pause-by-harness field: the pause marker's content, when it has
|
|
/// any. The operator's pause writes an empty marker, so it reports nothing.
|
|
fn hive_pause() -> HivePause {
|
|
let reason = std::fs::read_to_string(crate::paths::paused_marker()).unwrap_or_default();
|
|
let reason = reason.trim();
|
|
HivePause {
|
|
paused_by_hive: (!reason.is_empty())
|
|
.then(|| reason.chars().take(PAUSE_REASON_MAX_CHARS).collect()),
|
|
}
|
|
}
|
|
|
|
/// The header's login fields for `state`.
|
|
fn login(state: LoginState) -> Login {
|
|
Login {
|
|
needs_login: Some(state == LoginState::NeedsLogin),
|
|
}
|
|
}
|
|
|
|
/// Build the current header from the bus, the login state and the pause
|
|
/// marker.
|
|
/// The capabilities this build reports.
|
|
fn capabilities() -> Capabilities {
|
|
Capabilities {
|
|
capabilities: vec![TOLERANT_PERMISSIONS.to_owned()],
|
|
}
|
|
}
|
|
|
|
fn snapshot(bus: &Bus, login_state: &Mutex<LoginState>) -> 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),
|
|
inference: Effective {
|
|
model,
|
|
effort: bus.effort(),
|
|
model_source: bus.model_source(),
|
|
effort_source: bus.effort_source(),
|
|
},
|
|
resolved_model: bus.last_resolved_model(),
|
|
ctx_usage: bus.last_ctx_usage(),
|
|
cost_usage: bus.last_cost_usage(),
|
|
login: login(*login_state.lock().unwrap_or_else(PoisonError::into_inner)),
|
|
hive_pause: hive_pause(),
|
|
capabilities: capabilities(),
|
|
}
|
|
}
|
|
|
|
/// Start the publish task, if this agent has a label to name its subject with.
|
|
///
|
|
/// Every writer of `login_state` emits a status on `bus` after the write, so
|
|
/// a login change goes out as an ordinary rebuild.
|
|
pub fn spawn(bus: &Bus, login_state: Arc<Mutex<LoginState>>) {
|
|
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(), login_state, 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 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.
|
|
///
|
|
/// A [`Trigger::Refresh`] re-sends the header even when it is unchanged.
|
|
async fn run(
|
|
mut rx: broadcast::Receiver<BusEvent>,
|
|
bus: Bus,
|
|
login_state: Arc<Mutex<LoginState>>,
|
|
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.
|
|
let mut last: Option<Vec<u8>> = None;
|
|
let mut refresh = refresh_timer();
|
|
while let Some(trigger) = next_trigger(&mut rx, &mut refresh).await {
|
|
let msg = snapshot(&bus, &login_state);
|
|
publish(&client, &subject, &msg, &trigger, &mut last).await;
|
|
}
|
|
}
|
|
|
|
/// Whether `payload` goes out: on a refresh always, and `last` is forgotten
|
|
/// so a refresh that fails is re-sent by the next event; otherwise only when
|
|
/// it differs from `last`.
|
|
fn due(trigger: &Trigger, last: &mut Option<Vec<u8>>, payload: &[u8]) -> bool {
|
|
if *trigger == Trigger::Refresh {
|
|
*last = None;
|
|
}
|
|
last.as_deref() != Some(payload)
|
|
}
|
|
|
|
/// 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<BusEvent>,
|
|
refresh: &mut Interval,
|
|
) -> Option<Trigger> {
|
|
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 [`due`]. 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,
|
|
trigger: &Trigger,
|
|
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 !due(trigger, last, &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");
|
|
crate::otel_turn_metrics::record_swarm_queue_publish_skipped("agent_state");
|
|
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, PAUSE_REASON_MAX_CHARS, REFRESH_INTERVAL, Trigger, capabilities, due,
|
|
hive_pause, login, next_trigger, refresh_timer, snapshot,
|
|
};
|
|
use crate::events::{BusEvent, LiveEvent, TurnState};
|
|
use crate::login::LoginState;
|
|
use std::sync::Mutex;
|
|
use swarm_wire::agent_inference::{Effective, Source};
|
|
use swarm_wire::agent_pause::HivePause;
|
|
use swarm_wire::hive_status::{Capabilities, TOLERANT_PERMISSIONS};
|
|
use swarm_wire::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
|
|
/// 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,
|
|
inference: Effective {
|
|
model: "sonnet".to_owned(),
|
|
effort: "high".to_owned(),
|
|
model_source: Source::Swarm,
|
|
effort_source: Source::AgentNix,
|
|
},
|
|
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,
|
|
}))),
|
|
login: login(LoginState::NeedsLogin),
|
|
hive_pause: HivePause {
|
|
paused_by_hive: Some("skipped get_loose_ends".to_owned()),
|
|
},
|
|
capabilities: capabilities(),
|
|
}
|
|
}
|
|
|
|
/// 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",
|
|
"capabilities",
|
|
"context_window_tokens",
|
|
"cost_usage",
|
|
"ctx_usage",
|
|
"effort",
|
|
"effort_source",
|
|
"model",
|
|
"model_source",
|
|
"needs_login",
|
|
"paused_by_hive",
|
|
"resolved_model",
|
|
"turn_state",
|
|
"turn_state_since",
|
|
]
|
|
);
|
|
}
|
|
|
|
/// swarm-controller reads the inference fields back out of a header.
|
|
#[test]
|
|
fn the_header_decodes_as_its_inference_settings() {
|
|
let bytes = serde_json::to_vec(&msg()).expect("serialises");
|
|
let decoded: Effective = serde_json::from_slice(&bytes).expect("decodes");
|
|
assert_eq!(decoded, msg().inference);
|
|
}
|
|
|
|
/// 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"));
|
|
}
|
|
|
|
/// swarm-controller reads the login flag back out of a header.
|
|
#[test]
|
|
fn the_header_decodes_as_its_login_flag() {
|
|
for state in [LoginState::NeedsLogin, LoginState::Online] {
|
|
let bytes = serde_json::to_vec(&AgentStateMsg {
|
|
login: login(state),
|
|
..msg()
|
|
})
|
|
.expect("serialises");
|
|
let decoded: swarm_wire::agent_login::Login =
|
|
serde_json::from_slice(&bytes).expect("decodes");
|
|
assert_eq!(
|
|
decoded.needs_login,
|
|
Some(state == LoginState::NeedsLogin),
|
|
"{state:?}"
|
|
);
|
|
}
|
|
}
|
|
|
|
/// The marker's content is the reason; an empty marker (an operator's pause) and
|
|
/// no marker report nothing, and swarm-controller reads it back with the wire type.
|
|
#[test]
|
|
fn the_pause_marker_content_is_the_reported_reason() {
|
|
crate::paths::with_harness_dir(|| {
|
|
assert_eq!(hive_pause().paused_by_hive, None, "no marker");
|
|
std::fs::write(crate::paths::paused_marker(), "").unwrap();
|
|
assert_eq!(hive_pause().paused_by_hive, None, "operator pause");
|
|
std::fs::write(crate::paths::paused_marker(), "x".repeat(500)).unwrap();
|
|
assert_eq!(
|
|
hive_pause().paused_by_hive.unwrap().len(),
|
|
PAUSE_REASON_MAX_CHARS
|
|
);
|
|
std::fs::write(crate::paths::paused_marker(), "skipped get_loose_ends\n").unwrap();
|
|
let bytes = serde_json::to_vec(&AgentStateMsg {
|
|
hive_pause: hive_pause(),
|
|
..msg()
|
|
})
|
|
.expect("serialises");
|
|
let decoded: HivePause = serde_json::from_slice(&bytes).expect("decodes");
|
|
assert_eq!(
|
|
decoded.paused_by_hive.as_deref(),
|
|
Some("skipped get_loose_ends")
|
|
);
|
|
});
|
|
}
|
|
|
|
/// swarm-controller reads the harness's capabilities back out of a header.
|
|
#[test]
|
|
fn the_header_advertises_tolerant_permissions() {
|
|
let bytes = serde_json::to_vec(&msg()).expect("serialises");
|
|
let decoded: Capabilities = serde_json::from_slice(&bytes).expect("decodes");
|
|
assert!(decoded.has(TOLERANT_PERMISSIONS));
|
|
}
|
|
|
|
/// A login landing, or a logout, changes the header the next bus event
|
|
/// rebuilds, so it goes out without waiting for a refresh.
|
|
#[test]
|
|
fn a_login_state_change_is_published_on_the_next_event() {
|
|
let (before, after) = crate::paths::with_harness_dir(|| {
|
|
let bus = crate::events::Bus::new();
|
|
let login_state = Mutex::new(LoginState::NeedsLogin);
|
|
let before = serde_json::to_vec(&snapshot(&bus, &login_state)).expect("serialises");
|
|
*login_state.lock().unwrap() = LoginState::Online;
|
|
let after = serde_json::to_vec(&snapshot(&bus, &login_state)).expect("serialises");
|
|
(before, after)
|
|
});
|
|
let mut last = Some(before.clone());
|
|
assert!(!due(&Trigger::Event, &mut last, &before));
|
|
assert!(due(&Trigger::Event, &mut last, &after));
|
|
let after: serde_json::Value = serde_json::from_slice(&after).expect("valid json");
|
|
assert_eq!(after["needs_login"], serde_json::json!(false));
|
|
}
|
|
|
|
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);
|
|
}
|
|
|
|
#[test]
|
|
fn an_unchanged_header_goes_out_on_a_refresh_only() {
|
|
let sent = b"{\"turn_state\":\"idle\"}";
|
|
let mut last = Some(sent.to_vec());
|
|
assert!(!due(&Trigger::Event, &mut last, sent));
|
|
assert!(due(&Trigger::Refresh, &mut last, sent));
|
|
}
|
|
|
|
#[test]
|
|
fn a_changed_header_goes_out_on_either_trigger() {
|
|
let changed = b"{\"turn_state\":\"thinking\"}";
|
|
for trigger in [Trigger::Event, Trigger::Refresh] {
|
|
let mut last = Some(b"{\"turn_state\":\"idle\"}".to_vec());
|
|
assert!(due(&trigger, &mut last, changed), "{trigger:?}");
|
|
}
|
|
// Nothing sent yet.
|
|
assert!(due(&Trigger::Event, &mut None, changed));
|
|
}
|
|
|
|
/// A refresh the client then fails to send leaves nothing to dedup the
|
|
/// next event against, so that event re-sends the header.
|
|
#[test]
|
|
fn a_failed_refresh_is_re_sent_by_the_next_event() {
|
|
let sent = b"{\"turn_state\":\"idle\"}";
|
|
let mut last = Some(sent.to_vec());
|
|
assert!(due(&Trigger::Refresh, &mut last, sent));
|
|
// `publish` records `last` only once the client accepts the payload.
|
|
assert!(due(&Trigger::Event, &mut last, sent));
|
|
}
|
|
}
|