vinzenz/hyperhive
vinzenz/hyperhive
Watch
0
0
Fork
You've already forked hyperhive
 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

fix(broker): route all wakes through sqlite, remove transient ping buffer

Browse source
This commit is contained in:
damocles 2026-07-05 11:55:40 +02:00
commit fac326aa35
9 changed files with 84 additions and 215 deletions
Download patch file Download diff file Expand all files Collapse all files

6
hive-ag3nt/src/bin/hive.rs
View file

@ -391,11 +391,7 @@ impl Surface for AgentSurface {
async fn wake_external(socket: &Path, from: String, body: String) -> Result<()> { async fn wake_external(socket: &Path, from: String, body: String) -> Result<()> {
let resp: AgentResponse = client::request( let resp: AgentResponse = client::request(
socket, socket,
&AgentRequest::Wake { &AgentRequest::Wake { from, body },
from,
body,
transient: false,
},
) )
.await?; .await?;
match resp { match resp {

1
hive-ag3nt/src/forge_notify.rs
View file

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

5
hive-ag3nt/src/serve_common.rs
View file

@ -11,9 +11,8 @@ use crate::turn_stats::TurnStatRow;
/// Assemble the per-turn wake prompt string. The role/tools/etc. live in the /// 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 /// 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 /// id, rendered as a `[msg #<id>]` marker so the agent can reference it in
/// `ack_until` (transient pings carry the sentinel 0 and render without it). /// `ack_until`. `unread` is the inbox depth after this message was popped.
/// `unread` is the inbox depth after this message was popped. `redelivered` /// `redelivered` prepends a "may already be handled" banner.
/// prepends a "may already be handled" banner.
#[must_use] #[must_use]
pub fn format_wake_prompt( pub fn format_wake_prompt(
id: i64, id: i64,

1
hive-bash-mcp/src/runner.rs
View file

@ -688,7 +688,6 @@ pub(crate) async fn send_wake(
let req = hive_sh4re::AgentRequest::Wake { let req = hive_sh4re::AgentRequest::Wake {
from: format!("bash-task-{id}"), from: format!("bash-task-{id}"),
body, body,
transient: true,
}; };
match UnixStream::connect(socket).await { match UnixStream::connect(socket).await {

184
hive-c0re/src/broker.rs
View file

@ -1,7 +1,7 @@
//! Sqlite-backed message broker. Survives `hive-c0re` restart, and taps every //! Sqlite-backed message broker. Survives `hive-c0re` restart, and taps every
//! send/recv onto a broadcast channel so the dashboard can stream it. //! 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::path::Path;
use std::sync::Mutex; use std::sync::Mutex;
use std::time::{SystemTime, UNIX_EPOCH}; use std::time::{SystemTime, UNIX_EPOCH};
@ -124,15 +124,6 @@ pub enum MessageEvent {
at: i64, at: i64,
in_reply_to: Option<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 /// Per-recipient in-memory bookkeeping for the deliver-then-ack
@ -154,13 +145,6 @@ struct RecipientInflight {
requeued_ids: HashSet<i64>, 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 { pub struct Broker {
conn: Mutex<Connection>, conn: Mutex<Connection>,
events: broadcast::Sender<MessageEvent>, events: broadcast::Sender<MessageEvent>,
@ -169,15 +153,6 @@ pub struct Broker {
/// boot, which rebuilds the `requeued_ids` set from the DB and /// boot, which rebuilds the `requeued_ids` set from the DB and
/// clears any stale `unacked_ids`). /// clears any stale `unacked_ids`).
inflight: Mutex<HashMap<String, RecipientInflight>>, 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 { impl Broker {
@ -196,7 +171,6 @@ impl Broker {
conn: Mutex::new(conn), conn: Mutex::new(conn),
events, events,
inflight: Mutex::new(HashMap::new()), inflight: Mutex::new(HashMap::new()),
pending_pings: Mutex::new(HashMap::new()),
}) })
} }
@ -270,77 +244,6 @@ impl Broker {
Ok(()) 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. /// Unread (unacked) messages addressed to `recipient`, newest-first.
/// Filters to `acked_at IS NULL` so the agent inbox view clears after /// Filters to `acked_at IS NULL` so the agent inbox view clears after
/// "mark all read" — mirroring exactly what `mark_all_read` will drain. /// "mark all read" — mirroring exactly what `mark_all_read` will drain.
@ -575,9 +478,8 @@ impl Broker {
return Ok(Vec::new()); return Ok(Vec::new());
} }
let mut rx = self.subscribe(); let mut rx = self.subscribe();
// Immediate check: buffered pings (incl. any fired while no recv was // Immediate check: any pending sqlite messages.
// parked) + queued sqlite mail. let batch = self.recv_batch(recipient, max)?;
let batch = self.collect_batch(recipient, max)?;
if !batch.is_empty() { if !batch.is_empty() {
return Ok(batch); return Ok(batch);
} }
@ -589,17 +491,14 @@ impl Broker {
}; };
match tokio::time::timeout(remaining, rx.recv()).await { match tokio::time::timeout(remaining, rx.recv()).await {
Err(_) => return Ok(Vec::new()), 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). // (in case we missed our notification while behind).
Ok(Err(_)) => return self.collect_batch(recipient, max), Ok(Err(_)) => return self.recv_batch(recipient, max),
// A relevant event landed (real message or transient ping). // A message was sent to this recipient (or a wake injected).
// Re-collect from the buffers; the ping payload on the event // Re-poll sqlite; the event payload is ignored — recv_batch is
// is ignored — `drain_pings` is the source of truth, so a // the source of truth and prevents double-delivery.
// ping can't be double-delivered. Ok(Ok(MessageEvent::Sent { to, .. })) if to == recipient => {
Ok(Ok(MessageEvent::Sent { to, .. } | MessageEvent::Ping { to, .. })) let batch = self.recv_batch(recipient, max)?;
if to == recipient =>
{
let batch = self.collect_batch(recipient, max)?;
if !batch.is_empty() { if !batch.is_empty() {
return Ok(batch); return Ok(batch);
} }
@ -1520,45 +1419,50 @@ mod tests {
assert_eq!(broker.ack_turn("b").unwrap(), 5); assert_eq!(broker.ack_turn("b").unwrap(), 5);
} }
/// Transient-wake regression guard: a `ping` fired while no `recv` is parked /// Wake messages go through sqlite and get real row ids that the
/// must NOT be lost — it's buffered and drained by the next collect. /// 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] #[test]
fn transient_ping_buffered_when_no_receiver_parked() { fn wake_persisted_with_real_id_and_ackable() {
let h = open_broker(); let h = open_broker();
let broker = &h.broker; let broker = &h.broker;
// Nobody is parked on recv when these fire (the old broadcast-only // Two wakes arrive (simulating bash-task completions).
// path dropped them here). broker
broker.ping("b", "matrix", "wake1"); .send(&msg("bash-task-1", "b", "bash task `1` finished: exit=0"))
broker.ping("b", "bash-task", "wake2"); .unwrap();
let batch = broker.collect_batch("b", 10).unwrap(); broker
let got: Vec<_> = batch .send(&msg("bash-task-2", "b", "bash task `2` finished: exit=0"))
.iter() .unwrap();
.map(|d| (d.id, d.message.from.as_str(), d.message.body.as_str())) let batch = broker.recv_batch("b", 10).unwrap();
.collect(); assert_eq!(batch.len(), 2);
assert_eq!( // Every delivery has a real (non-zero) id.
got, assert!(batch[0].id > 0);
vec![(0, "matrix", "wake1"), (0, "bash-task", "wake2")], assert!(batch[1].id > 0);
"buffered pings drain FIFO as id=0 deliveries" // Agent can ack them explicitly via ack_until.
); let max_id = batch.iter().map(|d| d.id).max().unwrap();
// Drained — the buffer is now empty. assert_eq!(broker.ack_turn("b").unwrap(), 2);
assert!(broker.collect_batch("b", 10).unwrap().is_empty()); // Confirming the rows are now closed.
// Pings are sentinel id=0, so ack_turn has nothing to close out. assert!(pop_one(broker, "b").is_none());
assert_eq!(broker.ack_turn("b").unwrap(), 0); // 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] #[test]
fn pending_pings_surface_before_queued_mail() { fn wake_and_mail_ordered_fifo() {
let h = open_broker(); let h = open_broker();
let broker = &h.broker; let broker = &h.broker;
broker.send(&msg("a", "b", "mail")).unwrap(); broker.send(&msg("a", "b", "mail")).unwrap();
broker.ping("b", "matrix", "wake"); broker
let batch = broker.collect_batch("b", 10).unwrap(); .send(&msg("matrix", "b", "matrix wake"))
.unwrap();
let batch = broker.recv_batch("b", 10).unwrap();
assert_eq!(batch.len(), 2); assert_eq!(batch.len(), 2);
assert_eq!(batch[0].id, 0, "ping first"); // FIFO: mail was inserted first, comes out first.
assert_eq!(batch[0].message.body, "wake"); assert_eq!(batch[0].message.body, "mail");
assert!(batch[1].id > 0, "sqlite mail after"); assert_eq!(batch[1].message.body, "matrix wake");
assert_eq!(batch[1].message.body, "mail"); assert!(batch[0].id > 0);
assert!(batch[1].id > 0);
} }
/// `recv_batch` with no pending traffic returns an empty vec /// `recv_batch` with no pending traffic returns an empty vec

43
hive-c0re/src/dashboard.rs
View file

@ -1082,7 +1082,7 @@ async fn dashboard_history(State(state): State<AppState>) -> Response {
messages.reverse(); messages.reverse();
let events: Vec<crate::dashboard_events::DashboardEvent> = messages let events: Vec<crate::dashboard_events::DashboardEvent> = messages
.into_iter() .into_iter()
.filter_map(|m| match m { .map(|m| match m {
crate::broker::MessageEvent::Sent { crate::broker::MessageEvent::Sent {
id, id,
from, from,
@ -1092,7 +1092,7 @@ async fn dashboard_history(State(state): State<AppState>) -> Response {
in_reply_to, in_reply_to,
} => { } => {
let file_refs = scan_validated_paths(&body); let file_refs = scan_validated_paths(&body);
Some(crate::dashboard_events::DashboardEvent::Sent { crate::dashboard_events::DashboardEvent::Sent {
seq: 0, seq: 0,
id, id,
from, from,
@ -1101,7 +1101,7 @@ async fn dashboard_history(State(state): State<AppState>) -> Response {
at: hive_sh4re::wire_time::from_secs(at), 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::Delivered {
id, id,
@ -1112,7 +1112,7 @@ async fn dashboard_history(State(state): State<AppState>) -> Response {
in_reply_to, in_reply_to,
} => { } => {
let file_refs = scan_validated_paths(&body); let file_refs = scan_validated_paths(&body);
Some(crate::dashboard_events::DashboardEvent::Delivered { crate::dashboard_events::DashboardEvent::Delivered {
seq: 0, seq: 0,
id, id,
from, from,
@ -1121,11 +1121,8 @@ async fn dashboard_history(State(state): State<AppState>) -> Response {
at: hive_sh4re::wire_time::from_secs(at), at: hive_sh4re::wire_time::from_secs(at),
in_reply_to, in_reply_to,
file_refs, 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(); .collect();
axum::Json(serde_json::json!({ "seq": seq, "events": events })).into_response() axum::Json(serde_json::json!({ "seq": seq, "events": events })).into_response()
@ -1359,27 +1356,27 @@ async fn api_operator_inbox(State(state): State<AppState>) -> Response {
Ok(messages) => { Ok(messages) => {
let items: Vec<serde_json::Value> = messages let items: Vec<serde_json::Value> = messages
.into_iter() .into_iter()
.filter_map(|m| match m { .filter_map(|m| {
crate::broker::MessageEvent::Sent { let crate::broker::MessageEvent::Sent {
id, id,
from, from,
body, body,
at, at,
in_reply_to, in_reply_to,
.. ..
} => { } = m
let file_refs = scan_validated_paths(&body); else {
Some(serde_json::json!({ return None;
"id": id, };
"from": from, let file_refs = scan_validated_paths(&body);
"body": body, Some(serde_json::json!({
"at": hive_sh4re::wire_time::from_secs(at), "id": id,
"in_reply_to": in_reply_to, "from": from,
"file_refs": file_refs, "body": body,
})) "at": hive_sh4re::wire_time::from_secs(at),
} "in_reply_to": in_reply_to,
crate::broker::MessageEvent::Delivered { .. } "file_refs": file_refs,
| crate::broker::MessageEvent::Ping { .. } => None, }))
}) })
.collect(); .collect();
axum::Json(serde_json::json!({ "messages": items })).into_response() axum::Json(serde_json::json!({ "messages": items })).into_response()

3
hive-c0re/src/main.rs
View file

@ -522,9 +522,6 @@ fn spawn_broker_to_dashboard_forwarder(coord: Arc<Coordinator>) {
file_refs, 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)) => { Err(tokio::sync::broadcast::error::RecvError::Lagged(n)) => {
tracing::warn!(skipped = n, "broker-to-dashboard forwarder lagged"); tracing::warn!(skipped = n, "broker-to-dashboard forwarder lagged");
} }

40
hive-c0re/src/socket_server.rs
View file

@ -185,11 +185,7 @@ pub(crate) async fn dispatch_shared(
} }
hive_sh4re::Request::Status => handle_status(coord, agent), hive_sh4re::Request::Status => handle_status(coord, agent),
hive_sh4re::Request::OperatorMsg { body } => handle_operator_msg(coord, agent, body), hive_sh4re::Request::OperatorMsg { body } => handle_operator_msg(coord, agent, body),
hive_sh4re::Request::Wake { hive_sh4re::Request::Wake { from, body } => handle_wake(coord, agent, from, body),
from,
body,
transient,
} => handle_wake(coord, agent, from, body, *transient),
hive_sh4re::Request::Recent { limit } => handle_recent(coord, agent, *limit), hive_sh4re::Request::Recent { limit } => handle_recent(coord, agent, *limit),
hive_sh4re::Request::Ask { hive_sh4re::Request::Ask {
question, question,
@ -311,33 +307,25 @@ async fn handle_recv(
} }
} }
/// `Wake` — inject a wake into `agent`'s own inbox. Transient wakes /// `Wake` — inject a wake into `agent`'s own inbox. Persisted through
/// fire the broadcast channel only (no sqlite row, no redelivery on /// the sqlite broker like any other message so the agent can ack it
/// restart — used by bash-task completions); durable wakes persist /// via `AckUntil` and it appears in message history for post-mortem.
/// through the broker like any other message.
fn handle_wake( fn handle_wake(
coord: &Arc<Coordinator>, coord: &Arc<Coordinator>,
agent: &str, agent: &str,
from: &str, from: &str,
body: &str, body: &str,
transient: bool,
) -> hive_sh4re::Response { ) -> hive_sh4re::Response {
let broker = &coord.broker; match coord.broker.send(&Message {
if transient { from: from.to_owned(),
broker.ping(agent, from, body); to: agent.to_owned(),
hive_sh4re::Response::Ok body: body.to_owned(),
} else { in_reply_to: None,
match broker.send(&Message { }) {
from: from.to_owned(), Ok(()) => hive_sh4re::Response::Ok,
to: agent.to_owned(), Err(e) => hive_sh4re::Response::Err {
body: body.to_owned(), message: format!("{e:#}"),
in_reply_to: None, },
}) {
Ok(()) => hive_sh4re::Response::Ok,
Err(e) => hive_sh4re::Response::Err {
message: format!("{e:#}"),
},
}
} }
} }

16
hive-sh4re/src/lib.rs
View file

@ -579,19 +579,9 @@ pub enum Request {
/// Wake-up event injected from inside the container. Recipient is /// Wake-up event injected from inside the container. Recipient is
/// implicit (this agent); `from` is caller-chosen. See /// implicit (this agent); `from` is caller-chosen. See
/// `docs/conventions.md::Wake injection` for the trust model and /// `docs/conventions.md::Wake injection` for the trust model and
/// typical callers. /// typical callers. The wake is persisted in the sqlite broker
/// /// like any other message — the agent can ack it via `AckUntil`.
/// When `transient` is `true` the server delivers the wake signal Wake { from: String, body: String },
/// 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. /// Last `limit` messages addressed to this agent, newest-first.
/// Non-mutating — pulls from the broker without delivering. The /// Non-mutating — pulls from the broker without delivering. The
/// per-agent web UI uses this to render its own inbox section. /// per-agent web UI uses this to render its own inbox section.