From 1c3b5e52cfc8baf139b7e4a2b670bf9314e21c52 Mon Sep 17 00:00:00 2001 From: damocles Date: Sun, 8 Nov 2026 02:35:34 +0100 Subject: [PATCH] drop the host side of the agent broker send 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. --- docs/5-project/process/conventions.md | 36 ++---- hive-agent-mcp/src/mcp/render.rs | 43 +------ hive-agent/src/main.rs | 1 - hive-agent/src/reminder_timer.rs | 6 +- hive-c0re/src/agent_config/limits.rs | 62 +--------- hive-c0re/src/socket_server/mod.rs | 162 ++++---------------------- hive-c0re/src/stores/broker.rs | 109 ++--------------- hive-core-agent-sock/src/lib.rs | 63 ++-------- hive-sh4re/src/inbox.rs | 10 -- hive-sh4re/src/manager.rs | 12 +- 10 files changed, 58 insertions(+), 446 deletions(-) diff --git a/docs/5-project/process/conventions.md b/docs/5-project/process/conventions.md index 38783457..f9b80332 100644 --- a/docs/5-project/process/conventions.md +++ b/docs/5-project/process/conventions.md @@ -94,11 +94,9 @@ name validation rejects any character outside `[a-z0-9_-]`, so the angle-bracket and asterisk shapes below are structurally safe. - `*` — broadcast: deliver to every running agent except the sender - (`socket_server::handle_send` fans out via `Coordinator::broadcast_send`). + (`Coordinator::broadcast_send`, used by hive-c0re's own system broadcasts). - `operator` — the human at the dashboard. Reserved so no agent can take - the name, and refused as a recipient: the broker answers a `send` to it - with `OPERATOR_SEND_REFUSAL` (`hive-sh4re/src/manager.rs`). It appears - only as the sender of operator-authored messages. + the name. It appears only as the sender of operator-authored messages. The broker stores the recipient exactly as the agent passed it. @@ -115,14 +113,10 @@ in `hive-c0re::dashboard_events::DashboardEvent`. ### Broker delivery + ack cycle `AgentRequest::Recv` is the only path that delivers messages to an -agent. Always returns a list (`Messages { messages }`) — empty when -nothing's pending, single-pop when `max = None` (default 1, the -single-message behaviour), batched up to `max` when caller asks for -more (server-side cap is 5; values above clamp silently). The wire -request still carries an optional `wait_seconds` (long-poll the first -message, once one arrives — or one is already pending — the call -drains up to `max` in total): the harness's own turn-driving loop -uses it internally (`hive-agent`'s `recv_next`, 180s). The +agent. It returns a list (`Messages { messages }`) of zero or one +message: empty when nothing is pending. The request carries an optional +`wait_seconds` (long-poll for the next message); the harness's own +turn-driving loop uses it (`hive-agent`'s `recv_next`, 180s). The agent-facing MCP surface has no `recv` tool at all. Per-row bookkeeping inside the broker: @@ -153,21 +147,9 @@ each id in a per-recipient in-memory set so the next `Recv` can tag the row with `redelivered: true`. Idempotent + cheap when there's nothing in flight, so the at-boot fire is unconditional. -`AgentRequest::AckUntil { up_to }` is the agent-facing bulk-triage -escape hatch (`mcp__hyperhive__ack_until`). Unlike `AckTurn` **it's** -visible to claude: each recv row and wake prompt carries a -`[msg #]` marker (the broker row id; transient pings show no -marker — their sentinel id 0 has nothing to ack), and -`ack_until(up_to: n)` marks every one of the agent's rows with -`id <= n` handled in a single UPDATE — pending and delivered alike. -This bounds the redelivered-flood cost after a restart: instead of -popping dozens of already-handled messages one turn at a time, the -agent notes the highest id it has seen and acks up to it. -Recipient-scoped (an agent can only ack its own rows); also drains -the in-memory `unacked_ids` / `requeued_ids` bookkeeping below the -cutoff so a later `AckTurn` doesn't double-update and a stale -redelivery tag can't outlive its row. The operator-side sibling is -the dashboard's "mark all read" (unbounded, per-agent). +Agents have no bulk-ack of their own: the harness acks a turn's message +with `AckTurn`. The operator-side bulk action is the dashboard's "mark all +read" (unbounded, per-agent). ### Loose-ends wire shape diff --git a/hive-agent-mcp/src/mcp/render.rs b/hive-agent-mcp/src/mcp/render.rs index b4204f2a..a5cdc21d 100644 --- a/hive-agent-mcp/src/mcp/render.rs +++ b/hive-agent-mcp/src/mcp/render.rs @@ -33,18 +33,6 @@ pub fn format_ack( ) -> String { match resp { Ok(hive_core_agent_sock::Response::Ok) => ok_msg, - // Succeeded, with something the caller needs to read. Rendered - // after the success line rather than instead of it: the operation - // DID happen, and a warning shown as though it were a failure - // invites a retry that would only queue a second one. - Ok(hive_core_agent_sock::Response::OkWarn { warnings }) => { - let mut out = ok_msg; - for w in &warnings { - out.push_str("\n⚠️ "); - out.push_str(w); - } - out - } other => reply_err(other, tool), } } @@ -457,7 +445,7 @@ pub fn annotate_retries(mut s: String, retries: u32) -> String { #[cfg(test)] mod tests { - use super::{dial_socket, format_ack}; + use super::dial_socket; #[tokio::test] async fn a_socket_that_accepts_and_never_answers_is_given_up_on() { @@ -487,33 +475,4 @@ mod tests { assert!(resp.is_none()); } - - #[test] - fn ok_warn_keeps_the_success_line_and_appends_each_warning() { - let out = format_ack( - Ok(hive_core_agent_sock::Response::OkWarn { - warnings: vec!["name is reserved".to_owned(), "second thing".to_owned()], - }), - "remind", - "reminder set".to_owned(), - ); - // The operation HAPPENED — dropping the success line would read as a - // failure and invite a retry that repeats it. - assert!(out.starts_with("reminder set")); - assert!(out.contains("⚠️ name is reserved")); - assert!(out.contains("⚠️ second thing")); - } - - #[test] - fn plain_ok_is_untouched_by_the_warning_path() { - // Absence arm: without it, a renderer that always appended a - // warning marker would pass the test above. - let out = format_ack( - Ok(hive_core_agent_sock::Response::Ok), - "remind", - "sent".to_owned(), - ); - assert_eq!(out, "sent"); - assert!(!out.contains('⚠')); - } } diff --git a/hive-agent/src/main.rs b/hive-agent/src/main.rs index 2d88b3e8..8df9af9e 100644 --- a/hive-agent/src/main.rs +++ b/hive-agent/src/main.rs @@ -374,7 +374,6 @@ impl Surface for AgentSurface { socket, &Request::Recv { wait_seconds: Some(RECV_WAIT.as_secs()), - max: None, }, CONTROL_SOCKET_RETRY, RECV_RESPONSE_TIMEOUT, diff --git a/hive-agent/src/reminder_timer.rs b/hive-agent/src/reminder_timer.rs index 0b561136..37ca8a88 100644 --- a/hive-agent/src/reminder_timer.rs +++ b/hive-agent/src/reminder_timer.rs @@ -33,11 +33,7 @@ const REMINDER_BATCH_LIMIT: u64 = 100; /// Poll interval — matches the old c0re `POLL_INTERVAL`. const POLL_INTERVAL: Duration = Duration::from_secs(5); -/// Same cap the broker enforces on `send`/`remind` bodies -/// (`hive-c0re`'s `agent_config::limits::MESSAGE_MAX_BYTES`, not -/// reachable from here — hive-agent doesn't depend on hive-c0re). -/// Duplicated rather than shared: this is the last remaining reminder -/// caller of that constant once the c0re-side store is deleted (commit 6). +/// Cap on the inline form of a `remind` body; a longer body goes to a file. const REMINDER_BODY_MAX_BYTES: usize = 4096; /// Maximum pending (undelivered) reminders this agent may hold at once. diff --git a/hive-c0re/src/agent_config/limits.rs b/hive-c0re/src/agent_config/limits.rs index 279a3050..b15982ee 100644 --- a/hive-c0re/src/agent_config/limits.rs +++ b/hive-c0re/src/agent_config/limits.rs @@ -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//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()); diff --git a/hive-c0re/src/socket_server/mod.rs b/hive-c0re/src/socket_server/mod.rs index d79bc78d..6b21837e 100644 --- a/hive-c0re/src/socket_server/mod.rs +++ b/hive-c0re/src/socket_server/mod.rs @@ -190,11 +190,6 @@ async fn serve(stream: UnixStream, agent: String, coord: Arc) -> 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) -> 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, ) -> 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, agent: &str, wait_seconds: Option, - max: Option, ) -> 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, agent: &str, @@ -428,21 +411,6 @@ fn handle_ack_turn(coord: &Arc, 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, - 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, 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, - agent: &str, - to: &str, - body: &str, - in_reply_to: Option, -) -> 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(); diff --git a/hive-c0re/src/stores/broker.rs b/hive-c0re/src/stores/broker.rs index e7bc0cda..4cb40221 100644 --- a/hive-c0re/src/stores/broker.rs +++ b/hive-c0re/src/stores/broker.rs @@ -229,8 +229,9 @@ impl Broker { /// lines without popping the queue. pub fn count_pending(&self, recipient: &str) -> Result { 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 { - // 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. diff --git a/hive-core-agent-sock/src/lib.rs b/hive-core-agent-sock/src/lib.rs index 65f18676..b2ecc268 100644 --- a/hive-core-agent-sock/src/lib.rs +++ b/hive-core-agent-sock/src/lib.rs @@ -22,24 +22,12 @@ fn default_true() -> bool { #[derive(Debug, Clone, Serialize, Deserialize)] #[serde(tag = "cmd", rename_all = "snake_case")] pub enum Request { - /// Send a message to another agent. - Send { - to: String, - body: String, - /// Optional id of the message being replied to. Stored in the - /// broker DB and returned on `Recv` so the dashboard can render - /// threads. Ignored if the id is unknown or out of retention. - #[serde(default, skip_serializing_if = "Option::is_none")] - in_reply_to: Option, - }, /// Pop pending messages from this agent's inbox. /// Delivery + ack cycle: see /// `docs/5-project/process/conventions.md::Broker delivery + ack cycle`. Recv { #[serde(default)] wait_seconds: Option, - #[serde(default)] - max: Option, }, /// Non-mutating: how many pending messages are addressed to me? /// Used by the harness to render a status line after each tool call. @@ -48,7 +36,7 @@ pub enum Request { /// implicit (this agent); `from` is caller-chosen. See /// `docs/5-project/process/conventions.md::Wake injection` for the trust model and /// typical callers. The wake is persisted in the sqlite broker - /// like any other message — the agent can ack it via `AckUntil`. + /// like any other message. Wake { from: String, body: String }, /// Last `limit` messages addressed to this agent, newest-first. /// Non-mutating — pulls from the broker without delivering. The @@ -73,15 +61,6 @@ pub enum Request { /// Harness↔broker pairing fired after `TurnOutcome::Ok`. See /// `docs/5-project/process/conventions.md::Broker delivery + ack cycle`. AckTurn, - /// Mark every inbox message with broker row id `<= up_to` as - /// handled (`acked_at` set), whether still pending or already - /// delivered. The agent-facing bulk-triage escape hatch for a - /// redelivered / accumulated backlog: instead of popping and - /// re-reading dozens of already-handled messages one turn at a - /// time, the agent acks everything up to the id it has seen. - /// Recipient-scoped — an agent can only ack its own rows. See - /// `docs/5-project/process/conventions.md::Broker delivery + ack cycle`. - AckUntil { up_to: i64 }, /// Requeue every popped-but-unacked message back into the inbox. /// Harness fires this once at boot to recover from /// crashed-mid-turn sessions. See @@ -103,43 +82,25 @@ pub enum Request { #[derive(Debug, Clone, Serialize, Deserialize)] #[serde(tag = "kind", rename_all = "snake_case")] pub enum Response { - /// `Send` succeeded. + /// The request succeeded. Ok, - /// The operation succeeded **and** raised advisories the caller - /// should see. - /// - /// Distinct from `Err` on purpose: a check that WARNS rather than - /// refuses has no other way to reach the caller. Folding it into - /// `Err` would refuse; folding it into `Ok` would drop it, leaving - /// the daemon's log as the only record — and the person who can act - /// on the advisory is the one holding this response, not the one - /// reading the journal. - /// - /// Additive: every existing site returning `Ok` keeps returning it, - /// so a caller that never matches this variant is unaffected. - OkWarn { warnings: Vec }, - /// Either `Send` failed or `Recv` errored. + /// A request failed (`Recv` errored, a store write failed, ...). Err { message: String }, - /// `Recv` result: zero or more messages, FIFO-ordered, never - /// longer than the `max` the caller passed. Empty vec = nothing - /// pending (the "(empty)" path for the formatter). Per-row `id` + - /// `redelivered` carry the broker's row id (tracked by the harness - /// for `AckTurn`, and surfaced to claude as a `[msg #]` marker - /// so `AckUntil` has something to reference) and the "previously - /// popped, not acked" flag — see `DeliveredMessage` for details. - /// `remaining` is the inbox depth *after* this batch was popped — - /// how many still-pending messages the caller could drain next. The - /// harness surfaces it to claude ("N more pending") so an in-turn - /// `recv` learns whether the inbox is drained, mirroring the count - /// the wake prompt already carries. + /// `Recv` result: zero or one message. Empty vec = nothing pending. + /// Per-row `id` + `redelivered` carry the broker's row id (tracked by + /// the harness for `AckTurn`) and the "previously popped, not acked" + /// flag — see `DeliveredMessage` for details. `remaining` is the inbox + /// depth after the pop. No reader uses it any more, but a harness built + /// before it stopped being used still requires the field, so hive-c0re + /// keeps sending it until every agent has rebuilt; `default` lets a new + /// harness read an answer from a c0re that has dropped it. Messages { messages: Vec, + #[serde(default)] remaining: u64, }, /// `Status` result: how many pending messages are in this agent's inbox. Status { unread: u64 }, - /// `AckUntil` result: how many rows were newly marked handled. - Acked { count: u64 }, /// `Recent` result: newest-first inbox rows. Recent { rows: Vec }, /// `GetLooseEnds` result: list of loose ends pending against diff --git a/hive-sh4re/src/inbox.rs b/hive-sh4re/src/inbox.rs index d7cec80b..15b25850 100644 --- a/hive-sh4re/src/inbox.rs +++ b/hive-sh4re/src/inbox.rs @@ -7,16 +7,6 @@ use chrono::{DateTime, Utc}; use hive_types::Ident; use serde::{Deserialize, Serialize}; -/// Server-side hard cap on `Recv.max` (see the `Recv` request). Bounds -/// the size of a single round-trip so a confused caller can't drain the -/// entire inbox in one go and blow past wire-buffer sizes; everything -/// above the cap silently clamps. 5 keeps individual turns small — a big -/// backlog is drained over several recv calls instead of one giant pop. -/// Lives here so both the enforcing side (hive-c0re's `socket_server`) and -/// the hinting side (hive-agent's wake prompt + tool docs) reference one -/// constant instead of a scattered magic value. -pub const RECV_BATCH_MAX: u32 = 5; - /// Banner prepended to a wake prompt when the message was redelivered after /// a harness restart (the turn that first drove it never acked). pub const REDELIVERY_HINT: &str = "[redelivered after harness restart — may already be handled]\n"; diff --git a/hive-sh4re/src/manager.rs b/hive-sh4re/src/manager.rs index 9798f36f..1d76f22d 100644 --- a/hive-sh4re/src/manager.rs +++ b/hive-sh4re/src/manager.rs @@ -3,19 +3,9 @@ use hive_types::Ident; /// Logical name of the human operator: the sender of operator-authored -/// messages. Not a recipient — `send` to it fails with -/// [`OPERATOR_SEND_REFUSAL`]. +/// messages. Not a recipient. pub const OPERATOR_RECIPIENT: &str = "operator"; -/// The error a `send` to [`OPERATOR_RECIPIENT`] returns. One text for every -/// agent, so it covers both an agent with the matrix MCP tools and one -/// without them. -pub const OPERATOR_SEND_REFUSAL: &str = "send failed: `operator` is not a send recipient. \ - If you have the matrix MCP tools, reach the operator over matrix \ - (`mcp__matrix__send_dm`, or `mcp__matrix__send_message` to a room the operator is in). \ - Without them there is no direct channel to the operator: they read your `set_status` \ - line and turn output on the dashboard, and your logs."; - /// Sender hive-c0re uses for messages it pushes into an agent's inbox. pub const SYSTEM_SENDER: &str = "system";