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

drop the hyperhive send/recv/ack_until tools: agents message over matrix

Browse source
This commit is contained in:
damocles 2026-11-07 11:55:45 +01:00 • committed by mara
commit bffb2c07a8
3 changed files with 52 additions and 397 deletions
Download patch file Download diff file Expand all files Collapse all files

38
hive-agent-mcp/src/mcp/args.rs
View file

@ -4,44 +4,6 @@
use rmcp::schemars;
#[derive(Debug, serde::Deserialize, schemars::JsonSchema)]
pub struct SendArgs {
/// Logical agent name to deliver the message to (e.g. `"bob"`,
/// `"alice"`). `"operator"` is refused: reach the operator over matrix
/// if you have it.
pub to: String,
/// Message body. Plain text; the broker doesn't parse it.
pub body: String,
/// Optional broker row-id of the message this is a reply to. Lets
/// the dashboard render conversation threads. Pass the `id` from the
/// `DeliveredMessage` you're responding to; omit for new threads.
/// Silently ignored if the id is unknown or out of retention.
#[serde(default)]
pub in_reply_to: Option<i64>,
}
#[derive(Debug, serde::Deserialize, schemars::JsonSchema)]
pub struct RecvArgs {
/// Maximum number of messages to pop in this round-trip. Default
/// (None) is 1 (single-message behaviour — exactly what you want
/// when you're called to drive a turn off the first wake). Pass
/// a higher value (capped at 5 server-side) when you've been
/// told the inbox has more queued (the wake prompt mentions
/// pending count) and want to drain everything in one tool call.
#[serde(default)]
pub max: Option<u32>,
}
/// MCP tool args for `ack_until`.
#[derive(Debug, serde::Deserialize, schemars::JsonSchema)]
pub struct AckUntilArgs {
/// Highest broker message id to mark handled: every inbox message
/// with `id <= up_to` (ids show as `[msg #<id>]` in recv output)
/// is acked in one sweep. Pass the highest id you've actually
/// seen/triaged — anything above it stays queued for later turns.
pub up_to: i64,
}
/// MCP tool args for `mark_todos_done`.
#[derive(Debug, serde::Deserialize, schemars::JsonSchema)]
pub struct MarkTodosDoneArgs {

91
hive-agent-mcp/src/mcp/mod.rs
View file

@ -24,10 +24,9 @@ mod args;
mod render;
pub use args::{
AckUntilArgs, CancelLooseEndArgs, CompactArgs, GetAgentMetaArgs, MarkTodosDoneArgs, RecvArgs,
RemindArgs, SendArgs, SetStatusArgs,
CancelLooseEndArgs, CompactArgs, GetAgentMetaArgs, MarkTodosDoneArgs, RemindArgs, SetStatusArgs,
};
pub use render::{annotate_retries, format_ack, format_agent_meta, format_recv};
pub use render::{annotate_retries, format_ack, format_agent_meta};
use render::{
dial_agent_socket, format_matrix_summary, local_reminders, local_todos, mark_local_todo_done,
@ -140,92 +139,6 @@ impl AgentServer {
// non-interactive `--print` mode with "permissions not granted yet".
#[tool_router]
impl AgentServer {
#[tool(
description = "Send a message through the hive-local broker to another hyperhive \
agent on this hive. To talk to a peer agent, prefer matrix (a DM to the peer's \
`user_id` from `get_agent_meta`) when you have the matrix MCP tools; use this \
when you have none or the peer has no matrix account. Not for the operator: reach \
them over matrix if you have the matrix MCP tools."
)]
async fn send(&self, Parameters(args): Parameters<SendArgs>) -> String {
let log = format!("{args:?}");
let to = args.to.clone();
run_tool_envelope("send", log, async move {
let (resp, retries) = self
.dispatch(hive_core_agent_sock::Request::Send {
to: args.to,
body: args.body,
in_reply_to: args.in_reply_to,
})
.await;
annotate_retries(format_ack(resp, "send", format!("sent to {to}")), retries)
})
.await
}
#[tool(
description = "Pop messages from this agent's inbox. Returns one or more messages, or \
an empty marker if nothing is waiting. Always an immediate 'anything pending?' peek \
— never blocks. \n\n\
**Single-message default**: with no args (or `max: 1`) you get the next message — \
same behaviour the harness uses to drive a turn. \n\n\
**When idle, prefer ending the turn over polling here.** Ending the turn is the \
ONLY path that observes an in-container todo wake (bash-task completions, matrix \
unread, forge activity) — that signal reaches the harness loop between turns, never \
a live `recv` call, so looping on `recv` while idle means sitting blind until \
something else wakes you. Ending the turn also checkpoints your session, and costs \
no latency vs. polling for a real inbox message either way — the broker wakes the \
next turn just as fast. \n\n\
**Batch drain**: pass `max: N` (capped at 5) to drain up to N messages in one \
round-trip. Use this when the wake prompt told you the inbox has more queued, or \
any time you expect a burst — one tool call beats N consecutive single recvs. 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."
)]
async fn recv(&self, Parameters(args): Parameters<RecvArgs>) -> String {
let log = format!("{args:?}");
run_tool_envelope("recv", log, async move {
let (resp, retries) = self
.dispatch(hive_core_agent_sock::Request::Recv {
wait_seconds: None,
max: args.max,
})
.await;
annotate_retries(format_recv(resp), retries)
})
.await
}
#[tool(
description = "Bulk-mark inbox messages handled: every message with broker id \
<= `up_to` (ids show as `[msg #<id>]` in recv output and wake prompts) is \
acked in one call — pending and already-delivered rows alike. Use this to \
clear a backlog you've already triaged (e.g. a redelivered flood after a \
container restart, or a pile of stale notifications) instead of draining it \
one recv at a time: note the highest `[msg #N]` you've seen, then \
`ack_until(up_to: N)`. Acked messages never redeliver. Only affects YOUR \
inbox rows; messages newer than `up_to` stay queued. Returns how many rows \
were newly acked."
)]
async fn ack_until(&self, Parameters(args): Parameters<AckUntilArgs>) -> String {
let log = format!("{args:?}");
run_tool_envelope("ack_until", log, async move {
let (resp, retries) = self
.dispatch(hive_core_agent_sock::Request::AckUntil { up_to: args.up_to })
.await;
let rendered = match resp {
Ok(hive_core_agent_sock::Response::Acked { count }) => {
format!("acked {count} message(s) up to id {}", args.up_to)
}
other => reply_err(other, "ack_until"),
};
annotate_retries(rendered, retries)
})
.await
}
#[tool(
description = "List loose ends pending against this agent: pending reminders \
you scheduled. Also lists active \

166
hive-agent-mcp/src/mcp/render.rs
View file

@ -49,108 +49,6 @@ pub fn format_ack(
}
}
/// Format helper for `recv`: renders zero, one, or many popped
/// messages. Empty list collapses to "(empty)" so claude doesn't go
/// hunting for content. A single message renders as the historical
/// `from: X\n\nbody` block (banner first if `redelivered`). A
/// multi-message batch renders with a `popped N message(s):` header
/// and `---` separators between bodies so the model can tell where
/// one ends and the next begins; per-message redelivery banners
/// included.
#[must_use]
pub fn format_recv(resp: Result<hive_core_agent_sock::Response, anyhow::Error>) -> String {
match resp {
Ok(hive_core_agent_sock::Response::Messages {
messages,
remaining,
}) => render_recv_messages(&messages, remaining),
// A graceful stop is pending — the inbox is fenced. Render a single
// explicit directive (not an empty inbox, which claude might just
// recv again) so every recv during the 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_core_agent_sock::Response::GracefulStop) => {
render_recv_messages(&[graceful_stop_message()], 0)
}
other => reply_err(other, "recv"),
}
}
/// The synthetic single-message directive rendered for a fenced (graceful-stop)
/// inbox — see the `GracefulStop` arm of [`format_recv`].
fn graceful_stop_message() -> hive_sh4re::inbox::DeliveredMessage {
hive_sh4re::inbox::DeliveredMessage {
from: "graceful-stop".into(),
body: "⛔ GRACEFUL STOP IN PROGRESS — the container shuts down as soon as this turn \
ends. Flush anything worth keeping to your durable /state files, then END \
YOUR TURN now. Do NOT call recv again: the inbox is fenced and recv will \
only keep returning this same notice."
.into(),
id: 0,
redelivered: false,
in_reply_to: None,
}
}
/// Render the popped-message payload of a successful `recv` (see `format_recv`
/// 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::inbox::DeliveredMessage],
remaining: u64,
) -> String {
use std::fmt::Write as _;
if messages.is_empty() {
return "(empty)".to_owned();
}
let mut out = if messages.len() == 1 {
let m = &messages[0];
let banner = if m.redelivered {
hive_sh4re::inbox::REDELIVERY_HINT
} else {
""
};
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 {
hive_sh4re::inbox::REDELIVERY_HINT
} else {
""
};
let _ = write!(
out,
"{banner}{}from: {}\n\n{}",
msg_id_tag(m.id),
m.from,
m.body
);
}
out
};
out.push_str(&hive_sh4re::inbox::pending_hint(remaining));
out
}
/// `[msg #<id>] ` marker prefixed to each recv row so the agent knows
/// what to pass to `ack_until` when bulk-triaging a backlog. Transient
/// pings carry the sentinel id 0 (in-memory only, nothing in the
/// broker to ack) and render without the marker.
fn msg_id_tag(id: i64) -> String {
if id > 0 {
format!("[msg #{id}] ")
} else {
String::new()
}
}
/// Hard cap on how many individual `Todo` lines `render_loose_ends` will
/// emit. Reminders stay naturally bounded (they're triaged interactively
/// and don't self-multiply), but todos are pushed by
@ -559,7 +457,7 @@ pub fn annotate_retries(mut s: String, retries: u32) -> String {
#[cfg(test)]
mod tests {
use super::{dial_socket, format_ack, format_recv};
use super::{dial_socket, format_ack};
#[tokio::test]
async fn a_socket_that_accepts_and_never_answers_is_given_up_on() {
@ -590,70 +488,18 @@ mod tests {
assert!(resp.is_none());
}
fn msg(id: i64, from: &str, body: &str) -> hive_sh4re::inbox::DeliveredMessage {
hive_sh4re::inbox::DeliveredMessage {
from: from.to_owned(),
body: body.to_owned(),
id,
redelivered: false,
in_reply_to: None,
}
}
#[test]
fn empty_recv_renders_bare_empty_marker() {
let out = format_recv(Ok(hive_core_agent_sock::Response::Messages {
messages: vec![],
remaining: 0,
}));
assert_eq!(out, "(empty)");
}
#[test]
fn single_recv_with_remaining_appends_pending_hint() {
let out = format_recv(Ok(hive_core_agent_sock::Response::Messages {
messages: vec![msg(7, "alice", "hi")],
remaining: 3,
}));
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_core_agent_sock::Response::Messages {
messages: vec![msg(7, "alice", "hi")],
remaining: 0,
}));
assert!(!out.contains("more message(s) pending"));
}
#[test]
fn batch_recv_with_remaining_appends_pending_hint_once() {
let out = format_recv(Ok(hive_core_agent_sock::Response::Messages {
messages: vec![msg(7, "alice", "hi"), msg(8, "bob", "yo")],
remaining: 9,
}));
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::inbox::RECV_BATCH_MAX));
assert!(out.contains(&format!("max: {batch}")));
}
#[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()],
}),
"send",
"sent to iris".to_owned(),
"remind",
"reminder set".to_owned(),
);
// The operation HAPPENED — dropping the success line would read as a
// failure and invite a retry that sends a second message.
assert!(out.starts_with("sent to iris"));
// 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"));
}
@ -664,7 +510,7 @@ mod tests {
// warning marker would pass the test above.
let out = format_ack(
Ok(hive_core_agent_sock::Response::Ok),
"send",
"remind",
"sent".to_owned(),
);
assert_eq!(out, "sent");