feat(#2300): recv reports remaining inbox depth so agents know how many messages are left
This commit is contained in:
parent
371f888640
commit
493face93c
6 changed files with 153 additions and 52 deletions
|
|
@ -315,7 +315,7 @@ impl Surface for AgentSurface {
|
||||||
)
|
)
|
||||||
.await;
|
.await;
|
||||||
match recv {
|
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");
|
let first = messages.into_iter().next().expect("checked non-empty");
|
||||||
RecvOutcome::Message(first)
|
RecvOutcome::Message(first)
|
||||||
}
|
}
|
||||||
|
|
|
||||||
|
|
@ -248,6 +248,9 @@ impl AgentServer {
|
||||||
any time you expect a burst — one tool call beats N consecutive single recvs. \
|
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 \
|
`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\
|
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 \
|
Typical pattern: when you have nothing else useful to do, call \
|
||||||
`recv(wait_seconds: 180)` to park until something arrives."
|
`recv(wait_seconds: 180)` to park until something arrives."
|
||||||
)]
|
)]
|
||||||
|
|
|
||||||
|
|
@ -46,14 +46,19 @@ pub fn format_ack(
|
||||||
#[must_use]
|
#[must_use]
|
||||||
pub fn format_recv(resp: Result<hive_sh4re::Response, anyhow::Error>, waited: bool) -> String {
|
pub fn format_recv(resp: Result<hive_sh4re::Response, anyhow::Error>, waited: bool) -> String {
|
||||||
match resp {
|
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
|
// A graceful stop is pending — the inbox is fenced. Render a single
|
||||||
// explicit directive (not an empty inbox, which claude's "park on recv"
|
// explicit directive (not an empty inbox, which claude's "park on recv"
|
||||||
// habit would long-poll again, stalling the stop-checkpoint turn until
|
// 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
|
// 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) => {
|
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"),
|
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`
|
/// Render the popped-message payload of a successful `recv` (see `format_recv`
|
||||||
/// for the empty/single/batch shapes).
|
/// for the empty/single/batch shapes). `remaining` is the post-pop inbox
|
||||||
fn render_recv_messages(messages: &[hive_sh4re::DeliveredMessage], waited: bool) -> String {
|
/// 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 _;
|
use std::fmt::Write as _;
|
||||||
if messages.is_empty() {
|
if messages.is_empty() {
|
||||||
return if waited {
|
return if waited {
|
||||||
|
|
@ -86,26 +98,29 @@ fn render_recv_messages(messages: &[hive_sh4re::DeliveredMessage], waited: bool)
|
||||||
"(empty)".to_owned()
|
"(empty)".to_owned()
|
||||||
};
|
};
|
||||||
}
|
}
|
||||||
if messages.len() == 1 {
|
let mut out = if messages.len() == 1 {
|
||||||
let m = &messages[0];
|
let m = &messages[0];
|
||||||
let banner = if m.redelivered { REDELIVERY_HINT } else { "" };
|
let banner = if m.redelivered { REDELIVERY_HINT } else { "" };
|
||||||
return format!("{banner}{}from: {}\n\n{}", msg_id_tag(m.id), m.from, m.body);
|
format!("{banner}{}from: {}\n\n{}", msg_id_tag(m.id), m.from, m.body)
|
||||||
}
|
} else {
|
||||||
let n = messages.len();
|
let n = messages.len();
|
||||||
let mut out = format!("popped {n} message(s):\n\n");
|
let mut out = format!("popped {n} message(s):\n\n");
|
||||||
for (i, m) in messages.iter().enumerate() {
|
for (i, m) in messages.iter().enumerate() {
|
||||||
if i > 0 {
|
if i > 0 {
|
||||||
out.push_str("\n---\n\n");
|
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 { "" };
|
out
|
||||||
let _ = write!(
|
};
|
||||||
out,
|
out.push_str(&crate::serve_common::pending_hint(remaining));
|
||||||
"{banner}{}from: {}\n\n{}",
|
|
||||||
msg_id_tag(m.id),
|
|
||||||
m.from,
|
|
||||||
m.body
|
|
||||||
);
|
|
||||||
}
|
|
||||||
out
|
out
|
||||||
}
|
}
|
||||||
|
|
||||||
|
|
@ -427,10 +442,23 @@ pub fn annotate_retries(mut s: String, retries: u32) -> String {
|
||||||
mod tests {
|
mod tests {
|
||||||
use super::{IDLE_WAIT_HINT, format_recv};
|
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]
|
#[test]
|
||||||
fn empty_recv_after_wait_appends_idle_hint() {
|
fn empty_recv_after_wait_appends_idle_hint() {
|
||||||
let out = format_recv(
|
let out = format_recv(
|
||||||
Ok(hive_sh4re::Response::Messages { messages: vec![] }),
|
Ok(hive_sh4re::Response::Messages {
|
||||||
|
messages: vec![],
|
||||||
|
remaining: 0,
|
||||||
|
}),
|
||||||
true,
|
true,
|
||||||
);
|
);
|
||||||
assert!(out.starts_with("(empty)"));
|
assert!(out.starts_with("(empty)"));
|
||||||
|
|
@ -440,9 +468,54 @@ mod tests {
|
||||||
#[test]
|
#[test]
|
||||||
fn empty_recv_without_wait_has_no_hint() {
|
fn empty_recv_without_wait_has_no_hint() {
|
||||||
let out = format_recv(
|
let out = format_recv(
|
||||||
Ok(hive_sh4re::Response::Messages { messages: vec![] }),
|
Ok(hive_sh4re::Response::Messages {
|
||||||
|
messages: vec![],
|
||||||
|
remaining: 0,
|
||||||
|
}),
|
||||||
false,
|
false,
|
||||||
);
|
);
|
||||||
assert_eq!(out, "(empty)");
|
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}")));
|
||||||
|
}
|
||||||
}
|
}
|
||||||
|
|
|
||||||
|
|
@ -29,23 +29,32 @@ pub fn format_wake_prompt(
|
||||||
} else {
|
} else {
|
||||||
String::new()
|
String::new()
|
||||||
};
|
};
|
||||||
let pending = if unread == 0 {
|
let pending = pending_hint(unread);
|
||||||
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: <highest [msg #N] seen>)` \
|
|
||||||
clears everything up to that id in one call instead.)"
|
|
||||||
)
|
|
||||||
};
|
|
||||||
format!("{banner}{tag}Incoming message from `{from}`:\n---\n{body}\n---{pending}")
|
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: <highest [msg #N] seen>)` \
|
||||||
|
clears everything up to that id in one call instead.)"
|
||||||
|
)
|
||||||
|
}
|
||||||
|
|
||||||
/// Field-named args for [`build_row`]. Mirrors the turn-stats row
|
/// Field-named args for [`build_row`]. Mirrors the turn-stats row
|
||||||
/// columns; `outcome` and `bus` borrow for the duration of the call.
|
/// columns; `outcome` and `bus` borrow for the duration of the call.
|
||||||
pub struct TurnRowArgs<'a> {
|
pub struct TurnRowArgs<'a> {
|
||||||
|
|
|
||||||
|
|
@ -309,18 +309,26 @@ async fn handle_recv(
|
||||||
.recv_blocking_batch(agent, recv_timeout(wait_seconds), cap)
|
.recv_blocking_batch(agent, recv_timeout(wait_seconds), cap)
|
||||||
.await
|
.await
|
||||||
{
|
{
|
||||||
Ok(deliveries) => hive_sh4re::Response::Messages {
|
Ok(deliveries) => {
|
||||||
messages: deliveries
|
// `recv_batch` stamps `delivered_at` on every popped row, so a
|
||||||
.into_iter()
|
// `count_pending` here excludes the just-popped batch and reports
|
||||||
.map(|d| hive_sh4re::DeliveredMessage {
|
// exactly how many still-pending messages remain to drain. A count
|
||||||
from: d.message.from,
|
// error is non-fatal — fall back to 0 rather than fail the recv.
|
||||||
body: d.message.body,
|
let remaining = coord.broker.count_pending(agent).unwrap_or(0);
|
||||||
id: d.id,
|
hive_sh4re::Response::Messages {
|
||||||
redelivered: d.redelivered,
|
messages: deliveries
|
||||||
in_reply_to: d.message.in_reply_to,
|
.into_iter()
|
||||||
})
|
.map(|d| hive_sh4re::DeliveredMessage {
|
||||||
.collect(),
|
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 {
|
Err(e) => hive_sh4re::Response::Err {
|
||||||
message: format!("{e:#}"),
|
message: format!("{e:#}"),
|
||||||
},
|
},
|
||||||
|
|
|
||||||
|
|
@ -861,7 +861,15 @@ pub enum Response {
|
||||||
/// for `AckTurn`, and surfaced to claude as a `[msg #<id>]` marker
|
/// for `AckTurn`, and surfaced to claude as a `[msg #<id>]` marker
|
||||||
/// so `AckUntil` has something to reference) and the "previously
|
/// so `AckUntil` has something to reference) and the "previously
|
||||||
/// popped, not acked" flag — see `DeliveredMessage` for details.
|
/// popped, not acked" flag — see `DeliveredMessage` for details.
|
||||||
Messages { messages: Vec<DeliveredMessage> },
|
/// `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<DeliveredMessage>,
|
||||||
|
remaining: u64,
|
||||||
|
},
|
||||||
/// `Status` result: how many pending messages are in this agent's inbox.
|
/// `Status` result: how many pending messages are in this agent's inbox.
|
||||||
Status { unread: u64 },
|
Status { unread: u64 },
|
||||||
/// `AckUntil` result: how many rows were newly marked handled.
|
/// `AckUntil` result: how many rows were newly marked handled.
|
||||||
|
|
|
||||||
Loading…
Reference in a new issue