vinzenz/hyperhive
vinzenz/hyperhive
Watch
0
0
Fork
You've already forked hyperhive
 0
Code Activity

hive-agent: re-send the turn-state header once a minute

Browse source 
The header on `$SWARM.agent-state.<agent>` went out only when a bus event
changed it, and the core subject keeps nothing for a late subscriber. An
idle agent therefore said nothing to a reader that attached after its
last change: swarm-controller after a restart showed `effective: null`
for that agent until its next turn, which can be hours.

The publish loop now also wakes every 60 s and re-sends the current
header through the same `publish` path (clearing the dedup state first).
Transitions still go out as they happen. Same subject, same grant.

An agent whose queue connection is down now logs `swarm agent state:
publish skipped` once a minute even while idle, so the
`hh-swarm-agent-state-publish-skipped` alert also fires for idle
disconnected agents, not only busy ones.

Refs #5034
This commit is contained in:
atlas 2026-10-10 00:25:13 +02:00 • committed by mara
commit 19a045a401
8 changed files with 152 additions and 59 deletions
Download patch file Download diff file Expand all files Collapse all files

148
hive-agent/src/swarm_agent_state.rs
View file

@ -8,9 +8,8 @@
//! offers the same values upward on their own subject so a swarm-level
//! header can render an agent without reaching into its hive.
//!
//! **Published on transition, not on a timer.** The whole reason this
//! exists is that the once-a-minute bucket is too stale, so a second
//! periodic publisher would reproduce the problem it is here to fix. This
//! **Published on transition.** The whole reason this exists is that the
//! once-a-minute bucket is too stale for "is it thinking right now". This
//! subscribes to the event bus and republishes whenever the snapshot it
//! builds differs from the one it last sent — see [`run`] for why the
//! comparison is on the serialized bytes rather than on an event allow-list.
@ -18,18 +17,20 @@
//! **A core subject, not `JetStream`**, for the reason
//! [`crate::swarm_term`] gives at more length: nothing here is worth the
//! durability of the notices stream, and it keeps the agent's grant to a
//! plain publish. It does cost this subject the one thing a header would
//! like and a terminal would not — a late subscriber sees nothing until the
//! next transition, rather than the current value. Accepted, because the
//! alternative is a KV bucket written on every turn-state flip, and the
//! swarm already has one of those at the resolution it can afford.
//! plain publish. The queue keeps no header for a late subscriber, so the
//! current header is also re-sent every [`REFRESH_INTERVAL`]: a reader that
//! attaches while the agent is idle (swarm-controller after a restart) has
//! it within that interval. Transitions still go out as they happen.
//!
//! **Best-effort per transition.** The turn loop and the web UI never wait on
//! the queue, so the bus receiver is the only thing this task blocks on; a
//! failed publish is a log line and the next transition is still attempted.
use std::time::Duration;
use serde::Serialize;
use tokio::sync::broadcast;
use tokio::time::{Interval, MissedTickBehavior};
use swarm_queue_client::agent_inference::Effective;
use swarm_queue_client::wanted::AgentState;
@ -42,6 +43,9 @@ use crate::term_msg::iso8601_utc;
/// Same margin and same reasoning as `swarm_term::HEADROOM`.
const HEADROOM: usize = 1024;
/// How often the current header is re-sent when nothing changed it.
const REFRESH_INTERVAL: Duration = Duration::from_mins(1);
/// One turn-state header, as a swarm-level reader receives it.
///
/// The field names here are the published contract; two of them
@ -170,38 +174,75 @@ pub fn spawn(bus: &Bus) {
/// and it carries nothing this header reads. Every other variant, including
/// any added later, funnels into the comparison and costs nothing when it
/// changes nothing.
///
/// A [`Trigger::Refresh`] re-sends the header even when it is unchanged.
async fn run(mut rx: broadcast::Receiver<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.
// dropped here instead of on the wire. `None` until the first publish.
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,
let mut refresh = refresh_timer();
while let Some(trigger) = next_trigger(&mut rx, &mut refresh).await {
if trigger == Trigger::Refresh {
last = None;
}
publish(&client, &subject, &snapshot(&bus), &mut last).await;
}
}
/// Why a header is offered.
#[derive(Debug, PartialEq, Eq)]
enum Trigger {
/// A bus event that may have changed it.
Event,
/// [`REFRESH_INTERVAL`] passed.
Refresh,
}
/// Ticks every [`REFRESH_INTERVAL`], the first one an interval from now,
/// since the harness's boot events already send the first header.
fn refresh_timer() -> Interval {
let mut timer = tokio::time::interval_at(
tokio::time::Instant::now() + REFRESH_INTERVAL,
REFRESH_INTERVAL,
);
timer.set_missed_tick_behavior(MissedTickBehavior::Delay);
timer
}
/// The next reason to offer a header; `None` once the bus is closed.
async fn next_trigger(
rx: &mut broadcast::Receiver<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 it differs from the last one sent. Every failure is
/// terminal for that publish and for nothing else.
///
@ -258,10 +299,12 @@ async fn publish(
#[cfg(test)]
mod tests {
use super::AgentStateMsg;
use crate::events::TurnState;
use super::{AgentStateMsg, REFRESH_INTERVAL, Trigger, next_trigger, refresh_timer};
use crate::events::{BusEvent, LiveEvent, TurnState};
use swarm_queue_client::agent_inference::{Effective, Source};
use swarm_queue_client::wanted::AgentState;
use tokio::sync::broadcast;
use tokio::time::{Duration, Instant};
/// `TokenUsage` is `#[non_exhaustive]`, so a downstream crate cannot
/// write one as a struct expression at all — see `turn_stats`'s note on
@ -404,4 +447,49 @@ mod tests {
assert_eq!(value["turn_state"], serde_json::json!("idle"));
assert_eq!(value["agent_state"], serde_json::json!("up"));
}
fn event(event: LiveEvent) -> BusEvent {
BusEvent {
seq: 0,
ts: 0,
event,
}
}
/// An idle bus still yields a refresh once per interval; a bus event
/// yields at once, a stream line alone yields nothing, and a closed bus
/// ends the loop.
#[tokio::test(start_paused = true)]
async fn an_idle_header_is_refreshed_once_per_interval() {
let (tx, mut rx) = broadcast::channel(16);
let mut refresh = refresh_timer();
let start = Instant::now();
assert_eq!(
next_trigger(&mut rx, &mut refresh).await,
Some(Trigger::Refresh)
);
assert_eq!(start.elapsed(), REFRESH_INTERVAL);
tx.send(event(LiveEvent::Stream(serde_json::json!({}))))
.unwrap();
tx.send(event(LiveEvent::Note {
text: "claude exited 0".to_owned(),
}))
.unwrap();
let at = Instant::now();
assert_eq!(
next_trigger(&mut rx, &mut refresh).await,
Some(Trigger::Event)
);
assert_eq!(at.elapsed(), Duration::ZERO);
assert_eq!(
next_trigger(&mut rx, &mut refresh).await,
Some(Trigger::Refresh)
);
assert_eq!(start.elapsed(), 2 * REFRESH_INTERVAL);
drop(tx);
assert_eq!(next_trigger(&mut rx, &mut refresh).await, None);
}
}