From 493face93c84fa1eea10e3a2aebeae48b055d1ce Mon Sep 17 00:00:00 2001 From: damocles Date: Fri, 10 Jul 2026 01:34:22 +0200 Subject: [PATCH] feat(#2300): recv reports remaining inbox depth so agents know how many messages are left --- hive-ag3nt/src/bin/hive-agent.rs | 2 +- hive-ag3nt/src/mcp/mod.rs | 3 + hive-ag3nt/src/mcp/render.rs | 121 +++++++++++++++++++++++------ hive-ag3nt/src/serve_common.rs | 37 +++++---- hive-c0re/src/socket_server/mod.rs | 32 +++++--- hive-sh4re/src/lib.rs | 10 ++- 6 files changed, 153 insertions(+), 52 deletions(-) diff --git a/hive-ag3nt/src/bin/hive-agent.rs b/hive-ag3nt/src/bin/hive-agent.rs index d1aa5a1e..34771ceb 100644 --- a/hive-ag3nt/src/bin/hive-agent.rs +++ b/hive-ag3nt/src/bin/hive-agent.rs @@ -315,7 +315,7 @@ impl Surface for AgentSurface { ) .await; match recv { - Ok(AgentResponse::Messages { messages }) if !messages.is_empty() => { + Ok(AgentResponse::Messages { messages, .. }) if !messages.is_empty() => { let first = messages.into_iter().next().expect("checked non-empty"); RecvOutcome::Message(first) } diff --git a/hive-ag3nt/src/mcp/mod.rs b/hive-ag3nt/src/mcp/mod.rs index 7bd00c49..693a6322 100644 --- a/hive-ag3nt/src/mcp/mod.rs +++ b/hive-ag3nt/src/mcp/mod.rs @@ -248,6 +248,9 @@ impl AgentServer { any time you expect a burst — one tool call beats N consecutive single recvs. \ `wait_seconds` still applies to the FIRST message; once one arrives the call drains \ up to `max` in total. Empty result reported the same way regardless of `max`. \n\n\ + After popping, the result appends a `(N more message(s) pending …)` line whenever the \ + inbox still has queued messages — so you know whether to drain again (or `ack_until`) \ + without a separate status check. No line means the inbox is empty. \n\n\ Typical pattern: when you have nothing else useful to do, call \ `recv(wait_seconds: 180)` to park until something arrives." )] diff --git a/hive-ag3nt/src/mcp/render.rs b/hive-ag3nt/src/mcp/render.rs index 0fe22983..fdc172a8 100644 --- a/hive-ag3nt/src/mcp/render.rs +++ b/hive-ag3nt/src/mcp/render.rs @@ -46,14 +46,19 @@ pub fn format_ack( #[must_use] pub fn format_recv(resp: Result, waited: bool) -> String { match resp { - Ok(hive_sh4re::Response::Messages { messages }) => render_recv_messages(&messages, waited), + Ok(hive_sh4re::Response::Messages { + messages, + remaining, + }) => render_recv_messages(&messages, remaining, waited), // A graceful stop is pending — the inbox is fenced. Render a single // explicit directive (not an empty inbox, which claude's "park on recv" // habit would long-poll again, stalling the stop-checkpoint turn until // the drain wait times out into a hard stop) so every recv during the - // stop unmissably tells claude to flush + end. + // stop unmissably tells claude to flush + end. `remaining` is forced + // to 0 — the inbox is fenced, so a "N more pending" hint would be + // misleading. Ok(hive_sh4re::Response::GracefulStop) => { - render_recv_messages(&[graceful_stop_message()], waited) + render_recv_messages(&[graceful_stop_message()], 0, waited) } other => reply_err(other, "recv"), } @@ -76,8 +81,15 @@ fn graceful_stop_message() -> hive_sh4re::DeliveredMessage { } /// Render the popped-message payload of a successful `recv` (see `format_recv` -/// for the empty/single/batch shapes). -fn render_recv_messages(messages: &[hive_sh4re::DeliveredMessage], waited: bool) -> String { +/// for the empty/single/batch shapes). `remaining` is the post-pop inbox +/// depth; when non-zero a shared "(N more pending …)" hint (identical to the +/// wake prompt's) is appended so an in-turn drain knows more is queued. The +/// empty path never carries the hint (nothing was popped). +fn render_recv_messages( + messages: &[hive_sh4re::DeliveredMessage], + remaining: u64, + waited: bool, +) -> String { use std::fmt::Write as _; if messages.is_empty() { return if waited { @@ -86,26 +98,29 @@ fn render_recv_messages(messages: &[hive_sh4re::DeliveredMessage], waited: bool) "(empty)".to_owned() }; } - if messages.len() == 1 { + let mut out = if messages.len() == 1 { let m = &messages[0]; let banner = if m.redelivered { REDELIVERY_HINT } else { "" }; - return format!("{banner}{}from: {}\n\n{}", msg_id_tag(m.id), m.from, m.body); - } - let n = messages.len(); - let mut out = format!("popped {n} message(s):\n\n"); - for (i, m) in messages.iter().enumerate() { - if i > 0 { - out.push_str("\n---\n\n"); + format!("{banner}{}from: {}\n\n{}", msg_id_tag(m.id), m.from, m.body) + } else { + let n = messages.len(); + let mut out = format!("popped {n} message(s):\n\n"); + for (i, m) in messages.iter().enumerate() { + if i > 0 { + out.push_str("\n---\n\n"); + } + let banner = if m.redelivered { REDELIVERY_HINT } else { "" }; + let _ = write!( + out, + "{banner}{}from: {}\n\n{}", + msg_id_tag(m.id), + m.from, + m.body + ); } - let banner = if m.redelivered { REDELIVERY_HINT } else { "" }; - let _ = write!( - out, - "{banner}{}from: {}\n\n{}", - msg_id_tag(m.id), - m.from, - m.body - ); - } + out + }; + out.push_str(&crate::serve_common::pending_hint(remaining)); out } @@ -427,10 +442,23 @@ pub fn annotate_retries(mut s: String, retries: u32) -> String { mod tests { use super::{IDLE_WAIT_HINT, format_recv}; + fn msg(id: i64, from: &str, body: &str) -> hive_sh4re::DeliveredMessage { + hive_sh4re::DeliveredMessage { + from: from.to_owned(), + body: body.to_owned(), + id, + redelivered: false, + in_reply_to: None, + } + } + #[test] fn empty_recv_after_wait_appends_idle_hint() { let out = format_recv( - Ok(hive_sh4re::Response::Messages { messages: vec![] }), + Ok(hive_sh4re::Response::Messages { + messages: vec![], + remaining: 0, + }), true, ); assert!(out.starts_with("(empty)")); @@ -440,9 +468,54 @@ mod tests { #[test] fn empty_recv_without_wait_has_no_hint() { let out = format_recv( - Ok(hive_sh4re::Response::Messages { messages: vec![] }), + Ok(hive_sh4re::Response::Messages { + messages: vec![], + remaining: 0, + }), false, ); assert_eq!(out, "(empty)"); } + + #[test] + fn single_recv_with_remaining_appends_pending_hint() { + let out = format_recv( + Ok(hive_sh4re::Response::Messages { + messages: vec![msg(7, "alice", "hi")], + remaining: 3, + }), + false, + ); + assert!(out.starts_with("[msg #7] from: alice")); + assert!(out.contains("3 more message(s) pending")); + assert!(out.contains("max: 3")); + } + + #[test] + fn single_recv_no_remaining_has_no_pending_hint() { + let out = format_recv( + Ok(hive_sh4re::Response::Messages { + messages: vec![msg(7, "alice", "hi")], + remaining: 0, + }), + false, + ); + assert!(!out.contains("more message(s) pending")); + } + + #[test] + fn batch_recv_with_remaining_appends_pending_hint_once() { + let out = format_recv( + Ok(hive_sh4re::Response::Messages { + messages: vec![msg(7, "alice", "hi"), msg(8, "bob", "yo")], + remaining: 9, + }), + false, + ); + assert!(out.starts_with("popped 2 message(s):")); + assert_eq!(out.matches("more message(s) pending").count(), 1); + // `max` suggestion is clamped to the server-side recv cap. + let batch = 9u64.min(u64::from(hive_sh4re::RECV_BATCH_MAX)); + assert!(out.contains(&format!("max: {batch}"))); + } } diff --git a/hive-ag3nt/src/serve_common.rs b/hive-ag3nt/src/serve_common.rs index c43af732..5178147b 100644 --- a/hive-ag3nt/src/serve_common.rs +++ b/hive-ag3nt/src/serve_common.rs @@ -29,23 +29,32 @@ pub fn format_wake_prompt( } else { String::new() }; - let pending = if unread == 0 { - String::new() - } else { - // Suggested batch size is clamped to the server-side recv cap - // so the hint never asks for more than one round-trip can - // deliver. - let batch = unread.min(u64::from(hive_sh4re::RECV_BATCH_MAX)); - format!( - "\n\n({unread} more message(s) pending in your inbox — call `mcp__hyperhive__recv` \ - with `max: {batch}` to drain the next batch before acting. If the \ - backlog is stale/already handled, `ack_until(up_to: )` \ - clears everything up to that id in one call instead.)" - ) - }; + let pending = pending_hint(unread); format!("{banner}{tag}Incoming message from `{from}`:\n---\n{body}\n---{pending}") } +/// Shared "(N more message(s) pending …)" advisory appended after both the +/// wake prompt body and the `recv` tool result whenever the inbox still has +/// queued messages once the current message/batch is popped. Returns an empty +/// string when `remaining == 0`. The leading `\n\n` separates it from the +/// preceding body/message block, and the suggested `max` is clamped to the +/// server-side recv cap so the hint never asks for more than one round-trip +/// can deliver. One builder so the wake prompt and the in-turn recv result +/// stay identical. +#[must_use] +pub fn pending_hint(remaining: u64) -> String { + if remaining == 0 { + return String::new(); + } + let batch = remaining.min(u64::from(hive_sh4re::RECV_BATCH_MAX)); + format!( + "\n\n({remaining} more message(s) pending in your inbox — call `mcp__hyperhive__recv` \ + with `max: {batch}` to drain the next batch before acting. If the \ + backlog is stale/already handled, `ack_until(up_to: )` \ + clears everything up to that id in one call instead.)" + ) +} + /// Field-named args for [`build_row`]. Mirrors the turn-stats row /// columns; `outcome` and `bus` borrow for the duration of the call. pub struct TurnRowArgs<'a> { diff --git a/hive-c0re/src/socket_server/mod.rs b/hive-c0re/src/socket_server/mod.rs index faf94249..5d06604b 100644 --- a/hive-c0re/src/socket_server/mod.rs +++ b/hive-c0re/src/socket_server/mod.rs @@ -309,18 +309,26 @@ async fn handle_recv( .recv_blocking_batch(agent, recv_timeout(wait_seconds), cap) .await { - Ok(deliveries) => hive_sh4re::Response::Messages { - messages: deliveries - .into_iter() - .map(|d| hive_sh4re::DeliveredMessage { - from: d.message.from, - body: d.message.body, - id: d.id, - redelivered: d.redelivered, - in_reply_to: d.message.in_reply_to, - }) - .collect(), - }, + 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_sh4re::Response::Messages { + messages: deliveries + .into_iter() + .map(|d| hive_sh4re::DeliveredMessage { + from: d.message.from, + body: d.message.body, + id: d.id, + redelivered: d.redelivered, + in_reply_to: d.message.in_reply_to, + }) + .collect(), + remaining, + } + } Err(e) => hive_sh4re::Response::Err { message: format!("{e:#}"), }, diff --git a/hive-sh4re/src/lib.rs b/hive-sh4re/src/lib.rs index 0f40a40a..dd95e010 100644 --- a/hive-sh4re/src/lib.rs +++ b/hive-sh4re/src/lib.rs @@ -861,7 +861,15 @@ pub enum Response { /// 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. - Messages { messages: Vec }, + /// `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. + Messages { + messages: Vec, + 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.