diff --git a/hive-ag3nt/src/bin/hive.rs b/hive-ag3nt/src/bin/hive.rs index c9046cd7..0a09c101 100644 --- a/hive-ag3nt/src/bin/hive.rs +++ b/hive-ag3nt/src/bin/hive.rs @@ -391,11 +391,7 @@ impl Surface for AgentSurface { async fn wake_external(socket: &Path, from: String, body: String) -> Result<()> { let resp: AgentResponse = client::request( socket, - &AgentRequest::Wake { - from, - body, - transient: false, - }, + &AgentRequest::Wake { from, body }, ) .await?; match resp { diff --git a/hive-ag3nt/src/forge_notify.rs b/hive-ag3nt/src/forge_notify.rs index ddfdd1e8..855c3f87 100644 --- a/hive-ag3nt/src/forge_notify.rs +++ b/hive-ag3nt/src/forge_notify.rs @@ -818,7 +818,6 @@ async fn poll_once( let req = hive_sh4re::Request::Wake { from: "forge".to_owned(), body, - transient: false, }; let deliver_result = crate::client::request::<_, hive_sh4re::Response>(socket, &req) .await diff --git a/hive-ag3nt/src/serve_common.rs b/hive-ag3nt/src/serve_common.rs index bdff6cf7..3a3b00cc 100644 --- a/hive-ag3nt/src/serve_common.rs +++ b/hive-ag3nt/src/serve_common.rs @@ -11,9 +11,8 @@ use crate::turn_stats::TurnStatRow; /// Assemble the per-turn wake prompt string. The role/tools/etc. live in the /// system prompt; this is just the wake signal body. `id` is the broker row /// id, rendered as a `[msg #]` marker so the agent can reference it in -/// `ack_until` (transient pings carry the sentinel 0 and render without it). -/// `unread` is the inbox depth after this message was popped. `redelivered` -/// prepends a "may already be handled" banner. +/// `ack_until`. `unread` is the inbox depth after this message was popped. +/// `redelivered` prepends a "may already be handled" banner. #[must_use] pub fn format_wake_prompt( id: i64, diff --git a/hive-bash-mcp/src/runner.rs b/hive-bash-mcp/src/runner.rs index 692f916c..e2e9e25f 100644 --- a/hive-bash-mcp/src/runner.rs +++ b/hive-bash-mcp/src/runner.rs @@ -688,7 +688,6 @@ pub(crate) async fn send_wake( let req = hive_sh4re::AgentRequest::Wake { from: format!("bash-task-{id}"), body, - transient: true, }; match UnixStream::connect(socket).await { diff --git a/hive-c0re/src/broker.rs b/hive-c0re/src/broker.rs index 269d7386..6a21a7e9 100644 --- a/hive-c0re/src/broker.rs +++ b/hive-c0re/src/broker.rs @@ -1,7 +1,7 @@ //! Sqlite-backed message broker. Survives `hive-c0re` restart, and taps every //! send/recv onto a broadcast channel so the dashboard can stream it. -use std::collections::{HashMap, HashSet, VecDeque}; +use std::collections::{HashMap, HashSet}; use std::path::Path; use std::sync::Mutex; use std::time::{SystemTime, UNIX_EPOCH}; @@ -124,15 +124,6 @@ pub enum MessageEvent { at: i64, in_reply_to: Option, }, - /// Transient wake signal — NOT persisted to sqlite. Wakes - /// `recv_blocking_batch` for the target agent but is not stored, - /// not re-delivered on restart, and not shown in message history. - /// Used for bash task completion notifications. - Ping { - to: String, - from: String, - body: String, - }, } /// Per-recipient in-memory bookkeeping for the deliver-then-ack @@ -154,13 +145,6 @@ struct RecipientInflight { requeued_ids: HashSet, } -/// Hard cap on buffered transient pings per recipient. Pings are -/// ephemeral wakes (bash completions, matrix events); an agent drains -/// them on its next `recv`, so the buffer is normally near-empty. The -/// cap only bounds memory if a recipient stops recv'ing entirely (e.g. -/// a stopped container) — past it the oldest ping is dropped. -const MAX_PENDING_PINGS: usize = 256; - pub struct Broker { conn: Mutex, events: broadcast::Sender, @@ -169,15 +153,6 @@ pub struct Broker { /// boot, which rebuilds the `requeued_ids` set from the DB and /// clears any stale `unacked_ids`). inflight: Mutex>, - /// Per-recipient buffer of transient wake signals (`ping`) that - /// haven't been consumed by a `recv` yet. The broadcast channel only - /// reaches a `recv_blocking_batch` that is *currently parked*; a ping - /// fired while the harness is mid-turn (no live subscriber) would - /// otherwise be lost. Buffering here makes a transient wake reliable - /// — the next `recv` drains it — without sqlite persistence or - /// redelivery-on-restart (the buffer is in-memory, dropped on - /// hive-c0re restart, which is fine: a wake that old is stale). - pending_pings: Mutex>>, } impl Broker { @@ -196,7 +171,6 @@ impl Broker { conn: Mutex::new(conn), events, inflight: Mutex::new(HashMap::new()), - pending_pings: Mutex::new(HashMap::new()), }) } @@ -270,77 +244,6 @@ impl Broker { Ok(()) } - /// Deliver a transient wake signal to `to` without writing to sqlite. - /// The signal wakes the target agent's next `recv_blocking_batch` but is - /// not persisted, not redelivered on restart, and not shown in message - /// history. Use for ephemeral notifications (bash task completions, - /// matrix events) where persistence would cause duplicate delivery. - /// - /// The ping is buffered in `pending_pings` *before* the broadcast so a - /// `recv` that wasn't parked at fire time still drains it on its next - /// call — the broadcast alone reaches only a currently-parked receiver. - pub fn ping(&self, to: &str, from: &str, body: &str) { - { - let mut pending = self.pending_pings.lock().unwrap(); - let queue = pending.entry(to.to_owned()).or_default(); - queue.push_back((from.to_owned(), body.to_owned())); - // Bound memory if the recipient never recv's (stopped container): - // drop the oldest so the most recent wakes survive. - while queue.len() > MAX_PENDING_PINGS { - queue.pop_front(); - } - } - let _ = self.events.send(MessageEvent::Ping { - to: to.to_owned(), - from: from.to_owned(), - body: body.to_owned(), - }); - } - - /// Drain up to `max` buffered transient pings for `recipient`, mapping - /// each to a `Delivery` with the sentinel `id = 0` (never pushed to - /// `unacked_ids`, so `ack_turn` ignores it). FIFO; leaves any remainder - /// buffered for the next call. - fn drain_pings(&self, recipient: &str, max: usize) -> Vec { - if max == 0 { - return Vec::new(); - } - let mut pending = self.pending_pings.lock().unwrap(); - let Some(queue) = pending.get_mut(recipient) else { - return Vec::new(); - }; - let take = queue.len().min(max); - let drained: Vec = queue - .drain(..take) - .map(|(from, body)| Delivery { - id: 0, - redelivered: false, - message: Message { - from, - to: recipient.to_owned(), - body, - in_reply_to: None, - }, - }) - .collect(); - if queue.is_empty() { - pending.remove(recipient); - } - drained - } - - /// Collect a batch for `recipient`: buffered transient pings first - /// (they're wakes — surface them ahead of queued mail), then sqlite - /// messages up to the remaining budget. Shared by the immediate check - /// and the post-wake path of `recv_blocking_batch`. - fn collect_batch(&self, recipient: &str, max: usize) -> Result> { - let mut batch = self.drain_pings(recipient, max); - if batch.len() < max { - batch.extend(self.recv_batch(recipient, max - batch.len())?); - } - Ok(batch) - } - /// Unread (unacked) messages addressed to `recipient`, newest-first. /// Filters to `acked_at IS NULL` so the agent inbox view clears after /// "mark all read" — mirroring exactly what `mark_all_read` will drain. @@ -575,9 +478,8 @@ impl Broker { return Ok(Vec::new()); } let mut rx = self.subscribe(); - // Immediate check: buffered pings (incl. any fired while no recv was - // parked) + queued sqlite mail. - let batch = self.collect_batch(recipient, max)?; + // Immediate check: any pending sqlite messages. + let batch = self.recv_batch(recipient, max)?; if !batch.is_empty() { return Ok(batch); } @@ -589,17 +491,14 @@ impl Broker { }; match tokio::time::timeout(remaining, rx.recv()).await { Err(_) => return Ok(Vec::new()), - // Channel lagged or closed — fall back to a direct collect + // Channel lagged or closed — fall back to a direct recv_batch // (in case we missed our notification while behind). - Ok(Err(_)) => return self.collect_batch(recipient, max), - // A relevant event landed (real message or transient ping). - // Re-collect from the buffers; the ping payload on the event - // is ignored — `drain_pings` is the source of truth, so a - // ping can't be double-delivered. - Ok(Ok(MessageEvent::Sent { to, .. } | MessageEvent::Ping { to, .. })) - if to == recipient => - { - let batch = self.collect_batch(recipient, max)?; + Ok(Err(_)) => return self.recv_batch(recipient, max), + // A message was sent to this recipient (or a wake injected). + // Re-poll sqlite; the event payload is ignored — recv_batch is + // the source of truth and prevents double-delivery. + Ok(Ok(MessageEvent::Sent { to, .. })) if to == recipient => { + let batch = self.recv_batch(recipient, max)?; if !batch.is_empty() { return Ok(batch); } @@ -1520,45 +1419,50 @@ mod tests { assert_eq!(broker.ack_turn("b").unwrap(), 5); } - /// Transient-wake regression guard: a `ping` fired while no `recv` is parked - /// must NOT be lost — it's buffered and drained by the next collect. + /// Wake messages go through sqlite and get real row ids that the + /// agent can pass to `ack_until`. A wake sent when no recv is parked + /// is not lost — it sits in sqlite until the next recv pops it. #[test] - fn transient_ping_buffered_when_no_receiver_parked() { + fn wake_persisted_with_real_id_and_ackable() { let h = open_broker(); let broker = &h.broker; - // Nobody is parked on recv when these fire (the old broadcast-only - // path dropped them here). - broker.ping("b", "matrix", "wake1"); - broker.ping("b", "bash-task", "wake2"); - let batch = broker.collect_batch("b", 10).unwrap(); - let got: Vec<_> = batch - .iter() - .map(|d| (d.id, d.message.from.as_str(), d.message.body.as_str())) - .collect(); - assert_eq!( - got, - vec![(0, "matrix", "wake1"), (0, "bash-task", "wake2")], - "buffered pings drain FIFO as id=0 deliveries" - ); - // Drained — the buffer is now empty. - assert!(broker.collect_batch("b", 10).unwrap().is_empty()); - // Pings are sentinel id=0, so ack_turn has nothing to close out. - assert_eq!(broker.ack_turn("b").unwrap(), 0); + // Two wakes arrive (simulating bash-task completions). + broker + .send(&msg("bash-task-1", "b", "bash task `1` finished: exit=0")) + .unwrap(); + broker + .send(&msg("bash-task-2", "b", "bash task `2` finished: exit=0")) + .unwrap(); + let batch = broker.recv_batch("b", 10).unwrap(); + assert_eq!(batch.len(), 2); + // Every delivery has a real (non-zero) id. + assert!(batch[0].id > 0); + assert!(batch[1].id > 0); + // Agent can ack them explicitly via ack_until. + let max_id = batch.iter().map(|d| d.id).max().unwrap(); + assert_eq!(broker.ack_turn("b").unwrap(), 2); + // Confirming the rows are now closed. + assert!(pop_one(broker, "b").is_none()); + // ack_until on already-acked ids is a no-op. + assert_eq!(broker.ack_until("b", max_id).unwrap(), 0); } - /// Buffered pings surface ahead of queued sqlite mail in one batch. + /// Wakes and regular mail both go through sqlite: FIFO by insertion order. #[test] - fn pending_pings_surface_before_queued_mail() { + fn wake_and_mail_ordered_fifo() { let h = open_broker(); let broker = &h.broker; broker.send(&msg("a", "b", "mail")).unwrap(); - broker.ping("b", "matrix", "wake"); - let batch = broker.collect_batch("b", 10).unwrap(); + broker + .send(&msg("matrix", "b", "matrix wake")) + .unwrap(); + let batch = broker.recv_batch("b", 10).unwrap(); assert_eq!(batch.len(), 2); - assert_eq!(batch[0].id, 0, "ping first"); - assert_eq!(batch[0].message.body, "wake"); - assert!(batch[1].id > 0, "sqlite mail after"); - assert_eq!(batch[1].message.body, "mail"); + // FIFO: mail was inserted first, comes out first. + assert_eq!(batch[0].message.body, "mail"); + assert_eq!(batch[1].message.body, "matrix wake"); + assert!(batch[0].id > 0); + assert!(batch[1].id > 0); } /// `recv_batch` with no pending traffic returns an empty vec diff --git a/hive-c0re/src/dashboard.rs b/hive-c0re/src/dashboard.rs index d4789852..9ba989db 100644 --- a/hive-c0re/src/dashboard.rs +++ b/hive-c0re/src/dashboard.rs @@ -1082,7 +1082,7 @@ async fn dashboard_history(State(state): State) -> Response { messages.reverse(); let events: Vec = messages .into_iter() - .filter_map(|m| match m { + .map(|m| match m { crate::broker::MessageEvent::Sent { id, from, @@ -1092,7 +1092,7 @@ async fn dashboard_history(State(state): State) -> Response { in_reply_to, } => { let file_refs = scan_validated_paths(&body); - Some(crate::dashboard_events::DashboardEvent::Sent { + crate::dashboard_events::DashboardEvent::Sent { seq: 0, id, from, @@ -1101,7 +1101,7 @@ async fn dashboard_history(State(state): State) -> Response { at: hive_sh4re::wire_time::from_secs(at), in_reply_to, file_refs, - }) + } } crate::broker::MessageEvent::Delivered { id, @@ -1112,7 +1112,7 @@ async fn dashboard_history(State(state): State) -> Response { in_reply_to, } => { let file_refs = scan_validated_paths(&body); - Some(crate::dashboard_events::DashboardEvent::Delivered { + crate::dashboard_events::DashboardEvent::Delivered { seq: 0, id, from, @@ -1121,11 +1121,8 @@ async fn dashboard_history(State(state): State) -> Response { at: hive_sh4re::wire_time::from_secs(at), in_reply_to, file_refs, - }) + } } - // Ping events are never persisted to sqlite — this arm is - // unreachable in practice but required for exhaustiveness. - crate::broker::MessageEvent::Ping { .. } => None, }) .collect(); axum::Json(serde_json::json!({ "seq": seq, "events": events })).into_response() @@ -1359,27 +1356,27 @@ async fn api_operator_inbox(State(state): State) -> Response { Ok(messages) => { let items: Vec = messages .into_iter() - .filter_map(|m| match m { - crate::broker::MessageEvent::Sent { + .filter_map(|m| { + let crate::broker::MessageEvent::Sent { id, from, body, at, in_reply_to, .. - } => { - let file_refs = scan_validated_paths(&body); - Some(serde_json::json!({ - "id": id, - "from": from, - "body": body, - "at": hive_sh4re::wire_time::from_secs(at), - "in_reply_to": in_reply_to, - "file_refs": file_refs, - })) - } - crate::broker::MessageEvent::Delivered { .. } - | crate::broker::MessageEvent::Ping { .. } => None, + } = m + else { + return None; + }; + let file_refs = scan_validated_paths(&body); + Some(serde_json::json!({ + "id": id, + "from": from, + "body": body, + "at": hive_sh4re::wire_time::from_secs(at), + "in_reply_to": in_reply_to, + "file_refs": file_refs, + })) }) .collect(); axum::Json(serde_json::json!({ "messages": items })).into_response() diff --git a/hive-c0re/src/main.rs b/hive-c0re/src/main.rs index a7546d63..f39fb1e4 100644 --- a/hive-c0re/src/main.rs +++ b/hive-c0re/src/main.rs @@ -522,9 +522,6 @@ fn spawn_broker_to_dashboard_forwarder(coord: Arc) { file_refs, }); } - // Transient pings are not persisted and not shown in the - // dashboard message history — ignore silently. - Ok(MessageEvent::Ping { .. }) => {} Err(tokio::sync::broadcast::error::RecvError::Lagged(n)) => { tracing::warn!(skipped = n, "broker-to-dashboard forwarder lagged"); } diff --git a/hive-c0re/src/socket_server.rs b/hive-c0re/src/socket_server.rs index a839f86c..8e192239 100644 --- a/hive-c0re/src/socket_server.rs +++ b/hive-c0re/src/socket_server.rs @@ -185,11 +185,7 @@ pub(crate) async fn dispatch_shared( } hive_sh4re::Request::Status => handle_status(coord, agent), hive_sh4re::Request::OperatorMsg { body } => handle_operator_msg(coord, agent, body), - hive_sh4re::Request::Wake { - from, - body, - transient, - } => handle_wake(coord, agent, from, body, *transient), + hive_sh4re::Request::Wake { from, body } => handle_wake(coord, agent, from, body), hive_sh4re::Request::Recent { limit } => handle_recent(coord, agent, *limit), hive_sh4re::Request::Ask { question, @@ -311,33 +307,25 @@ async fn handle_recv( } } -/// `Wake` — inject a wake into `agent`'s own inbox. Transient wakes -/// fire the broadcast channel only (no sqlite row, no redelivery on -/// restart — used by bash-task completions); durable wakes persist -/// through the broker like any other message. +/// `Wake` — inject a wake into `agent`'s own inbox. Persisted through +/// the sqlite broker like any other message so the agent can ack it +/// via `AckUntil` and it appears in message history for post-mortem. fn handle_wake( coord: &Arc, agent: &str, from: &str, body: &str, - transient: bool, ) -> hive_sh4re::Response { - let broker = &coord.broker; - if transient { - broker.ping(agent, from, body); - hive_sh4re::Response::Ok - } else { - match broker.send(&Message { - from: from.to_owned(), - to: agent.to_owned(), - body: body.to_owned(), - in_reply_to: None, - }) { - Ok(()) => hive_sh4re::Response::Ok, - Err(e) => hive_sh4re::Response::Err { - message: format!("{e:#}"), - }, - } + match coord.broker.send(&Message { + from: from.to_owned(), + to: agent.to_owned(), + body: body.to_owned(), + in_reply_to: None, + }) { + Ok(()) => hive_sh4re::Response::Ok, + Err(e) => hive_sh4re::Response::Err { + message: format!("{e:#}"), + }, } } diff --git a/hive-sh4re/src/lib.rs b/hive-sh4re/src/lib.rs index d8193b91..a9f1d540 100644 --- a/hive-sh4re/src/lib.rs +++ b/hive-sh4re/src/lib.rs @@ -579,19 +579,9 @@ pub enum Request { /// Wake-up event injected from inside the container. Recipient is /// implicit (this agent); `from` is caller-chosen. See /// `docs/conventions.md::Wake injection` for the trust model and - /// typical callers. - /// - /// When `transient` is `true` the server delivers the wake signal - /// through an in-process channel only — no sqlite write, no - /// redelivery on restart. Use this for ephemeral notifications (e.g. - /// bash task completions) where persistence is unnecessary and would - /// cause duplicate delivery after a harness restart. - Wake { - from: String, - body: String, - #[serde(default)] - transient: bool, - }, + /// typical callers. The wake is persisted in the sqlite broker + /// like any other message — the agent can ack it via `AckUntil`. + Wake { from: String, body: String }, /// Last `limit` messages addressed to this agent, newest-first. /// Non-mutating — pulls from the broker without delivering. The /// per-agent web UI uses this to render its own inbox section.