diff --git a/hive-ag3nt/src/bin/hive.rs b/hive-ag3nt/src/bin/hive.rs index 967ff414..c9046cd7 100644 --- a/hive-ag3nt/src/bin/hive.rs +++ b/hive-ag3nt/src/bin/hive.rs @@ -389,8 +389,15 @@ 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 }).await?; + let resp: AgentResponse = client::request( + socket, + &AgentRequest::Wake { + from, + body, + transient: false, + }, + ) + .await?; match resp { AgentResponse::Ok => Ok(()), AgentResponse::Err { message } => anyhow::bail!("wake: {message}"), diff --git a/hive-ag3nt/src/forge_notify.rs b/hive-ag3nt/src/forge_notify.rs index 855c3f87..ddfdd1e8 100644 --- a/hive-ag3nt/src/forge_notify.rs +++ b/hive-ag3nt/src/forge_notify.rs @@ -818,6 +818,7 @@ 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 3a3b00cc..bdff6cf7 100644 --- a/hive-ag3nt/src/serve_common.rs +++ b/hive-ag3nt/src/serve_common.rs @@ -11,8 +11,9 @@ 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`. `unread` is the inbox depth after this message was popped. -/// `redelivered` prepends a "may already be handled" banner. +/// `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. #[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 e2e9e25f..692f916c 100644 --- a/hive-bash-mcp/src/runner.rs +++ b/hive-bash-mcp/src/runner.rs @@ -688,6 +688,7 @@ 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 c80d1ef4..269d7386 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}; +use std::collections::{HashMap, HashSet, VecDeque}; use std::path::Path; use std::sync::Mutex; use std::time::{SystemTime, UNIX_EPOCH}; @@ -124,6 +124,15 @@ 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 @@ -145,6 +154,13 @@ 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, @@ -153,6 +169,15 @@ 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 { @@ -171,6 +196,7 @@ impl Broker { conn: Mutex::new(conn), events, inflight: Mutex::new(HashMap::new()), + pending_pings: Mutex::new(HashMap::new()), }) } @@ -244,6 +270,77 @@ 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. @@ -478,8 +575,9 @@ impl Broker { return Ok(Vec::new()); } let mut rx = self.subscribe(); - // Immediate check: any pending sqlite messages. - let batch = self.recv_batch(recipient, max)?; + // Immediate check: buffered pings (incl. any fired while no recv was + // parked) + queued sqlite mail. + let batch = self.collect_batch(recipient, max)?; if !batch.is_empty() { return Ok(batch); } @@ -491,14 +589,17 @@ impl Broker { }; match tokio::time::timeout(remaining, rx.recv()).await { Err(_) => return Ok(Vec::new()), - // Channel lagged or closed — fall back to a direct recv_batch + // Channel lagged or closed — fall back to a direct collect // (in case we missed our notification while behind). - 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)?; + 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)?; if !batch.is_empty() { return Ok(batch); } @@ -1419,48 +1520,45 @@ mod tests { assert_eq!(broker.ack_turn("b").unwrap(), 5); } - /// 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. + /// 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. #[test] - fn wake_persisted_with_real_id_and_ackable() { + fn transient_ping_buffered_when_no_receiver_parked() { let h = open_broker(); let broker = &h.broker; - // 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); + // 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); } - /// Wakes and regular mail both go through sqlite: FIFO by insertion order. + /// Buffered pings surface ahead of queued sqlite mail in one batch. #[test] - fn wake_and_mail_ordered_fifo() { + fn pending_pings_surface_before_queued_mail() { let h = open_broker(); let broker = &h.broker; broker.send(&msg("a", "b", "mail")).unwrap(); - broker.send(&msg("matrix", "b", "matrix wake")).unwrap(); - let batch = broker.recv_batch("b", 10).unwrap(); + broker.ping("b", "matrix", "wake"); + let batch = broker.collect_batch("b", 10).unwrap(); assert_eq!(batch.len(), 2); - // 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); + 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"); } /// `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 9ba989db..d4789852 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() - .map(|m| match m { + .filter_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); - crate::dashboard_events::DashboardEvent::Sent { + Some(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); - crate::dashboard_events::DashboardEvent::Delivered { + Some(crate::dashboard_events::DashboardEvent::Delivered { seq: 0, id, from, @@ -1121,8 +1121,11 @@ 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() @@ -1356,27 +1359,27 @@ async fn api_operator_inbox(State(state): State) -> Response { Ok(messages) => { let items: Vec = messages .into_iter() - .filter_map(|m| { - let crate::broker::MessageEvent::Sent { + .filter_map(|m| match m { + crate::broker::MessageEvent::Sent { id, from, body, at, in_reply_to, .. - } = 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, - })) + } => { + 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, }) .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 f39fb1e4..a7546d63 100644 --- a/hive-c0re/src/main.rs +++ b/hive-c0re/src/main.rs @@ -522,6 +522,9 @@ 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 8e192239..a839f86c 100644 --- a/hive-c0re/src/socket_server.rs +++ b/hive-c0re/src/socket_server.rs @@ -185,7 +185,11 @@ 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 } => handle_wake(coord, agent, from, body), + hive_sh4re::Request::Wake { + from, + body, + transient, + } => handle_wake(coord, agent, from, body, *transient), hive_sh4re::Request::Recent { limit } => handle_recent(coord, agent, *limit), hive_sh4re::Request::Ask { question, @@ -307,25 +311,33 @@ async fn handle_recv( } } -/// `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. +/// `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. fn handle_wake( coord: &Arc, agent: &str, from: &str, body: &str, + transient: bool, ) -> hive_sh4re::Response { - 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:#}"), - }, + 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:#}"), + }, + } } } diff --git a/hive-sh4re/src/lib.rs b/hive-sh4re/src/lib.rs index a9f1d540..d8193b91 100644 --- a/hive-sh4re/src/lib.rs +++ b/hive-sh4re/src/lib.rs @@ -579,9 +579,19 @@ 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. The wake is persisted in the sqlite broker - /// like any other message — the agent can ack it via `AckUntil`. - Wake { from: String, body: String }, + /// 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, + }, /// 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.