diff --git a/docs/conventions.md b/docs/conventions.md index cefdbcf3..72ce9b31 100644 --- a/docs/conventions.md +++ b/docs/conventions.md @@ -152,22 +152,6 @@ 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 c0a849e4..353cdd15 100644 --- a/hive-ag3nt/prompts/system.md +++ b/hive-ag3nt/prompts/system.md @@ -3,7 +3,6 @@ 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 c9046cd7..ae594e96 100644 --- a/hive-ag3nt/src/bin/hive.rs +++ b/hive-ag3nt/src/bin/hive.rs @@ -588,7 +588,6 @@ 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; @@ -601,7 +600,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(msg_id, &from, &body, unread, redelivered); + let prompt = serve_common::format_wake_prompt(&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 87f432d3..da8dac22 100644 --- a/hive-ag3nt/src/mcp.rs +++ b/hive-ag3nt/src/mcp.rs @@ -33,14 +33,12 @@ 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 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. + /// 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. Messages(Vec), Status(u64), - /// `ack_until` result: rows newly marked handled. - Acked(u64), QuestionQueued(i64), Recent(Vec), Logs(String), @@ -76,7 +74,6 @@ 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), @@ -222,7 +219,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{}", msg_id_tag(m.id), m.from, m.body); + return format!("{banner}from: {}\n\n{}", m.from, m.body); } let n = messages.len(); let mut out = format!("popped {n} message(s):\n\n"); @@ -231,29 +228,11 @@ 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{}", - msg_id_tag(m.id), - m.from, - m.body - ); + let _ = write!(out, "{banner}from: {}\n\n{}", 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 @@ -622,16 +601,6 @@ 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 @@ -829,36 +798,6 @@ 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 abc460f9..5710d89c 100644 --- a/hive-ag3nt/src/serve_common.rs +++ b/hive-ag3nt/src/serve_common.rs @@ -9,36 +9,21 @@ 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. `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. +/// 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. #[must_use] -pub fn format_wake_prompt( - id: i64, - from: &str, - body: &str, - unread: u64, - redelivered: bool, -) -> String { +pub fn format_wake_prompt(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. If the \ - backlog is stale/already handled, `ack_until(up_to: )` \ - clears everything up to that id in one call instead.)" + with `max: {unread}` to drain them all in one round-trip before acting.)" ) }; - format!("{banner}{tag}Incoming message from `{from}`:\n---\n{body}\n---{pending}") + format!("{banner}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 b2caef09..754b1fb3 100644 --- a/hive-c0re/src/broker.rs +++ b/hive-c0re/src/broker.rs @@ -441,11 +441,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. let n: i64 = conn.query_row( "SELECT COUNT(*) FROM messages - WHERE recipient = ?1 AND delivered_at IS NULL AND acked_at IS NULL", + WHERE recipient = ?1 AND delivered_at IS NULL", params![recipient], |row| row.get(0), )?; @@ -459,12 +457,9 @@ impl Broker { /// bodies differ. pub fn has_pending_with_body(&self, recipient: &str, sender: &str, body: &str) -> Result { let conn = self.conn.lock().unwrap(); - // An `ack_until`-closed pending row is dead — it must not - // suppress a fresh scheduled delivery of the same body. let n: i64 = conn.query_row( "SELECT COUNT(*) FROM messages - WHERE recipient = ?1 AND sender = ?2 AND body = ?3 - AND delivered_at IS NULL AND acked_at IS NULL", + WHERE recipient = ?1 AND sender = ?2 AND body = ?3 AND delivered_at IS NULL", params![recipient, sender, body], |row| row.get(0), )?; @@ -497,7 +492,6 @@ impl Broker { AND sender = ?2 AND body LIKE ?3 AND delivered_at IS NULL - AND acked_at IS NULL LIMIT 1", params![child, hive_sh4re::SYSTEM_SENDER, format!("{PREFIX}%")], |row| Ok((row.get::<_, i64>(0)?, row.get::<_, String>(1)?)), @@ -650,13 +644,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. let mut stmt = conn.prepare( "SELECT id, sender, recipient, body, in_reply_to FROM messages - WHERE recipient = ?1 AND delivered_at IS NULL AND acked_at IS NULL + WHERE recipient = ?1 AND delivered_at IS NULL ORDER BY id ASC LIMIT ?2", )?; @@ -760,38 +751,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![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 / @@ -1345,61 +1304,6 @@ 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 df590644..3801da7a 100644 --- a/hive-c0re/src/socket_server.rs +++ b/hive-c0re/src/socket_server.rs @@ -237,7 +237,6 @@ 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 @@ -484,17 +483,6 @@ 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 f9b5332d..ebe397ef 100644 --- a/hive-sh4re/src/lib.rs +++ b/hive-sh4re/src/lib.rs @@ -679,15 +679,6 @@ 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 @@ -820,15 +811,12 @@ 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 (tracked by the harness - /// for `AckTurn`, and surfaced to claude as a `[msg #]` marker - /// so `AckUntil` has something to reference) and the "previously + /// `redelivered` carry the broker's row id (opaque to claude; + /// tracked by the harness for `AckTurn`) 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 @@ -1144,7 +1132,7 @@ impl ToolGroup { #[must_use] pub fn tools(self) -> &'static [&'static str] { match self { - Self::Messaging => &["send", "recv", "ack_until", "ask", "answer"], + Self::Messaging => &["send", "recv", "ask", "answer"], Self::Meta => &["get_agent_meta"], Self::Inbox => &[ "get_loose_ends",