From 666c32ae6518064436e4b81b5344169d87fb8155 Mon Sep 17 00:00:00 2001 From: damocles Date: Fri, 29 May 2026 11:39:21 +0200 Subject: [PATCH] broker: add mark_all_read + POST /api/agent/{name}/mark-all-read (#559 backend half) --- hive-c0re/src/broker.rs | 123 +++++++++++++++++++++++++++++++++++++ hive-c0re/src/dashboard.rs | 20 ++++++ 2 files changed, 143 insertions(+) diff --git a/hive-c0re/src/broker.rs b/hive-c0re/src/broker.rs index aa414ea8..5387112a 100644 --- a/hive-c0re/src/broker.rs +++ b/hive-c0re/src/broker.rs @@ -513,6 +513,51 @@ impl Broker { Ok(u64::try_from(ids.len()).unwrap_or(0)) } + /// Operator-driven "clear the inbox": mark every message addressed + /// to `recipient` as acked. Backfills `delivered_at = NOW` for any + /// row that was still pending (undelivered), so the row doesn't + /// become impossible to vacuum later — `vacuum_delivered` requires + /// both timestamps set. Also clears the in-memory inflight state + /// for the recipient so a subsequent `ack_turn` doesn't try to + /// re-mark ids that are already acked. Returns the number of rows + /// affected (zero is normal — inbox already empty). + /// + /// Distinct from `ack_turn` (which acks only the per-turn unacked + /// ids the harness pulled via `recv_batch`) and `requeue_inflight` + /// (which puts inflight-but-unacked rows BACK on the queue). This + /// is the destructive "drain everything for this agent" path the + /// dashboard surfaces as the side-panel "mark all read" button + /// (#559). Backs `POST /api/agent/{name}/mark-all-read`. + pub fn mark_all_read(&self, recipient: &str) -> Result { + let mut inflight = self.inflight.lock().unwrap(); + let conn = self.conn.lock().unwrap(); + let now = now_unix(); + // Two-axis update in one statement: set acked_at on every + // row for the recipient that doesn't have it yet, AND backfill + // delivered_at if it was NULL so the row is fully consumed + // (vacuum_delivered's `acked_at IS NOT NULL AND acked_at < ?` + // predicate then collects it on the normal hourly sweep). + let n = conn.execute( + "UPDATE messages + SET delivered_at = COALESCE(delivered_at, ?1), + acked_at = ?1 + WHERE recipient = ?2 + AND acked_at IS NULL", + params![now, recipient], + )?; + // Drop in-memory inflight bookkeeping for this recipient: the + // ids we just acked might still be in `unacked_ids` from a + // prior `recv_batch`; leaving them would cause the next + // `ack_turn` to re-issue an UPDATE against rows that no longer + // need it (correct but wasteful) and the requeue path would + // see stale ids. Cleanest to reset. + if let Some(slot) = inflight.get_mut(recipient) { + slot.unacked_ids.clear(); + slot.requeued_ids.clear(); + } + Ok(u64::try_from(n).unwrap_or(0)) + } + /// Store a new reminder. Returns the reminder id. pub fn store_reminder( &self, @@ -1113,6 +1158,84 @@ mod tests { assert!(!d.redelivered); } + /// `mark_all_read` covers a mix of pending + delivered + acked rows: + /// pending rows get both `delivered_at` and `acked_at` backfilled, + /// delivered-but-unacked rows just get `acked_at` set, already-acked + /// rows pass through untouched. Returns the count of rows mutated. + #[test] + fn mark_all_read_drains_all_states_for_recipient() { + let h = open_broker(); + let broker = &h.broker; + // Set up three rows in three different states: + // r1 — pending (never popped, both timestamps NULL) + // r2 — delivered + unacked (popped, harness didn't ack yet) + // r3 — delivered + acked (popped + ack_turn ran) + broker.send(&msg("a", "b", "pending")).unwrap(); + broker.send(&msg("a", "b", "delivered")).unwrap(); + broker.send(&msg("a", "b", "acked")).unwrap(); + // Pop both deliverable rows, then ack only the last so r2 stays + // delivered-but-unacked. After pop r3 is still in the unacked + // list; recv pops in FIFO order so first pop = "pending", but + // we want THAT row to remain undelivered. Workaround: pop two + // rows (so "pending" and "delivered" come off the queue) and + // requeue the first to put "pending" back. Then send a fourth + // "acked" and pop+ack just that. + // + // Simpler approach: bypass the queue helpers and craft the row + // states directly via send + recv + ack. FIFO order is by + // insertion; we pop two, ack only the second. + let _ = pop_one(broker, "b").expect("pop 1: pending → now delivered"); + let _ = pop_one(broker, "b").expect("pop 2: delivered"); + let _ = pop_one(broker, "b").expect("pop 3: acked-soon"); + assert_eq!(broker.ack_turn("b").unwrap(), 3); + // Now reshape: requeue first two so they're pending again. + // (Hack — easier: just call mark_all_read on the state we + // have, which is "three rows already acked". Should return + // zero because no row has acked_at IS NULL.) + assert_eq!(broker.mark_all_read("b").unwrap(), 0); + // Add a fresh pending row + a delivered-but-unacked row. + broker.send(&msg("a", "b", "new pending")).unwrap(); + broker.send(&msg("a", "b", "new delivered")).unwrap(); + let _ = pop_one(broker, "b").expect("pop new pending → delivered"); + // Don't ack — leaves it delivered+unacked. + // Now: one row is pending (delivered_at IS NULL), one is + // delivered+unacked. mark_all_read should hit both. + assert_eq!(broker.mark_all_read("b").unwrap(), 2); + // Second call: nothing pending now. + assert_eq!(broker.mark_all_read("b").unwrap(), 0); + // Confirm via recv — inbox is empty. + assert!(pop_one(broker, "b").is_none()); + } + + /// Per-recipient isolation: marking alice doesn't touch bob's inbox. + #[test] + fn mark_all_read_is_per_recipient() { + let h = open_broker(); + let broker = &h.broker; + broker.send(&msg("x", "alice", "for alice")).unwrap(); + broker.send(&msg("x", "bob", "for bob")).unwrap(); + assert_eq!(broker.mark_all_read("alice").unwrap(), 1); + // bob's row still pending — pop succeeds. + let d = pop_one(broker, "bob").expect("bob pop"); + assert_eq!(d.message.body, "for bob"); + } + + /// After `mark_all_read`, a subsequent `ack_turn` from a stale + /// in-memory unacked list MUST NOT panic or double-ack. The + /// inflight bookkeeping is cleared by `mark_all_read`. + #[test] + fn mark_all_read_clears_inflight_so_ack_turn_is_noop() { + let h = open_broker(); + let broker = &h.broker; + broker.send(&msg("a", "b", "hi")).unwrap(); + let _ = pop_one(broker, "b").expect("popped"); + // Now there's an id in the unacked_ids list. mark_all_read + // should clear it so the harness's next ack_turn (which still + // thinks the id is unacked) returns 0 cleanly. + assert_eq!(broker.mark_all_read("b").unwrap(), 1); + assert_eq!(broker.ack_turn("b").unwrap(), 0); + } + /// Per-recipient isolation: `requeue_inflight("a")` doesn't touch /// b's inflight rows. #[test] diff --git a/hive-c0re/src/dashboard.rs b/hive-c0re/src/dashboard.rs index 39b284a0..2288396d 100644 --- a/hive-c0re/src/dashboard.rs +++ b/hive-c0re/src/dashboard.rs @@ -69,6 +69,7 @@ pub async fn serve(port: u16, coord: Arc) -> Result<()> { .route("/api/state-file", get(get_state_file)) .route("/api/reminders", get(api_reminders)) .route("/api/agent/{name}/links", get(get_agent_links)) + .route("/api/agent/{name}/mark-all-read", post(post_mark_all_read)) .route("/cancel-reminder/{id}", post(post_cancel_reminder)) .route("/retry-reminder/{id}", post(post_retry_reminder)) .route("/request-spawn", post(post_request_spawn)) @@ -1747,6 +1748,25 @@ async fn post_retry_reminder( } } +/// Operator-driven "clear this agent's inbox" — backs the side-panel +/// "mark all read" button (#559). Marks every message addressed to the +/// agent as acked (backfilling `delivered_at` for any still-pending +/// rows so vacuum can collect them). Returns `{ "marked": N }` so the +/// frontend can show "cleared N messages" feedback without an extra +/// fetch. +async fn post_mark_all_read( + State(state): State, + AxumPath(name): AxumPath, +) -> Response { + match state.coord.broker.mark_all_read(&name) { + Ok(n) => { + tracing::info!(%name, marked = n, "operator marked all messages read"); + axum::Json(serde_json::json!({ "marked": n })).into_response() + } + Err(e) => error_response(&format!("mark-all-read {name} failed: {e:#}")), + } +} + async fn post_purge_tombstone( State(state): State, AxumPath(name): AxumPath,