mara/hyperhive
mara/hyperhive
Watch
0
0
Fork
You've already forked hyperhive
 0
Code Activity

drop the host side of the agent broker send

Browse source 
Remove the agent-side Send and AckUntil requests, their hive-c0re handlers
(handle_send with its 4 KiB size cap, handle_ack_until) and Broker::ack_until.
Everything that existed only for them goes with them: the OkWarn response
(its sole producer was Send's reserved-name warning), Recv's max parameter
and RECV_BATCH_MAX (the harness never asked for more than one message),
OPERATOR_SEND_REFUSAL, and the MESSAGE_MAX_BYTES / check_size limit (the
reminder timer keeps its own copy of the cap, so its comment is reworded).

Messages.remaining stays on the wire (default on read) because a harness
built before this change still requires it. Update the conventions docs.
This commit is contained in:
damocles 2026-11-08 02:35:34 +01:00 • committed by mara
commit 1c3b5e52cf
3 changed files with 58 additions and 446 deletions
Download patch file Download diff file Expand all files Collapse all files

62
hive-c0re/src/agent_config/limits.rs
View file

@ -1,45 +1,4 @@
//! Wire-protocol size limits shared across the agent + manager
//! sockets. Caps on inline message bodies stop a single chatty agent
//! (or a misbehaving extra-MCP server) from flooding the broker
//! sqlite with megabyte-sized rows that then bloat every recipient's
//! wake-prompt context. Anything genuinely larger should be written
//! to a state file and the path sent as the body.
//!
//! Reminders get a separate auto-file escape hatch (see
//! `socket_server::handle_remind`) so callers don't have to think
//! about it — oversized reminder bodies get persisted to disk
//! transparently and the inbox sees a pointer.
/// Per-message body cap. Applies to `send` bodies and the stored
/// inline form of a reminder. 4 KiB
/// catches the bulk of conversational overflow (status reports,
/// bullet-list summaries, short proposals) while staying small
/// enough that a backed-up inbox of ~10 unread messages only adds
/// ~40 KiB to the recipient's wake-prompt context. Genuinely
/// long-form artifacts (audit reports, full diffs, transcripts)
/// still belong in a state file — the error message on overflow
/// points callers at that escape hatch.
pub const MESSAGE_MAX_BYTES: usize = 4096;
/// Validate that `body` fits under [`MESSAGE_MAX_BYTES`]. Returns a
/// caller-ready error string (caller wraps in
/// `Response::Err`) on failure.
///
/// `label` shows up in the error message verbatim — pass a short
/// noun like `"send"` or `"broadcast"` so the model can tell which
/// call got rejected.
pub fn check_size(label: &str, body: &str) -> Result<(), String> {
if body.len() > MESSAGE_MAX_BYTES {
Err(format!(
"{label} body too long ({} bytes, max {MESSAGE_MAX_BYTES}); write the \
payload to a file under your `/agents/<you>/state/` dir and send the \
path as the body instead",
body.len()
))
} else {
Ok(())
}
}
//! Wire-protocol size limits shared across the agent sockets.
/// Per-status soft cap. `set_status` renders as a short chip on the
/// dashboard agent card — the front-end truncates long strings to
@ -97,25 +56,6 @@ pub fn check_status_text(text: &str) -> Result<(), String> {
mod tests {
use super::*;
#[test]
fn accepts_short_body() {
assert!(check_size("send", "hello").is_ok());
assert!(check_size("send", &"x".repeat(MESSAGE_MAX_BYTES)).is_ok());
}
#[test]
fn rejects_oversize_body() {
let err = check_size("send", &"x".repeat(MESSAGE_MAX_BYTES + 1)).unwrap_err();
assert!(err.contains("send body too long"));
assert!(err.contains(&format!("max {MESSAGE_MAX_BYTES}")));
}
#[test]
fn label_threads_through() {
let err = check_size("broadcast", &"x".repeat(MESSAGE_MAX_BYTES + 1)).unwrap_err();
assert!(err.starts_with("broadcast body too long"));
}
#[test]
fn check_status_accepts_short_single_line() {
assert!(check_status_text("idle").is_ok());

162
hive-c0re/src/socket_server/mod.rs
View file

@ -190,11 +190,6 @@ async fn serve(stream: UnixStream, agent: String, coord: Arc<Coordinator>) -> Re
/// positive `wait_seconds`.
pub(crate) const RECV_LONG_POLL_MAX: std::time::Duration = std::time::Duration::from_mins(3);
/// Server-side hard cap on `Recv.max` — canonical value lives in
/// `hive_sh4re::inbox::RECV_BATCH_MAX` so the harness's wake-prompt hint and
/// this enforcement site can't drift apart.
pub(crate) const RECV_BATCH_MAX: u32 = hive_sh4re::inbox::RECV_BATCH_MAX;
pub(crate) fn recv_timeout(wait_seconds: Option<u64>) -> std::time::Duration {
match wait_seconds {
Some(s) => std::time::Duration::from_secs(s).min(RECV_LONG_POLL_MAX),
@ -210,13 +205,8 @@ async fn dispatch(
coord: &Arc<Coordinator>,
) -> hive_core_agent_sock::Response {
match req {
hive_core_agent_sock::Request::Send {
to,
body,
in_reply_to,
} => handle_send(coord, agent, to, body, *in_reply_to),
hive_core_agent_sock::Request::Recv { wait_seconds, max } => {
handle_recv(coord, agent, *wait_seconds, *max).await
hive_core_agent_sock::Request::Recv { wait_seconds } => {
handle_recv(coord, agent, *wait_seconds).await
}
hive_core_agent_sock::Request::Status => handle_status(coord, agent),
hive_core_agent_sock::Request::Wake { from, body } => handle_wake(coord, agent, from, body),
@ -226,7 +216,6 @@ async fn dispatch(
handle_get_agent_meta(coord, agent, name.as_ref()).await
}
hive_core_agent_sock::Request::AckTurn => handle_ack_turn(coord, agent),
hive_core_agent_sock::Request::AckUntil { up_to } => handle_ack_until(coord, agent, *up_to),
hive_core_agent_sock::Request::RequeueInflight => handle_requeue_inflight(coord, agent),
hive_core_agent_sock::Request::GracefulStopComplete => {
// Harness drained + is exiting: clear the fence so the
@ -246,13 +235,12 @@ async fn dispatch(
}
}
/// `Recv` — long-poll the broker for up to `max` messages (capped at
/// `RECV_BATCH_MAX`), mapping deliveries onto the wire response.
/// `Recv` — long-poll the broker for the next message, mapping the delivery
/// onto the wire response.
async fn handle_recv(
coord: &Arc<Coordinator>,
agent: &str,
wait_seconds: Option<u64>,
max: Option<u32>,
) -> hive_core_agent_sock::Response {
// Graceful-stop fence: while a graceful stop is pending for this agent,
// return `GracefulStop` instead of polling the broker. The harness runs
@ -263,32 +251,27 @@ async fn handle_recv(
if coord.is_graceful_stop_pending(agent) {
return hive_core_agent_sock::Response::GracefulStop;
}
let cap = max.unwrap_or(1).min(RECV_BATCH_MAX) as usize;
match coord
.broker
.recv_blocking_batch(agent, recv_timeout(wait_seconds), cap)
.recv_blocking_batch(agent, recv_timeout(wait_seconds), 1)
.await
{
Ok(deliveries) => {
// `recv_batch` stamps `delivered_at` on every popped row, so a
// `count_pending` here excludes the just-popped batch and reports
// exactly how many still-pending messages remain to drain. A count
// error is non-fatal — fall back to 0 rather than fail the recv.
let remaining = coord.broker.count_pending(agent).unwrap_or(0);
hive_core_agent_sock::Response::Messages {
messages: deliveries
.into_iter()
.map(|d| hive_sh4re::inbox::DeliveredMessage {
from: d.message.from.to_string(),
body: d.message.body,
id: d.id,
redelivered: d.redelivered,
in_reply_to: d.message.in_reply_to,
})
.collect(),
remaining,
}
}
Ok(deliveries) => hive_core_agent_sock::Response::Messages {
messages: deliveries
.into_iter()
.map(|d| hive_sh4re::inbox::DeliveredMessage {
from: d.message.from.to_string(),
body: d.message.body,
id: d.id,
redelivered: d.redelivered,
in_reply_to: d.message.in_reply_to,
})
.collect(),
// Still sent: a harness built before the field went unused
// requires it (see the `Messages` doc). A count error is
// non-fatal, so fall back to 0 rather than fail the recv.
remaining: coord.broker.count_pending(agent).unwrap_or(0),
},
Err(e) => hive_core_agent_sock::Response::Err {
message: format!("{e:#}"),
},
@ -296,8 +279,8 @@ 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.
/// the sqlite broker like any other message so it appears in message history
/// for post-mortem.
fn handle_wake(
coord: &Arc<Coordinator>,
agent: &str,
@ -428,21 +411,6 @@ fn handle_ack_turn(coord: &Arc<Coordinator>, agent: &str) -> hive_core_agent_soc
}
}
/// `AckUntil` — bulk-ack every message addressed to `agent` with row
/// id `<= up_to` (the agent-side backlog-triage escape hatch).
fn handle_ack_until(
coord: &Arc<Coordinator>,
agent: &str,
up_to: i64,
) -> hive_core_agent_sock::Response {
match coord.broker.ack_until(agent, up_to) {
Ok(count) => hive_core_agent_sock::Response::Acked { count },
Err(e) => hive_core_agent_sock::Response::Err {
message: format!("{e:#}"),
},
}
}
/// `RequeueInflight` — resurface `agent`'s unacked in-flight messages
/// (crash recovery on harness boot).
fn handle_requeue_inflight(
@ -472,73 +440,6 @@ fn handle_get_loose_ends(coord: &Arc<Coordinator>, agent: &str) -> Response {
}
}
/// `Send` handler. Applies the
/// 4 KiB body cap, then routes broadcast (`to == "*"`) / unicast through
/// their respective broker calls.
pub(crate) fn handle_send(
coord: &Arc<Coordinator>,
agent: &str,
to: &str,
body: &str,
in_reply_to: Option<i64>,
) -> Response {
if let Err(message) = crate::limits::check_size("send", body) {
return Response::Err { message };
}
if to == "*" {
let errors = coord.broadcast_send(agent, body);
return if errors.is_empty() {
Response::Ok
} else {
Response::Err {
message: format!("broadcast failed for agents: {}", errors.join(", ")),
}
};
}
// Validate that the recipient is a known local agent. Without this
// check a typo in `to` silently queues a message nobody will ever read.
//
// Cross-hive messaging (`name@hive` qualified names) and the operator
// are not reached through the broker — the Matrix MCP tools are.
if to.contains('@') {
return Response::Err {
message: format!(
"send failed: cross-hive recipient `{to}` is not supported \
via the broker — use Matrix MCP tools for cross-hive messaging"
),
};
}
if to == hive_sh4re::manager::OPERATOR_RECIPIENT {
return Response::Err {
message: hive_sh4re::manager::OPERATOR_SEND_REFUSAL.to_owned(),
};
}
// A name that doesn't parse as an Ident can't be a local agent, so
// it collapses into the same "unknown recipient" error as a valid
// name with no state dir.
let exists =
hive_types::Ident::parse(to).is_ok_and(|id| crate::paths::agent_state_dir(&id).exists());
if !exists {
return Response::Err {
message: format!(
"send failed: unknown recipient `{to}` \
(no agent with that name exists on this hive)"
),
};
}
match coord.broker.send(&Message {
from: hive_sh4re::manager::trusted_sender(agent),
to: to.to_owned(),
body: body.to_owned(),
in_reply_to,
}) {
Ok(()) => Response::Ok,
Err(e) => Response::Err {
message: format!("{e:#}"),
},
}
}
#[cfg(test)]
mod tests {
use hive_sock_client::REQUEST_LINE_MAX_BYTES;
@ -560,23 +461,6 @@ mod tests {
(dir, Arc::new(coord))
}
#[test]
fn send_to_operator_is_refused_naming_matrix_and_the_no_matrix_fallback() {
let (_dir, coord) = coordinator();
let resp = handle_send(
&coord,
"iris",
hive_sh4re::manager::OPERATOR_RECIPIENT,
"hi",
None,
);
assert!(
matches!(&resp, Response::Err { message }
if message.contains("mcp__matrix__") && message.contains("set_status")),
"{resp:?}"
);
}
#[tokio::test]
async fn serve_answers_an_over_long_line_with_an_error_and_closes() {
let (_dir, coord) = coordinator();

109
hive-c0re/src/stores/broker.rs
View file

@ -229,8 +229,9 @@ impl Broker {
/// lines without popping the queue.
pub fn count_pending(&self, recipient: &str) -> Result<u64> {
let conn = self.conn.lock().unwrap();
// Skip rows closed by `ack_until` while still pending — they
// will never pop, so counting them would show phantom unread.
// Skip rows acked while still pending (the removed `ack_until` could
// leave such rows in an older db) — they never pop, so counting them
// would show phantom unread.
let n: i64 = conn.query_row(
"SELECT COUNT(*) FROM messages
WHERE recipient = ?1 AND delivered_at IS NULL AND acked_at IS NULL",
@ -345,9 +346,10 @@ impl Broker {
let mut inflight = self.inflight.lock().unwrap();
let conn = self.conn.lock().unwrap();
let max_i = i64::try_from(max).unwrap_or(i64::MAX);
// `acked_at IS NULL` matters for rows closed by `ack_until`
// while still pending (never delivered): they carry an ack but
// no `delivered_at`, and must not pop.
// `acked_at IS NULL` matters for rows acked while still pending
// (never delivered; the removed `ack_until` could leave such rows in
// an older db): they carry an ack but no `delivered_at`, and must
// not pop.
let mut stmt = conn.prepare(
"SELECT id, sender, recipient, body, in_reply_to
FROM messages
@ -460,38 +462,6 @@ impl Broker {
Ok(u64::try_from(n).unwrap_or(0))
}
/// Bulk-ack every message addressed to `recipient` with row id
/// `<= up_to` that isn't acked yet — pending AND delivered rows
/// alike. The agent-facing triage tool behind `AckUntil`: after a
/// redelivered flood the agent acks everything up to the highest id
/// it has seen instead of re-popping each row. Recipient-scoped by
/// the WHERE clause, so an agent can never touch another agent's
/// rows. Also drains the in-memory `unacked_ids` / `requeued_ids`
/// bookkeeping below the cutoff so a later `ack_turn` doesn't
/// re-update rows this call already closed and a stale redelivery
/// tag doesn't outlive its row. Returns the number of rows newly
/// acked.
///
/// # Errors
///
/// Propagates sqlite errors from the `UPDATE`.
pub fn ack_until(&self, recipient: &str, up_to: i64) -> Result<u64> {
// Same lock order as `recv` / `ack_turn` / `requeue_inflight`:
// `inflight` FIRST, then `conn`.
let mut inflight = self.inflight.lock().unwrap();
if let Some(state) = inflight.get_mut(recipient) {
state.unacked_ids.retain(|&id| id > up_to);
state.requeued_ids.retain(|&id| id > up_to);
}
let conn = self.conn.lock().unwrap();
let n = conn.execute(
"UPDATE messages SET acked_at = ?1
WHERE recipient = ?2 AND id <= ?3 AND acked_at IS NULL",
params![Utc::now().timestamp(), recipient, up_to],
)?;
Ok(u64::try_from(n).unwrap_or(0))
}
/// Resurface every message the broker previously handed to this
/// recipient that never got `acked_at` set. Used by the harness at
/// boot to recover from the crashed-mid-turn / OOM-killed /
@ -711,61 +681,6 @@ mod tests {
assert_eq!(broker.ack_turn("b").unwrap(), 0);
}
/// Bulk triage: three queued messages, agent acks up to the second
/// id — the first two never deliver again, the third still pops.
#[test]
fn ack_until_acks_only_rows_at_or_below_cutoff() {
let h = open_broker();
let broker = &h.broker;
broker.send(&msg("a", "b", "one")).unwrap();
broker.send(&msg("a", "b", "two")).unwrap();
broker.send(&msg("a", "b", "three")).unwrap();
// Pop the first two so we know their ids (FIFO).
let d1 = pop_one(broker, "b").expect("popped one");
let d2 = pop_one(broker, "b").expect("popped two");
assert_eq!(broker.ack_until("b", d2.id).unwrap(), 2);
// The cutoff also drained the in-memory unacked list, so a
// turn-level ack right after finds nothing left to do.
assert_eq!(broker.ack_turn("b").unwrap(), 0);
// A restart-style requeue finds nothing below the cutoff …
assert_eq!(broker.requeue_inflight("b").unwrap(), 0);
// … and the third message (id above the cutoff) still pops.
let d3 = pop_one(broker, "b").expect("third still pending");
assert_eq!(d3.message.body, "three");
assert!(d1.id < d2.id && d2.id < d3.id);
}
/// Pending (never-delivered) rows below the cutoff are acked too —
/// that's the whole point for a stale backlog the agent never
/// popped individually.
#[test]
fn ack_until_covers_pending_rows() {
let h = open_broker();
let broker = &h.broker;
broker.send(&msg("a", "b", "stale-1")).unwrap();
broker.send(&msg("a", "b", "stale-2")).unwrap();
// Learn the highest id by peeking via recent_for (non-mutating).
let rows = broker.recent_for("b", 10).unwrap();
let max_id = rows.iter().map(|r| r.id).max().expect("rows");
assert_eq!(broker.ack_until("b", max_id).unwrap(), 2);
assert!(pop_one(broker, "b").is_none(), "backlog cleared");
}
/// Recipient scoping: acking b's inbox never touches c's rows,
/// even when c's ids fall below the cutoff.
#[test]
fn ack_until_is_recipient_scoped() {
let h = open_broker();
let broker = &h.broker;
broker.send(&msg("a", "c", "for-c")).unwrap();
broker.send(&msg("a", "b", "for-b")).unwrap();
let rows = broker.recent_for("b", 10).unwrap();
let max_id = rows.iter().map(|r| r.id).max().expect("rows");
assert_eq!(broker.ack_until("b", max_id).unwrap(), 1);
let d = pop_one(broker, "c").expect("c's message untouched");
assert_eq!(d.message.body, "for-c");
}
/// Crash-recovery: send → recv → (no ack) → `requeue_inflight`
/// resets `delivered_at` + tags the next pop as redelivered. After
/// that `ack_turn` closes it out cleanly.
@ -884,9 +799,9 @@ 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.
/// Wake messages go through sqlite and get real row ids. A wake sent
/// when no recv is parked is not lost — it sits in sqlite until the
/// next recv pops it.
#[test]
fn wake_persisted_with_real_id_and_ackable() {
let h = open_broker();
@ -903,13 +818,9 @@ mod tests {
// 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);
}
/// Wakes and regular mail both go through sqlite: FIFO by insertion order.