diff --git a/hive-ag3nt/src/mcp.rs b/hive-ag3nt/src/mcp.rs index 21435b21..7604bca6 100644 --- a/hive-ag3nt/src/mcp.rs +++ b/hive-ag3nt/src/mcp.rs @@ -24,117 +24,6 @@ use rmcp::{ use crate::client; -/// Wire-protocol-agnostic view of a hyperhive socket response. Both flavors -/// of `AgentServer` convert into this so the tool formatters can be shared. -#[derive(Debug)] -pub enum SocketReply { - Ok, - Err(String), - /// 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. - Messages(Vec), - Status(u64), - /// `ack_until` result: rows newly marked handled. - Acked(u64), - QuestionQueued(i64), - Recent(Vec), - Logs(String), - HostJournal(String), - /// `list_schedules` result — returned by `list_schedules` (scheduling tool group). - Schedules(Vec), - /// `list_containers` result — descendant containers with running status. - Containers(Vec), - LooseEnds(Vec), - PendingRemindersCount(u64), - ReminderRollup(hive_sh4re::ReminderStats), - AgentMeta { - name: String, - running: bool, - hyperhive_rev: Option, - status_text: Option, - status_set_at: Option, - hive_name: Option, - swarm_name: Option, - matrix_accounts: Vec, - }, - /// `create_repo` result — the new repo's full name + clone URL. - RepoCreated { - full_name: String, - clone_url: String, - }, -} - -impl From for SocketReply { - fn from(r: hive_sh4re::Response) -> Self { - match r { - hive_sh4re::Response::Ok => Self::Ok, - 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), - hive_sh4re::Response::PendingRemindersCount { count } => { - Self::PendingRemindersCount(count) - } - hive_sh4re::Response::ReminderRollup(stats) => Self::ReminderRollup(stats), - hive_sh4re::Response::Logs { content } => Self::Logs(content), - hive_sh4re::Response::HostJournal { content } => Self::HostJournal(content), - hive_sh4re::Response::Schedules { schedules } => Self::Schedules(schedules), - hive_sh4re::Response::Containers { containers } => Self::Containers(containers), - // A graceful stop is pending — the inbox is fenced. Returning an - // *empty* inbox here is ambiguous: claude's "park on recv" habit - // makes it long-poll again instead of ending the turn, so the - // stop-checkpoint turn never finishes and the drain wait times out - // into a hard stop. Return a single explicit directive instead, so - // every recv during the stop unmissably tells claude to flush + end. - hive_sh4re::Response::GracefulStop => Self::Messages(vec![hive_sh4re::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, - }]), - hive_sh4re::Response::AgentMeta { - name, - running, - hyperhive_rev, - status_text, - status_set_at, - hive_name, - swarm_name, - matrix_accounts, - } => Self::AgentMeta { - name, - running, - hyperhive_rev, - status_text, - status_set_at, - hive_name, - swarm_name, - matrix_accounts, - }, - hive_sh4re::Response::RepoCreated { - full_name, - clone_url, - } => Self::RepoCreated { - full_name, - clone_url, - }, - } - } -} - /// Write (or remove) the status file in the agent's own `state/` directory. /// Called by `AgentServer::set_status` for both agent and manager flavors /// before dispatching the wire `SetStatus` request (which only triggers a @@ -186,9 +75,9 @@ fn write_status_file(text: &str) -> Result<(), String> { /// transport error: …"`. Handlers match their own happy-path variant and route /// everything else here via a catch-all arm (`other => reply_err(other, tool)`), /// so the triplet lives in exactly one place. -fn reply_err(resp: Result, tool: &str) -> String { +fn reply_err(resp: Result, tool: &str) -> String { match resp { - Ok(SocketReply::Err(m)) => format!("{tool} failed: {m}"), + Ok(hive_sh4re::Response::Err { message }) => format!("{tool} failed: {message}"), Ok(other) => format!("{tool} unexpected response: {other:?}"), Err(e) => format!("{tool} transport error: {e:#}"), } @@ -198,9 +87,13 @@ fn reply_err(resp: Result, tool: &str) -> String { /// `tool` and `ok_msg` only appear in the result string; they don't change /// behavior. #[must_use] -pub fn format_ack(resp: Result, tool: &str, ok_msg: String) -> String { +pub fn format_ack( + resp: Result, + tool: &str, + ok_msg: String, +) -> String { match resp { - Ok(SocketReply::Ok) => ok_msg, + Ok(hive_sh4re::Response::Ok) => ok_msg, other => reply_err(other, tool), } } @@ -216,13 +109,37 @@ pub fn format_ack(resp: Result, tool: &str, ok_msg: /// 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, waited: bool) -> String { +pub fn format_recv(resp: Result, waited: bool) -> String { match resp { - Ok(SocketReply::Messages(m)) => render_recv_messages(&m, waited), + Ok(hive_sh4re::Response::Messages { messages }) => render_recv_messages(&messages, 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. + Ok(hive_sh4re::Response::GracefulStop) => { + render_recv_messages(&[graceful_stop_message()], waited) + } 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::DeliveredMessage { + hive_sh4re::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). fn render_recv_messages(messages: &[hive_sh4re::DeliveredMessage], waited: bool) -> String { @@ -284,10 +201,9 @@ pub const IDLE_WAIT_HINT: &str = " — nothing arrived before the wait timed out If you have other useful work (assigned issues, in-flight PRs, a docs sweep, \ notes to update), do that now rather than immediately parking on recv again."; -/// Inner renderer for a `Vec` already extracted from the -/// socket reply. Called by both `format_loose_ends` (which handles the -/// `Result` wrapper) and the augmented `get_loose_ends` -/// handler (which injects the `UnreadMatrix` entry before formatting). +/// Inner renderer for a `Vec` already extracted from the socket +/// reply. Called by the `get_loose_ends` handler, which injects the +/// `UnreadMatrix` entry before formatting. fn render_loose_ends(loose_ends: &[hive_sh4re::LooseEnd]) -> String { use std::fmt::Write as _; if loose_ends.is_empty() { @@ -366,21 +282,6 @@ fn render_loose_ends(loose_ends: &[hive_sh4re::LooseEnd]) -> String { out } -/// Format helper for `get_loose_ends`: renders a short bulleted list -/// of pending approvals + questions + reminders. Empty list collapses -/// to a clear marker so claude doesn't go hunting for a payload that -/// isn't there. -#[must_use] -pub fn format_loose_ends(resp: Result) -> String { - let loose_ends = match resp { - Ok(SocketReply::LooseEnds(t)) => t, - Ok(SocketReply::Err(m)) => return format!("get_loose_ends failed: {m}"), - Ok(other) => return format!("get_loose_ends unexpected response: {other:?}"), - Err(e) => return format!("get_loose_ends transport error: {e:#}"), - }; - render_loose_ends(&loose_ends) -} - /// Per-room unread entry returned by `matrix_unread_summary`. Mirrors /// `hive-matrix-mcp`'s `RoomUnread` but defined locally to avoid a /// cross-crate dep on the matrix-sdk crate tree. @@ -481,9 +382,9 @@ fn loose_end_kind_label(kind: hive_sh4re::CancelLooseEndKind) -> &'static str { /// `running: no` line tells the caller WHY. See /// `docs/turn-loop.md::Sub-agent tools` (`get_agent_meta`). #[must_use] -pub fn format_agent_meta(resp: Result) -> String { +pub fn format_agent_meta(resp: Result) -> String { match resp { - Ok(SocketReply::AgentMeta { + Ok(hive_sh4re::Response::AgentMeta { name, running, hyperhive_rev, @@ -693,18 +594,18 @@ impl AgentServer { Self { socket } } - /// Issue any `Request` through the retry-aware client and pull - /// the reply through `SocketReply`. Returns the retry count so tool - /// handlers can annotate their result (see `annotate_retries`). + /// Issue any `Request` through the retry-aware client. Returns the raw + /// `Response` plus the retry count so tool handlers can annotate their + /// result (see `annotate_retries`). /// /// `AgentRequest` / `ManagerRequest` / `Request` are all the same type /// (hive-sh4re type aliases), so this single method covers both sockets. async fn dispatch( &self, req: hive_sh4re::Request, - ) -> (Result, u32) { + ) -> (Result, u32) { match client::request_retried::<_, hive_sh4re::Response>(&self.socket, &req).await { - Ok((r, n)) => (Ok(SocketReply::from(r)), n), + Ok((r, n)) => (Ok(r), n), Err(e) => (Err(e), 0), } } @@ -770,7 +671,7 @@ impl AgentServer { }) .await; let s = match resp { - Ok(SocketReply::QuestionQueued(id)) => format!( + Ok(hive_sh4re::Response::QuestionQueued { id }) => format!( "question queued (id={id}); answer will arrive as a system \ `question_answered` event in your inbox" ), @@ -858,7 +759,7 @@ impl AgentServer { .dispatch(hive_sh4re::Request::AckUntil { up_to: args.up_to }) .await; let rendered = match resp { - Ok(SocketReply::Acked(count)) => { + Ok(hive_sh4re::Response::Acked { count }) => { format!("acked {count} message(s) up to id {}", args.up_to) } other => reply_err(other, "ack_until"), @@ -891,7 +792,7 @@ impl AgentServer { .await; // Extract the vec so we can augment before rendering. let mut loose_ends = match resp { - Ok(SocketReply::LooseEnds(t)) => t, + Ok(hive_sh4re::Response::LooseEnds { loose_ends }) => loose_ends, other => return annotate_retries(reply_err(other, "get_loose_ends"), retries), }; // Prepend matrix unread entry for self-queries only (can't @@ -1030,7 +931,7 @@ impl AgentServer { .dispatch(hive_sh4re::Request::CreateRepo { repo: args.repo }) .await; let s = match resp { - Ok(SocketReply::RepoCreated { + Ok(hive_sh4re::Response::RepoCreated { full_name, clone_url, }) => format!("created repo {full_name} — clone: {clone_url}"), @@ -1191,7 +1092,7 @@ impl AgentServer { run_tool_envelope("list_containers", String::new(), async move { let (resp, retries) = self.dispatch(hive_sh4re::Request::ListDescendants).await; let body = match resp { - Ok(SocketReply::Containers(containers)) => { + Ok(hive_sh4re::Response::Containers { containers }) => { if containers.is_empty() { "no descendant containers".to_owned() } else { @@ -1245,7 +1146,7 @@ impl AgentServer { }) .await; let result = match resp { - Ok(SocketReply::HostJournal(content)) => content, + Ok(hive_sh4re::Response::HostJournal { content }) => content, other => reply_err(other, "get_host_journal"), }; annotate_retries(result, retries) @@ -1370,7 +1271,7 @@ impl AgentServer { }) .await; let s = match resp { - Ok(SocketReply::Logs(content)) => { + Ok(hive_sh4re::Response::Logs { content }) => { if content.is_empty() { format!("(no journal output for {agent})") } else { @@ -1563,7 +1464,7 @@ impl AgentServer { run_tool_envelope("list_schedules", String::new(), async move { let (resp, retries) = self.dispatch(hive_sh4re::Request::ListSchedules).await; let body = match resp { - Ok(SocketReply::Schedules(schedules)) => serde_json::to_string(&schedules) + Ok(hive_sh4re::Response::Schedules { schedules }) => serde_json::to_string(&schedules) .unwrap_or_else(|e| format!("list_schedules: serialise: {e:#}")), other => reply_err(other, "list_schedules"), }; @@ -1580,27 +1481,23 @@ impl AgentServer { )] impl ServerHandler for AgentServer {} -/// Run an MCP server over stdio for the given flavor. Returns when the client disconnects. +/// Run the MCP server over stdio. Used by all roles. Returns when the client +/// disconnects. /// /// # Errors /// /// Returns an error if the MCP server fails to initialize or the transport /// encounters a fatal error. -pub async fn serve_stdio(socket: PathBuf) -> Result<()> { +pub async fn serve_agent_stdio(socket: PathBuf) -> Result<()> { let server = AgentServer::new(socket); let service = server.serve(stdio()).await?; service.waiting().await?; Ok(()) } -/// Run the MCP server over stdio. Used by all roles. -pub async fn serve_agent_stdio(socket: PathBuf) -> Result<()> { - serve_stdio(socket).await -} - /// Run the MCP server over HTTP (rmcp streamable-http transport) on `addr`. /// -/// Unlike [`serve_stdio`] — a fresh stdio child claude respawns every turn — +/// Unlike [`serve_agent_stdio`] — a fresh stdio child claude respawns every turn — /// this is meant to run as a long-lived in-container daemon. claude reconnects /// to the stable URL each turn instead of respawning and re-registering a stdio /// subprocess, which removes the per-turn MCP registration race that can strand @@ -1750,17 +1647,6 @@ pub struct CancelLooseEndArgs { pub id: i64, } -#[derive(Debug, serde::Deserialize, schemars::JsonSchema)] -pub struct GetLooseEndsArgs { - /// Whose loose ends to list. Omit (or `null`) for your own: approvals - /// you submitted + questions where you are asker/target + your own - /// pending reminders. Pass `"*"` for a hive-wide view of EVERY pending - /// approval, unanswered question, and reminder across the swarm. Pass a - /// specific agent name to inspect just that agent's threads. - #[serde(default)] - pub agent: Option, -} - #[derive(Debug, serde::Deserialize, schemars::JsonSchema)] pub struct AgentGetLooseEndsArgs { /// Whose loose ends to list. Omit (or `null`) for your own. You may @@ -2316,20 +2202,20 @@ pub fn render_claude_config(agent_binary: &str, socket: &std::path::Path) -> Str #[cfg(test)] mod tests { - use super::{IDLE_WAIT_HINT, SocketReply, format_recv}; + use super::{IDLE_WAIT_HINT, format_recv}; use super::{SERVER_NAME, allowed_mcp_tools}; use hive_sh4re::ToolGroup; #[test] fn empty_recv_after_wait_appends_idle_hint() { - let out = format_recv(Ok(SocketReply::Messages(vec![])), true); + let out = format_recv(Ok(hive_sh4re::Response::Messages { messages: vec![] }), true); assert!(out.starts_with("(empty)")); assert!(out.contains(IDLE_WAIT_HINT)); } #[test] fn empty_recv_without_wait_has_no_hint() { - let out = format_recv(Ok(SocketReply::Messages(vec![])), false); + let out = format_recv(Ok(hive_sh4re::Response::Messages { messages: vec![] }), false); assert_eq!(out, "(empty)"); }