From c634dab2a6f9bd979d021d2675949c61f9f3115e Mon Sep 17 00:00:00 2001 From: damocles Date: Thu, 4 Jun 2026 00:07:07 +0200 Subject: [PATCH] refactor: unify AgentServer + ManagerServer into HiveServer {socket, flavor} --- docs/conventions.md | 8 +- hive-ag3nt/src/mcp.rs | 1074 +++++++++++---------------------- hive-c0re/src/agent_server.rs | 23 + 3 files changed, 379 insertions(+), 726 deletions(-) diff --git a/docs/conventions.md b/docs/conventions.md index 1a7dcb65..ba6714ed 100644 --- a/docs/conventions.md +++ b/docs/conventions.md @@ -311,10 +311,10 @@ from `tool-groups.json`). Unrecognised tokens are logged and skipped. Falls back to `ToolGroup::AGENT_DEFAULT` (`messaging`, `meta`, `inbox`, `execution`) or `ToolGroup::MANAGER_DEFAULT` (all groups) when the var is absent or empty. -**Updating the surface** — when a new `#[tool]` fn is added to `AgentServer` -or `ManagerServer` in `hive-ag3nt/src/mcp.rs`, add its name to the matching -`ToolGroup::tools()` slice in `hive-sh4re/src/lib.rs`. That's the single -source of truth; `allowed_mcp_tools` reads it at session start. +**Updating the surface** — when a new `#[tool]` fn is added to `HiveServer` +in `hive-ag3nt/src/mcp.rs`, add its name to the matching `ToolGroup::tools()` +slice in `hive-sh4re/src/lib.rs`. That's the single source of truth; +`allowed_mcp_tools` reads it at session start. ## Capabilities diff --git a/hive-ag3nt/src/mcp.rs b/hive-ag3nt/src/mcp.rs index 6e97f76c..3751aa3c 100644 --- a/hive-ag3nt/src/mcp.rs +++ b/hive-ag3nt/src/mcp.rs @@ -8,13 +8,13 @@ //! broker-routed protocol. Unaffected by this module. //! - **MCP stdio** owned by this module — what claude actually speaks. //! -//! Two server flavors: -//! - `AgentServer` — sub-agent tools (`send`, `recv`). -//! - `ManagerServer` — agent tools + lifecycle (`kill`, -//! `request_init_config`, `request_apply_commit`). +//! One `HiveServer { socket, flavor }` with two public type aliases: +//! - `AgentServer` — sub-agent flavor; restricted send allow-list. +//! - `ManagerServer` — manager flavor; manager-only tools unlocked. //! -//! Both go through the same `run_tool_envelope` helper so logging + status -//! line stay uniform. +//! All tools live in one `#[tool_router] impl HiveServer`; manager-only +//! tools guard via `require_manager()`. Both go through the same +//! `run_tool_envelope` helper so logging + status line stay uniform. use std::future::Future; use std::path::PathBuf; @@ -27,9 +27,8 @@ use rmcp::{ use crate::client; -/// Wire-protocol-agnostic view of a hyperhive socket response. Both -/// `AgentResponse` and `ManagerResponse` convert into this so the tool -/// formatters can be shared between `AgentServer` and `ManagerServer`. +/// Wire-protocol-agnostic view of a hyperhive socket response. Both flavors +/// of `HiveServer` convert into this so the tool formatters can be shared. #[derive(Debug)] pub enum SocketReply { Ok, @@ -106,7 +105,7 @@ impl From for SocketReply { } /// Write (or remove) the status file in the agent's own `state/` directory. -/// Called by both `AgentServer::set_status` and `ManagerServer::set_status` +/// Called by `HiveServer::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 @@ -572,31 +571,57 @@ pub struct RemindArgs { pub file_path: Option, } -/// Per-agent tool surface. Holds the socket path so each tool call doesn't -/// re-derive it; the socket itself is the per-container `/run/hive/mcp.sock`. +/// Unified MCP tool surface for both sub-agent and manager roles. +/// +/// `AgentRequest = ManagerRequest = Request` and `AgentResponse = +/// ManagerResponse = Response` are type aliases in hive-sh4re, so a single +/// `dispatch` call covers both sockets — the only real difference is which +/// socket path is used and which tools the flavor enables. +/// +/// Manager-only tools return a clear error when called from an agent context. +/// In practice the `--allowedTools` gate prevents that — this is belt-and-suspenders. #[derive(Debug, Clone)] -pub struct AgentServer { +pub struct HiveServer { socket: PathBuf, + flavor: Flavor, } -impl AgentServer { +// Keep the old names as type aliases so any external code that references them +// still compiles without changes. +pub type AgentServer = HiveServer; +pub type ManagerServer = HiveServer; + +impl HiveServer { #[must_use] - pub fn new(socket: PathBuf) -> Self { - Self { socket } + pub fn new(socket: PathBuf, flavor: Flavor) -> Self { + Self { socket, flavor } } - /// Issue any `AgentRequest` through the retry-aware client and pull + /// 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`). + /// + /// `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::AgentRequest, + req: hive_sh4re::Request, ) -> (Result, u32) { - match client::request_retried::<_, hive_sh4re::AgentResponse>(&self.socket, &req).await { + match client::request_retried::<_, hive_sh4re::Response>(&self.socket, &req).await { Ok((r, n)) => (Ok(SocketReply::from(r)), n), Err(e) => (Err(e), 0), } } + + /// Returns an error string when called from a non-manager flavor. + /// Manager-only tool methods call this first and early-return on Some. + fn require_manager(&self) -> Option { + if !matches!(self.flavor, Flavor::Manager) { + Some("error: this tool is only available to manager agents".to_owned()) + } else { + None + } + } } // IMPORTANT: when adding a new `#[tool]` fn to this impl, also add @@ -604,7 +629,7 @@ impl AgentServer { // Claude Code's permission gate refuses uninlisted MCP tools in // non-interactive `--print` mode with "permissions not granted yet". #[tool_router] -impl AgentServer { +impl HiveServer { #[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." @@ -612,12 +637,16 @@ impl AgentServer { async fn send(&self, Parameters(args): Parameters) -> String { let log = format!("{args:?}"); let to = args.to.clone(); - if let Err(refusal) = check_send_allowed(&to) { - return run_tool_envelope("send", log, async move { refusal }).await; + // Agent flavor: check per-agent allow-list (hyperhive.allowedRecipients). + // Manager flavor: unrestricted send. + if matches!(self.flavor, Flavor::Agent) { + if let Err(refusal) = 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_sh4re::AgentRequest::Send { + .dispatch(hive_sh4re::Request::Send { to: args.to, body: args.body, in_reply_to: args.in_reply_to, @@ -648,7 +677,7 @@ impl AgentServer { let log = format!("{args:?}"); run_tool_envelope("ask", log, async move { let (resp, retries) = self - .dispatch(hive_sh4re::AgentRequest::Ask { + .dispatch(hive_sh4re::Request::Ask { question: args.question, options: args.options, multi: args.multi, @@ -684,7 +713,7 @@ impl AgentServer { let id = args.id; run_tool_envelope("answer", log, async move { let (resp, retries) = self - .dispatch(hive_sh4re::AgentRequest::Answer { + .dispatch(hive_sh4re::Request::Answer { id, answer: args.answer, }) @@ -718,7 +747,7 @@ impl AgentServer { let log = format!("{args:?}"); run_tool_envelope("recv", log, async move { let (resp, retries) = self - .dispatch(hive_sh4re::AgentRequest::Recv { + .dispatch(hive_sh4re::Request::Recv { wait_seconds: args.wait_seconds, max: args.max, }) @@ -746,7 +775,7 @@ impl AgentServer { run_tool_envelope("get_loose_ends", String::new(), async move { let is_self_query = args.agent.is_none(); let (resp, retries) = self - .dispatch(hive_sh4re::AgentRequest::GetLooseEnds { agent: args.agent }) + .dispatch(hive_sh4re::Request::GetLooseEnds { agent: args.agent }) .await; // Extract the vec so we can augment before rendering. let mut loose_ends = match resp { @@ -815,7 +844,7 @@ impl AgentServer { return e; } let (resp, retries) = self - .dispatch(hive_sh4re::AgentRequest::SetStatus { text: args.text }) + .dispatch(hive_sh4re::Request::SetStatus { text: args.text }) .await; annotate_retries( format_ack(resp, "set_status", "status updated".to_owned()), @@ -840,7 +869,7 @@ impl AgentServer { let log = args.name.clone().unwrap_or_else(|| "".to_owned()); run_tool_envelope("get_agent_meta", log, async move { let (resp, retries) = self - .dispatch(hive_sh4re::AgentRequest::GetAgentMeta { name: args.name }) + .dispatch(hive_sh4re::Request::GetAgentMeta { name: args.name }) .await; annotate_retries(format_agent_meta(resp), retries) }) @@ -865,7 +894,7 @@ impl AgentServer { }; let kind_label = loose_end_kind_label(kind); let (resp, retries) = self - .dispatch(hive_sh4re::AgentRequest::CancelLooseEnd { kind, id }) + .dispatch(hive_sh4re::Request::CancelLooseEnd { kind, id }) .await; annotate_retries( format_ack( @@ -908,7 +937,7 @@ impl AgentServer { (None, Some(t)) => hive_sh4re::ReminderTiming::At { unix_timestamp: t }, }; let (resp, retries) = self - .dispatch(hive_sh4re::AgentRequest::Remind { + .dispatch(hive_sh4re::Request::Remind { message: args.message, timing, file_path: args.file_path, @@ -959,7 +988,7 @@ impl AgentServer { let name = args.name.clone(); run_tool_envelope("restart", log, async move { let (resp, retries) = self - .dispatch(hive_sh4re::AgentRequest::Restart { name: args.name }) + .dispatch(hive_sh4re::Request::Restart { name: args.name }) .await; annotate_retries( format_ack(resp, "restart", format!("restarted {name}")), @@ -983,7 +1012,7 @@ impl AgentServer { let name = args.name.clone(); run_tool_envelope("kill", log, async move { let (resp, retries) = self - .dispatch(hive_sh4re::AgentRequest::Kill { name: args.name }) + .dispatch(hive_sh4re::Request::Kill { name: args.name }) .await; annotate_retries(format_ack(resp, "kill", format!("killed {name}")), retries) }) @@ -1004,7 +1033,7 @@ impl AgentServer { let name = args.name.clone(); run_tool_envelope("update", log, async move { let (resp, retries) = self - .dispatch(hive_sh4re::AgentRequest::Update { name: args.name }) + .dispatch(hive_sh4re::Request::Update { name: args.name }) .await; annotate_retries(format_ack(resp, "update", format!("updated {name}")), retries) }) @@ -1024,7 +1053,7 @@ impl AgentServer { async fn list_containers(&self) -> String { run_tool_envelope("list_containers", String::new(), async move { let (resp, retries) = self - .dispatch(hive_sh4re::AgentRequest::ListDescendants) + .dispatch(hive_sh4re::Request::ListDescendants) .await; let body = match resp { Ok(SocketReply::Containers(containers)) => { @@ -1069,7 +1098,7 @@ impl AgentServer { let log = format!("{args:?}"); run_tool_envelope("get_host_journal", log, async move { let (resp, retries) = self - .dispatch(hive_sh4re::AgentRequest::GetHostJournal { + .dispatch(hive_sh4re::Request::GetHostJournal { unit: args.unit, container: args.container, lines: args.lines, @@ -1111,7 +1140,7 @@ impl AgentServer { let name = args.name.clone(); run_tool_envelope("request_init_config", log, async move { let (resp, retries) = self - .dispatch(hive_sh4re::AgentRequest::RequestInitConfig { + .dispatch(hive_sh4re::Request::RequestInitConfig { name: args.name, description: args.description, }) @@ -1149,7 +1178,7 @@ impl AgentServer { let commit_ref = args.commit_ref.clone(); run_tool_envelope("request_apply_commit", log, async move { let (resp, retries) = self - .dispatch(hive_sh4re::AgentRequest::RequestApplyCommit { + .dispatch(hive_sh4re::Request::RequestApplyCommit { agent: args.agent, commit_ref: args.commit_ref, description: args.description, @@ -1166,6 +1195,280 @@ impl AgentServer { }) .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_sh4re::Request::Start { name: args.name }) + .await; + annotate_retries(format_ack(resp, "start", format!("started {name}")), retries) + }) + .await + } + + // ------------------------------------------------------------------------- + // Manager-only tools — guarded by require_manager(). The `--allowedTools` + // gate prevents agents from ever calling these in practice; the guard is + // belt-and-suspenders in case the gate is misconfigured. + // ------------------------------------------------------------------------- + + #[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 { + if let Some(e) = self.require_manager() { + return e; + } + 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_sh4re::Request::GetLogs { + agent: agent.clone(), + lines, + }) + .await; + let s = match resp { + Ok(SocketReply::Logs(content)) => { + if content.is_empty() { + format!("(no journal output for {agent})") + } else { + content + } + } + Ok(SocketReply::Err(m)) => format!("get_logs failed: {m}"), + Ok(other) => format!("get_logs unexpected response: {other:?}"), + Err(e) => format!("get_logs transport error: {e:#}"), + }; + 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 { + if let Some(e) = self.require_manager() { + return e; + } + 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_sh4re::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 { + if let Some(e) = self.require_manager() { + return e; + } + 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_sh4re::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 { + if let Some(e) = self.require_manager() { + return e; + } + let log = format!("{args:?}"); + run_tool_envelope("fire_schedule_now", log, async move { + let id = args.id; + let (resp, retries) = self + .dispatch(hive_sh4re::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 { + if let Some(e) = self.require_manager() { + return e; + } + let log = format!("{args:?}"); + run_tool_envelope("cancel_schedule", log, async move { + let id = args.id; + let (resp, retries) = self + .dispatch(hive_sh4re::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 { + if let Some(e) = self.require_manager() { + return e; + } + let log = format!("{args:?}"); + run_tool_envelope("edit_schedule", log, async move { + let id = args.id; + let (resp, retries) = self + .dispatch(hive_sh4re::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 { + if let Some(e) = self.require_manager() { + return e; + } + 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) + .unwrap_or_else(|e| format!("list_schedules: serialise: {e:#}")), + Ok(SocketReply::Err(m)) => format!("list_schedules: {m}"), + Ok(other) => format!("list_schedules unexpected response: {other:?}"), + Err(e) => format!("list_schedules transport error: {e:#}"), + }; + annotate_retries(body, retries) + }) + .await + } } #[tool_handler( @@ -1173,29 +1476,29 @@ impl AgentServer { 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 {} +impl ServerHandler for HiveServer {} -/// Run the agent MCP server over stdio. Returns when the client disconnects. +/// Run an MCP server over stdio for the given flavor. 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_agent_stdio(socket: PathBuf) -> Result<()> { - let server = AgentServer::new(socket); +pub async fn serve_stdio(socket: PathBuf, flavor: Flavor) -> Result<()> { + let server = HiveServer::new(socket, flavor); let service = server.serve(stdio()).await?; service.waiting().await?; Ok(()) } -/// Run the manager MCP server over stdio. Same idea, different tool surface. -/// -/// # Errors -/// -/// Returns an error if the MCP server fails to initialize or the transport -/// encounters a fatal error. +/// Convenience wrapper: run the agent MCP server over stdio. +pub async fn serve_agent_stdio(socket: PathBuf) -> Result<()> { + serve_stdio(socket, Flavor::Agent).await +} + +/// Convenience wrapper: run the manager MCP server over stdio. pub async fn serve_manager_stdio(socket: PathBuf) -> Result<()> { - let server = ManagerServer::new(socket); + let server = HiveServer::new(socket, Flavor::Manager); let service = server.serve(stdio()).await?; service.waiting().await?; Ok(()) @@ -1482,679 +1785,6 @@ pub struct GetHostJournalArgs { pub until: Option, } -#[derive(Debug, Clone)] -pub struct ManagerServer { - socket: PathBuf, -} - -impl ManagerServer { - #[must_use] - pub fn new(socket: PathBuf) -> Self { - Self { socket } - } - - /// Helper: issue any `ManagerRequest` through the retry-aware - /// client, convert the reply through `SocketReply`, and return the - /// retry count alongside so the tool handler can `annotate_retries` - /// on the final string. - async fn dispatch( - &self, - req: hive_sh4re::ManagerRequest, - ) -> (Result, u32) { - match client::request_retried::<_, hive_sh4re::ManagerResponse>(&self.socket, &req).await { - Ok((r, n)) => (Ok(SocketReply::from(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 ManagerServer { - #[tool( - description = "Send a message to a sub-agent (by logical name), to another agent, \ - or to the operator (recipient `operator`, surfaces in the dashboard)." - )] - async fn send(&self, Parameters(args): Parameters) -> String { - let log = format!("{args:?}"); - let to = args.to.clone(); - run_tool_envelope("send", log, async move { - let (resp, retries) = self - .dispatch(hive_sh4re::ManagerRequest::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 = "Pop messages from the manager inbox. Default returns one (sender + \ - body) or empty. Without `wait_seconds` (or 0) returns immediately — a cheap inbox \ - peek. Pass a positive value (capped at 180) to park until either a message arrives \ - or the timeout fires; prefer a long wait (120 or 180) over ending a turn early \ - when you have nothing else to do. \n\n\ - Pass `max: N` (capped at 32) to drain up to N messages in one round-trip — useful \ - when the wake prompt tells you the inbox has more queued. `wait_seconds` still \ - applies to the FIRST message; once one lands the call drains up to `max` in total." - )] - 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_sh4re::ManagerRequest::Recv { - wait_seconds: args.wait_seconds, - max: args.max, - }) - .await; - annotate_retries(format_recv(resp), retries) - }) - .await - } - - #[tool( - description = "Step 1 of 2 for creating a new agent: initialise the proposed config \ - repo and queue an InitConfig approval. On operator approval hive-c0re seeds \ - `/agents//config/agent.nix` with the default template so the manager can \ - customise it before spawning. After the ConfigReady helper event arrives, edit \ - agent.nix, commit the changes, then call `request_apply_commit` with the commit \ - sha — that's what creates the container. Fails if a config repo for this name \ - already exists (use `request_apply_commit` directly to update an existing agent)." - )] - 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_sh4re::ManagerRequest::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 - } - - #[tool( - description = "Stop a sub-agent container (graceful). The state dir is kept; \ - recreating reuses prior config + Claude credentials. No approval required." - )] - 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_sh4re::ManagerRequest::Kill { name: args.name }) - .await; - annotate_retries(format_ack(resp, "kill", format!("killed {name}")), retries) - }) - .await - } - - #[tool( - description = "Start a stopped sub-agent container. No approval required — \ - lifecycle ops on existing containers are at the manager's discretion." - )] - 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_sh4re::ManagerRequest::Start { name: args.name }) - .await; - annotate_retries( - format_ack(resp, "start", format!("started {name}")), - retries, - ) - }) - .await - } - - #[tool(description = "Restart a sub-agent container (stop + start). No approval required.")] - 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_sh4re::ManagerRequest::Restart { name: args.name }) - .await; - annotate_retries( - format_ack(resp, "restart", format!("restarted {name}")), - retries, - ) - }) - .await - } - - #[tool( - description = "Rebuild a sub-agent: re-applies the current hyperhive flake + agent.nix \ - and restarts the container. No approval required — idempotent. Use when you receive a \ - `needs_update` system event for an agent." - )] - 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_sh4re::ManagerRequest::Update { name: args.name }) - .await; - annotate_retries( - format_ack(resp, "update", format!("updated {name}")), - 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_sh4re::ManagerRequest::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_sh4re::ManagerRequest::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_sh4re::ManagerRequest::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_sh4re::ManagerRequest::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_sh4re::ManagerRequest::EditSchedule { - id: args.id, - body: args.body, - // The agent-side args use plain Option; the - // manager wire type's `Some(None)` ("set to - // null") cases stay operator-exclusive, so we - // promote agent-supplied values into - // `Some(Some(v))` and omit when the agent - // didn't pass a value. - 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_sh4re::ManagerRequest::ListSchedules) - .await; - let body = match resp { - Ok(SocketReply::Schedules(schedules)) => serde_json::to_string(&schedules) - .unwrap_or_else(|e| format!("list_schedules: serialise: {e:#}")), - Ok(SocketReply::Err(m)) => format!("list_schedules: {m}"), - Ok(other) => format!("list_schedules unexpected response: {other:?}"), - Err(e) => format!("list_schedules transport error: {e:#}"), - }; - annotate_retries(body, retries) - }) - .await - } - - #[tool( - description = "Surface a structured question to either the operator OR a sub-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 sub-agent — they receive a \ - `question_asked` event in their inbox and answer via their `mcp__hyperhive__answer` \ - tool. Useful for delegating decisions / clarifications without losing the \ - question id correlation. \n\n\ - `options` is advisory: pass a short fixed-choice list when applicable, otherwise \ - leave empty for free text. Set `multi: true` to render checkboxes; the answer \ - comes back as a comma-separated string. Set `ttl_seconds` to auto-cancel — 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 (resp, retries) = self - .dispatch(hive_sh4re::ManagerRequest::Ask { - question: args.question, - options: args.options, - multi: args.multi, - ttl_seconds: args.ttl_seconds, - to: args.to, - }) - .await; - let s = match resp { - Ok(SocketReply::QuestionQueued(id)) => format!( - "question queued (id={id}); answer will arrive as a system \ - `question_answered` event in your inbox" - ), - Ok(SocketReply::Err(m)) => format!("ask failed: {m}"), - Ok(other) => format!("ask unexpected response: {other:?}"), - Err(e) => format!("ask transport error: {e:#}"), - }; - annotate_retries(s, retries) - }) - .await - } - - #[tool( - description = "Answer a question that was routed to the manager via a `question_asked` \ - system event in the manager's inbox (i.e. a sub-agent did `ask(to: \"manager\", \ - ...)`). Pass the `id` from the event and your `answer`. The answer surfaces in the \ - asker's inbox as a `question_answered` event." - )] - 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_sh4re::ManagerRequest::Answer { - id, - answer: args.answer, - }) - .await; - annotate_retries( - format_ack(resp, "answer", format!("answered question {id}")), - retries, - ) - }) - .await - } - - #[tool( - description = "Submit a config change for operator approval. Pass the agent name \ - (e.g. `alice`) and a commit sha (7-40 hex \ - chars, full or short) in that agent's proposed config repo — a branch/tag name like \ - `main` is rejected, the approval pins the exact commit. On approval hive-c0re \ - rebuilds the container." - )] - async fn request_apply_commit( - &self, - Parameters(args): Parameters, - ) -> String { - let log = format!("{args:?}"); - let agent = args.agent.clone(); - let commit_ref = args.commit_ref.clone(); - run_tool_envelope("request_apply_commit", log, async move { - let (resp, retries) = self - .dispatch(hive_sh4re::ManagerRequest::RequestApplyCommit { - agent: args.agent, - commit_ref: args.commit_ref, - description: args.description, - }) - .await; - annotate_retries( - format_ack( - resp, - "request_apply_commit", - format!("apply approval queued for {agent} @ {commit_ref}"), - ), - retries, - ) - }) - .await - } - - #[tool( - description = "Schedule a reminder that lands in the manager's own inbox at a future \ - time (sender will appear as `reminder`). Use for self-paced manager follow-ups: \ - 'recheck pending approval in 10m', 'nudge alice if she hasn't replied by 14:00 \ - UTC'. 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 `/state/reminders/` \ - (the manager's own state mount) and the inbox message becomes a short pointer. \ - Pass `file_path` if you want to control the destination yourself." - )] - 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 }, - }; - let (resp, retries) = self - .dispatch(hive_sh4re::ManagerRequest::Remind { - message: args.message, - timing, - file_path: args.file_path, - }) - .await; - annotate_retries( - format_ack(resp, "remind", "reminder scheduled".to_string()), - retries, - ) - }) - .await - } - - #[tool( - description = "List loose ends. By default returns your OWN — the manager's: \ - pending approvals you submitted + unanswered questions where you are \ - asker/target + your own pending reminders. Pass `agent: \"*\"` for a \ - hive-wide scan (EVERY pending approval, unanswered question, and reminder \ - across the swarm) — use it to spot stalled coordination, e.g. questions \ - sub-agents asked each other that nobody's answering. Pass `agent: \ - \"\"` to inspect one agent's threads. Cancel any question or reminder \ - row via `cancel_loose_end` (manager bypasses the owner check)." - )] - async fn get_loose_ends(&self, Parameters(args): Parameters) -> String { - run_tool_envelope("get_loose_ends", String::new(), async move { - let (resp, retries) = self - .dispatch(hive_sh4re::ManagerRequest::GetLooseEnds { agent: args.agent }) - .await; - annotate_retries(format_loose_ends(resp), 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. \ - Pass an empty string to clear. 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_sh4re::ManagerRequest::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, and the target's self-reported `status` text (set via \ - `set_status`) plus how long ago it was set. Pass `name` to query a sub-agent or \ - peer manager; omit `name` for the manager's own identity stamp — useful for \ - boot announcements, state-file headers, or cross-agent attribution that won't \ - drift across renames. Status reads `` when the target has never called \ - `set_status` or has cleared it." - )] - 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 (resp, retries) = self - .dispatch(hive_sh4re::ManagerRequest::GetAgentMeta { name: args.name }) - .await; - annotate_retries(format_agent_meta(resp), retries) - }) - .await - } - - #[tool( - description = "Cancel any open thread in the swarm — a `question` (cancels \ - with the operator-override sentinel so the asker unblocks), a `reminder` \ - (hard-deleted before fire), or an `approval` (withdraws a pending approval \ - you submitted; the dashboard pulls the card from pending and the row resolves \ - as `cancelled` instead of approved/denied/failed). `kind` is \ - `\"question\"`, `\"reminder\"`, or `\"approval\"`; `id` is the row id from \ - `get_loose_ends` or the original submission reply. Manager surface bypasses \ - the owner check on the sub-agent flavour — use for hive-wide cleanup of \ - stuck or stale threads, or to drop your own approvals that got superseded." - )] - 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 { - let kind = match parse_loose_end_kind(&args.kind) { - Ok(k) => k, - Err(e) => return e, - }; - let kind_label = loose_end_kind_label(kind); - let (resp, retries) = self - .dispatch(hive_sh4re::ManagerRequest::CancelLooseEnd { kind, id }) - .await; - annotate_retries( - format_ack( - resp, - "cancel_loose_end", - format!("cancelled {kind_label} {id}"), - ), - 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_sh4re::ManagerRequest::GetLogs { - agent: agent.clone(), - lines, - }) - .await; - let s = match resp { - Ok(SocketReply::Logs(content)) => { - if content.is_empty() { - format!("(no journal output for {agent})") - } else { - content - } - } - Ok(SocketReply::Err(m)) => format!("get_logs failed: {m}"), - Ok(other) => format!("get_logs unexpected response: {other:?}"), - Err(e) => format!("get_logs transport error: {e:#}"), - }; - annotate_retries(s, retries) - }) - .await - } -} - -#[tool_handler( - instructions = "You are the hyperhive manager (root). You coordinate sub-agents and \ - relay between them and the operator. Use `send` to talk to agents/operator, `recv` \ - to drain your inbox. Privileged: `request_init_config` (step 1 of new-agent \ - creation — seeds the proposed config repo so you can customise agent.nix; \ - operator-approved), `kill` (graceful stop), `request_apply_commit` (config \ - change for any agent including yourself — also doubles as step 2 of new-agent \ - creation: the first ApplyCommit on a freshly-init'd config creates the \ - container), `ask` (structured question to the operator or a \ - sub-agent — non-blocking, answer arrives later as a `question_answered` event), \ - `answer` (respond to a `question_asked` event directed at you), \ - `get_loose_ends` (hive-wide loose ends — pending approvals + unanswered \ - questions + pending reminders across the swarm), `cancel_loose_end` (cancel any \ - question or reminder row by id), `set_status` / `get_agent_meta` (publish your \ - own status text + query identity/status of any agent — `get_agent_meta` with \ - no arg replaces the old `whoami` self-introspection)." -)] -impl ServerHandler for ManagerServer {} - /// Name of the hyperhive MCP server inside claude's view. Claude prefixes /// tools as `mcp____` (e.g. `mcp__hyperhive__send`). pub const SERVER_NAME: &str = "hyperhive"; diff --git a/hive-c0re/src/agent_server.rs b/hive-c0re/src/agent_server.rs index ba653733..7d44ea27 100644 --- a/hive-c0re/src/agent_server.rs +++ b/hive-c0re/src/agent_server.rs @@ -345,6 +345,29 @@ async fn dispatch(req: &AgentRequest, agent: &str, coord: &Arc) -> Err(message) => AgentResponse::Err { message }, } } + AgentRequest::Start { name } => { + if !crate::topology::children_of(agent) + .iter() + .any(|c| c == name) + { + return AgentResponse::Err { + message: format!( + "agent `{agent}` cannot start `{name}`: \ + not a direct child in the topology tree" + ), + }; + } + tracing::info!(%agent, %name, "agent: start child"); + match crate::lifecycle::start(name).await { + Ok(()) => { + coord.kick_agent(name, "container started"); + AgentResponse::Ok + } + Err(e) => AgentResponse::Err { + message: format!("{e:#}"), + }, + } + } AgentRequest::Restart { name } => { // Topology check: the caller must be the direct parent of the // target. This is the only authorisation criterion — no