diff --git a/docs/conventions.md b/docs/conventions.md index ba6714ed..1a7dcb65 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 `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. +**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. ## Capabilities diff --git a/hive-ag3nt/src/bin/hive.rs b/hive-ag3nt/src/bin/hive.rs index bcd7c71e..b880bb11 100644 --- a/hive-ag3nt/src/bin/hive.rs +++ b/hive-ag3nt/src/bin/hive.rs @@ -104,7 +104,7 @@ async fn main() -> Result<()> { serve_main::(&cli.socket, poll_ms).await } (Role::Agent, Cmd::Mcp) => mcp::serve_agent_stdio(cli.socket).await, - (Role::Manager, Cmd::Mcp) => mcp::serve_agent_stdio(cli.socket).await, + (Role::Manager, Cmd::Mcp) => mcp::serve_manager_stdio(cli.socket).await, (Role::Agent, Cmd::Wake { from, body }) => { wake::(&cli.socket, from, body).await } diff --git a/hive-ag3nt/src/mcp.rs b/hive-ag3nt/src/mcp.rs index dee9bee4..6e97f76c 100644 --- a/hive-ag3nt/src/mcp.rs +++ b/hive-ag3nt/src/mcp.rs @@ -8,10 +8,13 @@ //! broker-routed protocol. Unaffected by this module. //! - **MCP stdio** 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. +//! Two server flavors: +//! - `AgentServer` — sub-agent tools (`send`, `recv`). +//! - `ManagerServer` — agent tools + lifecycle (`kill`, +//! `request_init_config`, `request_apply_commit`). +//! +//! Both go through the same `run_tool_envelope` helper so logging + status +//! line stay uniform. use std::future::Future; use std::path::PathBuf; @@ -24,8 +27,9 @@ use rmcp::{ use crate::client; -/// Wire-protocol-agnostic view of a hyperhive socket response. Both flavors -/// of `AgentServer` convert into this so the tool formatters can be shared. +/// 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`. #[derive(Debug)] pub enum SocketReply { Ok, @@ -102,7 +106,7 @@ impl From for SocketReply { } /// Write (or remove) the status file in the agent's own `state/` directory. -/// Called by `AgentServer::set_status` for both agent and manager flavors +/// Called by both `AgentServer::set_status` and `ManagerServer::set_status` /// 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 @@ -568,13 +572,8 @@ pub struct RemindArgs { pub file_path: Option, } -/// 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. -/// +/// 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`. #[derive(Debug, Clone)] pub struct AgentServer { socket: PathBuf, @@ -586,17 +585,14 @@ impl AgentServer { Self { socket } } - /// Issue any `Request` through the retry-aware client and pull + /// Issue any `AgentRequest` 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::Request, + req: hive_sh4re::AgentRequest, ) -> (Result, u32) { - match client::request_retried::<_, hive_sh4re::Response>(&self.socket, &req).await { + match client::request_retried::<_, hive_sh4re::AgentResponse>(&self.socket, &req).await { Ok((r, n)) => (Ok(SocketReply::from(r)), n), Err(e) => (Err(e), 0), } @@ -616,14 +612,12 @@ impl AgentServer { 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) = 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::Request::Send { + .dispatch(hive_sh4re::AgentRequest::Send { to: args.to, body: args.body, in_reply_to: args.in_reply_to, @@ -654,7 +648,7 @@ impl AgentServer { let log = format!("{args:?}"); run_tool_envelope("ask", log, async move { let (resp, retries) = self - .dispatch(hive_sh4re::Request::Ask { + .dispatch(hive_sh4re::AgentRequest::Ask { question: args.question, options: args.options, multi: args.multi, @@ -690,7 +684,7 @@ impl AgentServer { let id = args.id; run_tool_envelope("answer", log, async move { let (resp, retries) = self - .dispatch(hive_sh4re::Request::Answer { + .dispatch(hive_sh4re::AgentRequest::Answer { id, answer: args.answer, }) @@ -724,7 +718,7 @@ impl AgentServer { let log = format!("{args:?}"); run_tool_envelope("recv", log, async move { let (resp, retries) = self - .dispatch(hive_sh4re::Request::Recv { + .dispatch(hive_sh4re::AgentRequest::Recv { wait_seconds: args.wait_seconds, max: args.max, }) @@ -752,7 +746,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::Request::GetLooseEnds { agent: args.agent }) + .dispatch(hive_sh4re::AgentRequest::GetLooseEnds { agent: args.agent }) .await; // Extract the vec so we can augment before rendering. let mut loose_ends = match resp { @@ -821,7 +815,7 @@ impl AgentServer { return e; } let (resp, retries) = self - .dispatch(hive_sh4re::Request::SetStatus { text: args.text }) + .dispatch(hive_sh4re::AgentRequest::SetStatus { text: args.text }) .await; annotate_retries( format_ack(resp, "set_status", "status updated".to_owned()), @@ -846,7 +840,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::Request::GetAgentMeta { name: args.name }) + .dispatch(hive_sh4re::AgentRequest::GetAgentMeta { name: args.name }) .await; annotate_retries(format_agent_meta(resp), retries) }) @@ -871,7 +865,7 @@ impl AgentServer { }; let kind_label = loose_end_kind_label(kind); let (resp, retries) = self - .dispatch(hive_sh4re::Request::CancelLooseEnd { kind, id }) + .dispatch(hive_sh4re::AgentRequest::CancelLooseEnd { kind, id }) .await; annotate_retries( format_ack( @@ -914,7 +908,7 @@ impl AgentServer { (None, Some(t)) => hive_sh4re::ReminderTiming::At { unix_timestamp: t }, }; let (resp, retries) = self - .dispatch(hive_sh4re::Request::Remind { + .dispatch(hive_sh4re::AgentRequest::Remind { message: args.message, timing, file_path: args.file_path, @@ -965,7 +959,7 @@ impl AgentServer { let name = args.name.clone(); run_tool_envelope("restart", log, async move { let (resp, retries) = self - .dispatch(hive_sh4re::Request::Restart { name: args.name }) + .dispatch(hive_sh4re::AgentRequest::Restart { name: args.name }) .await; annotate_retries( format_ack(resp, "restart", format!("restarted {name}")), @@ -989,7 +983,7 @@ impl AgentServer { let name = args.name.clone(); run_tool_envelope("kill", log, async move { let (resp, retries) = self - .dispatch(hive_sh4re::Request::Kill { name: args.name }) + .dispatch(hive_sh4re::AgentRequest::Kill { name: args.name }) .await; annotate_retries(format_ack(resp, "kill", format!("killed {name}")), retries) }) @@ -1010,7 +1004,7 @@ impl AgentServer { let name = args.name.clone(); run_tool_envelope("update", log, async move { let (resp, retries) = self - .dispatch(hive_sh4re::Request::Update { name: args.name }) + .dispatch(hive_sh4re::AgentRequest::Update { name: args.name }) .await; annotate_retries(format_ack(resp, "update", format!("updated {name}")), retries) }) @@ -1030,7 +1024,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::Request::ListDescendants) + .dispatch(hive_sh4re::AgentRequest::ListDescendants) .await; let body = match resp { Ok(SocketReply::Containers(containers)) => { @@ -1075,7 +1069,7 @@ impl AgentServer { let log = format!("{args:?}"); run_tool_envelope("get_host_journal", log, async move { let (resp, retries) = self - .dispatch(hive_sh4re::Request::GetHostJournal { + .dispatch(hive_sh4re::AgentRequest::GetHostJournal { unit: args.unit, container: args.container, lines: args.lines, @@ -1117,7 +1111,7 @@ impl AgentServer { let name = args.name.clone(); run_tool_envelope("request_init_config", log, async move { let (resp, retries) = self - .dispatch(hive_sh4re::Request::RequestInitConfig { + .dispatch(hive_sh4re::AgentRequest::RequestInitConfig { name: args.name, description: args.description, }) @@ -1155,7 +1149,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::Request::RequestApplyCommit { + .dispatch(hive_sh4re::AgentRequest::RequestApplyCommit { agent: args.agent, commit_ref: args.commit_ref, description: args.description, @@ -1172,254 +1166,6 @@ 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 - } - - - #[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::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 { - 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 { - 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 { - 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 { - 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 { - 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 { - 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( @@ -1429,22 +1175,30 @@ impl AgentServer { )] impl ServerHandler for AgentServer {} -/// Run an MCP server over stdio for the given flavor. Returns when the client disconnects. +/// Run the agent MCP server over stdio. Returns when the client disconnects. /// /// # Errors /// /// Returns an error if the MCP server fails to initialize or the transport /// encounters a fatal error. -pub async fn serve_stdio(socket: PathBuf) -> Result<()> { +pub async fn serve_agent_stdio(socket: PathBuf) -> Result<()> { let server = AgentServer::new(socket); let service = server.serve(stdio()).await?; service.waiting().await?; Ok(()) } -/// Run the MCP server over stdio. Used by all roles. -pub async fn serve_agent_stdio(socket: PathBuf) -> Result<()> { - serve_stdio(socket).await +/// Run the 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. +pub async fn serve_manager_stdio(socket: PathBuf) -> Result<()> { + let server = ManagerServer::new(socket); + let service = server.serve(stdio()).await?; + service.waiting().await?; + Ok(()) } // ----------------------------------------------------------------------------- @@ -1728,6 +1482,679 @@ 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 7d44ea27..ba653733 100644 --- a/hive-c0re/src/agent_server.rs +++ b/hive-c0re/src/agent_server.rs @@ -345,29 +345,6 @@ 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