hive-agent: test the turn-state refresh's re-send decision

Whether a header goes out was split between `run()` (clear `last` on a
refresh) and `publish()` (skip a payload equal to `last`), and only the
refresh cadence was tested: deleting the clear kept every test green.
The decision is now the pure `due(trigger, last, payload)`, which
`publish()` calls, with tests that an unchanged payload is not sent on
an event, is sent on a refresh, a changed one is sent on either, and a
refresh that then fails to send is re-sent by the next event.
Behaviour is unchanged.

Refs #5034

hive-agent/src/swarm_agent_state.rs

@@ -186,13 +186,20 @@ async fn run(mut rx: broadcast::Receiver<BusEvent>, bus: Bus, agent: String) {
     let mut last: Option<Vec<u8>> = 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<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 {
@@ -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<Vec<u8>>,
 ) {
     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));
+    }
 }