From 970ee83a9392a75d788a99ffd001297dd92810a6 Mon Sep 17 00:00:00 2001
From: atlas <atlas@hyperhive.local>
Date: Fri, 6 Nov 2026 02:29:46 +0100
Subject: [PATCH] broker: keep unacked ids until the ack update succeeds

Closes #6092
---
 hive-c0re/src/stores/broker.rs | 32 +++++++++++++++++++++++++++++---
 1 file changed, 29 insertions(+), 3 deletions(-)

diff --git a/hive-c0re/src/stores/broker.rs b/hive-c0re/src/stores/broker.rs
index 60003c25..4b67ba9d 100644
--- a/hive-c0re/src/stores/broker.rs
+++ b/hive-c0re/src/stores/broker.rs
@@ -465,8 +465,9 @@ impl Broker {
         Ok(deliveries)
     }
 
-    /// Drain the per-recipient unacked-id list and mark every row
-    /// `acked_at = NOW`. Fired by the harness after `TurnOutcome::Ok`.
+    /// Mark every row in the per-recipient unacked-id list
+    /// `acked_at = NOW`, then drain the list. On a sqlite error the list
+    /// is kept so a retry acks the same rows. Fired by the harness after `TurnOutcome::Ok`.
     /// Returns the number of rows acked (zero is normal — claude
     /// may have not called recv during the turn). Tolerant of ids
     /// that no longer exist in the DB (vacuumed, manually deleted)
@@ -476,7 +477,7 @@ impl Broker {
         let mut inflight = self.inflight.lock().unwrap();
         let ids: Vec<i64> = inflight
             .get_mut(recipient)
-            .map(|s| std::mem::take(&mut s.unacked_ids))
+            .map(|s| s.unacked_ids.clone())
             .unwrap_or_default();
         if ids.is_empty() {
             return Ok(0);
@@ -497,6 +498,10 @@ impl Broker {
             params_vec.push(id);
         }
         let n = conn.execute(&sql, params_vec.as_slice())?;
+        // Clear only after the UPDATE succeeded, so a failed ack can be retried.
+        if let Some(s) = inflight.get_mut(recipient) {
+            s.unacked_ids.clear();
+        }
         Ok(u64::try_from(n).unwrap_or(0))
     }
 
@@ -712,6 +717,27 @@ mod tests {
         assert!(pop_one(broker, "b").is_none());
     }
 
+    /// A failed UPDATE keeps the unacked ids: the retry acks the same rows.
+    #[test]
+    fn ack_turn_keeps_ids_when_update_fails() {
+        let h = open_broker();
+        let broker = &h.broker;
+        broker.send(&msg("a", "b", "hi")).unwrap();
+        pop_one(broker, "b").expect("popped");
+        broker
+            .conn
+            .lock()
+            .unwrap()
+            .busy_timeout(std::time::Duration::ZERO)
+            .unwrap();
+        let blocker = rusqlite::Connection::open(&h.path).unwrap();
+        blocker.execute_batch("BEGIN IMMEDIATE").unwrap();
+        assert!(broker.ack_turn("b").is_err());
+        blocker.execute_batch("ROLLBACK").unwrap();
+        assert_eq!(broker.ack_turn("b").unwrap(), 1);
+        assert_eq!(broker.ack_turn("b").unwrap(), 0);
+    }
+
     /// Bulk triage: three queued messages, agent acks up to the second
     /// id — the first two never deliver again, the third still pops.
     #[test]