From a1c6736ba54657c16b0db0413952da92adb951c0 Mon Sep 17 00:00:00 2001 From: damocles Date: Wed, 3 Jun 2026 10:40:33 +0200 Subject: [PATCH] =?UTF-8?q?feat(#1106):=20transient=20wake=20for=20bash=20?= =?UTF-8?q?tasks=20=E2=80=94=20bypass=20broker=20sqlite?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- hive-ag3nt/src/bash_runner.rs | 5 ++++ hive-ag3nt/src/bin/hive.rs | 6 +++-- hive-ag3nt/src/forge_notify.rs | 1 + hive-c0re/src/agent_server.rs | 36 +++++++++++++++++++--------- hive-c0re/src/broker.rs | 44 ++++++++++++++++++++++++++++++++++ hive-c0re/src/dashboard.rs | 13 ++++++---- hive-c0re/src/main.rs | 3 +++ hive-sh4re/src/lib.rs | 13 +++++++++- 8 files changed, 102 insertions(+), 19 deletions(-) diff --git a/hive-ag3nt/src/bash_runner.rs b/hive-ag3nt/src/bash_runner.rs index 2f123450..23dc202d 100644 --- a/hive-ag3nt/src/bash_runner.rs +++ b/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"), diff --git a/hive-ag3nt/src/bin/hive.rs b/hive-ag3nt/src/bin/hive.rs index cf2d3c50..8f899ed6 100644 --- a/hive-ag3nt/src/bin/hive.rs +++ b/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}"), diff --git a/hive-ag3nt/src/forge_notify.rs b/hive-ag3nt/src/forge_notify.rs index 829e3584..544ebdf7 100644 --- a/hive-ag3nt/src/forge_notify.rs +++ b/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 diff --git a/hive-c0re/src/agent_server.rs b/hive-c0re/src/agent_server.rs index 93c7d81a..9a5ce8f1 100644 --- a/hive-c0re/src/agent_server.rs +++ b/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 { diff --git a/hive-c0re/src/broker.rs b/hive-c0re/src/broker.rs index 088556b9..1d4d7e10 100644 --- a/hive-c0re/src/broker.rs +++ b/hive-c0re/src/broker.rs @@ -117,6 +117,11 @@ 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 @@ -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(_)) => {} } } diff --git a/hive-c0re/src/dashboard.rs b/hive-c0re/src/dashboard.rs index 07ec47ce..7c4a763d 100644 --- a/hive-c0re/src/dashboard.rs +++ b/hive-c0re/src/dashboard.rs @@ -899,7 +899,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, @@ -909,7 +909,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, @@ -918,7 +918,7 @@ async fn dashboard_history(State(state): State) -> Response { at, in_reply_to, file_refs, - } + }) } crate::broker::MessageEvent::Delivered { id, @@ -929,7 +929,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, @@ -938,8 +938,11 @@ async fn dashboard_history(State(state): State) -> 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() diff --git a/hive-c0re/src/main.rs b/hive-c0re/src/main.rs index d463ba0a..a41cb326 100644 --- a/hive-c0re/src/main.rs +++ b/hive-c0re/src/main.rs @@ -408,6 +408,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-sh4re/src/lib.rs b/hive-sh4re/src/lib.rs index 6f861623..9c7df6dc 100644 --- a/hive-sh4re/src/lib.rs +++ b/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.