add ack_until: bulk-ack inbox messages by id + surface msg ids in wake prompts and recv (closes #2125)

hive-c0re/src/broker.rs

@@ -751,6 +751,38 @@ impl Broker {
         Ok(u64::try_from(n).unwrap_or(0))
     }
 
+    /// Bulk-ack every message addressed to `recipient` with row id
+    /// `<= up_to` that isn't acked yet — pending AND delivered rows
+    /// alike. The agent-facing triage tool behind `AckUntil`: after a
+    /// redelivered flood the agent acks everything up to the highest id
+    /// it has seen instead of re-popping each row. Recipient-scoped by
+    /// the WHERE clause, so an agent can never touch another agent's
+    /// rows. Also drains the in-memory `unacked_ids` / `requeued_ids`
+    /// bookkeeping below the cutoff so a later `ack_turn` doesn't
+    /// re-update rows this call already closed and a stale redelivery
+    /// tag doesn't outlive its row. Returns the number of rows newly
+    /// acked.
+    ///
+    /// # Errors
+    ///
+    /// Propagates sqlite errors from the `UPDATE`.
+    pub fn ack_until(&self, recipient: &str, up_to: i64) -> Result<u64> {
+        // Same lock order as `recv` / `ack_turn` / `requeue_inflight`:
+        // `inflight` FIRST, then `conn`.
+        let mut inflight = self.inflight.lock().unwrap();
+        if let Some(state) = inflight.get_mut(recipient) {
+            state.unacked_ids.retain(|&id| id > up_to);
+            state.requeued_ids.retain(|&id| id > up_to);
+        }
+        let conn = self.conn.lock().unwrap();
+        let n = conn.execute(
+            "UPDATE messages SET acked_at = ?1
+              WHERE recipient = ?2 AND id <= ?3 AND acked_at IS NULL",
+            params![now_unix(), recipient, up_to],
+        )?;
+        Ok(u64::try_from(n).unwrap_or(0))
+    }
+
     /// Resurface every message the broker previously handed to this
     /// recipient that never got `acked_at` set. Used by the harness at
     /// boot to recover from the crashed-mid-turn / OOM-killed /
@@ -1304,6 +1336,61 @@ mod tests {
         assert!(pop_one(broker, "b").is_none());
     }
 
+    /// Bulk triage: three queued messages, agent acks up to the second
+    /// id — the first two never deliver again, the third still pops.
+    #[test]
+    fn ack_until_acks_only_rows_at_or_below_cutoff() {
+        let h = open_broker();
+        let broker = &h.broker;
+        broker.send(&msg("a", "b", "one")).unwrap();
+        broker.send(&msg("a", "b", "two")).unwrap();
+        broker.send(&msg("a", "b", "three")).unwrap();
+        // Pop the first two so we know their ids (FIFO).
+        let d1 = pop_one(broker, "b").expect("popped one");
+        let d2 = pop_one(broker, "b").expect("popped two");
+        assert_eq!(broker.ack_until("b", d2.id).unwrap(), 2);
+        // The cutoff also drained the in-memory unacked list, so a
+        // turn-level ack right after finds nothing left to do.
+        assert_eq!(broker.ack_turn("b").unwrap(), 0);
+        // A restart-style requeue finds nothing below the cutoff …
+        assert_eq!(broker.requeue_inflight("b").unwrap(), 0);
+        // … and the third message (id above the cutoff) still pops.
+        let d3 = pop_one(broker, "b").expect("third still pending");
+        assert_eq!(d3.message.body, "three");
+        assert!(d1.id < d2.id && d2.id < d3.id);
+    }
+
+    /// Pending (never-delivered) rows below the cutoff are acked too —
+    /// that's the whole point for a stale backlog the agent never
+    /// popped individually.
+    #[test]
+    fn ack_until_covers_pending_rows() {
+        let h = open_broker();
+        let broker = &h.broker;
+        broker.send(&msg("a", "b", "stale-1")).unwrap();
+        broker.send(&msg("a", "b", "stale-2")).unwrap();
+        // Learn the highest id by peeking via recent_for (non-mutating).
+        let rows = broker.recent_for("b", 10).unwrap();
+        let max_id = rows.iter().map(|r| r.id).max().expect("rows");
+        assert_eq!(broker.ack_until("b", max_id).unwrap(), 2);
+        assert!(pop_one(broker, "b").is_none(), "backlog cleared");
+    }
+
+    /// Recipient scoping: acking b's inbox never touches c's rows,
+    /// even when c's ids fall below the cutoff.
+    #[test]
+    fn ack_until_is_recipient_scoped() {
+        let h = open_broker();
+        let broker = &h.broker;
+        broker.send(&msg("a", "c", "for-c")).unwrap();
+        broker.send(&msg("a", "b", "for-b")).unwrap();
+        let rows = broker.recent_for("b", 10).unwrap();
+        let max_id = rows.iter().map(|r| r.id).max().expect("rows");
+        assert_eq!(broker.ack_until("b", max_id).unwrap(), 1);
+        let d = pop_one(broker, "c").expect("c's message untouched");
+        assert_eq!(d.message.body, "for-c");
+    }
+
     /// Crash-recovery: send → recv → (no ack) → `requeue_inflight`
     /// resets `delivered_at` + tags the next pop as redelivered. After
     /// that `ack_turn` closes it out cleanly.

hive-c0re/src/socket_server.rs

@@ -237,6 +237,7 @@ pub(crate) async fn dispatch_shared(
         }
         hive_sh4re::Request::CreateRepo { repo } => handle_create_repo(agent, repo).await,
         hive_sh4re::Request::AckTurn => handle_ack_turn(coord, agent),
+        hive_sh4re::Request::AckUntil { up_to } => handle_ack_until(coord, agent, *up_to),
         hive_sh4re::Request::RequeueInflight => handle_requeue_inflight(coord, agent),
         hive_sh4re::Request::GracefulStopComplete => {
             // Harness drained + is exiting: clear the fence so the
@@ -483,6 +484,17 @@ fn handle_ack_turn(coord: &Arc<Coordinator>, agent: &str) -> hive_sh4re::Respons
     }
 }
 
+/// `AckUntil` — bulk-ack every message addressed to `agent` with row
+/// id `<= up_to` (the agent-side backlog-triage escape hatch).
+fn handle_ack_until(coord: &Arc<Coordinator>, agent: &str, up_to: i64) -> hive_sh4re::Response {
+    match coord.broker.ack_until(agent, up_to) {
+        Ok(count) => hive_sh4re::Response::Acked { count },
+        Err(e) => hive_sh4re::Response::Err {
+            message: format!("{e:#}"),
+        },
+    }
+}
+
 /// `RequeueInflight` — resurface `agent`'s unacked in-flight messages
 /// (crash recovery on harness boot).
 fn handle_requeue_inflight(coord: &Arc<Coordinator>, agent: &str) -> hive_sh4re::Response {