diff --git a/hive-agent/src/swarm_agent_state.rs b/hive-agent/src/swarm_agent_state.rs index 5794b38b..880592bc 100644 --- a/hive-agent/src/swarm_agent_state.rs +++ b/hive-agent/src/swarm_agent_state.rs @@ -186,13 +186,20 @@ async fn run(mut rx: broadcast::Receiver, bus: Bus, agent: String) { let mut last: Option> = None; 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; + publish(&client, &subject, &snapshot(&bus), &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>, 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 { @@ -243,8 +250,8 @@ async fn next_trigger( } } -/// Offer one header, if it differs from the last one sent. Every failure is -/// terminal for that publish and for nothing else. +/// 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 @@ -253,6 +260,7 @@ async fn publish( client: &async_nats::Client, subject: &str, msg: &AgentStateMsg, + trigger: &Trigger, last: &mut Option>, ) { let payload = match serde_json::to_vec(msg) { @@ -262,7 +270,7 @@ async fn publish( return; } }; - if last.as_ref() == Some(&payload) { + if !due(trigger, last, &payload) { return; } // An unconnected client does not fail a publish, it buffers it — and the @@ -299,7 +307,7 @@ async fn publish( #[cfg(test)] mod tests { - use super::{AgentStateMsg, REFRESH_INTERVAL, Trigger, next_trigger, refresh_timer}; + use super::{AgentStateMsg, REFRESH_INTERVAL, Trigger, due, next_trigger, refresh_timer}; use crate::events::{BusEvent, LiveEvent, TurnState}; use swarm_queue_client::agent_inference::{Effective, Source}; use swarm_queue_client::wanted::AgentState; @@ -492,4 +500,34 @@ mod tests { 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)); + } }