//! Embedded MCP server. Claude Code (running inside the agent container) //! connects to this over streamable-HTTP via `--mcp-config` (the long-lived //! `hive-mcp-http` daemon); tool calls land here and are translated to //! `Request::*` against hyperhive's own //! per-container unix socket at `/run/hive/mcp.sock`. //! //! Two protocols, two surfaces: //! - **hyperhive socket** at `/run/hive/mcp.sock` — JSON-line, our //! broker-routed protocol. Unaffected by this module. //! - **MCP over HTTP** owned by this module — what claude actually speaks. //! //! One `AgentServer { socket }` struct for all roles. //! Tool access is gated upstream by `--allowedTools` (derived from the //! agent's `ToolGroup` config); the server itself is a dumb dispatcher. //! All tools go through the same `run_tool_envelope` helper. use std::future::Future; use std::path::PathBuf; use anyhow::Result; use rmcp::{ServerHandler, handler::server::wrapper::Parameters, tool, tool_handler, tool_router}; mod args; mod render; pub use args::{ AckUntilArgs, AgentGetLooseEndsArgs, AnswerArgs, AskArgs, CancelLooseEndArgs, CancelScheduleArgs, CreateRepoArgs, EditScheduleArgs, FireScheduleNowArgs, GetAgentMetaArgs, GetHostJournalArgs, GetLogsArgs, KillArgs, MarkTodosDoneArgs, RecvArgs, RemindArgs, RequestInitConfigArgs, RequestSchedulePromptArgs, RestartArgs, SendArgs, SetStatusArgs, StartArgs, UpdateArgs, UpdateMetaInputsArgs, }; pub use render::{annotate_retries, format_ack, format_agent_meta, format_recv}; use render::{ dial_agent_socket, format_matrix_summary, local_questions, local_reminders, local_todos, loose_end_kind_label, mark_local_todo_done, mark_local_todos_done, matrix_unread_summary, parse_loose_end_kind, render_loose_ends, reply_err, }; /// 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 /// dashboard rescan on the host side — file I/O moved here because the /// harness runs as the agent user and has write access to `state/`, whereas /// hive-c0re's `hive-core` user does not after the privsep migration). /// /// Mirrors the validation in `hive-c0re::limits::check_status_text` so /// the file is never written with text the server would later reject (which /// would leave a stale invalid entry on disk). fn write_status_file(text: &str) -> Result<(), String> { // 200 chars mirrors STATUS_MAX_CHARS in hive-c0re/src/limits.rs. // Keep in sync if that constant changes. const STATUS_MAX_CHARS: usize = 200; let trimmed = text.trim(); if !trimmed.is_empty() { if trimmed.contains('\n') || trimmed.contains('\r') { return Err( "set_status text must be a single line — write multi-line context to \ a file under your state/ dir and reference that path from the chip instead" .to_owned(), ); } let len = trimmed.chars().count(); if len > STATUS_MAX_CHARS { return Err(format!( "set_status text too long ({len} chars, max {STATUS_MAX_CHARS}); trim to a short summary" )); } } let path = crate::paths::state_dir().join("hyperhive-status"); let result = if trimmed.is_empty() { std::fs::remove_file(&path).or_else(|e| { if e.kind() == std::io::ErrorKind::NotFound { Ok(()) } else { Err(e) } }) } else { std::fs::write(&path, format!("{trimmed}\n")) }; result.map_err(|e| format!("set_status write failed: {e}")) } /// Common envelope around every MCP tool handler: pre-log → run → /// post-log. Tool results stay clean — the inbox-status hint lives in /// the wake prompt + UI header, not appended here. pub async fn run_tool_envelope(tool: &'static str, args: String, body: F) -> String where F: Future, { tracing::info!(tool, %args, "tool: request"); let result = body.await; tracing::info!(tool, result = %result, "tool: result"); result } /// Unified MCP tool surface for both sub-agent and manager roles. /// /// Both sockets speak the same `Request` / `Response` wire, so a single /// `dispatch` call covers both sockets — the only real difference is which /// socket path is used and which tools the flavor enables. #[derive(Debug, Clone)] pub struct AgentServer { socket: PathBuf, } impl AgentServer { #[must_use] pub fn new(socket: PathBuf) -> Self { Self { socket } } /// 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`). /// /// Both sockets speak the same `Request` type, so this single method /// covers both. async fn dispatch( &self, req: hive_core_agent_sock::Request, ) -> (Result, u32) { // Ride out a hive-c0re restart rather than surface it: the caller // here is claude, and a tool error costs a whole retry turn's // tokens where 60s of in-daemon patience costs nothing. match hive_sock_client::request_retried::<_, hive_core_agent_sock::Response>( &self.socket, &req, hive_sock_client::Retry::RideOutRestart, ) .await { Ok((r, n)) => (Ok(r), n), Err(e) => (Err(e), 0), } } } // IMPORTANT: when adding a new `#[tool]` fn to this impl, also add // its name to the matching `ToolGroup::tools()` slice in hive-sh4re. // Claude Code's permission gate refuses uninlisted MCP tools in // non-interactive `--print` mode with "permissions not granted yet". #[tool_router] impl AgentServer { #[tool( description = "Send a message to another hyperhive agent (or to the operator). \ Use this to talk to peers or to surface output for the human at the dashboard." )] async fn send(&self, Parameters(args): Parameters) -> String { let log = format!("{args:?}"); let to = args.to.clone(); // Check per-agent allow-list (hyperhive.allowedRecipients). When no // policy file is present (e.g. manager containers) the check is a no-op. if let Err(refusal) = crate::send_allow::check_send_allowed(&to) { return run_tool_envelope("send", log, async move { refusal }).await; } 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 = "Surface a structured question to either the operator OR a peer agent. \ 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. \n\n\ Recipient: omit `to` (or set `to: \"operator\"`) for the human operator on the \ dashboard. Set `to: \"\"` to ask a peer agent — they receive a \ `question_asked { id, asker, question, options, multi }` event in their inbox \ and answer via `mcp__hyperhive__answer`. \n\n\ `options` is advisory: pass a short fixed-choice list when applicable, otherwise \ leave empty for free text. Set `multi: true` to let the answerer pick multiple \ options (checkboxes on the dashboard, hint to the agent otherwise) — answer comes \ back as a comma-separated string. Set `ttl_seconds` to auto-cancel a \ no-longer-relevant question — on expiry the answer is `[expired]` (with \ `answerer: \"ttl-watchdog\"`) and the same `question_answered` event fires." )] async fn ask(&self, Parameters(args): Parameters) -> String { let log = format!("{args:?}"); run_tool_envelope("ask", log, async move { let to = match args.to.map(|t| hive_types::Ident::parse(&t)).transpose() { Ok(to) => to, Err(reason) => return format!("invalid `to` agent name: {reason}"), }; let question = args.question; let target = to .as_ref() .map_or_else(|| "operator".to_owned(), std::string::ToString::to_string); let (resp, retries) = self .dispatch(hive_core_agent_sock::Request::Ask { question: question.clone(), options: args.options, multi: args.multi, ttl_seconds: args.ttl_seconds, to, }) .await; let s = match resp { Ok(hive_core_agent_sock::Response::QuestionQueued { id }) => { // Best-effort local questions-mirror record — a dial // failure just means `get_loose_ends` won't show this // row locally; the actual question is already queued // in c0re regardless. let _ = dial_agent_socket(&hive_agent_sock::Request::RecordAskedQuestion { id, target, question, }) .await; format!( "question queued (id={id}); answer will arrive as a system \ `question_answered` event in your inbox" ) } other => reply_err(other, "ask"), }; annotate_retries(s, retries) }) .await } #[tool( description = "Answer a question that was routed to YOU via a `question_asked` system \ event in your inbox. Pass the `id` from that event and your `answer` string. The \ answer will surface in the asker's inbox as a `question_answered { id, question, \ answer, answerer: }` event. \n\n\ Authorisation is strict — you can only answer questions where you are the declared \ target (i.e. the asker did `ask(to: \"\", ...)`). Trying to answer an \ operator-targeted question or a question addressed to a different agent will fail." )] async fn answer(&self, Parameters(args): Parameters) -> String { let log = format!("{args:?}"); let id = args.id; run_tool_envelope("answer", log, async move { let (resp, retries) = self .dispatch(hive_core_agent_sock::Request::Answer { id, answer: args.answer, }) .await; // Clear the local `answering` mirror row whenever this agent is // done owing a reply for `id`: either a genuine success, or // c0re telling us the question already resolved without us // (the asker cancelled/answered it first, surfacing as an // "already answered"/"not found" rejection) — in both cases // nothing is still owed, so the row would otherwise linger // stale. Any other rejection (e.g. wrong answerer) means the // question is still genuinely outstanding, so leave it be. let should_clear = match &resp { Ok(hive_core_agent_sock::Response::Ok) => true, Ok(hive_core_agent_sock::Response::Err { message }) => { message.contains("not found") || message.contains("already answered") } _ => false, }; if should_clear { let _ = dial_agent_socket(&hive_agent_sock::Request::ClearQuestion { id }).await; } annotate_retries( format_ack(resp, "answer", format!("answered question {id}")), 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) -> 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 #]` 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_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: unanswered questions \ where you are the asker (waiting on someone) or the target (someone's waiting on \ you), pending reminders you scheduled, plus — for the manager only — pending \ approvals you submitted that the operator hasn't acted on yet (agents with the \ `approvals` tool group also see their own pending approvals). Also lists active \ local tasks published by external MCP daemons (e.g. running bash tasks). Cheap sweep, no args. Useful \ at turn start to remember what you owe / what's owed to you without scrolling \ inbox history. Output is a short bulleted list with ids, ages in seconds, and \ the relevant context. Each `question` or `reminder` row can be cancelled by \ passing its id + kind to `cancel_loose_end`. Empty result is reported clearly. Todos \ cap at 40 rows; a trailer line says how many more are pending — clear the shown ones \ with `mark_todos_done`, then call again for the rest.\n\ Pass `agent: \"\"` to inspect a specific peer agent's threads. Direct \ child agents are always accessible. For non-children, the `query_agent_state` \ capability is required — without it the request is rejected with an error." )] async fn get_loose_ends(&self, Parameters(args): Parameters) -> String { run_tool_envelope("get_loose_ends", String::new(), async move { let is_self_query = args.agent.is_none(); let (resp, retries) = self .dispatch(hive_core_agent_sock::Request::GetLooseEnds { agent: args.agent }) .await; // Extract the vec so we can augment before rendering. let mut loose_ends = match resp { Ok(hive_core_agent_sock::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 // reach another agent's matrix daemon from here). if is_self_query && let Some(unread_rooms) = matrix_unread_summary().await { let total = u32::try_from(unread_rooms.len()).unwrap_or(u32::MAX); if total > 0 { let summary = format_matrix_summary(&unread_rooms); loose_ends.insert( 0, hive_sh4re::LooseEnd::UnreadMatrix { rooms: total, summary, }, ); } } // Merge the harness's local todos (loose-ends v2) for self-queries. // The harness owns the todo store in-container; another agent's // todos aren't reachable from here (same as matrix above). if is_self_query && let Some(todos) = local_todos().await { loose_ends.extend(todos); } // Merge local pending reminders — same self-query-only // restriction: a manager asking for a child's loose-ends no longer // sees the child's reminders, matching the todos precedent above). if is_self_query && let Some(reminders) = local_reminders().await { loose_ends.extend(reminders); } // Merge local mirrored questions — same self-query-only // restriction as todos/reminders above. c0re no longer sources // `Question` rows for `for_agent`/`hive_wide` (see // `hive-c0re::loose_ends` doc), so this is the only place a // self-query sees its own questions now; a manager query for a // child still sees that child's approvals (unaffected) but no // longer its questions, matching the reminders precedent. if is_self_query && let Some(questions) = local_questions().await { loose_ends.extend(questions); } annotate_retries(render_loose_ends(&loose_ends), retries) }) .await } #[tool( description = "Set a free-text status string visible on the operator dashboard. \ Call this at the START of every task to describe what you're working on (e.g. \ `\"processing matrix messages\"`, `\"fixing bitburner crash\"`, `\"idle\"`). Pass an empty \ string to clear. The status is shown on your dashboard card and persists across \ harness restarts." )] async fn set_status(&self, Parameters(args): Parameters) -> String { run_tool_envelope("set_status", args.text.clone(), async move { if let Err(e) = write_status_file(&args.text) { return e; } let (resp, retries) = self .dispatch(hive_core_agent_sock::Request::SetStatus { text: args.text }) .await; annotate_retries( format_ack(resp, "set_status", "status updated".to_owned()), retries, ) }) .await } #[tool( description = "Fetch identity + status metadata for an agent. Returns canonical \ `name`, the current `hyperhive_rev` hive-c0re is running against, `running` \ (bool — whether the container is currently up; when false, `status_text` and \ `status_set_at` are stale pre-stop values and should not be treated as live), \ and the target's self-reported `status` text (set via `set_status`) plus how \ long ago it was set. Also returns the hive + swarm display names (`hive_name`, \ `swarm_name`) when the operator has configured \ `services.hyperhive.hiveName` / `services.hyperhive.swarm.name`; both lines \ omitted when unset. Pass `name` to query a peer (e.g. \ check whether iris is idle before pinging them); omit `name` to get your own \ identity stamp — handy for state files / commit messages / cross-agent \ attribution that won't drift across renames or session-continue boundaries \ where the system-prompt label could be stale. Status reads `` when the \ target has never called `set_status` or has cleared it. Also returns \ `matrix_accounts` — a list of matrix identities the agent can act as \ (`name`, `user_id?`, `homeserver`); omitted for agents with no matrix \ provisioning." )] async fn get_agent_meta(&self, Parameters(args): Parameters) -> String { let log = args.name.clone().unwrap_or_else(|| "".to_owned()); run_tool_envelope("get_agent_meta", log, async move { let name = match args.name.map(|n| hive_types::Ident::parse(&n)).transpose() { Ok(name) => name, Err(reason) => return format!("invalid agent name: {reason}"), }; let (resp, retries) = self .dispatch(hive_core_agent_sock::Request::GetAgentMeta { name }) .await; annotate_retries(format_agent_meta(resp), retries) }) .await } #[tool( description = "Cancel an open thread you own — a `question` you asked (the \ asker gets `[cancelled by ]` as the answer and unblocks) or a `reminder` \ you scheduled (hard-deleted before it fires). `kind` is `\"question\"` or \ `\"reminder\"`; `id` is the row id from the matching `get_loose_ends` entry \ or the `question_queued` reply you got when you submitted. Auth: you can only \ cancel rows where you're the asker / owner. Returns `ok` or an error string.\n\ `kind` may also be `\"approval\"` to withdraw a pending approval you submitted \ (before the operator acts on it) — root agent (`ruth`) only; the server rejects \ `approval` kind for all other callers.\n\ `kind` may also be `\"todo\"` to clear one of your own loose-ends-v2 todos (the id \ from a `get_loose_ends` `todo #N` line) — dials the in-agent socket directly, safe \ to call repeatedly." )] async fn cancel_loose_end(&self, Parameters(args): Parameters) -> String { let log = format!("{args:?}"); let id = args.id; run_tool_envelope("cancel_loose_end", log, async move { if args.kind.trim().eq_ignore_ascii_case("todo") { return match mark_local_todo_done(id).await { Some(hive_agent_sock::Response::Acked { count }) if count > 0 => { format!("cleared todo {id}") } Some(hive_agent_sock::Response::Acked { .. }) => { format!("no such todo {id} (already cleared, or never existed)") } Some(hive_agent_sock::Response::Err { message }) => { format!("cancel_loose_end failed: {message}") } Some(other) => format!("cancel_loose_end unexpected response: {other:?}"), None => "cancel_loose_end: local todo socket unavailable \ (HIVE_AGENT_SOCKET unset or harness unreachable)" .to_owned(), }; } let kind = match parse_loose_end_kind(&args.kind) { Ok(k) => k, Err(e) => return e, }; let kind_label = loose_end_kind_label(kind); // Reminders are harness-local — dial the in-agent socket // directly instead of the broker; every other kind // (question/approval) still lives in c0re. if kind == hive_sh4re::CancelLooseEndKind::Reminder { return match dial_agent_socket(&hive_agent_sock::Request::CancelReminder { id }) .await { Some(hive_agent_sock::Response::Acked { count }) if count > 0 => { format!("cancelled {kind_label} {id}") } Some(hive_agent_sock::Response::Acked { .. }) => { format!("cancel_loose_end failed: no pending {kind_label} {id}") } Some(hive_agent_sock::Response::Err { message }) => { format!("cancel_loose_end failed: {message}") } Some(other) => format!("cancel_loose_end unexpected response: {other:?}"), None => "cancel_loose_end failed: in-agent socket unavailable".to_owned(), }; } let (resp, retries) = self .dispatch(hive_core_agent_sock::Request::CancelLooseEnd { kind, id }) .await; if resp.is_ok() && kind == hive_sh4re::CancelLooseEndKind::Question { // Best-effort — cancel is ownership-gated to the asker on // the c0re side, so a successful cancel here always means // *this* agent's own `asked` mirror row for `id`. Known gap // (documented on `Questions`): the target isn't notified, // so their `answering` row lingers until they call // `answer()` or it goes stale. let _ = dial_agent_socket(&hive_agent_sock::Request::ClearQuestion { id }).await; } annotate_retries( format_ack( resp, "cancel_loose_end", format!("cancelled {kind_label} {id}"), ), retries, ) }) .await } #[tool( description = "Bulk-clear specific todos by id in one call, instead of \ `cancel_loose_end`ing each one. List-based, not range-based — no `ack_until`-style \ 'clear below id N' for todos, since a blind range-clear risks acking something you \ never saw. Pass the ids you've actually reviewed (typically what `get_loose_ends` \ just showed); unknown/already-acked ids are silently skipped. Returns how many were \ newly acked." )] async fn mark_todos_done(&self, Parameters(args): Parameters) -> String { let log = format!("{args:?}"); run_tool_envelope("mark_todos_done", log, async move { match mark_local_todos_done(args.ids).await { Some(hive_agent_sock::Response::Acked { count }) => { format!("acked {count} todo(s)") } Some(hive_agent_sock::Response::Err { message }) => { format!("mark_todos_done failed: {message}") } Some(other) => format!("mark_todos_done unexpected response: {other:?}"), None => "mark_todos_done: local todo socket unavailable \ (HIVE_AGENT_SOCKET unset or harness unreachable)" .to_owned(), } }) .await } #[tool( description = "Create a git repo through hive-c0re. You CANNOT create repos with your \ own forge token (creation is disabled) — this is the only path. The repo is created in \ the c0re-owned `agents` org, you're added as a write collaborator (not owner), and the \ default branch gets branch protection so merges require an operator-team approval — you \ cannot merge your own PRs. `repo` is a single name segment (letters, digits, `-`, `_`, \ `.`). Returns the new repo's full name + clone URL; clone it over \ `$HIVE_FORGE_URL/agents/.git` and push/open PRs as normal." )] async fn create_repo(&self, Parameters(args): Parameters) -> String { let log = format!("{args:?}"); run_tool_envelope("create_repo", log, async move { let (resp, retries) = self .dispatch(hive_core_agent_sock::Request::CreateRepo { repo: args.repo }) .await; let s = match resp { Ok(hive_core_agent_sock::Response::RepoCreated { full_name, clone_url, }) => format!("created repo {full_name} — clone: {clone_url}"), other => reply_err(other, "create_repo"), }; annotate_retries(s, retries) }) .await } #[tool( description = "Schedule a reminder that lands in this agent's own inbox at a future \ time (sender will appear as `reminder`). Use for self-paced follow-ups: 'check task \ status in 60s', 'retry failed deploy at 14:00 UTC', 'nudge me when the operator's \ deploy window opens'. Set EXACTLY ONE of `delay_seconds` (fire N seconds from now) \ or `at_unix_timestamp` (fire at absolute epoch second). Body soft-caps at 4 KiB \ inline — anything larger gets auto-persisted to a file under your \ `/agents//state/reminders/` dir and the inbox message becomes a short pointer; \ pass `file_path` if you want to control the destination yourself. Returns \ immediately — the reminder lives in the broker until due." )] async fn remind(&self, Parameters(args): Parameters) -> String { let log = format!("{args:?}"); run_tool_envelope("remind", log, async move { let timing = match (args.delay_seconds, args.at_unix_timestamp) { (Some(_), Some(_)) => { return "remind failed: pass exactly one of `delay_seconds` or \ `at_unix_timestamp`, not both" .to_string(); } (None, None) => { return "remind failed: pass exactly one of `delay_seconds` or \ `at_unix_timestamp`" .to_string(); } (Some(s), None) => hive_sh4re::ReminderTiming::InSeconds { seconds: s }, (None, Some(t)) => hive_sh4re::ReminderTiming::At { unix_timestamp: t }, }; // Reminders are harness-local — dial the in-agent socket // directly instead of the broker. match dial_agent_socket(&hive_agent_sock::Request::StoreReminder { message: args.message, timing, file_path: args.file_path, }) .await { Some(hive_agent_sock::Response::Ok) => "reminder scheduled".to_owned(), Some(hive_agent_sock::Response::Err { message }) => { format!("remind failed: {message}") } Some(other) => format!("remind unexpected response: {other:?}"), None => "remind failed: in-agent socket unavailable".to_owned(), } }) .await } #[tool( description = "Compact the current session's context, mirroring the operator's \ dashboard `/compact` button. Gated: only honoured when this agent's last \ completed turn used more than 66% of the effective context window — below \ that the call is refused with an explanation and has no effect. On a pass, \ queues compaction for the end of the current turn (same deferred mechanism \ the dashboard button uses, so it never races a live claude process); the \ usual pre-compaction notes-checkpoint turn still fires first. No args." )] async fn compact(&self) -> String { run_tool_envelope("compact", String::new(), async move { match dial_agent_socket(&hive_agent_sock::Request::Compact).await { Some(hive_agent_sock::Response::Ok) => { "compact queued — will run at the end of the current turn".to_owned() } Some(hive_agent_sock::Response::Err { message }) => message, Some(other) => format!("compact: unexpected response: {other:?}"), None => "compact failed: in-agent socket unavailable".to_owned(), } }) .await } // IMPORTANT: this tool is only available when the `lifecycle` tool group // is granted to this agent. hive-c0re enforces the topology check // server-side: the call is rejected unless `name` is a direct child. #[tool( description = "Restart a direct child sub-agent container (stop + start). \ Only succeeds if `name` is a direct child of this agent in the topology \ tree — the server enforces this. No approval required. \ Agents holding the `infra_admin` capability may also pass a hive \ infrastructure container name (`hive-ci`, `hive-gateway`, `hive-forge`) \ to restart it directly via the privileged helper." )] async fn restart(&self, Parameters(args): Parameters) -> String { let log = format!("{args:?}"); let name = args.name.clone(); run_tool_envelope("restart", log, async move { let (resp, retries) = self .dispatch(hive_core_agent_sock::Request::Restart { name: args.name }) .await; annotate_retries( format_ack(resp, "restart", format!("restarted {name}")), retries, ) }) .await } // IMPORTANT: this tool is only available when the `lifecycle` tool group // is granted to this agent. hive-c0re enforces the topology check // server-side: the call is rejected unless `name` is a direct child. #[tool(description = "Stop a direct child sub-agent container (graceful). \ Only succeeds if `name` is a direct child of this agent in the topology \ tree — the server enforces this. No approval required. \ State dir is kept; recreating the agent reuses prior config + credentials.")] async fn kill(&self, Parameters(args): Parameters) -> String { let log = format!("{args:?}"); let name = args.name.clone(); run_tool_envelope("kill", log, async move { let (resp, retries) = self .dispatch(hive_core_agent_sock::Request::Kill { name: args.name }) .await; annotate_retries(format_ack(resp, "kill", format!("killed {name}")), retries) }) .await } // IMPORTANT: this tool is only available when the `lifecycle` tool group // is granted to this agent. hive-c0re enforces the topology check // server-side: the call is rejected unless `name` is a direct child. #[tool( description = "Rebuild a direct child sub-agent: re-applies the current hyperhive \ flake + agent.nix and restarts the container. Only succeeds if `name` is a direct \ child of this agent in the topology tree — the server enforces this. \ No approval required. Idempotent — use when a child needs its config reapplied." )] async fn update(&self, Parameters(args): Parameters) -> String { let log = format!("{args:?}"); let name = args.name.clone(); run_tool_envelope("update", log, async move { let (resp, retries) = self .dispatch(hive_core_agent_sock::Request::Update { name: args.name }) .await; annotate_retries( format_ack(resp, "update", format!("updated {name}")), retries, ) }) .await } // IMPORTANT: this tool is only available when the `lifecycle` tool group // is granted to this agent. Returns all topological descendants of the // calling agent with their running status. #[tool( description = "List all containers that are topological descendants of this agent \ (direct children + their subtrees). Requires the `lifecycle` tool group. \ Returns every known descendant regardless of running state — check the `running` \ field to distinguish live from stopped containers. Ordered by topology depth \ (parents before children), then alphabetically within each tier." )] async fn list_containers(&self) -> String { run_tool_envelope("list_containers", String::new(), async move { let (resp, retries) = self .dispatch(hive_core_agent_sock::Request::ListDescendants) .await; let body = match resp { Ok(hive_core_agent_sock::Response::Containers { containers }) => { if containers.is_empty() { "no descendant containers".to_owned() } else { containers .iter() .map(|c| { let status = if c.running { "running" } else { "stopped" }; format!("{} ({})", c.name, status) }) .collect::>() .join("\n") } } other => reply_err(other, "list_containers"), }; annotate_retries(body, retries) }) .await } // IMPORTANT: this tool is capability-gated (`read_host_journal`). // It is added to `--allowedTools` by `allowed_capability_tools` only // when `HIVE_CAPABILITIES` contains `read_host_journal`. hive-c0re // performs a second capability check server-side before running journalctl. #[tool( description = "Fetch recent lines from the host journal (requires `read_host_journal` \ capability). All filters are optional - omit to get the last N host journal lines. \ `unit`: filter to a systemd unit (e.g. `hive-c0re.service`). \ `container`: nspawn machine name verbatim (e.g. `h-iris`). Omit for host journal. \ Agent containers use the `h-` prefix (e.g. `h-iris`, `h-atlas`); \ infrastructure containers use their full name (e.g. `hive-ci`, `hive-forge`, \ `hive-matrix`, `hive-gateway`). \ `lines`: how many lines (default 30, max 100). \ `priority`: minimum syslog level enum. \ `grep`: regex matched against log message fields (journalctl --grep). \ `since`: show entries on or newer than this (e.g. `-1h`, `2024-01-01 12:00:00`). \ `until`: show entries on or older than this." )] async fn get_host_journal(&self, Parameters(args): Parameters) -> String { let log = format!("{args:?}"); run_tool_envelope("get_host_journal", log, async move { let (resp, retries) = self .dispatch(hive_core_agent_sock::Request::GetHostJournal { unit: args.unit, container: args.container, lines: args.lines, priority: args.priority, grep: args.grep, since: args.since, until: args.until, }) .await; let result = match resp { Ok(hive_core_agent_sock::Response::HostJournal { content }) => content, other => reply_err(other, "get_host_journal"), }; annotate_retries(result, retries) }) .await } // IMPORTANT: this tool is only available when the `approvals` tool group // is configured for the agent (`HIVE_TOOL_GROUPS` contains `approvals`). // hive-c0re performs a topology check server-side: only direct children // of the calling agent are accepted; all other names are rejected. #[tool( description = "Create a brand-new direct child agent's config repo and queue an \ `InitConfig` approval for the operator to review. Requires the `approvals` tool \ group. `name` must be a direct child of this agent in the topology tree. Fails if a \ config repo for that child already exists. This tool **creates the repo** and \ nothing else: on approval hive-c0re seeds it with a default `agent.nix`, and \ approving the follow-up `Spawn` creates the container. \ Every config change — the child's first one included — goes through a PR on its \ config repo, made from a clone you take yourself, reviewed + approved by the \ operator. `/agents//config` is a **read-only copy** for reading a config, \ never an editing surface." )] async fn request_init_config( &self, Parameters(args): Parameters, ) -> String { let log = format!("{args:?}"); let name = args.name.clone(); run_tool_envelope("request_init_config", log, async move { let (resp, retries) = self .dispatch(hive_core_agent_sock::Request::RequestInitConfig { name: args.name, description: args.description, }) .await; annotate_retries( format_ack( resp, "request_init_config", format!("init_config approval queued for {name}"), ), retries, ) }) .await } // IMPORTANT: this tool is only available when the `lifecycle` tool group // is granted to this agent. hive-c0re enforces the topology check // server-side: the call is rejected unless `name` is a direct child. #[tool(description = "Start a stopped direct child sub-agent container. \ Only succeeds if `name` is a direct child of this agent in the topology \ tree — the server enforces this. No approval required.")] async fn start(&self, Parameters(args): Parameters) -> String { let log = format!("{args:?}"); let name = args.name.clone(); run_tool_envelope("start", log, async move { let (resp, retries) = self .dispatch(hive_core_agent_sock::Request::Start { name: args.name }) .await; annotate_retries( format_ack(resp, "start", format!("started {name}")), retries, ) }) .await } #[tool( description = "Fetch recent journal log lines for a sub-agent container. Useful \ for diagnosing MCP server registration failures, startup crashes, plugin install \ errors, or any harness issue you can't see from inside the container. Pass the \ plain logical agent name (e.g. `gui`) — hive-c0re resolves the machine name. \ `lines` defaults to 50 (max capped at 500 on the host side)." )] async fn get_logs(&self, Parameters(args): Parameters) -> String { let log = format!("{args:?}"); let agent = args.agent.clone(); run_tool_envelope("get_logs", log, async move { let lines = args.lines.map(|n| n.min(500)); let (resp, retries) = self .dispatch(hive_core_agent_sock::Request::GetLogs { agent: agent.clone(), lines, }) .await; let s = match resp { Ok(hive_core_agent_sock::Response::Logs { content }) => { if content.is_empty() { format!("(no journal output for {agent})") } else { content } } other => reply_err(other, "get_logs"), }; annotate_retries(s, retries) }) .await } #[tool( description = "Queue an approval for the operator to run `nix flake update` on the \ meta flake and commit the resulting lock changes. Pass specific input names to update \ only those inputs (e.g. `[\"bitburner-agent\"]`), or pass an empty list to update ALL \ inputs. Returns immediately — the lock update runs when the operator approves. \ Does NOT trigger container rebuilds — call `update` on each affected agent \ separately after the approval resolves." )] async fn request_update_meta_inputs( &self, Parameters(args): Parameters, ) -> String { let log = format!("{args:?}"); run_tool_envelope("request_update_meta_inputs", log, async move { let label = if args.inputs.is_empty() { "all inputs".to_string() } else { args.inputs.join(", ") }; let (resp, retries) = self .dispatch(hive_core_agent_sock::Request::RequestUpdateMetaInputs { inputs: args.inputs, description: args.description, }) .await; annotate_retries( format_ack( resp, "request_update_meta_inputs", format!("approval queued: {label}"), ), retries, ) }) .await } #[tool( description = "Queue an approval to add a scheduled prompt — one body delivered to \ N agent inboxes at a target time, optionally recurring every `interval_seconds`. \ The operator approves; on approve hive-c0re inserts the schedule and the worker \ fans it out. Even self-targeted schedules go through this flow (the operator pays \ for the wake-up tokens); the existing `remind` MCP tool stays the quick \ no-approval self-wake path. \n\n\ Catch-up clamp: if hive-c0re is down across multiple intervals, only ONE delayed \ fire happens on resume (per recurring schedule). The skipped-cycle count surfaces \ in the per-target `last_result` for the operator's audit trail. \n\n\ Per-target failure: a target name that doesn't resolve to a live agent → operator \ gets a one-line advisory `Message` from `system`; the schedule keeps firing for \ the other (live) targets." )] async fn request_schedule_prompt( &self, Parameters(args): Parameters, ) -> String { let log = format!("{args:?}"); run_tool_envelope("request_schedule_prompt", log, async move { let target_count = args.targets.len(); let (resp, retries) = self .dispatch(hive_core_agent_sock::Request::RequestSchedulePrompt( hive_sh4re::SchedulePromptPayload { targets: args.targets, body: args.body, first_fire_at_unix: args.first_fire_at_unix, interval_seconds: args.interval_seconds, description: args.description, }, )) .await; annotate_retries( format_ack( resp, "request_schedule_prompt", format!("approval queued: {target_count} target(s)"), ), retries, ) }) .await } #[tool( description = "Fire a scheduled prompt out of band — runs the per-target fan-out \ once immediately without disturbing the schedule's cadence. Recurring schedules \ keep their next_fire_at unchanged (the manual fire is additive). One-shot \ schedules are CONSUMED by the manual fire (cancelled afterwards): the operator's \ intent on a one-shot is 'send this now, the scheduled time was wrong'. \n\n\ Authorization mirrors `cancel_schedule`: you can fire your own schedules + any \ owned by a sub-agent in your subtree per topology.json." )] async fn fire_schedule_now(&self, Parameters(args): Parameters) -> String { let log = format!("{args:?}"); run_tool_envelope("fire_schedule_now", log, async move { let id = args.id; let (resp, retries) = self .dispatch(hive_core_agent_sock::Request::FireScheduleNow { id }) .await; annotate_retries( format_ack(resp, "fire_schedule_now", format!("fired #{id} now")), retries, ) }) .await } #[tool( description = "Cancel a scheduled prompt. With no `targets` field, cancels the \ whole schedule (all recipients flipped). With a non-empty `targets` list, cancels \ just those recipients; the schedule keeps firing for any remaining active targets \ and auto-cancels its parent row when every target is cancelled. \n\n\ Authorization: the manager can cancel its own schedules + any schedule whose \ owner is one of its sub-agents per topology.json. Other owners are refused." )] async fn cancel_schedule(&self, Parameters(args): Parameters) -> String { let log = format!("{args:?}"); run_tool_envelope("cancel_schedule", log, async move { let id = args.id; let (resp, retries) = self .dispatch(hive_core_agent_sock::Request::CancelSchedule { id: args.id, targets: args.targets, }) .await; annotate_retries( format_ack(resp, "cancel_schedule", format!("cancelled #{id}")), retries, ) }) .await } #[tool( description = "Edit an existing scheduled prompt's mutable fields. Pass only \ the fields you want to change — anything omitted keeps its current value. Editable: \ `body`, `description`, `interval_seconds` (positive only via this tool; flipping \ recurring→one-shot is operator-only via the dashboard), `next_fire_at_unix`, and \ the target set via `targets_add` / `targets_remove`. Both target lists are \ applied in the same transaction with removes-before-adds, so a single edit can \ swap a target atomically. Re-adding a previously-removed target starts a fresh \ per-target history (drops the tombstone). Draining all targets auto-cancels the \ parent schedule. \n\n\ Authorization mirrors `cancel_schedule` / `fire_schedule_now`: you can edit your \ own schedules + any owned by a sub-agent in your subtree per topology.json. \ Refuses cancelled schedules (the row's terminal — submit a fresh one)." )] async fn edit_schedule(&self, Parameters(args): Parameters) -> String { let log = format!("{args:?}"); run_tool_envelope("edit_schedule", log, async move { let id = args.id; let (resp, retries) = self .dispatch(hive_core_agent_sock::Request::EditSchedule { id: args.id, body: args.body, description: args.description.map(Some), interval_seconds: args.interval_seconds.map(Some), next_fire_at_unix: args.next_fire_at_unix, targets_add: args.targets_add, targets_remove: args.targets_remove, }) .await; annotate_retries( format_ack(resp, "edit_schedule", format!("edited #{id}")), retries, ) }) .await } #[tool( description = "List every scheduled prompt in the queue (active + cancelled but \ not yet reaped). Returns the full snapshot — schedule id, owner, body, target set \ with per-target last_fired_at + last_result, next fire time, recurring interval. \ Use this to look up an id before calling `cancel_schedule`, or to audit what \ the swarm is going to be woken up about next." )] async fn list_schedules(&self) -> String { run_tool_envelope("list_schedules", String::new(), async move { let (resp, retries) = self .dispatch(hive_core_agent_sock::Request::ListSchedules) .await; let body = match resp { Ok(hive_core_agent_sock::Response::Schedules { schedules }) => { serde_json::to_string(&schedules) .unwrap_or_else(|e| format!("list_schedules: serialise: {e:#}")) } other => reply_err(other, "list_schedules"), }; annotate_retries(body, retries) }) .await } } #[tool_handler( instructions = "You are a hyperhive agent. Use `send` to talk to peers (by their logical \ name) or to the operator (recipient `operator`). Use `recv` to drain your inbox one \ message at a time. Use `remind` to schedule a future wake-up message for yourself." )] impl ServerHandler for AgentServer {} /// Run the MCP server over HTTP (rmcp streamable-http transport) on `addr`. /// /// This is the sole transport for the built-in hyperhive surface. It runs 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 /// an agent when the async `initialize`/`tools/list` loses to claude's first /// tool call. `socket` is the hyperhive control socket every tool call dials /// fresh (the handler holds only the path), so a host-side hive-c0re restart is /// transparent — the next call just reconnects. /// /// Binds loopback only in practice; the default `allowed_hosts` /// (`localhost`/`127.0.0.1`/`::1`) rejects Host headers from anywhere else. /// /// # Errors /// /// Returns an error if the listener cannot bind `addr` or the HTTP server /// exits with a fatal error. pub async fn serve_http(socket: PathBuf, addr: std::net::SocketAddr) -> Result<()> { use rmcp::transport::streamable_http_server::{ StreamableHttpServerConfig, StreamableHttpService, session::local::LocalSessionManager, }; let session_manager = std::sync::Arc::new(LocalSessionManager::default()); let service = StreamableHttpService::new( move || Ok(AgentServer::new(socket.clone())), session_manager, StreamableHttpServerConfig::default(), ); let app = axum::Router::new().nest_service("/mcp", service); let listener = tokio::net::TcpListener::bind(addr).await?; tracing::info!(%addr, "serving hyperhive MCP over streamable-http at /mcp"); axum::serve(listener, app).await?; Ok(()) }