feat(#1106): transient wake for bash tasks — bypass broker sqlite

hive-ag3nt/src/bash_runner.rs

@@ -465,6 +465,11 @@ async fn send_wake(socket: &Path, id: &str, summary: &str, output: Option<(&str,
     let req = hive_sh4re::AgentRequest::Wake {
         from: format!("bash-task-{id}"),
         body,
+        // Transient: do not persist to the message broker. Bash task
+        // completions are fire-and-forget — the output is already on disk
+        // in harness/bash-tasks/; persisting would cause duplicate delivery
+        // after a harness restart (issue #1103).
+        transient: true,
     };
     match crate::client::request::<_, hive_sh4re::AgentResponse>(socket, &req).await {
         Ok(_) => tracing::info!(id = %id, "bash_runner: wake delivered"),

hive-ag3nt/src/bin/hive.rs

@@ -316,6 +316,7 @@ impl Surface for AgentSurface {
             &AgentRequest::Wake {
                 from: "self".into(),
                 body: "continue".into(),
+                transient: false,
             },
         )
         .await;
@@ -365,7 +366,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 }).await?;
+            client::request(socket, &AgentRequest::Wake { from, body, transient: false }).await?;
         match resp {
             AgentResponse::Ok => Ok(()),
             AgentResponse::Err { message } => anyhow::bail!("wake: {message}"),
@@ -456,6 +457,7 @@ impl Surface for ManagerSurface {
             &ManagerRequest::Wake {
                 from: "self".into(),
                 body: "continue".into(),
+                transient: false,
             },
         )
         .await;
@@ -505,7 +507,7 @@ impl Surface for ManagerSurface {
 
     async fn wake_external(socket: &Path, from: String, body: String) -> Result<()> {
         let resp: ManagerResponse =
-            client::request(socket, &ManagerRequest::Wake { from, body }).await?;
+            client::request(socket, &ManagerRequest::Wake { from, body, transient: false }).await?;
         match resp {
             ManagerResponse::Ok => Ok(()),
             ManagerResponse::Err { message } => anyhow::bail!("wake: {message}"),

hive-ag3nt/src/forge_notify.rs

@@ -684,6 +684,7 @@ async fn poll_once(
         let req = hive_sh4re::Request::Wake {
             from: "forge".to_owned(),
             body,
+            transient: false,
         };
         let delivered = crate::client::request::<_, hive_sh4re::Response>(socket, &req)
             .await

hive-c0re/src/agent_server.rs

@@ -174,17 +174,31 @@ pub(crate) async fn dispatch_shared(
                 message: format!("{e:#}"),
             },
         },
-        hive_sh4re::Request::Wake { from, body } => match broker.send(&Message {
-            from: from.clone(),
-            to: agent.to_owned(),
-            body: body.clone(),
-            in_reply_to: None,
-        }) {
-            Ok(()) => hive_sh4re::Response::Ok,
-            Err(e) => hive_sh4re::Response::Err {
-                message: format!("{e:#}"),
-            },
-        },
+        hive_sh4re::Request::Wake {
+            from,
+            body,
+            transient,
+        } => {
+            if *transient {
+                // Transient wakes bypass sqlite — they fire the broadcast
+                // channel only. No redelivery on restart; no message history
+                // entry. Used by bash task completions.
+                broker.ping(agent, from, body);
+                hive_sh4re::Response::Ok
+            } else {
+                match broker.send(&Message {
+                    from: from.clone(),
+                    to: agent.to_owned(),
+                    body: body.clone(),
+                    in_reply_to: None,
+                }) {
+                    Ok(()) => hive_sh4re::Response::Ok,
+                    Err(e) => hive_sh4re::Response::Err {
+                        message: format!("{e:#}"),
+                    },
+                }
+            }
+        }
         hive_sh4re::Request::Recent { limit } => match broker.recent_for(agent, *limit) {
             Ok(rows) => hive_sh4re::Response::Recent { rows },
             Err(e) => hive_sh4re::Response::Err {

hive-c0re/src/broker.rs

@@ -117,6 +117,11 @@ 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
@@ -191,6 +196,19 @@ impl Broker {
         Ok(())
     }
 
+    /// Deliver a transient wake signal to `to` without writing to sqlite.
+    /// The signal wakes a long-polling `recv_blocking_batch` for the target
+    /// agent but is not persisted, not redelivered on restart, and not shown
+    /// in message history. Use for ephemeral notifications (bash task
+    /// completions) where persistence would cause duplicate delivery.
+    pub fn ping(&self, to: &str, from: &str, body: &str) {
+        let _ = self.events.send(MessageEvent::Ping {
+            to: to.to_owned(),
+            from: from.to_owned(),
+            body: body.to_owned(),
+        });
+    }
+
     /// Latest `limit` messages addressed to `recipient`, newest-first.
     /// Includes delivered + undelivered alike — used for the operator
     /// inbox view on the dashboard. Caller decides what to show.
@@ -402,6 +420,32 @@ impl Broker {
                     }
                     // Lost a race (concurrent recv elsewhere). Keep waiting.
                 }
+                // Transient ping — not sqlite-backed. Return it directly as
+                // a Delivery with id=0 (sentinel: never pushed to unacked_ids
+                // so ack_turn silently ignores it).
+                Ok(Ok(MessageEvent::Ping {
+                    to,
+                    from,
+                    body,
+                })) if to == recipient => {
+                    // Also drain any real sqlite messages that may have landed
+                    // concurrently; prepend the ping so the agent sees both.
+                    let mut batch = self.recv_batch(recipient, max.saturating_sub(1))?;
+                    batch.insert(
+                        0,
+                        Delivery {
+                            id: 0,
+                            redelivered: false,
+                            message: Message {
+                                from,
+                                to,
+                                body,
+                                in_reply_to: None,
+                            },
+                        },
+                    );
+                    return Ok(batch);
+                }
                 Ok(Ok(_)) => {}
             }
         }

hive-c0re/src/dashboard.rs

@@ -899,7 +899,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,
@@ -909,7 +909,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,
@@ -918,7 +918,7 @@ async fn dashboard_history(State(state): State<AppState>) -> Response {
                             at,
                             in_reply_to,
                             file_refs,
-                        }
+                        })
                     }
                     crate::broker::MessageEvent::Delivered {
                         id,
@@ -929,7 +929,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,
@@ -938,8 +938,11 @@ async fn dashboard_history(State(state): State<AppState>) -> Response {
                             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()

hive-c0re/src/main.rs

@@ -408,6 +408,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-sh4re/src/lib.rs

@@ -344,7 +344,18 @@ pub enum Request {
     /// implicit (this agent); `from` is caller-chosen. See
     /// `docs/conventions.md::Wake injection` for the trust model and
     /// typical callers.
-    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.