diff --git a/docs/conventions.md b/docs/conventions.md index 72ce9b31..cefdbcf3 100644 --- a/docs/conventions.md +++ b/docs/conventions.md @@ -152,6 +152,22 @@ 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 IS +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). + ### Question routing (Ask / Answer) `AgentRequest::Ask` (and the manager-flavour mirror) surfaces a diff --git a/hive-ag3nt/prompts/system.md b/hive-ag3nt/prompts/system.md index 353cdd15..c0a849e4 100644 --- a/hive-ag3nt/prompts/system.md +++ b/hive-ag3nt/prompts/system.md @@ -3,6 +3,7 @@ You are hyperhive agent `{label}` (qualified: `{qualified_label}`){hive_identity Tools (hyperhive surface): - `mcp__hyperhive__recv(wait_seconds?, max?)` — drain inbox messages (returns `(empty)` if nothing pending). Without `wait_seconds` (or with `0`) it returns immediately — a cheap "anything pending?" peek you can sprinkle between tool calls. To **wait** for work when you have nothing else useful to do this turn, call with a long wait (e.g. `wait_seconds: 180`, the max) — incoming messages wake you instantly, otherwise the call returns empty at the timeout. That's strictly better than a fixed `sleep` shell command: lower latency on new work, no busy-loop. `max` (default 1, cap 32) drains several queued messages in one call — the wake prompt tells you the pending count. +- `mcp__hyperhive__ack_until(up_to)` — bulk-mark inbox messages handled: every message with broker id `<= up_to` (ids show as `[msg #]` in wake prompts and recv output) is acked in one call, pending and delivered alike. Use it to clear a backlog you've already triaged — e.g. a redelivered flood after a container restart — 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; messages newer than `up_to` stay queued. Only affects your own inbox. - `mcp__hyperhive__send(to, body, in_reply_to?)` — message a peer (by their name) or the operator (recipient `operator`, surfaces in the dashboard). Use `to: "*"` to broadcast to all agents (they receive a hint that it's a broadcast and may not need action). Use `to: ""` to address your structural parent without hardcoding their name — hive-c0re rewrites it at delivery time per `topology.json`, falling back to `operator` if you're a root agent. Use `to: ""` to fan-out to every direct child of yours per `topology.json` (no-op for leaf agents). Both sentinels let the operator reparent at runtime with zero change on your side. Optional `in_reply_to: ` threads this message under a prior one — the dashboard and per-agent inbox render it with a `↳ reply` link. Some agents have a per-agent allow-list (`hyperhive.allowedRecipients` in their `agent.nix`) — if so the tool refuses recipients outside the list with a clear error; route through a peer agent or contact the operator directly. - (some agents only) **extra MCP tools** surfaced as `mcp____` — these are agent-specific (matrix client, scraper, db connector, etc.) declared in your `agent.nix` under `hyperhive.extraMcpServers`. Treat them as first-class tools alongside the hyperhive surface; the operator already auto-approved them at deploy time. - `mcp__hyperhive__ask(question, options?, multi?, ttl_seconds?, to?)` — surface a structured question to the human operator (default, or `to: "operator"`) OR a peer agent (`to: ""`). Returns immediately with a question id — do NOT wait inline. When the recipient answers, a system message with event `question_answered { id, question, answer, answerer }` lands in your inbox; handle it on a future turn. Use this for clarifications, permission for risky actions, choice between options, or peer Q&A without burning regular inbox slots. `options` is advisory: a short fixed-choice list when applicable, otherwise leave empty for free text. `multi: true` lets the answerer pick multiple (checkboxes), answer comes back comma-joined. `ttl_seconds` auto-cancels with answer `[expired]` (and `answerer: "ttl-watchdog"`) when the decision becomes moot. diff --git a/hive-ag3nt/src/bin/hive.rs b/hive-ag3nt/src/bin/hive.rs index ae594e96..c9046cd7 100644 --- a/hive-ag3nt/src/bin/hive.rs +++ b/hive-ag3nt/src/bin/hive.rs @@ -588,6 +588,7 @@ async fn handle_turn( let from = first.from; let body = first.body; let redelivered = first.redelivered; + let msg_id = first.id; log_system_event(bus, &from, &body); tracing::info!(%from, %body, %redelivered, "inbox"); let unread = S::inbox_unread(socket).await; @@ -600,7 +601,7 @@ async fn handle_turn( let started_at = serve_common::now_unix(); let started_instant = std::time::Instant::now(); let model_at_start = bus.model(); - let prompt = serve_common::format_wake_prompt(&from, &body, unread, redelivered); + let prompt = serve_common::format_wake_prompt(msg_id, &from, &body, unread, redelivered); let outcome = { let _guard = turn_lock.lock().await; turn::drive_turn(&prompt, files, bus).await diff --git a/hive-ag3nt/src/mcp.rs b/hive-ag3nt/src/mcp.rs index da8dac22..87f432d3 100644 --- a/hive-ag3nt/src/mcp.rs +++ b/hive-ag3nt/src/mcp.rs @@ -33,12 +33,14 @@ pub enum SocketReply { /// Unified `recv` result: zero or more messages popped in one /// round-trip. Empty vec = "(empty)" path; single-message = the /// standard wake body; multi = batch render with per-message - /// separators. Per-row `id` is opaque to claude (the bin loops - /// drive ack via `AckTurn`, not per-id); `redelivered` triggers - /// the "may already be handled" banner in `format_recv` for that - /// specific row. + /// separators. Per-row `id` is rendered as a `[msg #]` marker + /// so claude can bulk-triage via `ack_until` (turn-level ack still + /// rides `AckTurn`); `redelivered` triggers the "may already be + /// handled" banner in `format_recv` for that specific row. Messages(Vec), Status(u64), + /// `ack_until` result: rows newly marked handled. + Acked(u64), QuestionQueued(i64), Recent(Vec), Logs(String), @@ -74,6 +76,7 @@ impl From for SocketReply { hive_sh4re::Response::Err { message } => Self::Err(message), hive_sh4re::Response::Messages { messages } => Self::Messages(messages), hive_sh4re::Response::Status { unread } => Self::Status(unread), + hive_sh4re::Response::Acked { count } => Self::Acked(count), hive_sh4re::Response::Recent { rows } => Self::Recent(rows), hive_sh4re::Response::QuestionQueued { id } => Self::QuestionQueued(id), hive_sh4re::Response::LooseEnds { loose_ends } => Self::LooseEnds(loose_ends), @@ -219,7 +222,7 @@ pub fn format_recv(resp: Result, waited: bool) -> St if messages.len() == 1 { let m = &messages[0]; let banner = if m.redelivered { REDELIVERY_HINT } else { "" }; - return format!("{banner}from: {}\n\n{}", m.from, m.body); + 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"); @@ -228,11 +231,29 @@ pub fn format_recv(resp: Result, waited: bool) -> St out.push_str("\n---\n\n"); } let banner = if m.redelivered { REDELIVERY_HINT } else { "" }; - let _ = write!(out, "{banner}from: {}\n\n{}", m.from, m.body); + let _ = write!( + out, + "{banner}{}from: {}\n\n{}", + msg_id_tag(m.id), + m.from, + m.body + ); } out } +/// `[msg #] ` 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() + } +} + /// Header prepended to message bodies that were popped by a prior /// harness session, never acked (turn crash / OOM / restart), and /// resurfaced by `RequeueInflight` on this session's boot. Same @@ -601,6 +622,16 @@ pub struct RecvArgs { pub max: Option, } +/// 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 #]` 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 `remind`. Exactly one of `delay_seconds` or /// `at_unix_timestamp` must be set; both / neither is a tool-side error. /// Hides the tagged `ReminderTiming` enum behind a flatter schema so the @@ -798,6 +829,36 @@ impl AgentServer { .await } + #[tool( + description = "Bulk-mark inbox messages handled: every message with broker id \ + <= `up_to` (ids show as `[msg #]` 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) -> String { + let log = format!("{args:?}"); + run_tool_envelope("ack_until", log, async move { + let (resp, retries) = self + .dispatch(hive_sh4re::Request::AckUntil { up_to: args.up_to }) + .await; + let rendered = match resp { + Ok(SocketReply::Acked(count)) => { + format!("acked {count} message(s) up to id {}", args.up_to) + } + Ok(SocketReply::Err(m)) => format!("ack_until failed: {m}"), + Ok(other) => format!("ack_until unexpected response: {other:?}"), + Err(e) => format!("ack_until transport error: {e:#}"), + }; + annotate_retries(rendered, retries) + }) + .await + } + #[tool( description = "List loose ends pending against this agent: unanswered questions \ where you are the asker (waiting on someone) or the target (someone's waiting on \ diff --git a/hive-ag3nt/src/serve_common.rs b/hive-ag3nt/src/serve_common.rs index 5710d89c..abc460f9 100644 --- a/hive-ag3nt/src/serve_common.rs +++ b/hive-ag3nt/src/serve_common.rs @@ -9,21 +9,36 @@ use crate::turn::TurnOutcome; use crate::turn_stats::TurnStatRow; /// Assemble the per-turn wake prompt string. The role/tools/etc. live in the -/// system prompt; this is just the wake signal body. `unread` is the inbox -/// depth after this message was popped. `redelivered` prepends a "may already -/// be handled" banner. +/// system prompt; this is just the wake signal body. `id` is the broker row +/// id, rendered as a `[msg #]` marker so the agent can reference it in +/// `ack_until` (transient pings carry the sentinel 0 and render without it). +/// `unread` is the inbox depth after this message was popped. `redelivered` +/// prepends a "may already be handled" banner. #[must_use] -pub fn format_wake_prompt(from: &str, body: &str, unread: u64, redelivered: bool) -> String { +pub fn format_wake_prompt( + id: i64, + from: &str, + body: &str, + unread: u64, + redelivered: bool, +) -> String { let banner = if redelivered { REDELIVERY_HINT } else { "" }; + let tag = if id > 0 { + format!("[msg #{id}] ") + } else { + String::new() + }; let pending = if unread == 0 { String::new() } else { format!( "\n\n({unread} more message(s) pending in your inbox — call `mcp__hyperhive__recv` \ - with `max: {unread}` to drain them all in one round-trip before acting.)" + with `max: {unread}` to drain them all in one round-trip before acting. If the \ + backlog is stale/already handled, `ack_until(up_to: )` \ + clears everything up to that id in one call instead.)" ) }; - format!("{banner}Incoming message from `{from}`:\n---\n{body}\n---{pending}") + format!("{banner}{tag}Incoming message from `{from}`:\n---\n{body}\n---{pending}") } /// Current time as a Unix timestamp (seconds). Returns 0 on any error. diff --git a/hive-c0re/src/broker.rs b/hive-c0re/src/broker.rs index 754b1fb3..8f103862 100644 --- a/hive-c0re/src/broker.rs +++ b/hive-c0re/src/broker.rs @@ -751,6 +751,38 @@ 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![now_unix(), 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 / @@ -1304,6 +1336,61 @@ mod tests { assert!(pop_one(broker, "b").is_none()); } + /// 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. diff --git a/hive-c0re/src/socket_server.rs b/hive-c0re/src/socket_server.rs index 3801da7a..df590644 100644 --- a/hive-c0re/src/socket_server.rs +++ b/hive-c0re/src/socket_server.rs @@ -237,6 +237,7 @@ pub(crate) async fn dispatch_shared( } hive_sh4re::Request::CreateRepo { repo } => handle_create_repo(agent, repo).await, hive_sh4re::Request::AckTurn => handle_ack_turn(coord, agent), + hive_sh4re::Request::AckUntil { up_to } => handle_ack_until(coord, agent, *up_to), hive_sh4re::Request::RequeueInflight => handle_requeue_inflight(coord, agent), hive_sh4re::Request::GracefulStopComplete => { // Harness drained + is exiting: clear the fence so the @@ -483,6 +484,17 @@ fn handle_ack_turn(coord: &Arc, agent: &str) -> hive_sh4re::Respons } } +/// `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_sh4re::Response { + match coord.broker.ack_until(agent, up_to) { + Ok(count) => hive_sh4re::Response::Acked { count }, + Err(e) => hive_sh4re::Response::Err { + message: format!("{e:#}"), + }, + } +} + /// `RequeueInflight` — resurface `agent`'s unacked in-flight messages /// (crash recovery on harness boot). fn handle_requeue_inflight(coord: &Arc, agent: &str) -> hive_sh4re::Response { diff --git a/hive-sh4re/src/lib.rs b/hive-sh4re/src/lib.rs index ebe397ef..f9b5332d 100644 --- a/hive-sh4re/src/lib.rs +++ b/hive-sh4re/src/lib.rs @@ -679,6 +679,15 @@ pub enum Request { /// Harness↔broker pairing fired after `TurnOutcome::Ok`. See /// `docs/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/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 @@ -811,12 +820,15 @@ pub enum Response { /// `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 (opaque to claude; - /// tracked by the harness for `AckTurn`) and the "previously + /// `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. Messages { messages: Vec }, /// `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 }, /// `Ask` result: the queued question id. The answer lands later @@ -1132,7 +1144,7 @@ impl ToolGroup { #[must_use] pub fn tools(self) -> &'static [&'static str] { match self { - Self::Messaging => &["send", "recv", "ask", "answer"], + Self::Messaging => &["send", "recv", "ack_until", "ask", "answer"], Self::Meta => &["get_agent_meta"], Self::Inbox => &[ "get_loose_ends",