hyperhive/hive-subagent-mcp/src/mcp.rs

//! The MCP tool surface: `start` / `continue` / `status` / `interrupt`,
//! served directly over streamable-http — no stdio bridge, no round-trip
//! socket.
//!
//! Two surfaces, two routes. `/mcp` is the parent's: the four tools above.
//! `/signal/mcp/<token>` is the *subagent's*, and carries `goal_reached` and
//! `need_help` only — it is what a subagent is handed in its own
//! `--mcp-config` (see [`crate::mcp_config`]), so the ability to report on
//! its own run can't be the ability to spawn a nested one.
//!
//! **The token in that path is the subagent's identity, and neither signal
//! tool takes a session name.** One route is registered, once, at startup
//! (sessions come and go far faster than a `Router` can be rebuilt); every
//! request resolves its token to a session and dispatches to a handler bound
//! to *that* session, so the name a signal lands against is never a value the
//! sender chose. A token that resolves to nothing — never minted, or revoked
//! when its run ended — is a bare 404, which is also why the route says
//! nothing that would let a subagent enumerate its siblings.

use std::collections::HashMap;
use std::sync::{Arc, Mutex, PoisonError};

use axum::extract::{Path, Request, State as RouteState};
use axum::response::{IntoResponse as _, Response};
use rmcp::transport::streamable_http_server::{
    StreamableHttpServerConfig, StreamableHttpService, session::local::LocalSessionManager,
};
use rmcp::{
    ServerHandler,
    handler::server::wrapper::Parameters,
    schemars::{self, JsonSchema},
    tool, tool_handler, tool_router,
};
use serde::Deserialize;

use crate::session::{self, State};

#[derive(Debug, Deserialize, JsonSchema)]
struct StartArgs {
    /// Session name — this daemon's tracking key while it's alive, and the
    /// identity to `continue`/`status`/`interrupt` it by afterward. Same
    /// identifier rules as the `bash` server's task names: lowercase,
    /// digits, hyphen, max 63 chars. Reusable once a prior *finished*
    /// session under that name is done — rejected while one under the same
    /// name is still running.
    name: String,
    /// Which model the subagent's own session runs. Omit for claude's own
    /// default. The `base:claude-subagents` skill's "cheaper-than-you"
    /// guidance still applies here.
    #[serde(default)]
    model: Option<String>,
    /// Which reasoning effort level the subagent's own session runs at
    /// (`--effort`). Omit to default to `medium` — a deliberate hive
    /// policy for subagent work, not claude's own default (`high` on most
    /// models). Independent of `model`, so a cheap model at high effort or
    /// an expensive one at low effort are both valid combinations, not
    /// just the two extremes. See the `base:claude-subagents` skill for
    /// Anthropic's own guidance on choosing between levels.
    #[serde(default)]
    effort: Option<String>,
    /// Path to a file holding the subagent's actual task instructions. A
    /// file, not an inline string, so a large recipe can't blow past a
    /// shell argument length limit.
    prompt_file: String,
    /// Written to the subagent's stdin as its first turn's prompt. Default:
    /// a generic "carry out your instructions" nudge — the real task detail
    /// belongs in `prompt_file`, not here.
    #[serde(default = "default_trigger")]
    trigger: String,
    /// Working directory for the subagent's session — e.g. a git worktree
    /// you've already prepared for it, so a parallel batch of subagents
    /// never race on the same working tree. Must exist. Omit to inherit
    /// this daemon's own working directory (today's default). The daemon
    /// remembers whichever `dir` you give here against `name`, so a later
    /// `continue`/`status` for the same name doesn't need to repeat it —
    /// only pass it there again if you want to point at a *different*
    /// directory. Forgotten on a daemon restart, same as everything else
    /// this daemon tracks in memory.
    #[serde(default)]
    dir: Option<String>,
    /// What this subagent is working *toward*, in its own words — set it and
    /// the daemon keeps giving it turns until it says it's done, says it's
    /// stuck, or runs out. Omit it and the session is a single turn, exactly
    /// as before. Written for the subagent to read: it's quoted back at it
    /// verbatim at the start of every continued turn, so "get `cargo clippy
    /// --workspace` to pass with no warnings" continues far better than "fix
    /// the lints".
    #[serde(default)]
    goal: Option<String>,
    /// How many turns the continuation may spend before the harness stops it
    /// itself. Default 5. Only meaningful alongside `goal` — without one
    /// there's nothing to re-prompt toward, so nothing to cap. The cap
    /// bounds *unattended* re-prompting: a `continue` you issue yourself
    /// starts the allowance over.
    #[serde(default)]
    max_turns: Option<u32>,
    /// Where the instructions in `prompt_file` told this subagent to write
    /// its report. The daemon appends the run's stop reason to that file when
    /// the run ends, so the artifact you were going to read anyway also says
    /// how it stopped. Nothing is inferred: if you don't pass it (and the
    /// subagent doesn't name it when it signals), no file is touched.
    #[serde(default)]
    report_file: Option<String>,
}

fn default_trigger() -> String {
    "Carry out the task described in your instructions.".to_owned()
}

/// No session name, deliberately: which session this signal is about is
/// decided by the URL it arrives on (see this module's doc), so there is no
/// field here in which one subagent could name another.
#[derive(Debug, Deserialize, JsonSchema)]
struct GoalReachedArgs {
    /// Optionally, what you did. It's shown to whoever spawned you.
    #[serde(default)]
    msg: Option<String>,
    /// Optionally, the path you wrote your report to, so the stop reason
    /// gets appended to it.
    #[serde(default)]
    report_file: Option<String>,
}

/// No session name here either — same reason as [`GoalReachedArgs`].
#[derive(Debug, Deserialize, JsonSchema)]
struct NeedHelpArgs {
    /// What is blocking you, specifically enough for someone else to act on
    /// it. This is the whole content of the signal, which is why it's
    /// required.
    msg: String,
    /// Optionally, the path you wrote your report to, so the stop reason
    /// gets appended to it.
    #[serde(default)]
    report_file: Option<String>,
}

#[derive(Debug, Deserialize, JsonSchema)]
struct ContinueArgs {
    /// The existing session's name (from a prior `start`).
    name: String,
    /// The new turn's prompt, written to the subagent's stdin.
    prompt: String,
    /// Which model this turn runs. Omit to let claude fall back to its own
    /// default — this does not have to match whatever model `start` used.
    #[serde(default)]
    model: Option<String>,
    /// Which reasoning effort level this turn runs at. Omit to default to
    /// `medium` (this daemon's own default, not claude's) — this does not
    /// have to match whatever effort `start` (or a prior `continue`) used.
    #[serde(default)]
    effort: Option<String>,
    /// Omit to reuse whatever `dir` `start` (or a prior `continue`) used for
    /// this name — the daemon remembers it until it restarts, and a restart
    /// is exactly when you're most likely to be reaching for `continue`. So
    /// pass it when pointing the session at a *different* directory than
    /// last time, and pass it again after a restart if the session lives
    /// anywhere other than the daemon's own working directory. A resume that
    /// finds nothing says which directory it searched.
    #[serde(default)]
    dir: Option<String>,
}

#[derive(Debug, Deserialize, JsonSchema)]
struct StatusArgs {
    /// The subagent name to check.
    name: String,
    /// Omit to use whatever `dir` was last remembered for this name (see
    /// `start`'s `dir` doc) — you only need this if nothing's running or
    /// reserved for `name` right now (the common "is it running" case never
    /// even looks at it) *and* you want to check a different directory's
    /// session than the one last remembered.
    #[serde(default)]
    dir: Option<String>,
}

#[derive(Debug, Deserialize, JsonSchema)]
struct InterruptArgs {
    /// The running session's name to signal.
    name: String,
    /// `true` sends SIGKILL immediately; `false` (default) sends SIGINT,
    /// letting claude shut down cleanly if it's already mid-response.
    #[serde(default)]
    force: bool,
}

#[derive(Clone)]
struct SubagentMcp {
    state: Arc<State>,
}

#[tool_router]
impl SubagentMcp {
    #[tool(
        description = "Start a fresh claude subagent session under `name`, running in the \
            background. Returns as soon as the process is confirmed running — not once it \
            finishes; this daemon pushes a todo when the turn ends, or use `continue` later to \
            give it another turn. A prior *finished* session under the same name is archived \
            first (real fresh start, not a silent resume); a *currently running* one is \
            refused. Runs unattended — every tool-call permission prompt is pre-approved rather \
            than interactively confirmed — with its MCP server set fixed to what this daemon \
            configures for it. Pass `goal` to make this a multi-turn run: the daemon re-prompts \
            the subagent toward that goal each time a turn ends, up to `max_turns` (default 5), \
            stopping early when the subagent reports the goal reached or asks for help. Whichever \
            way it stops, one todo is pushed at the end and `status` says which. See the \
            `base:claude-subagents` skill for when to reach for this."
    )]
    fn start(&self, Parameters(args): Parameters<StartArgs>) -> String {
        match session::start(
            &self.state,
            session::StartRequest {
                name: args.name,
                model: args.model,
                effort: args.effort,
                prompt_file: args.prompt_file,
                trigger: args.trigger,
                dir: args.dir,
                goal: args.goal,
                max_turns: args.max_turns,
                report_file: args.report_file,
            },
        ) {
            Ok(msg) => msg,
            Err(e) => format!("start error: {e:#}"),
        }
    }

    #[tool(
        name = "continue",
        description = "Give an existing named subagent session a new turn — whether that's \
            because its previous turn finished and you have a follow-up instruction, or you're \
            reattaching after this daemon restarted (the session itself survives independently \
            of the daemon that spawned it). Returns once the turn is underway rather than the \
            instant the process exists — a second or so, not the length of the turn — so a \
            reply saying the turn started means it started; a name already running is refused. \
            A name with no session to resume is not refused up front, because claude's own \
            `--resume` decides that: a miss comes back as this call's own error, naming the \
            directory that was searched, so check `dir` before concluding the session is gone. \
            Resuming a session whose last turn was killed is allowed — the reply says so, since \
            that turn's work stopped wherever it had got to."
    )]
    async fn r#continue(&self, Parameters(args): Parameters<ContinueArgs>) -> String {
        match session::continue_(
            &self.state,
            &args.name,
            args.prompt,
            args.model,
            args.effort,
            args.dir.as_deref(),
        )
        .await
        {
            Ok(msg) => msg,
            Err(e) => format!("continue error: {e:#}"),
        }
    }

    #[tool(
        description = "Signal a currently-running subagent session to stop. Only works once \
            it's actually running — a `start`/`continue` still in its brief window before the \
            process is confirmed spawned refuses interrupt too (nothing to signal yet; retry \
            shortly), same as a name with nothing tracked at all. `force: true` for SIGKILL, \
            otherwise SIGINT."
    )]
    fn interrupt(&self, Parameters(args): Parameters<InterruptArgs>) -> String {
        match session::interrupt(&self.state, &args.name, args.force) {
            Ok(msg) => msg,
            Err(e) => format!("interrupt error: {e:#}"),
        }
    }

    #[tool(
        description = "Report whether a subagent is currently running — a zero-cost check that \
            never launches a process, unlike `continue`. The answer says what state it found \
            and what to do about it. For a running one it also reports how long since that turn \
            last produced any output, which is how you tell a subagent that's working from one \
            that has wedged without resorting to `ps`. For a session started with a `goal` it \
            reports which turn of the budget it is on, and — once the run has stopped — why it \
            stopped: goal reported reached, blocked and needing help, or out of turns."
    )]
    fn status(&self, Parameters(args): Parameters<StatusArgs>) -> String {
        match session::status(&self.state, &args.name, args.dir.as_deref()) {
            Ok(msg) => msg,
            Err(e) => format!("status error: {e:#}"),
        }
    }
}

#[tool_handler]
impl ServerHandler for SubagentMcp {}

/// The surface a *subagent* gets, served on its own route: two tools, both
/// of them ways for a running subagent to say how its own run should end.
/// Separate handler rather than two more tools on [`SubagentMcp`] so the
/// split is structural — there is no route a subagent holds that `start` is
/// reachable from.
///
/// `session` is the whole point of the type: an instance is built only by
/// `signal_route`, from a token that had already resolved, and both tools
/// read the session they act on out of `self`. Nothing in an incoming
/// payload can reach that field.
#[derive(Clone)]
struct SubagentSignalMcp {
    state: Arc<State>,
    session: String,
}

#[tool_router]
impl SubagentSignalMcp {
    #[tool(
        description = "Report that you have reached the goal you were given. Stops the harness \
            from starting another turn to re-prompt you toward it, and extends the \"subagent \
            done\" message your parent gets with what you say here. It always applies to your \
            own session and takes no session name — the endpoint you are calling it on is yours \
            alone, so there is nothing to address it to. This records a claim, not a result: \
            whoever spawned you reads the diff and the gate output regardless, so calling it \
            does not make unfinished work finished. Call it when the goal is actually met — \
            otherwise keep working, or call `need_help` if you can't proceed."
    )]
    fn goal_reached(&self, Parameters(args): Parameters<GoalReachedArgs>) -> String {
        session::goal_reached(
            &self.state,
            &self.session,
            args.msg,
            args.report_file.as_deref(),
        )
    }

    #[tool(
        description = "Report that you cannot proceed, and why. Stops the harness from starting \
            another turn to re-prompt you toward your goal, marks your session as blocked so \
            whoever spawned you sees it in `status` without reading any file, and extends the \
            \"subagent done\" message with your reason. Like `goal_reached` it takes no session \
            name and always applies to your own session. Use it for a genuine block — a missing \
            credential, a decision that isn't yours, an instruction that contradicts what you \
            found — not for work that is merely hard. Say enough that someone else can act on it."
    )]
    fn need_help(&self, Parameters(args): Parameters<NeedHelpArgs>) -> String {
        session::need_help(
            &self.state,
            &self.session,
            args.msg,
            args.report_file.as_deref(),
        )
    }
}

#[tool_handler]
impl ServerHandler for SubagentSignalMcp {}

/// Prefix the subagent-facing signal surface is served under, and the part
/// of a subagent's URL that is *not* secret: every real endpoint is this plus
/// that session's own token. Kept here, next to the route that answers it, so
/// the two can't drift. Nothing is served at the bare prefix.
pub const SIGNAL_PATH: &str = "/signal/mcp";

/// Everything the one registered signal route needs to answer a request for
/// any session: the daemon state the token resolves against, and a session
/// manager per live token.
///
/// **Per token, not one shared manager.** rmcp keeps an MCP session (created
/// by `initialize`, named by an `mcp-session-id` header) inside its manager,
/// and a manager shared across tokens would make that header a second way to
/// pick which bound handler serves a request — an id is unguessable, but this
/// is the one property the route exists to make structural rather than
/// probabilistic. With a manager per token, an id minted under one token
/// simply does not exist under another.
#[derive(Clone)]
struct SignalRouting {
    state: Arc<State>,
    managers: Arc<Mutex<HashMap<String, Arc<LocalSessionManager>>>>,
}

impl SignalRouting {
    /// The session manager for `token`, created on its first request.
    ///
    /// Entries for tokens that no longer resolve are dropped on the way
    /// past — the map would otherwise grow for the daemon's whole life, and
    /// a revoked token's MCP sessions have nothing left to serve. It's a
    /// linear scan of a map with one entry per *live* subagent, on a request
    /// a subagent makes a handful of times per run.
    fn manager_for(&self, token: &str) -> Arc<LocalSessionManager> {
        let mut managers = self.managers.lock().unwrap_or_else(PoisonError::into_inner);
        managers.retain(|t, _| self.state.session_for_signal_token(t).is_some());
        Arc::clone(
            managers
                .entry(token.to_owned())
                .or_insert_with(session_manager),
        )
    }
}

/// A session manager with this daemon's keep-alive: a subagent turn can run
/// considerably longer than a bash command — same 24h rationale as the bash
/// and matrix daemons.
fn session_manager() -> Arc<LocalSessionManager> {
    let mut manager = LocalSessionManager::default();
    manager.session_config.keep_alive = Some(std::time::Duration::from_hours(24));
    Arc::new(manager)
}

/// Resolve the request's token to a session and serve it with a handler bound
/// to that session — or 404, for a token this daemon never minted or has
/// since revoked.
///
/// The `StreamableHttpService` is built per request, which is cheap (a few
/// `Arc`s) and is what lets the bound session come from the URL: rmcp's
/// service factory takes no request, so the binding has to happen outside it.
/// Only a request that starts an MCP session ever calls the factory; the rest
/// are answered from the token's own manager, against the handler built when
/// that session was initialized.
async fn signal_route(
    RouteState(routing): RouteState<SignalRouting>,
    Path(token): Path<String>,
    request: Request,
) -> Response {
    let Some(session) = routing.state.session_for_signal_token(&token) else {
        tracing::debug!("signal route: unknown or expired token, refusing with 404");
        return axum::http::StatusCode::NOT_FOUND.into_response();
    };
    let manager = routing.manager_for(&token);
    let state = Arc::clone(&routing.state);
    let service = StreamableHttpService::new(
        move || {
            Ok(SubagentSignalMcp {
                state: Arc::clone(&state),
                session: session.clone(),
            })
        },
        manager,
        StreamableHttpServerConfig::default(),
    );
    service.handle(request).await.into_response()
}

/// Run the MCP server over HTTP (rmcp streamable-http transport) on `addr`.
/// Loopback-only bind, one long-lived session — same shape as the bash and
/// matrix daemons' own `serve_http`.
///
/// Two routes off one listener: `/mcp` for the parent's four tools, and
/// `<SIGNAL_PATH>/{token}` for the subagent's two. Separate session managers
/// because they're separate MCP servers to separate clients — the parent's
/// harness on one, each subagent's own claude on its own token's.
///
/// The signal route is registered **once**, with the token as a path
/// parameter, because this `Router` is built at startup and subagents come
/// and go for the daemon's whole life: there is no mounting a fresh route per
/// session. Resolution therefore happens per request, in `signal_route`.
///
/// # Errors
///
/// Returns an error if the listener cannot bind `addr` or the HTTP server
/// exits with a fatal error.
pub async fn serve_http(addr: std::net::SocketAddr, state: Arc<State>) -> anyhow::Result<()> {
    let parent_state = Arc::clone(&state);
    let service = StreamableHttpService::new(
        move || {
            Ok(SubagentMcp {
                state: Arc::clone(&parent_state),
            })
        },
        session_manager(),
        StreamableHttpServerConfig::default(),
    );
    let routing = SignalRouting {
        state,
        managers: Arc::new(Mutex::new(HashMap::new())),
    };
    let app = axum::Router::new()
        .nest_service("/mcp", service)
        .route(
            &format!("{SIGNAL_PATH}/{{token}}"),
            axum::routing::any(signal_route),
        )
        .with_state(routing);
    let listener = tokio::net::TcpListener::bind(addr).await?;
    tracing::info!(
        %addr,
        signal = SIGNAL_PATH,
        "serving hive-subagent MCP over streamable-http at /mcp"
    );
    axum::serve(listener, app).await?;
    Ok(())
}

#[cfg(test)]
mod tests {
    use super::*;

    /// The state a route test resolves against, with `name` already minted a
    /// token — the same shape `start` leaves behind.
    fn state() -> Arc<State> {
        Arc::new(State::new(
            std::path::PathBuf::from("/dev/null"),
            format!("http://127.0.0.1:1{SIGNAL_PATH}"),
        ))
    }

    fn routing(state: Arc<State>) -> SignalRouting {
        SignalRouting {
            state,
            managers: Arc::new(Mutex::new(HashMap::new())),
        }
    }

    /// Mint `name` a signal URL the way a spawning run does, and hand back
    /// just the token — what the route would have taken out of the path.
    fn minted(state: &State, name: &str) -> String {
        let url = state.mint_signal_url(name);
        url.rsplit('/')
            .next()
            .expect("a minted URL always has a last segment")
            .to_owned()
    }

    fn post(uri: &str) -> Request {
        axum::http::Request::builder()
            .method("POST")
            .uri(uri)
            .body(axum::body::Body::empty())
            .expect("a well-formed test request")
    }

    #[test]
    fn the_signal_tools_take_no_session_name() {
        // The fix itself, pinned: a subagent has no field in which to name a
        // sibling. Asserted against the generated JSON schema rather than the
        // struct, because the schema is what actually reaches the subagent —
        // a `name` re-added under any serde rename would show up here.
        for (tool, schema) in [
            ("goal_reached", schemars::schema_for!(GoalReachedArgs)),
            ("need_help", schemars::schema_for!(NeedHelpArgs)),
        ] {
            let schema = serde_json::to_value(&schema).expect("a schema serializes");
            let properties = schema
                .get("properties")
                .and_then(serde_json::Value::as_object)
                .cloned()
                .unwrap_or_default();
            assert!(
                !properties.contains_key("name"),
                "`{tool}` must not take a session name — identity comes from the route's token, \
                 and an argument for it is an argument one subagent can fill in with another's \
                 name. Properties offered: {:?}",
                properties.keys().collect::<Vec<_>>()
            );
        }
    }

    #[tokio::test]
    async fn an_unminted_signal_token_is_a_bare_404() {
        // No enumeration: a token this daemon never issued gets the same
        // answer whether or not a session by any name exists.
        let state = state();
        minted(&state, "live");
        let response = signal_route(
            RouteState(routing(state)),
            Path("not-a-token".to_owned()),
            post("/signal/mcp/not-a-token"),
        )
        .await;
        assert_eq!(response.status(), axum::http::StatusCode::NOT_FOUND);
    }

    #[tokio::test]
    async fn a_minted_signal_token_reaches_the_surface_at_all() {
        // The other half of the 404 above: it is the *resolution* refusing,
        // not the route being unreachable. The body here is not a valid MCP
        // request, so rmcp rejects it — on its own terms, with some answer
        // that is not this route's "no such token".
        let state = state();
        let token = minted(&state, "live");
        let response = signal_route(
            RouteState(routing(state)),
            Path(token.clone()),
            post(&format!("/signal/mcp/{token}")),
        )
        .await;
        assert_ne!(
            response.status(),
            axum::http::StatusCode::NOT_FOUND,
            "a live session's own token must resolve"
        );
    }
}