hyperhive/hive-agent-mcp/src/mcp/mod.rs

//! Embedded MCP server. Claude Code (running inside the agent container)
//! connects to this over streamable-HTTP via `--mcp-config` (the long-lived
//! `hive-mcp-http` daemon); tool calls land here and are translated to
//! `Request::*` against hyperhive's own
//! per-container unix socket at `/run/hive/mcp.sock`.
//!
//! Two protocols, two surfaces:
//! - **hyperhive socket** at `/run/hive/mcp.sock` — JSON-line, our
//!   broker-routed protocol. Unaffected by this module.
//! - **MCP over HTTP** owned by this module — what claude actually speaks.
//!
//! One `AgentServer { socket }` struct for all roles.
//! Tool access is gated upstream by `--allowedTools` (derived from the
//! agent's `ToolGroup` config); the server itself is a dumb dispatcher.
//! All tools go through the same `run_tool_envelope` helper.

use std::future::Future;
use std::path::PathBuf;

use anyhow::Result;
use rmcp::{ServerHandler, handler::server::wrapper::Parameters, tool, tool_handler, tool_router};

mod args;
mod render;

pub use args::{
    AckUntilArgs, AgentGetLooseEndsArgs, CancelLooseEndArgs, CancelScheduleArgs, CompactArgs,
    CreateRepoArgs, EditScheduleArgs, FireScheduleNowArgs, GetAgentMetaArgs, GetHostJournalArgs,
    GetLogsArgs, KillArgs, MarkTodosDoneArgs, RecvArgs, RemindArgs, RequestInitConfigArgs,
    RequestSchedulePromptArgs, RestartArgs, SendArgs, SetStatusArgs, StartArgs, UpdateArgs,
    UpdateMetaInputsArgs,
};
pub use render::{annotate_retries, format_ack, format_agent_meta, format_recv};

use render::{
    dial_agent_socket, format_matrix_summary, local_reminders, local_todos, loose_end_kind_label,
    mark_local_todo_done, mark_local_todos_done, matrix_unread_summary, parse_loose_end_kind,
    render_loose_ends, reply_err,
};

/// Write (or remove) the status file in the agent's own `state/` directory.
/// Called by `AgentServer::set_status` for both agent and manager flavors
/// before dispatching the wire `SetStatus` request (which only triggers a
/// dashboard rescan on the host side — file I/O moved here because the
/// harness runs as the agent user and has write access to `state/`, whereas
/// hive-c0re's `hive-core` user does not after the privsep migration).
///
/// Mirrors the validation in `hive-c0re::agent_config::limits::check_status_text` so
/// the file is never written with text the server would later reject (which
/// would leave a stale invalid entry on disk).
fn write_status_file(text: &str) -> Result<(), String> {
    // 200 chars mirrors STATUS_MAX_CHARS in hive-c0re/src/agent_config/limits.rs.
    // Keep in sync if that constant changes.
    const STATUS_MAX_CHARS: usize = 200;
    let trimmed = text.trim();
    if !trimmed.is_empty() {
        if trimmed.contains('\n') || trimmed.contains('\r') {
            return Err(
                "set_status text must be a single line — write multi-line context to \
                 a file under your state/ dir and reference that path from the chip instead"
                    .to_owned(),
            );
        }
        let len = trimmed.chars().count();
        if len > STATUS_MAX_CHARS {
            return Err(format!(
                "set_status text too long ({len} chars, max {STATUS_MAX_CHARS}); trim to a short summary"
            ));
        }
    }
    let path = crate::paths::state_dir().join("hyperhive-status");
    let result = if trimmed.is_empty() {
        std::fs::remove_file(&path).or_else(|e| {
            if e.kind() == std::io::ErrorKind::NotFound {
                Ok(())
            } else {
                Err(e)
            }
        })
    } else {
        std::fs::write(&path, format!("{trimmed}\n"))
    };
    result.map_err(|e| format!("set_status write failed: {e}"))
}

/// Common envelope around every MCP tool handler: pre-log → run →
/// post-log. Tool results stay clean — the inbox-status hint lives in
/// the wake prompt + UI header, not appended here.
pub async fn run_tool_envelope<F>(tool: &'static str, args: String, body: F) -> String
where
    F: Future<Output = String>,
{
    tracing::info!(tool, %args, "tool: request");
    let result = body.await;
    tracing::info!(tool, result = %result, "tool: result");
    result
}

/// Unified MCP tool surface for both sub-agent and manager roles.
///
/// Both sockets speak the same `Request` / `Response` wire, so a single
/// `dispatch` call covers both sockets — the only real difference is which
/// socket path is used and which tools the flavor enables.
#[derive(Debug, Clone)]
pub struct AgentServer {
    socket: PathBuf,
}

impl AgentServer {
    #[must_use]
    pub fn new(socket: PathBuf) -> Self {
        Self { socket }
    }

    /// Issue any `Request` through the retry-aware client. Returns the raw
    /// `Response` plus the retry count so tool handlers can annotate their
    /// result (see `annotate_retries`).
    ///
    /// Both sockets speak the same `Request` type, so this single method
    /// covers both.
    async fn dispatch(
        &self,
        req: hive_core_agent_sock::Request,
    ) -> (Result<hive_core_agent_sock::Response, anyhow::Error>, u32) {
        // Ride out a hive-c0re restart rather than surface it: the caller
        // here is claude, and a tool error costs a whole retry turn's
        // tokens where 60s of in-daemon patience costs nothing.
        match hive_sock_client::request_retried::<_, hive_core_agent_sock::Response>(
            &self.socket,
            &req,
            hive_sock_client::Retry::RideOutRestart,
        )
        .await
        {
            Ok((r, n)) => (Ok(r), n),
            Err(e) => (Err(e), 0),
        }
    }
}

// IMPORTANT: when adding a new `#[tool]` fn to this impl, also add
// its name to the matching `ToolGroup::tools()` slice in hive-sh4re.
// Claude Code's permission gate refuses uninlisted MCP tools in
// non-interactive `--print` mode with "permissions not granted yet".
#[tool_router]
impl AgentServer {
    #[tool(
        description = "Send a message to another hyperhive agent (or to the operator). \
            Use this to talk to peers or to surface output for the human at the dashboard."
    )]
    async fn send(&self, Parameters(args): Parameters<SendArgs>) -> String {
        let log = format!("{args:?}");
        let to = args.to.clone();
        // Check per-agent allow-list (hyperhive.allowedRecipients). When no
        // policy file is present (e.g. manager containers) the check is a no-op.
        if let Err(refusal) = crate::send_allow::check_send_allowed(&to) {
            return run_tool_envelope("send", log, async move { refusal }).await;
        }
        run_tool_envelope("send", log, async move {
            let (resp, retries) = self
                .dispatch(hive_core_agent_sock::Request::Send {
                    to: args.to,
                    body: args.body,
                    in_reply_to: args.in_reply_to,
                })
                .await;
            annotate_retries(format_ack(resp, "send", format!("sent to {to}")), retries)
        })
        .await
    }

    #[tool(
        description = "Pop messages from this agent's inbox. Returns one or more messages, or \
            an empty marker if nothing is waiting. Always an immediate 'anything pending?' peek \
            — never blocks. \n\n\
            **Single-message default**: with no args (or `max: 1`) you get the next message — \
            same behaviour the harness uses to drive a turn. \n\n\
            **When idle, prefer ending the turn over polling here.** Ending the turn is the \
            ONLY path that observes an in-container todo wake (bash-task completions, matrix \
            unread, forge activity) — that signal reaches the harness loop between turns, never \
            a live `recv` call, so looping on `recv` while idle means sitting blind until \
            something else wakes you. Ending the turn also checkpoints your session, and costs \
            no latency vs. polling for a real inbox message either way — the broker wakes the \
            next turn just as fast. \n\n\
            **Batch drain**: pass `max: N` (capped at 5) to drain up to N messages in one \
            round-trip. Use this when the wake prompt told you the inbox has more queued, or \
            any time you expect a burst — one tool call beats N consecutive single recvs. Empty \
            result reported the same way regardless of `max`. \n\n\
            After popping, the result appends a `(N more message(s) pending …)` line whenever the \
            inbox still has queued messages — so you know whether to drain again (or `ack_until`) \
            without a separate status check. No line means the inbox is empty."
    )]
    async fn recv(&self, Parameters(args): Parameters<RecvArgs>) -> String {
        let log = format!("{args:?}");
        run_tool_envelope("recv", log, async move {
            let (resp, retries) = self
                .dispatch(hive_core_agent_sock::Request::Recv {
                    wait_seconds: None,
                    max: args.max,
                })
                .await;
            annotate_retries(format_recv(resp), retries)
        })
        .await
    }

    #[tool(
        description = "Bulk-mark inbox messages handled: every message with broker id \
            <= `up_to` (ids show as `[msg #<id>]` in recv output and wake prompts) is \
            acked in one call — pending and already-delivered rows alike. Use this to \
            clear a backlog you've already triaged (e.g. a redelivered flood after a \
            container restart, or a pile of stale notifications) instead of draining it \
            one recv at a time: note the highest `[msg #N]` you've seen, then \
            `ack_until(up_to: N)`. Acked messages never redeliver. Only affects YOUR \
            inbox rows; messages newer than `up_to` stay queued. Returns how many rows \
            were newly acked."
    )]
    async fn ack_until(&self, Parameters(args): Parameters<AckUntilArgs>) -> String {
        let log = format!("{args:?}");
        run_tool_envelope("ack_until", log, async move {
            let (resp, retries) = self
                .dispatch(hive_core_agent_sock::Request::AckUntil { up_to: args.up_to })
                .await;
            let rendered = match resp {
                Ok(hive_core_agent_sock::Response::Acked { count }) => {
                    format!("acked {count} message(s) up to id {}", args.up_to)
                }
                other => reply_err(other, "ack_until"),
            };
            annotate_retries(rendered, retries)
        })
        .await
    }

    #[tool(
        description = "List loose ends pending against this agent: pending reminders \
        you scheduled, plus — for the manager only — pending \
        approvals you submitted that the operator hasn't acted on yet (agents with the \
        `approvals` tool group also see their own pending approvals). Also lists active \
        local tasks published by external MCP daemons (e.g. running bash tasks). Cheap sweep, no args. Useful \
        at turn start to remember what you owe / what's owed to you without scrolling \
        inbox history. Output is a short bulleted list with ids, ages in seconds, and \
        the relevant context. A `reminder` row can be cancelled by \
        passing its id + kind to `cancel_loose_end`. Empty result is reported clearly.\n\
        Todos are a to-do list, not an inbox to empty on sight — a row that's still \
        genuinely in flight (a running bash task, say) is expected to sit there and \
        doesn't need marking done early; only mark done the ones you've actually \
        reviewed and resolved. A real content change (like a task finishing) resurfaces \
        the same row on its own regardless of whether you marked it done while it was \
        still running, so doing that buys nothing and just costs a redundant call later. \
        Todos cap at 40 rows; a trailer line says how many more are pending — mark the \
        reviewed ones done with `mark_todos_done`, then call again for the rest.\n\
        Pass `agent: \"<name>\"` to inspect a specific peer agent's threads. Direct \
        child agents are always accessible. For non-children, the `query_agent_state` \
        capability is required — without it the request is rejected with an error."
    )]
    async fn get_loose_ends(&self, Parameters(args): Parameters<AgentGetLooseEndsArgs>) -> String {
        run_tool_envelope("get_loose_ends", String::new(), async move {
            let is_self_query = args.agent.is_none();
            let (resp, retries) = self
                .dispatch(hive_core_agent_sock::Request::GetLooseEnds { agent: args.agent })
                .await;
            // Extract the vec so we can augment before rendering.
            let mut loose_ends = match resp {
                Ok(hive_core_agent_sock::Response::LooseEnds { loose_ends }) => loose_ends,
                other => return annotate_retries(reply_err(other, "get_loose_ends"), retries),
            };
            // Prepend matrix unread entry for self-queries only (can't
            // reach another agent's matrix daemon from here).
            if is_self_query && let Some(unread_rooms) = matrix_unread_summary().await {
                let total = u32::try_from(unread_rooms.len()).unwrap_or(u32::MAX);
                if total > 0 {
                    let summary = format_matrix_summary(&unread_rooms);
                    loose_ends.insert(
                        0,
                        hive_sh4re::inbox::LooseEnd::UnreadMatrix {
                            rooms: total,
                            summary,
                        },
                    );
                }
            }
            // Merge the harness's local todos (loose-ends v2) for self-queries.
            // The harness owns the todo store in-container; another agent's
            // todos aren't reachable from here (same as matrix above).
            if is_self_query && let Some(todos) = local_todos().await {
                loose_ends.extend(todos);
            }
            // Merge local pending reminders — same self-query-only
            // restriction: a manager asking for a child's loose-ends no longer
            // sees the child's reminders, matching the todos precedent above).
            if is_self_query && let Some(reminders) = local_reminders().await {
                loose_ends.extend(reminders);
            }
            annotate_retries(render_loose_ends(&loose_ends), retries)
        })
        .await
    }

    #[tool(
        description = "Set a free-text status string visible on the operator dashboard. \
        Call this at the START of every task to describe what you're working on (e.g. \
        `\"processing matrix messages\"`, `\"fixing bitburner crash\"`, `\"idle\"`). Pass an empty \
        string to clear. The status is shown on your dashboard card and persists across \
        harness restarts."
    )]
    async fn set_status(&self, Parameters(args): Parameters<SetStatusArgs>) -> String {
        run_tool_envelope("set_status", args.text.clone(), async move {
            if let Err(e) = write_status_file(&args.text) {
                return e;
            }
            let (resp, retries) = self
                .dispatch(hive_core_agent_sock::Request::SetStatus { text: args.text })
                .await;
            annotate_retries(
                format_ack(resp, "set_status", "status updated".to_owned()),
                retries,
            )
        })
        .await
    }

    #[tool(
        description = "Fetch identity + status metadata for an agent. Returns canonical \
        `name`, the current `hyperhive_rev` hive-c0re is running against, `running` \
        (bool — whether the container is currently up; when false, `status_text` and \
        `status_set_at` are stale pre-stop values and should not be treated as live), \
        and the target's self-reported `status` text (set via `set_status`) plus how \
        long ago it was set. Also returns the hive + swarm display names (`hive_name`, \
        `swarm_name`) when the operator has configured \
        `services.hyperhive.hiveName` / `services.hyperhive.swarm.name`; both lines \
        omitted when unset. Pass `name` to query a peer (e.g. \
        check whether iris is idle before pinging them); omit `name` to get your own \
        identity stamp — handy for state files / commit messages / cross-agent \
        attribution that won't drift across renames or session-continue boundaries \
        where the system-prompt label could be stale. Status reads `<none>` when the \
        target has never called `set_status` or has cleared it. Also returns \
        `matrix_accounts` — a list of matrix identities the agent can act as \
        (`name`, `user_id?`, `homeserver`); omitted for agents with no matrix \
        provisioning."
    )]
    async fn get_agent_meta(&self, Parameters(args): Parameters<GetAgentMetaArgs>) -> String {
        let log = args.name.clone().unwrap_or_else(|| "<self>".to_owned());
        run_tool_envelope("get_agent_meta", log, async move {
            let name = match args.name.map(|n| hive_types::Ident::parse(&n)).transpose() {
                Ok(name) => name,
                Err(reason) => return format!("invalid agent name: {reason}"),
            };
            let (resp, retries) = self
                .dispatch(hive_core_agent_sock::Request::GetAgentMeta { name })
                .await;
            annotate_retries(format_agent_meta(resp), retries)
        })
        .await
    }

    #[tool(description = "Cancel an open thread you own — a `reminder` \
        you scheduled (hard-deleted before it fires). `kind` is `\"reminder\"`; \
        `id` is the row id from the matching `get_loose_ends` entry. Auth: you can only \
        cancel rows where you're the owner. Returns `ok` or an error string.\n\
        `kind` may also be `\"approval\"` to withdraw a pending approval you submitted \
        (before the operator acts on it) — root agent (`ruth`) only; the server rejects \
        `approval` kind for all other callers.\n\
        `kind` may also be `\"todo\"` to clear one of your own loose-ends-v2 todos (the id \
        from a `get_loose_ends` `todo #N` line) — dials the in-agent socket directly, safe \
        to call repeatedly.")]
    async fn cancel_loose_end(&self, Parameters(args): Parameters<CancelLooseEndArgs>) -> String {
        let log = format!("{args:?}");
        let id = args.id;
        run_tool_envelope("cancel_loose_end", log, async move {
            if args.kind.trim().eq_ignore_ascii_case("todo") {
                return match mark_local_todo_done(id).await {
                    Some(hive_agent_sock::Response::Acked { count }) if count > 0 => {
                        format!("cleared todo {id}")
                    }
                    Some(hive_agent_sock::Response::Acked { .. }) => {
                        format!("no such todo {id} (already cleared, or never existed)")
                    }
                    Some(hive_agent_sock::Response::Err { message }) => {
                        format!("cancel_loose_end failed: {message}")
                    }
                    Some(other) => format!("cancel_loose_end unexpected response: {other:?}"),
                    None => "cancel_loose_end: local todo socket unavailable \
                             (HIVE_AGENT_SOCKET unset or harness unreachable)"
                        .to_owned(),
                };
            }
            let kind = match parse_loose_end_kind(&args.kind) {
                Ok(k) => k,
                Err(e) => return e,
            };
            let kind_label = loose_end_kind_label(kind);
            // Reminders are harness-local — dial the in-agent socket
            // directly instead of the broker; the other kind (approval)
            // still lives in c0re.
            if kind == hive_sh4re::inbox::CancelLooseEndKind::Reminder {
                return match dial_agent_socket(&hive_agent_sock::Request::CancelReminder { id })
                    .await
                {
                    Some(hive_agent_sock::Response::Acked { count }) if count > 0 => {
                        format!("cancelled {kind_label} {id}")
                    }
                    Some(hive_agent_sock::Response::Acked { .. }) => {
                        format!("cancel_loose_end failed: no pending {kind_label} {id}")
                    }
                    Some(hive_agent_sock::Response::Err { message }) => {
                        format!("cancel_loose_end failed: {message}")
                    }
                    Some(other) => format!("cancel_loose_end unexpected response: {other:?}"),
                    None => "cancel_loose_end failed: in-agent socket unavailable".to_owned(),
                };
            }
            let (resp, retries) = self
                .dispatch(hive_core_agent_sock::Request::CancelLooseEnd { kind, id })
                .await;
            annotate_retries(
                format_ack(
                    resp,
                    "cancel_loose_end",
                    format!("cancelled {kind_label} {id}"),
                ),
                retries,
            )
        })
        .await
    }

    #[tool(
        description = "Bulk-mark specific todos done by id in one call, instead of \
        `cancel_loose_end`ing each one. List-based, not range-based — no `ack_until`-style \
        'clear below id N' for todos, since a blind range-clear risks marking something \
        done that you never actually reviewed. Pass the ids you've actually reviewed \
        (typically what `get_loose_ends` just showed); unknown/already-done ids are \
        silently skipped. Returns how many were newly marked done."
    )]
    async fn mark_todos_done(&self, Parameters(args): Parameters<MarkTodosDoneArgs>) -> String {
        let log = format!("{args:?}");
        run_tool_envelope("mark_todos_done", log, async move {
            match mark_local_todos_done(args.ids).await {
                Some(hive_agent_sock::Response::Acked { count }) => {
                    format!("marked {count} todo(s) done")
                }
                Some(hive_agent_sock::Response::Err { message }) => {
                    format!("mark_todos_done failed: {message}")
                }
                Some(other) => format!("mark_todos_done unexpected response: {other:?}"),
                None => "mark_todos_done: local todo socket unavailable \
                         (HIVE_AGENT_SOCKET unset or harness unreachable)"
                    .to_owned(),
            }
        })
        .await
    }

    #[tool(
        description = "Create a git repo through hive-c0re. You CANNOT create repos with your \
        own forge token (creation is disabled) — this is the only path. The repo is created in \
        the c0re-owned `agents` org, you're added as a write collaborator (not owner), and the \
        default branch gets branch protection so merges require an operator-team approval — you \
        cannot merge your own PRs. `repo` is a single name segment (letters, digits, `-`, `_`, \
        `.`). Returns the new repo's full name + clone URL; clone it over \
        `$HIVE_FORGE_URL/agents/<repo>.git` and push/open PRs as normal."
    )]
    async fn create_repo(&self, Parameters(args): Parameters<CreateRepoArgs>) -> String {
        let log = format!("{args:?}");
        run_tool_envelope("create_repo", log, async move {
            let (resp, retries) = self
                .dispatch(hive_core_agent_sock::Request::CreateRepo { repo: args.repo })
                .await;
            let s = match resp {
                Ok(hive_core_agent_sock::Response::RepoCreated {
                    full_name,
                    clone_url,
                }) => format!("created repo {full_name} — clone: {clone_url}"),
                other => reply_err(other, "create_repo"),
            };
            annotate_retries(s, retries)
        })
        .await
    }

    #[tool(
        description = "Schedule a reminder that lands in this agent's own inbox at a future \
        time (sender will appear as `reminder`). Use for self-paced follow-ups: 'check task \
        status in 60s', 'retry failed deploy at 14:00 UTC', 'nudge me when the operator's \
        deploy window opens'. Set EXACTLY ONE of `delay_seconds` (fire N seconds from now) \
        or `at_unix_timestamp` (fire at absolute epoch second). Body soft-caps at 4 KiB \
        inline — anything larger gets auto-persisted to a file under your \
        `/agents/<you>/state/reminders/` dir and the inbox message becomes a short pointer; \
        pass `file_path` if you want to control the destination yourself. Returns \
        immediately — the reminder lives in the broker until due."
    )]
    async fn remind(&self, Parameters(args): Parameters<RemindArgs>) -> 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::inbox::ReminderTiming::InSeconds { seconds: s },
                (None, Some(t)) => hive_sh4re::inbox::ReminderTiming::At { unix_timestamp: t },
            };
            // Reminders are harness-local — dial the in-agent socket
            // directly instead of the broker.
            match dial_agent_socket(&hive_agent_sock::Request::StoreReminder {
                message: args.message,
                timing,
                file_path: args.file_path,
            })
            .await
            {
                Some(hive_agent_sock::Response::Ok) => "reminder scheduled".to_owned(),
                Some(hive_agent_sock::Response::Err { message }) => {
                    format!("remind failed: {message}")
                }
                Some(other) => format!("remind unexpected response: {other:?}"),
                None => "remind failed: in-agent socket unavailable".to_owned(),
            }
        })
        .await
    }

    #[tool(
        description = "Compact the current session's context, mirroring the operator's \
            dashboard `/compact` button. Gated: only honoured when this agent's last \
            completed turn used more than 66% of the effective context window — below \
            that the call is refused with an explanation and has no effect. On a pass, \
            queues compaction for the end of the current turn (same deferred mechanism \
            the dashboard button uses, so it never races a live claude process); the \
            usual pre-compaction notes-checkpoint turn still fires first. Pass \
            `wake_prompt` to have the harness drive one synthetic follow-up turn with \
            that body as soon as compaction finishes — without it you just go idle \
            waiting for the next external event, same as ending a turn normally."
    )]
    async fn compact(&self, Parameters(args): Parameters<CompactArgs>) -> String {
        let log = format!("{args:?}");
        run_tool_envelope("compact", log, async move {
            match dial_agent_socket(&hive_agent_sock::Request::Compact {
                wake_prompt: args.wake_prompt,
            })
            .await
            {
                Some(hive_agent_sock::Response::Ok) => {
                    "compact queued — will run at the end of the current turn".to_owned()
                }
                Some(hive_agent_sock::Response::Err { message }) => message,
                Some(other) => format!("compact: unexpected response: {other:?}"),
                None => "compact failed: in-agent socket unavailable".to_owned(),
            }
        })
        .await
    }

    // IMPORTANT: this tool is only available when the `lifecycle` tool group
    // is granted to this agent. hive-c0re enforces the topology check
    // server-side: the call is rejected unless `name` is a direct child.
    #[tool(
        description = "Restart a direct child sub-agent container (stop + start). \
            Only succeeds if `name` is a direct child of this agent in the topology \
            tree — the server enforces this. No approval required."
    )]
    async fn restart(&self, Parameters(args): Parameters<RestartArgs>) -> String {
        let log = format!("{args:?}");
        let name = args.name.clone();
        run_tool_envelope("restart", log, async move {
            let (resp, retries) = self
                .dispatch(hive_core_agent_sock::Request::Restart { name: args.name })
                .await;
            annotate_retries(
                format_ack(resp, "restart", format!("restarted {name}")),
                retries,
            )
        })
        .await
    }

    // IMPORTANT: this tool is only available when the `lifecycle` tool group
    // is granted to this agent. hive-c0re enforces the topology check
    // server-side: the call is rejected unless `name` is a direct child.
    #[tool(description = "Stop a direct child sub-agent container (graceful). \
            Only succeeds if `name` is a direct child of this agent in the topology \
            tree — the server enforces this. No approval required. \
            State dir is kept; recreating the agent reuses prior config + credentials.")]
    async fn kill(&self, Parameters(args): Parameters<KillArgs>) -> String {
        let log = format!("{args:?}");
        let name = args.name.clone();
        run_tool_envelope("kill", log, async move {
            let (resp, retries) = self
                .dispatch(hive_core_agent_sock::Request::Kill { name: args.name })
                .await;
            annotate_retries(format_ack(resp, "kill", format!("killed {name}")), retries)
        })
        .await
    }

    // IMPORTANT: this tool is only available when the `lifecycle` tool group
    // is granted to this agent. hive-c0re enforces the topology check
    // server-side: the call is rejected unless `name` is a direct child.
    #[tool(
        description = "Rebuild a direct child sub-agent: re-applies the current hyperhive \
            flake + agent.nix and restarts the container. Only succeeds if `name` is a direct \
            child of this agent in the topology tree — the server enforces this. \
            No approval required. Idempotent — use when a child needs its config reapplied."
    )]
    async fn update(&self, Parameters(args): Parameters<UpdateArgs>) -> String {
        let log = format!("{args:?}");
        let name = args.name.clone();
        run_tool_envelope("update", log, async move {
            let (resp, retries) = self
                .dispatch(hive_core_agent_sock::Request::Update { name: args.name })
                .await;
            annotate_retries(
                format_ack(resp, "update", format!("updated {name}")),
                retries,
            )
        })
        .await
    }

    // IMPORTANT: this tool is only available when the `lifecycle` tool group
    // is granted to this agent. Returns all topological descendants of the
    // calling agent with their running status.
    #[tool(
        description = "List all containers that are topological descendants of this agent \
            (direct children + their subtrees). Requires the `lifecycle` tool group. \
            Returns every known descendant regardless of running state — check the `running` \
            field to distinguish live from stopped containers. Ordered by topology depth \
            (parents before children), then alphabetically within each tier."
    )]
    async fn list_containers(&self) -> String {
        run_tool_envelope("list_containers", String::new(), async move {
            let (resp, retries) = self
                .dispatch(hive_core_agent_sock::Request::ListDescendants)
                .await;
            let body = match resp {
                Ok(hive_core_agent_sock::Response::Containers { containers }) => {
                    if containers.is_empty() {
                        "no descendant containers".to_owned()
                    } else {
                        containers
                            .iter()
                            .map(|c| {
                                let status = if c.running { "running" } else { "stopped" };
                                format!("{} ({})", c.name, status)
                            })
                            .collect::<Vec<_>>()
                            .join("\n")
                    }
                }
                other => reply_err(other, "list_containers"),
            };
            annotate_retries(body, retries)
        })
        .await
    }

    // IMPORTANT: this tool is capability-gated (`read_host_journal`).
    // It is added to `--allowedTools` by `allowed_capability_tools` only
    // when `HIVE_CAPABILITIES` contains `read_host_journal`. hive-c0re
    // performs a second capability check server-side before running journalctl.
    #[tool(
        description = "Fetch recent lines from the host journal (requires `read_host_journal` \
            capability). All filters are optional - omit to get the last N host journal lines. \
            `unit`: filter to a systemd unit (e.g. `hive-c0re.service`). \
            `container`: nspawn machine name verbatim (e.g. `h-iris`). Omit for host journal. \
            Agent containers use the `h-` prefix (e.g. `h-iris`, `h-atlas`); \
            infrastructure containers use their full name (e.g. `hive-ci`, `hive-forge`, \
            `hive-matrix`). The gateway is not a container: its nginx logs are in \
            the host journal, so omit `container` and pass `unit: nginx.service`. \
            `lines`: how many lines (default 30, max 100). \
            `priority`: minimum syslog level enum. \
            `grep`: regex matched against log message fields (journalctl --grep). \
            `since`: show entries on or newer than this (e.g. `-1h`, `2024-01-01 12:00:00`). \
            `until`: show entries on or older than this."
    )]
    async fn get_host_journal(&self, Parameters(args): Parameters<GetHostJournalArgs>) -> String {
        let log = format!("{args:?}");
        run_tool_envelope("get_host_journal", log, async move {
            let (resp, retries) = self
                .dispatch(hive_core_agent_sock::Request::GetHostJournal {
                    unit: args.unit,
                    container: args.container,
                    lines: args.lines,
                    priority: args.priority,
                    grep: args.grep,
                    since: args.since,
                    until: args.until,
                })
                .await;
            let result = match resp {
                Ok(hive_core_agent_sock::Response::HostJournal { content }) => content,
                other => reply_err(other, "get_host_journal"),
            };
            annotate_retries(result, retries)
        })
        .await
    }

    // IMPORTANT: this tool is only available when the `approvals` tool group
    // is configured for the agent (`HIVE_TOOL_GROUPS` contains `approvals`).
    // hive-c0re performs a topology check server-side: only direct children
    // of the calling agent are accepted; all other names are rejected.
    #[tool(
        description = "Create a brand-new direct child agent's config repo and queue an \
            `InitConfig` approval for the operator to review. Requires the `approvals` tool \
            group. `name` must be a direct child of this agent in the topology tree. Fails if a \
            config repo for that child already exists. This tool **creates the repo** and \
            nothing else: on approval hive-c0re seeds it with a default `agent.nix`, and \
            approving the follow-up `Spawn` creates the container. \
            Every config change — the child's first one included — goes through a PR on its \
            config repo, made from a clone you take yourself, reviewed + approved by the \
            operator. `/agents/<name>/config` is a **read-only copy** for reading a config, \
            never an editing surface."
    )]
    async fn request_init_config(
        &self,
        Parameters(args): Parameters<RequestInitConfigArgs>,
    ) -> String {
        let log = format!("{args:?}");
        let name = args.name.clone();
        run_tool_envelope("request_init_config", log, async move {
            let (resp, retries) = self
                .dispatch(hive_core_agent_sock::Request::RequestInitConfig {
                    name: args.name,
                    description: args.description,
                })
                .await;
            annotate_retries(
                format_ack(
                    resp,
                    "request_init_config",
                    format!("init_config approval queued for {name}"),
                ),
                retries,
            )
        })
        .await
    }

    // IMPORTANT: this tool is only available when the `lifecycle` tool group
    // is granted to this agent. hive-c0re enforces the topology check
    // server-side: the call is rejected unless `name` is a direct child.
    #[tool(description = "Start a stopped direct child sub-agent container. \
            Only succeeds if `name` is a direct child of this agent in the topology \
            tree — the server enforces this. No approval required.")]
    async fn start(&self, Parameters(args): Parameters<StartArgs>) -> String {
        let log = format!("{args:?}");
        let name = args.name.clone();
        run_tool_envelope("start", log, async move {
            let (resp, retries) = self
                .dispatch(hive_core_agent_sock::Request::Start { name: args.name })
                .await;
            annotate_retries(
                format_ack(resp, "start", format!("started {name}")),
                retries,
            )
        })
        .await
    }

    #[tool(
        description = "Fetch recent journal log lines for a sub-agent container. Useful \
        for diagnosing MCP server registration failures, startup crashes, plugin install \
        errors, or any harness issue you can't see from inside the container. Pass the \
        plain logical agent name (e.g. `gui`) — hive-c0re resolves the machine name. \
        `lines` defaults to 50 (max capped at 500 on the host side)."
    )]
    async fn get_logs(&self, Parameters(args): Parameters<GetLogsArgs>) -> String {
        let log = format!("{args:?}");
        let agent = args.agent.clone();
        run_tool_envelope("get_logs", log, async move {
            let lines = args.lines.map(|n| n.min(500));
            let (resp, retries) = self
                .dispatch(hive_core_agent_sock::Request::GetLogs {
                    agent: agent.clone(),
                    lines,
                })
                .await;
            let s = match resp {
                Ok(hive_core_agent_sock::Response::Logs { content }) => {
                    if content.is_empty() {
                        format!("(no journal output for {agent})")
                    } else {
                        content
                    }
                }
                other => reply_err(other, "get_logs"),
            };
            annotate_retries(s, retries)
        })
        .await
    }

    #[tool(
        description = "Queue an approval for the operator to run `nix flake update` on the \
        meta flake and commit the resulting lock changes. Pass specific input names to update \
        only those inputs (e.g. `[\"bitburner-agent\"]`), or pass an empty list to update ALL \
        inputs. Returns immediately — the lock update runs when the operator approves. \
        Does NOT trigger container rebuilds — call `update` on each affected agent \
        separately after the approval resolves."
    )]
    async fn request_update_meta_inputs(
        &self,
        Parameters(args): Parameters<UpdateMetaInputsArgs>,
    ) -> String {
        let log = format!("{args:?}");
        run_tool_envelope("request_update_meta_inputs", log, async move {
            let label = if args.inputs.is_empty() {
                "all inputs".to_string()
            } else {
                args.inputs.join(", ")
            };
            let (resp, retries) = self
                .dispatch(hive_core_agent_sock::Request::RequestUpdateMetaInputs {
                    inputs: args.inputs,
                    description: args.description,
                })
                .await;
            annotate_retries(
                format_ack(
                    resp,
                    "request_update_meta_inputs",
                    format!("approval queued: {label}"),
                ),
                retries,
            )
        })
        .await
    }

    #[tool(
        description = "Queue an approval to add a scheduled prompt — one body delivered to \
        N agent inboxes at a target time, optionally recurring every `interval_seconds`. \
        The operator approves; on approve hive-c0re inserts the schedule and the worker \
        fans it out. Even self-targeted schedules go through this flow (the operator pays \
        for the wake-up tokens); the existing `remind` MCP tool stays the quick \
        no-approval self-wake path. \n\n\
        Catch-up clamp: if hive-c0re is down across multiple intervals, only ONE delayed \
        fire happens on resume (per recurring schedule). The skipped-cycle count surfaces \
        in the per-target `last_result` for the operator's audit trail. \n\n\
        Per-target failure: a target name that doesn't resolve to a live agent → operator \
        gets a one-line advisory `Message` from `system`; the schedule keeps firing for \
        the other (live) targets."
    )]
    async fn request_schedule_prompt(
        &self,
        Parameters(args): Parameters<RequestSchedulePromptArgs>,
    ) -> String {
        let log = format!("{args:?}");
        run_tool_envelope("request_schedule_prompt", log, async move {
            let target_count = args.targets.len();
            let (resp, retries) = self
                .dispatch(hive_core_agent_sock::Request::RequestSchedulePrompt(
                    hive_sh4re::manager::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<FireScheduleNowArgs>) -> String {
        let log = format!("{args:?}");
        run_tool_envelope("fire_schedule_now", log, async move {
            let id = args.id;
            let (resp, retries) = self
                .dispatch(hive_core_agent_sock::Request::FireScheduleNow { id })
                .await;
            annotate_retries(
                format_ack(resp, "fire_schedule_now", format!("fired #{id} now")),
                retries,
            )
        })
        .await
    }

    #[tool(
        description = "Cancel a scheduled prompt. With no `targets` field, cancels the \
        whole schedule (all recipients flipped). With a non-empty `targets` list, cancels \
        just those recipients; the schedule keeps firing for any remaining active targets \
        and auto-cancels its parent row when every target is cancelled. \n\n\
        Authorization: the manager can cancel its own schedules + any schedule whose \
        owner is one of its sub-agents per topology.json. Other owners are refused."
    )]
    async fn cancel_schedule(&self, Parameters(args): Parameters<CancelScheduleArgs>) -> String {
        let log = format!("{args:?}");
        run_tool_envelope("cancel_schedule", log, async move {
            let id = args.id;
            let (resp, retries) = self
                .dispatch(hive_core_agent_sock::Request::CancelSchedule {
                    id: args.id,
                    targets: args.targets,
                })
                .await;
            annotate_retries(
                format_ack(resp, "cancel_schedule", format!("cancelled #{id}")),
                retries,
            )
        })
        .await
    }

    #[tool(
        description = "Edit an existing scheduled prompt's mutable fields. Pass only \
        the fields you want to change — anything omitted keeps its current value. Editable: \
        `body`, `description`, `interval_seconds` (positive only via this tool; flipping \
        recurring→one-shot is operator-only via the dashboard), `next_fire_at_unix`, and \
        the target set via `targets_add` / `targets_remove`. Both target lists are \
        applied in the same transaction with removes-before-adds, so a single edit can \
        swap a target atomically. Re-adding a previously-removed target starts a fresh \
        per-target history (drops the tombstone). Draining all targets auto-cancels the \
        parent schedule. \n\n\
        Authorization mirrors `cancel_schedule` / `fire_schedule_now`: you can edit your \
        own schedules + any owned by a sub-agent in your subtree per topology.json. \
        Refuses cancelled schedules (the row's terminal — submit a fresh one)."
    )]
    async fn edit_schedule(&self, Parameters(args): Parameters<EditScheduleArgs>) -> String {
        let log = format!("{args:?}");
        run_tool_envelope("edit_schedule", log, async move {
            let id = args.id;
            let (resp, retries) = self
                .dispatch(hive_core_agent_sock::Request::EditSchedule {
                    id: args.id,
                    body: args.body,
                    description: args.description.map(Some),
                    interval_seconds: args.interval_seconds.map(Some),
                    next_fire_at_unix: args.next_fire_at_unix,
                    targets_add: args.targets_add,
                    targets_remove: args.targets_remove,
                })
                .await;
            annotate_retries(
                format_ack(resp, "edit_schedule", format!("edited #{id}")),
                retries,
            )
        })
        .await
    }

    #[tool(
        description = "List every scheduled prompt in the queue (active + cancelled but \
        not yet reaped). Returns the full snapshot — schedule id, owner, body, target set \
        with per-target last_fired_at + last_result, next fire time, recurring interval. \
        Use this to look up an id before calling `cancel_schedule`, or to audit what \
        the swarm is going to be woken up about next."
    )]
    async fn list_schedules(&self) -> String {
        run_tool_envelope("list_schedules", String::new(), async move {
            let (resp, retries) = self
                .dispatch(hive_core_agent_sock::Request::ListSchedules)
                .await;
            let body = match resp {
                Ok(hive_core_agent_sock::Response::Schedules { schedules }) => {
                    serde_json::to_string(&schedules)
                        .unwrap_or_else(|e| format!("list_schedules: serialise: {e:#}"))
                }
                other => reply_err(other, "list_schedules"),
            };
            annotate_retries(body, retries)
        })
        .await
    }
}

#[tool_handler(
    instructions = "You are a hyperhive agent. Use `send` to talk to peers (by their logical \
        name) or to the operator (recipient `operator`). Use `recv` to drain your inbox one \
        message at a time. Use `remind` to schedule a future wake-up message for yourself."
)]
impl ServerHandler for AgentServer {}

/// Run the MCP server over HTTP (rmcp streamable-http transport) on `addr`.
///
/// This is the sole transport for the built-in hyperhive surface. It runs as a
/// long-lived in-container daemon. claude reconnects
/// to the stable URL each turn instead of respawning and re-registering a stdio
/// subprocess, which removes the per-turn MCP registration race that can strand
/// an agent when the async `initialize`/`tools/list` loses to claude's first
/// tool call. `socket` is the hyperhive control socket every tool call dials
/// fresh (the handler holds only the path), so a host-side hive-c0re restart is
/// transparent — the next call just reconnects.
///
/// Binds loopback only in practice; the default `allowed_hosts`
/// (`localhost`/`127.0.0.1`/`::1`) rejects Host headers from anywhere else.
///
/// # Errors
///
/// Returns an error if the listener cannot bind `addr` or the HTTP server
/// exits with a fatal error.
pub async fn serve_http(socket: PathBuf, addr: std::net::SocketAddr) -> Result<()> {
    use rmcp::transport::streamable_http_server::{
        StreamableHttpServerConfig, StreamableHttpService, session::local::LocalSessionManager,
    };
    let mut session_manager = LocalSessionManager::default();
    // rmcp's default session idle timeout is 5 minutes, meant to reap zombie
    // connections lost to an HTTP/2 RST_STREAM — far too short for a claude
    // turn that goes several minutes between MCP calls to this daemon (a long
    // Read/Edit stretch, a background-task wait). That trips a client-visible
    // "MCP server ... session expired" mid-turn; self-recovering on the next
    // call, but confusing. Give it real headroom instead of turning it off
    // outright.
    session_manager.session_config.keep_alive = Some(std::time::Duration::from_hours(24));
    let session_manager = std::sync::Arc::new(session_manager);
    let service = StreamableHttpService::new(
        move || Ok(AgentServer::new(socket.clone())),
        session_manager,
        StreamableHttpServerConfig::default(),
    );
    let app = axum::Router::new().nest_service("/mcp", service);
    let listener = tokio::net::TcpListener::bind(addr).await?;
    tracing::info!(%addr, "serving hyperhive MCP over streamable-http at /mcp");
    axum::serve(listener, app).await?;
    Ok(())
}