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 =
+        let resp: AgentResponse = client::request(
-            client::request(socket, &AgentRequest::Wake { from, body }).await?;
+            socket,
+            &AgentRequest::Wake {
+                from,
+                body,
+                transient: false,
+            },
+        )
+        .await?;
         match resp {
             AgentResponse::Ok => Ok(()),
             AgentResponse::Err { message } => anyhow::bail!("wake: {message}"),

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

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 #<id>]` marker so the agent can reference it in
-/// `ack_until`. `unread` is the inbox depth after this message was popped.
+/// `ack_until` (transient pings carry the sentinel 0 and render without it).
-/// `redelivered` prepends a "may already be handled" banner.
+/// `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,

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 {

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<i64>,
     },
+    /// 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<i64>,
 }
 
+/// 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<Connection>,
     events: broadcast::Sender<MessageEvent>,
@@ -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<HashMap<String, RecipientInflight>>,
+    /// 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<HashMap<String, VecDeque<(String, String)>>>,
 }
 
 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<Delivery> {
+        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<Delivery> = 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<Vec<Delivery>> {
+        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.
+        // Immediate check: buffered pings (incl. any fired while no recv was
-        let batch = self.recv_batch(recipient, max)?;
+        // 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),
+                Ok(Err(_)) => return self.collect_batch(recipient, max),
-                // A message was sent to this recipient (or a wake injected).
+                // A relevant event landed (real message or transient ping).
-                // Re-poll sqlite; the event payload is ignored — recv_batch is
+                // Re-collect from the buffers; the ping payload on the event
-                // the source of truth and prevents double-delivery.
+                // is ignored — `drain_pings` is the source of truth, so a
-                Ok(Ok(MessageEvent::Sent { to, .. })) if to == recipient => {
+                // ping can't be double-delivered.
-                    let batch = self.recv_batch(recipient, max)?;
+                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
+    /// Transient-wake regression guard: a `ping` fired while no `recv` is parked
-    /// agent can pass to `ack_until`. A wake sent when no recv is parked
+    /// must NOT be lost — it's buffered and drained by the next collect.
-    /// is not lost — it sits in sqlite until the next recv pops it.
     #[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).
+        // Nobody is parked on recv when these fire (the old broadcast-only
-        broker
+        // path dropped them here).
-            .send(&msg("bash-task-1", "b", "bash task `1` finished: exit=0"))
+        broker.ping("b", "matrix", "wake1");
-            .unwrap();
+        broker.ping("b", "bash-task", "wake2");
-        broker
+        let batch = broker.collect_batch("b", 10).unwrap();
-            .send(&msg("bash-task-2", "b", "bash task `2` finished: exit=0"))
+        let got: Vec<_> = batch
-            .unwrap();
+            .iter()
-        let batch = broker.recv_batch("b", 10).unwrap();
+            .map(|d| (d.id, d.message.from.as_str(), d.message.body.as_str()))
-        assert_eq!(batch.len(), 2);
+            .collect();
-        // Every delivery has a real (non-zero) id.
+        assert_eq!(
-        assert!(batch[0].id > 0);
+            got,
-        assert!(batch[1].id > 0);
+            vec![(0, "matrix", "wake1"), (0, "bash-task", "wake2")],
-        // Agent can ack them explicitly via ack_until.
+            "buffered pings drain FIFO as id=0 deliveries"
-        let max_id = batch.iter().map(|d| d.id).max().unwrap();
+        );
-        assert_eq!(broker.ack_turn("b").unwrap(), 2);
+        // Drained — the buffer is now empty.
-        // Confirming the rows are now closed.
+        assert!(broker.collect_batch("b", 10).unwrap().is_empty());
-        assert!(pop_one(broker, "b").is_none());
+        // Pings are sentinel id=0, so ack_turn has nothing to close out.
-        // ack_until on already-acked ids is a no-op.
+        assert_eq!(broker.ack_turn("b").unwrap(), 0);
-        assert_eq!(broker.ack_until("b", max_id).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();
+        broker.ping("b", "matrix", "wake");
-        let batch = broker.recv_batch("b", 10).unwrap();
+        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].id, 0, "ping first");
-        assert_eq!(batch[0].message.body, "mail");
+        assert_eq!(batch[0].message.body, "wake");
-        assert_eq!(batch[1].message.body, "matrix wake");
+        assert!(batch[1].id > 0, "sqlite mail after");
-        assert!(batch[0].id > 0);
+        assert_eq!(batch[1].message.body, "mail");
-        assert!(batch[1].id > 0);
     }
 
     /// `recv_batch` with no pending traffic returns an empty vec

hive-c0re/src/dashboard.rs

@@ -1082,7 +1082,7 @@ async fn dashboard_history(State(state): State<AppState>) -> Response {
             messages.reverse();
             let events: Vec<crate::dashboard_events::DashboardEvent> = 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<AppState>) -> 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<AppState>) -> 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<AppState>) -> 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<AppState>) -> 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<AppState>) -> Response {
         Ok(messages) => {
             let items: Vec<serde_json::Value> = messages
                 .into_iter()
-                .filter_map(|m| {
+                .filter_map(|m| match m {
-                    let crate::broker::MessageEvent::Sent {
+                    crate::broker::MessageEvent::Sent {
                         id,
                         from,
                         body,
                         at,
                         in_reply_to,
                         ..
-                    } = m
+                    } => {
-                    else {
+                        let file_refs = scan_validated_paths(&body);
-                        return None;
+                        Some(serde_json::json!({
-                    };
+                            "id": id,
-                    let file_refs = scan_validated_paths(&body);
+                            "from": from,
-                    Some(serde_json::json!({
+                            "body": body,
-                        "id": id,
+                            "at": hive_sh4re::wire_time::from_secs(at),
-                        "from": from,
+                            "in_reply_to": in_reply_to,
-                        "body": body,
+                            "file_refs": file_refs,
-                        "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()

hive-c0re/src/main.rs

@@ -522,6 +522,9 @@ fn spawn_broker_to_dashboard_forwarder(coord: Arc<Coordinator>) {
                         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");
                 }

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
+/// `Wake` — inject a wake into `agent`'s own inbox. Transient wakes
-/// the sqlite broker like any other message so the agent can ack it
+/// fire the broadcast channel only (no sqlite row, no redelivery on
-/// via `AckUntil` and it appears in message history for post-mortem.
+/// restart — used by bash-task completions); durable wakes persist
+/// through the broker like any other message.
 fn handle_wake(
     coord: &Arc<Coordinator>,
     agent: &str,
     from: &str,
     body: &str,
+    transient: bool,
 ) -> hive_sh4re::Response {
-    match coord.broker.send(&Message {
+    let broker = &coord.broker;
-        from: from.to_owned(),
+    if transient {
-        to: agent.to_owned(),
+        broker.ping(agent, from, body);
-        body: body.to_owned(),
+        hive_sh4re::Response::Ok
-        in_reply_to: None,
+    } else {
-    }) {
+        match broker.send(&Message {
-        Ok(()) => hive_sh4re::Response::Ok,
+            from: from.to_owned(),
-        Err(e) => hive_sh4re::Response::Err {
+            to: agent.to_owned(),
-            message: format!("{e:#}"),
+            body: body.to_owned(),
-        },
+            in_reply_to: None,
+        }) {
+            Ok(()) => hive_sh4re::Response::Ok,
+            Err(e) => hive_sh4re::Response::Err {
+                message: format!("{e:#}"),
+            },
+        }
     }
 }
 

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
+    /// typical callers.
-    /// like any other message — the agent can ack it via `AckUntil`.
+    ///
-    Wake { from: String, body: String },
+    /// 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.