vinzenz/hyperhive
vinzenz/hyperhive
Watch
0
0
Fork
You've already forked hyperhive
 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

refactor(mcp): drop vestigial SocketReply, match hive_sh4re::Response directly; remove dead code

Browse source
This commit is contained in:
müde 2026-07-05 22:12:45 +02:00
commit 9a0d525fdf
1 changed files with 59 additions and 173 deletions
Download patch file Download diff file Expand all files Collapse all files

232
hive-ag3nt/src/mcp.rs
View file

@ -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 #<id>]` 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<hive_sh4re::DeliveredMessage>),
Status(u64),
/// `ack_until` result: rows newly marked handled.
Acked(u64),
QuestionQueued(i64),
Recent(Vec<hive_sh4re::InboxRow>),
Logs(String),
HostJournal(String),
/// `list_schedules` result — returned by `list_schedules` (scheduling tool group).
Schedules(Vec<hive_sh4re::WireSchedule>),
/// `list_containers` result — descendant containers with running status.
Containers(Vec<hive_sh4re::ContainerInfo>),
LooseEnds(Vec<hive_sh4re::LooseEnd>),
PendingRemindersCount(u64),
ReminderRollup(hive_sh4re::ReminderStats),
AgentMeta {
name: String,
running: bool,
hyperhive_rev: Option<String>,
status_text: Option<String>,
status_set_at: Option<i64>,
hive_name: Option<String>,
swarm_name: Option<String>,
matrix_accounts: Vec<hive_sh4re::MatrixIdentity>,
},
/// `create_repo` result — the new repo's full name + clone URL.
RepoCreated {
full_name: String,
clone_url: String,
},
}
impl From<hive_sh4re::Response> 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<SocketReply, anyhow::Error>, tool: &str) -> String {
fn reply_err(resp: Result<hive_sh4re::Response, anyhow::Error>, 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<SocketReply, anyhow::Error>, 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<SocketReply, anyhow::Error>, tool: &str, ok_msg: String) -> String {
pub fn format_ack(
resp: Result<hive_sh4re::Response, anyhow::Error>,
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<SocketReply, anyhow::Error>, 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<SocketReply, anyhow::Error>, waited: bool) -> String {
pub fn format_recv(resp: Result<hive_sh4re::Response, anyhow::Error>, 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<LooseEnd>` already extracted from the
/// socket reply. Called by both `format_loose_ends` (which handles the
/// `Result<SocketReply>` wrapper) and the augmented `get_loose_ends`
/// handler (which injects the `UnreadMatrix` entry before formatting).
/// Inner renderer for a `Vec<LooseEnd>` 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<SocketReply, anyhow::Error>) -> 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<SocketReply, anyhow::Error>) -> String {
pub fn format_agent_meta(resp: Result<hive_sh4re::Response, anyhow::Error>) -> 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<SocketReply, anyhow::Error>, u32) {
) -> (Result<hive_sh4re::Response, anyhow::Error>, 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<String>,
}
#[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)");
}