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.

use std::sync::Arc;

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>,
    /// 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>,
}

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

#[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>,
    /// Omit to reuse whatever `dir` `start` (or a prior `continue`) used for
    /// this name — the daemon remembers it. Only pass this to point the
    /// session at a *different* directory than last time.
    #[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. 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,
            &args.name,
            args.model,
            &args.prompt_file,
            args.trigger,
            args.dir.as_deref(),
        ) {
            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 as soon as confirmed running, same as \
            `start`. Refuses a name with no session on disk at all, or one already running. \
            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."
    )]
    fn r#continue(&self, Parameters(args): Parameters<ContinueArgs>) -> String {
        match session::continue_(
            &self.state,
            &args.name,
            args.prompt,
            args.model,
            args.dir.as_deref(),
        ) {
            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."
    )]
    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 {}

/// 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`.
///
/// # 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<()> {
    use rmcp::transport::streamable_http_server::{
        StreamableHttpServerConfig, StreamableHttpService, session::local::LocalSessionManager,
    };
    let mut session_manager = LocalSessionManager::default();
    // A subagent turn can run considerably longer than a bash command —
    // same 24h keep-alive rationale as the bash/matrix daemons.
    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(SubagentMcp {
                state: Arc::clone(&state),
            })
        },
        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 hive-subagent MCP over streamable-http at /mcp");
    axum::serve(listener, app).await?;
    Ok(())
}