jumperless-mcp 0.1.0

//! MCP transport layer.
//!
//! Drives the MCP server loop for a [`crate::Subsystem`]. v1 wires Stdio only;
//! Tcp and Ws are reserved for Phase B (remote bench-brain over Tailscale).
//!
//! # Architecture
//!
//! [`serve_mcp`] is the single-subsystem entrypoint called by [`crate::run`].
//! It wraps the caller's [`Subsystem`] in [`SubsystemMcpServer`], which
//! implements [`rmcp::ServerHandler`] and delegates tool dispatch back to the
//! subsystem.
//!
//! [`FederationAggregator`](crate::FederationAggregator) also implements
//! [`Subsystem`], so `serve_mcp` handles the federation case transparently —
//! no separate code path needed.
//!
//! # Landmine resolutions (Phase 0b.4)
//!
//! **Landmine 1 — `.waiting()` is non-negotiable.**
//! `rmcp::serve_server()` returns a `RunningService`, NOT `()`. Dropping it
//! causes the binary to exit silently after the initialize handshake. We MUST
//! call `.waiting()` to block until the client disconnects. See the
//! `serve_stdio` function for the exact call site.
//!
//! **Landmine 2 — dotted tool names produce rmcp warnings but still register.**
//! In single-subsystem mode, `Subsystem::tools()` returns bare snake_case names
//! (e.g., `connect`, `dac_set`) — no dots. No warnings. In federation mode,
//! `FederationAggregator::tools()` returns internal dot-prefixed names
//! (`jumperless.connect`) which we translate via `WireFormat` BEFORE passing to
//! rmcp, so rmcp never sees dots. Landmine 2 is neutralised at the wire boundary.
//!
//! **Landmine 3 — rmcp-macros versions independently.**
//! We do NOT use rmcp-macros. The `ServerHandler` implementation is hand-written
//! using the trait-based API directly. No macro version skew is possible.

use std::sync::Arc;

use rmcp::{
    model::{
        CallToolRequestParams, CallToolResult, ClientCapabilities, Content, Implementation,
        InitializeRequestParams, InitializeResult, ListToolsResult, PaginatedRequestParams,
        ServerCapabilities, ServerInfo,
    },
    service::{RequestContext, RoleServer},
    ErrorData as RmcpError, ServerHandler,
};
use tokio::sync::Mutex;

use crate::base::discovery::{force_deferred_via_env, FORCE_DEFERRED_ENV_VAR};
use crate::base::errors::McpError;
use crate::base::subsystem::Subsystem;
use crate::base::wire::WireFormat;

// ── Discovery meta-tools ─────────────────────────────────────────────────────

/// Name of the discovery-mode search meta-tool (mirrors Python toolrack).
const DISCOVERY_TOOL_SEARCH: &str = "tools_search";
/// Name of the discovery-mode describe meta-tool (mirrors Python toolrack).
const DISCOVERY_TOOL_DESCRIBE: &str = "tools_describe_many";

/// Capability key clients declare in MCP initialize to request deferred
/// tool loading.
///
/// Sent under `client_capabilities.experimental.x-deferred-tools` per the MCP
/// spec convention for non-standard, vendor-specific extensions. (The Python
/// toolrack at toolrack/mcp_server.py currently reads this from a non-spec
/// location — `capabilities.tools.x-deferred-tools` — pending migration.
/// Cross-impl interop bridge until then is YGG_MCP_FORCE_DEFERRED.)
const DEFERRED_TOOLS_CAP_KEY: &str = "x-deferred-tools";

// ── Server state ─────────────────────────────────────────────────────────────

/// Negotiated per-connection mode, cached after `initialize` returns.
#[derive(Debug, Clone, Copy)]
enum NegotiatedMode {
    /// Client has not yet sent `initialize`. Waiting.
    Pending,
    /// Full catalog — all tools with schemas. Default.
    Full,
    /// Deferred — only meta-tools. Client declared `x-deferred-tools`.
    Discovery,
}

/// Wraps a [`Subsystem`] and adapts it to the [`rmcp::ServerHandler`] trait.
///
/// `S` must be `Send + Sync + 'static` (guaranteed by the `Subsystem` bound).
/// State is shared via `Arc`; the negotiated mode is behind a `Mutex` because
/// rmcp may call handler methods on multiple tasks concurrently.
///
/// # Wire format
/// `wire_format` controls outbound tool-name translation. For single-subsystem
/// deployments this is almost always the identity (tool names are bare
/// snake_case with no dots). For federation deployments, dot-prefixed internal
/// names (`jumperless.connect`) are translated to the wire form
/// (`jumperless_connect`) before rmcp ever sees them.
struct SubsystemMcpServer<S: Subsystem> {
    subsystem: Arc<S>,
    wire_format: WireFormat,
    /// Negotiated list mode. Set once during `initialize`; read in `list_tools`.
    mode: Mutex<NegotiatedMode>,
}

impl<S: Subsystem> SubsystemMcpServer<S> {
    fn new(subsystem: S, wire_format: WireFormat) -> Self {
        Self {
            subsystem: Arc::new(subsystem),
            wire_format,
            mode: Mutex::new(NegotiatedMode::Pending),
        }
    }

    /// Convert our [`ToolDescriptor`](crate::ToolDescriptor)s to rmcp's
    /// `Tool`s with wire-format name translation applied.
    ///
    /// In single-subsystem mode, tool names are bare snake_case and contain no
    /// dots, so `UnderscoreSeparated` translation is a no-op. In federation
    /// mode (via `FederationAggregator`), internal dotted names are translated
    /// here before being handed to rmcp.
    fn build_tool_list(&self) -> Vec<rmcp::model::Tool> {
        self.subsystem
            .tools()
            .into_iter()
            .map(|td| {
                // Apply WireFormat translation to the tool name, then convert
                // via From<ToolDescriptor> (handles schema conversion + error logging).
                let wire_name = self.wire_format.outbound(&td.name);
                // Direct struct literal intentionally bypasses try_new() validation:
                // - In UnderscoreSeparated mode, wire_name = td.name (already validated by try_new at construction).
                // - In DotSeparated mode, wire_name may contain dots (federation prefix); validation would reject these.
                debug_assert!(
                    matches!(self.wire_format, WireFormat::DotSeparated)
                        || !wire_name.contains('.'),
                    "wire_name '{}' contains dots in non-DotSeparated mode",
                    wire_name
                );
                let td_with_wire_name = crate::base::tool::ToolDescriptor {
                    name: wire_name,
                    ..td
                };
                rmcp::model::Tool::from(td_with_wire_name)
            })
            .collect()
    }

    /// Build the two discovery meta-tools returned in [`ListMode::Discovery`].
    ///
    /// Discovery meta-tools — the two well-known tool names a client invokes
    /// in deferred-tool mode to enumerate and describe the rest of the tool
    /// catalog without paying for the full schema upfront.
    fn discovery_meta_tools() -> Vec<rmcp::model::Tool> {
        // tools_search
        let search_schema = Arc::new({
            let mut m = serde_json::Map::new();
            m.insert("type".into(), serde_json::Value::String("object".into()));
            let mut props = serde_json::Map::new();
            let mut q = serde_json::Map::new();
            q.insert("type".into(), serde_json::Value::String("string".into()));
            q.insert(
                "description".into(),
                serde_json::Value::String("keyword query".into()),
            );
            props.insert("query".into(), serde_json::Value::Object(q));
            m.insert("properties".into(), serde_json::Value::Object(props));
            m.insert(
                "required".into(),
                serde_json::Value::Array(vec![serde_json::Value::String("query".into())]),
            );
            m
        });
        let search_tool = rmcp::model::Tool::new(
            DISCOVERY_TOOL_SEARCH,
            "Search available tools by keyword. Use this to discover what tools \
             are available before invoking them.",
            search_schema,
        );

        // tools_describe_many
        let describe_schema = Arc::new({
            let mut m = serde_json::Map::new();
            m.insert("type".into(), serde_json::Value::String("object".into()));
            let mut props = serde_json::Map::new();
            let mut names = serde_json::Map::new();
            names.insert("type".into(), serde_json::Value::String("array".into()));
            let mut items = serde_json::Map::new();
            items.insert("type".into(), serde_json::Value::String("string".into()));
            names.insert("items".into(), serde_json::Value::Object(items));
            names.insert(
                "description".into(),
                serde_json::Value::String("List of tool names to describe".into()),
            );
            props.insert("names".into(), serde_json::Value::Object(names));
            m.insert("properties".into(), serde_json::Value::Object(props));
            m.insert(
                "required".into(),
                serde_json::Value::Array(vec![serde_json::Value::String("names".into())]),
            );
            m
        });
        let describe_tool = rmcp::model::Tool::new(
            DISCOVERY_TOOL_DESCRIBE,
            "Fetch full schemas for a list of tool names discovered via \
             tools_search. Returns the complete input schema for each tool.",
            describe_schema,
        );

        vec![search_tool, describe_tool]
    }

    /// Select which tools to return based on the negotiated list mode.
    ///
    /// Factors out the mode-arm logic from `list_tools` so that all three arms —
    /// `Full`, `Discovery`, and `Pending` — can be exercised directly in unit tests
    /// without requiring a live rmcp `RequestContext`.
    fn select_tools_for_mode(&self, mode: NegotiatedMode) -> Vec<rmcp::model::Tool> {
        match mode {
            NegotiatedMode::Discovery => Self::discovery_meta_tools(),
            NegotiatedMode::Full => self.build_tool_list(),
            NegotiatedMode::Pending => {
                tracing::warn!(
                    subsystem = %self.subsystem.name(),
                    "tools/list called before initialize completed; defaulting to Full mode (client may be in inconsistent state)"
                );
                self.build_tool_list()
            }
        }
    }

    /// Dispatch an invocation of a discovery meta-tool.
    ///
    /// These are stub implementations for Phase 0b.4. Full keyword search
    /// lands in a follow-up pass once the search engine is wired to the
    /// subsystem's tool catalog.
    fn dispatch_discovery_meta(
        &self,
        name: &str,
        args: Option<serde_json::Map<String, serde_json::Value>>,
    ) -> CallToolResult {
        match name {
            DISCOVERY_TOOL_SEARCH => {
                // TODO(phase 0b.5): cache the wire-translated catalog rather than rebuilding
                // per call. At Phase 0b.4 the catalog is empty so the cost is negligible.

                // Validate query: must be a non-empty string.
                let query = match args.as_ref().and_then(|m| m.get("query")) {
                    Some(serde_json::Value::String(q)) if !q.is_empty() => q.clone(),
                    Some(serde_json::Value::String(_)) => {
                        return CallToolResult::error(vec![Content::text(
                            "tools_search: 'query' must be a non-empty string",
                        )])
                    }
                    Some(_) => {
                        return CallToolResult::error(vec![Content::text(
                            "tools_search: 'query' must be a string",
                        )])
                    }
                    None => {
                        return CallToolResult::error(vec![Content::text(
                            "tools_search: 'query' argument is required",
                        )])
                    }
                };

                // V1 keyword search: filter by name/description substring.
                // Inputs `names` MUST be wire-format names (the same names returned by
                // tools_search). In single-subsystem mode this is identical to internal
                // names. In federation mode, dotted internal names are translated to
                // underscore-joined wire names before exposure.
                let hits: Vec<serde_json::Value> = self
                    .subsystem
                    .tools()
                    .into_iter()
                    .filter(|td| {
                        let q = query.to_lowercase();
                        td.name.to_lowercase().contains(&q)
                            || td.description.to_lowercase().contains(&q)
                    })
                    .map(|td| {
                        let wire = self.wire_format.outbound(&td.name);
                        serde_json::json!({"name": wire, "description": td.description})
                    })
                    .collect();
                match serde_json::to_string_pretty(&hits) {
                    Ok(text) => CallToolResult::success(vec![Content::text(text)]),
                    Err(e) => {
                        tracing::error!(error = %e, "failed to serialize discovery results");
                        CallToolResult::error(vec![Content::text(format!(
                            "serialization error: {e}"
                        ))])
                    }
                }
            }
            DISCOVERY_TOOL_DESCRIBE => {
                // Inputs `names` MUST be wire-format names (the same names returned by
                // tools_search). In single-subsystem mode this is identical to internal
                // names. In federation mode, dotted internal names are translated to
                // underscore-joined wire names before exposure.
                let names_value = args.as_ref().and_then(|m| m.get("names"));
                let names_raw = match names_value {
                    Some(serde_json::Value::Array(arr)) => arr.clone(),
                    Some(_) => {
                        return CallToolResult::error(vec![Content::text(
                            "tools_describe_many: 'names' must be an array",
                        )])
                    }
                    None => {
                        return CallToolResult::error(vec![Content::text(
                            "tools_describe_many: 'names' argument is required",
                        )])
                    }
                };

                let filtered: Vec<&str> = names_raw.iter().filter_map(|v| v.as_str()).collect();
                if filtered.len() != names_raw.len() {
                    tracing::warn!(
                        requested = names_raw.len(),
                        valid = filtered.len(),
                        "tools_describe_many: input contained non-string values; ignoring"
                    );
                }

                let names: Vec<String> = filtered.iter().map(|s| s.to_string()).collect();
                let all_tools = self.build_tool_list();
                let matched_tools: Vec<_> = all_tools
                    .into_iter()
                    .filter(|t| names.contains(&t.name.to_string()))
                    .collect();
                if matched_tools.len() < names.len() {
                    let found: std::collections::HashSet<_> =
                        matched_tools.iter().map(|t| t.name.to_string()).collect();
                    let missing: Vec<&str> = names
                        .iter()
                        .filter(|n| !found.contains(*n))
                        .map(|n| n.as_str())
                        .collect();
                    tracing::warn!(
                        requested = names.len(),
                        found = matched_tools.len(),
                        ?missing,
                        "tools_describe_many: some requested names were not found in catalog"
                    );
                }
                let matched: Vec<serde_json::Value> = matched_tools
                    .into_iter()
                    .map(|t| {
                        serde_json::json!({
                            "name": t.name,
                            "description": t.description,
                            "inputSchema": *t.input_schema,
                        })
                    })
                    .collect();
                match serde_json::to_string_pretty(&matched) {
                    Ok(text) => CallToolResult::success(vec![Content::text(text)]),
                    Err(e) => {
                        tracing::error!(error = %e, "failed to serialize discovery results");
                        CallToolResult::error(vec![Content::text(format!(
                            "serialization error: {e}"
                        ))])
                    }
                }
            }
            other => {
                CallToolResult::error(vec![Content::text(format!("unknown meta-tool: {other}"))])
            }
        }
    }
}

// ── rmcp::ServerHandler impl ─────────────────────────────────────────────────

// rmcp::ServerHandler trait methods use `fn name() -> impl Future<Output = ...> + Send + '_`
// rather than `async fn name()` for object-safety reasons (async fn in traits has
// stabilization quirks with `dyn` and lifetime elaboration). The clippy
// manual_async_fn lint flags these but the trait contract requires this shape.
//
// DO NOT convert these to `async fn` — `cargo clippy --fix` will suggest it,
// but the resulting code breaks the ServerHandler object-safety requirement.
// The explicit `impl Future + Send + '_` signature is load-bearing.
#[allow(clippy::manual_async_fn)]
impl<S: Subsystem> ServerHandler for SubsystemMcpServer<S> {
    /// Server info returned during MCP initialize.
    fn get_info(&self) -> ServerInfo {
        // TODO(phase 0b.5): once discovery meta-tools are functional, advertise the
        // deferred-tool-loading capability in InitializeResult.capabilities.experimental:
        //     "x-deferred-loading": true
        //     "x-discovery-tools": [DISCOVERY_TOOL_SEARCH, DISCOVERY_TOOL_DESCRIBE]
        // This signals to clients that they can opt into deferred mode.
        InitializeResult::new(ServerCapabilities::builder().enable_tools().build())
            .with_server_info(Implementation::new(
                self.subsystem.name(),
                self.subsystem.version(),
            ))
    }

    /// Called when the client sends `initialize`.
    ///
    /// We inspect the client's `experimental` capabilities for the
    /// `x-deferred-tools` key. If present, we switch to
    /// [`NegotiatedMode::Discovery`]; otherwise, [`NegotiatedMode::Full`].
    ///
    /// This stores the negotiated mode so `list_tools` can honour it without
    /// re-reading the capabilities map on every request.
    fn initialize(
        &self,
        request: InitializeRequestParams,
        _context: RequestContext<RoleServer>,
    ) -> impl std::future::Future<Output = Result<InitializeResult, RmcpError>> + Send + '_ {
        async move {
            // rmcp's serve_server populates context.peer.peer_info() with InitializeRequestParams
            // during the handshake before invoking this handler. Validated against rmcp 1.5.0
            // (see Cargo.toml). If a future rmcp version returns None from peer_info() inside
            // this handler, this contract has changed — re-add explicit set_peer_info() here,
            // or read from request.params directly.

            // Negotiate deferred-mode from client capabilities.
            let mode = detect_list_mode(&request.capabilities);
            *self.mode.lock().await = mode;

            tracing::debug!(
                subsystem = self.subsystem.name(),
                client = request.client_info.name,
                ?mode,
                "MCP initialize negotiated"
            );

            Ok(self.get_info())
        }
    }

    /// Return the tool catalog, filtered by negotiated list mode.
    ///
    /// - [`NegotiatedMode::Full`]: all tools with schemas.
    /// - [`NegotiatedMode::Discovery`]: only the two meta-tools.
    /// - [`NegotiatedMode::Pending`]: falls back to Full (safety net; should
    ///   not occur in practice since `initialize` fires first).
    fn list_tools(
        &self,
        _request: Option<PaginatedRequestParams>,
        _context: RequestContext<RoleServer>,
    ) -> impl std::future::Future<Output = Result<ListToolsResult, RmcpError>> + Send + '_ {
        async move {
            let mode = *self.mode.lock().await;
            let tools = self.select_tools_for_mode(mode);
            tracing::debug!(
                subsystem = self.subsystem.name(),
                ?mode,
                tool_count = tools.len(),
                "tools/list"
            );
            Ok(ListToolsResult {
                tools,
                next_cursor: None,
                meta: None,
            })
        }
    }

    /// Dispatch a `tools/call` to the underlying [`Subsystem`].
    ///
    /// - Discovery meta-tools (`tools_search`, `tools_describe_many`)
    ///   are handled locally without touching the subsystem.
    /// - All other calls delegate to [`Subsystem::invoke`] with the raw
    ///   arguments JSON object.
    ///
    /// Tool-not-found and other `McpError`s from the subsystem are propagated
    /// as rmcp error responses — the transport never panics on bad invocations.
    fn call_tool(
        &self,
        request: CallToolRequestParams,
        _context: RequestContext<RoleServer>,
    ) -> impl std::future::Future<Output = Result<CallToolResult, RmcpError>> + Send + '_ {
        async move {
            let name: &str = &request.name;
            let args_value = request
                .arguments
                .map(serde_json::Value::Object)
                .unwrap_or(serde_json::Value::Null);

            tracing::debug!(tool = name, "tools/call");

            // Route discovery meta-tools locally.
            if name == DISCOVERY_TOOL_SEARCH || name == DISCOVERY_TOOL_DESCRIBE {
                let args_map = match args_value {
                    serde_json::Value::Object(m) => Some(m),
                    _ => None,
                };
                return Ok(self.dispatch_discovery_meta(name, args_map));
            }

            // In single-subsystem mode, no inbound name translation is applied because
            // the wire name IS the local tool name (no dots, no prefix). In federation
            // mode, self.subsystem is the FederationAggregator, which receives the wire
            // name and resolves it via its internal wire-to-internal mapping. Either
            // way, no translation happens HERE.
            match self.subsystem.invoke(name, args_value).await {
                Ok(result) => {
                    // Subsystem returned a JSON Value. Serialize it as text content.
                    let text = serde_json::to_string_pretty(&result)
                        .unwrap_or_else(|e| format!("serialization error: {e}"));
                    Ok(CallToolResult::success(vec![Content::text(text)]))
                }
                Err(McpError::ToolNotFound { name: n }) => {
                    tracing::debug!(tool = %n, "tools/call: tool not found");
                    Ok(CallToolResult::error(vec![Content::text(format!(
                        "tool not found: {n}"
                    ))]))
                }
                Err(e) => {
                    tracing::error!(tool = %name, error = %e, "tools/call: subsystem error");
                    Ok(CallToolResult::error(vec![Content::text(format!(
                        "subsystem error: {e}"
                    ))]))
                }
            }
        }
    }
}

// ── Capability negotiation ────────────────────────────────────────────────────

// FORCE_DEFERRED_ENV_VAR and force_deferred_via_env() are canonical in
// discovery.rs so that determine_default_mode (public API) and detect_list_mode
// (private transport) share a single source of truth. Imported at the top of
// this module via `use crate::base::discovery::{...}`.

/// Inspect client capabilities from the MCP initialize handshake and
/// determine the appropriate [`NegotiatedMode`].
///
/// The `x-deferred-tools` signal lives in `capabilities.experimental` (a
/// `BTreeMap<String, JsonObject>`) per the MCP spec convention for
/// non-standard vendor extensions. Its presence as a key (with any value,
/// including an empty object) triggers [`NegotiatedMode::Discovery`].
///
/// Deferred mode may also be forced via the `YGG_MCP_FORCE_DEFERRED` env var
/// (allowlist: `"1"`, `"true"`, `"yes"`, `"on"`, case-insensitive). This is a
/// transitional bridge for clients that don't yet support `x-deferred-tools`
/// (e.g. stock Claude Code, OpenCode Go). Once those clients adopt the
/// capability, the env var becomes redundant.
fn detect_list_mode(caps: &ClientCapabilities) -> NegotiatedMode {
    let client_opted_in = caps
        .experimental
        .as_ref()
        .map(|exp| exp.contains_key(DEFERRED_TOOLS_CAP_KEY))
        .unwrap_or(false);
    let forced = force_deferred_via_env();
    if forced {
        if client_opted_in {
            tracing::debug!(
                env_var = FORCE_DEFERRED_ENV_VAR,
                "env var set but client already opted in; no effect on negotiated mode"
            );
        } else {
            tracing::info!(
                env_var = FORCE_DEFERRED_ENV_VAR,
                "deferred mode forced by env var (client did not opt in)"
            );
        }
    }
    let deferred = client_opted_in || forced;
    if deferred {
        NegotiatedMode::Discovery
    } else {
        NegotiatedMode::Full
    }
}

// ── Stdio transport ───────────────────────────────────────────────────────────

/// Start the MCP server loop for `subsystem` using the given transport.
///
/// Blocks until the client disconnects or an unrecoverable error occurs.
///
/// # Transport selection
/// Only [`crate::base::cli::McpTransport::Stdio`] is implemented in v1. Passing
/// `Tcp` or `Ws` returns an error immediately (they are reserved for Phase B).
///
/// # Landmine 1 — `.waiting()` is mandatory
/// `rmcp::serve_server()` returns a `RunningService`. If it is dropped
/// without calling `.waiting()` the binary exits silently after the
/// initialize handshake. We call `.waiting()` at the ONLY return path from
/// `serve_stdio`, ensuring the guarantee is centrally enforced.
pub async fn serve_mcp<S: Subsystem>(
    subsystem: S,
    transport: crate::base::cli::McpTransport,
) -> Result<(), McpError> {
    match transport {
        crate::base::cli::McpTransport::Stdio => serve_stdio(subsystem).await,
        crate::base::cli::McpTransport::Tcp | crate::base::cli::McpTransport::Ws => {
            Err(McpError::Transport(
                "TCP and WebSocket transports are not implemented in v1; \
                 use --transport stdio"
                    .to_string(),
            ))
        }
    }
}

/// Wire up rmcp's stdio transport and block until the session ends.
///
/// # Landmine 1 (Landmine 2, Landmine 3): addressed here
/// - `.waiting()` is called unconditionally on the `RunningService`.
/// - `SubsystemMcpServer` translates tool names to wire format before rmcp
///   ever sees them, so rmcp never encounters dotted names.
/// - We do NOT use rmcp-macros; the handler is trait-based only.
async fn serve_stdio<S: Subsystem>(subsystem: S) -> Result<(), McpError> {
    let server = SubsystemMcpServer::new(subsystem, WireFormat::default());

    // rmcp::transport::stdio() returns (tokio::io::Stdin, tokio::io::Stdout).
    // rmcp's IntoTransport impl accepts any (AsyncRead, AsyncWrite) tuple.
    let transport = rmcp::transport::stdio();

    // serve_server drives the initialize handshake synchronously, then spawns
    // a background task for the ongoing message loop. The returned
    // RunningService holds the JoinHandle for that task.
    let running = rmcp::serve_server(server, transport).await.map_err(|e| {
        tracing::error!(error = %e, "rmcp::serve_server failed during initialize handshake");
        McpError::Transport(e.to_string())
    })?;

    tracing::info!("MCP server ready (stdio)");

    // LANDMINE 1: .waiting() blocks until the client disconnects.
    // Dropping `running` without calling this exits the process immediately
    // after initialize, silently swallowing all subsequent requests.
    running.waiting().await.map_err(|e| {
        tracing::error!(error = %e, "MCP server task exited with error");
        McpError::Transport(format!("server task join error: {e}"))
    })?;

    tracing::info!("MCP session ended");
    Ok(())
}

// ── Tests ────────────────────────────────────────────────────────────────────

#[cfg(test)]
mod tests {
    use super::*;
    use crate::base::health::{HealthLevel, HealthStatus};
    use crate::base::subsystem::ConnectArgs;
    use crate::base::tool::ToolDescriptor;
    use async_trait::async_trait;
    use serde_json::{json, Value};

    // ── Minimal test subsystem ───────────────────────────────────────────────

    struct MockSubsystem {
        name: &'static str,
        tools: Vec<ToolDescriptor>,
    }

    impl MockSubsystem {
        fn with_tools(name: &'static str, tools: Vec<ToolDescriptor>) -> Self {
            Self { name, tools }
        }
    }

    #[async_trait]
    impl Subsystem for MockSubsystem {
        fn name(&self) -> &str {
            self.name
        }
        fn tools(&self) -> Vec<ToolDescriptor> {
            self.tools.clone()
        }
        async fn connect(&mut self, _args: &ConnectArgs) -> Result<(), McpError> {
            Ok(())
        }
        async fn disconnect(&mut self) -> Result<(), McpError> {
            Ok(())
        }
        async fn health_check(&self) -> Result<HealthStatus, McpError> {
            Ok(HealthStatus {
                level: HealthLevel::Healthy,
                last_seen_unix_ms: 0,
                latency_ms: None,
                version: "0.0.0".into(),
                ring: None,
                subsystem_specific: Value::Null,
            })
        }
        async fn shutdown(&mut self) -> Result<(), McpError> {
            Ok(())
        }
        async fn invoke(&self, tool_name: &str, _args: Value) -> Result<Value, McpError> {
            Ok(json!({"tool": tool_name, "ok": true}))
        }
    }

    // ── Tool catalog construction ────────────────────────────────────────────

    /// Verify that ToolDescriptors are correctly converted to rmcp::model::Tool
    /// with the expected name, description, and input_schema shape.
    #[test]
    fn tool_catalog_construction_matches_expected_shape() {
        let tools = vec![
            ToolDescriptor::new(
                "connect",
                "Connect two breadboard nodes",
                json!({"type": "object", "properties": {"node1": {"type": "integer"}, "node2": {"type": "integer"}}, "required": ["node1", "node2"]}),
            ),
            ToolDescriptor::new(
                "dac_set",
                "Set DAC voltage output",
                json!({"type": "object", "properties": {"channel": {"type": "integer"}, "voltage_mv": {"type": "integer"}}, "required": ["channel", "voltage_mv"]}),
            ),
        ];
        let subsystem = MockSubsystem::with_tools("jumperless", tools);
        let server = SubsystemMcpServer::new(subsystem, WireFormat::default());

        let rmcp_tools = server.build_tool_list();

        assert_eq!(rmcp_tools.len(), 2);

        let connect = &rmcp_tools[0];
        assert_eq!(connect.name.as_ref(), "connect");
        assert_eq!(
            connect.description.as_deref(),
            Some("Connect two breadboard nodes")
        );
        // Schema should contain "type": "object"
        assert_eq!(
            connect.input_schema.get("type").and_then(|v| v.as_str()),
            Some("object")
        );

        let dac = &rmcp_tools[1];
        assert_eq!(dac.name.as_ref(), "dac_set");
    }

    /// Verify that federation-style dotted names are translated to wire format
    /// when a non-default WireFormat is not involved (single-subsystem uses
    /// bare names, no translation needed).
    ///
    /// This test exercises the path where the tool names are bare snake_case —
    /// the expected wire name equals the internal name (no dots present).
    #[test]
    fn bare_snake_case_names_pass_through_unchanged_under_underscore_fmt() {
        let tools = vec![ToolDescriptor::new(
            "ina_get_current",
            "Read INA current",
            json!({"type": "object", "properties": {}}),
        )];
        let subsystem = MockSubsystem::with_tools("bench", tools);
        let server = SubsystemMcpServer::new(subsystem, WireFormat::UnderscoreSeparated);

        let rmcp_tools = server.build_tool_list();
        // No dots in "ina_get_current", so UnderscoreSeparated is identity here.
        assert_eq!(rmcp_tools[0].name.as_ref(), "ina_get_current");
    }

    /// Verify that dotted internal names (as produced by FederationAggregator)
    /// are correctly translated to underscore-separated wire names.
    ///
    /// This is the core of D15 / Landmine 2: we NEVER pass dots to rmcp.
    #[test]
    fn dotted_internal_name_translates_to_underscore_wire_name() {
        // Simulate federation-style: a ToolDescriptor whose name is dotted
        // (as FederationAggregator.tools() would return them).
        // ToolDescriptor::new() validates snake_case and rejects dots,
        // so we bypass it here using direct construction.
        let td = ToolDescriptor {
            name: "jumperless.connect".to_string(),
            description: "Connect nodes".to_string(),
            input_schema: json!({"type": "object", "properties": {}}),
            timeout_ms: crate::base::tool::DEFAULT_TOOL_TIMEOUT_MS,
        };
        let server = SubsystemMcpServer::new(
            MockSubsystem::with_tools("fed", vec![]),
            WireFormat::UnderscoreSeparated,
        );

        // Manually apply outbound translation as build_tool_list() would.
        let wire_name = server.wire_format.outbound(&td.name);
        assert_eq!(wire_name, "jumperless_connect");

        // Confirm DotSeparated is identity.
        let dot_server = SubsystemMcpServer::new(
            MockSubsystem::with_tools("fed2", vec![]),
            WireFormat::DotSeparated,
        );
        let dot_wire = dot_server.wire_format.outbound(&td.name);
        assert_eq!(dot_wire, "jumperless.connect");
    }

    // ── Deferred-loading mode detection ─────────────────────────────────────

    /// Verify that a client declaring `x-deferred-tools` in experimental
    /// capabilities triggers Discovery mode.
    ///
    /// ClientCapabilities is #[non_exhaustive] so we deserialize from JSON
    /// rather than constructing via struct literal.
    #[test]
    fn deferred_tools_capability_triggers_discovery_mode() {
        // ClientCapabilities is #[non_exhaustive]; use serde_json round-trip
        // to construct a capabilities value with experimental.x-deferred-tools set.
        let caps: ClientCapabilities = serde_json::from_value(serde_json::json!({
            "experimental": {
                "x-deferred-tools": {}
            }
        }))
        .expect("failed to deserialize test ClientCapabilities");

        let mode = detect_list_mode(&caps);
        assert!(
            matches!(mode, NegotiatedMode::Discovery),
            "expected Discovery mode, got {mode:?}"
        );
    }

    /// Verify that a client WITHOUT `x-deferred-tools` gets Full mode.
    #[test]
    fn no_deferred_capability_yields_full_mode() {
        let caps = ClientCapabilities::default();
        let mode = detect_list_mode(&caps);
        assert!(
            matches!(mode, NegotiatedMode::Full),
            "expected Full mode, got {mode:?}"
        );
    }

    /// Verify that the discovery meta-tools list contains both expected tools
    /// with correct names and non-empty descriptions.
    #[test]
    fn discovery_meta_tools_have_correct_names() {
        let meta = SubsystemMcpServer::<MockSubsystem>::discovery_meta_tools();
        assert_eq!(meta.len(), 2);

        let names: Vec<&str> = meta.iter().map(|t| t.name.as_ref()).collect();
        assert!(names.contains(&DISCOVERY_TOOL_SEARCH));
        assert!(names.contains(&DISCOVERY_TOOL_DESCRIBE));

        // Both meta-tools must have non-empty descriptions.
        for tool in &meta {
            assert!(
                tool.description
                    .as_ref()
                    .map(|d| !d.is_empty())
                    .unwrap_or(false),
                "meta-tool '{}' has empty description",
                tool.name
            );
        }
    }

    /// Verify that list_tools in Pending mode (no initialize called) returns the
    /// full catalog, not the discovery meta-tool subset.
    #[tokio::test]
    async fn pending_mode_falls_back_to_full() {
        // Construct server without calling initialize — mode stays Pending.
        let tools = vec![
            ToolDescriptor::new(
                "tool_a",
                "First tool",
                json!({"type": "object", "properties": {}}),
            ),
            ToolDescriptor::new(
                "tool_b",
                "Second tool",
                json!({"type": "object", "properties": {}}),
            ),
        ];
        let subsystem = MockSubsystem::with_tools("pending-test", tools);
        let server = SubsystemMcpServer::new(subsystem, WireFormat::default());

        // Mode starts as Pending (no initialize called). build_tool_list() should
        // return the full catalog, not the two discovery meta-tools.
        let full_catalog = server.build_tool_list();
        assert_eq!(
            full_catalog.len(),
            2,
            "expected 2 subsystem tools, not discovery meta-tools"
        );
        let names: Vec<&str> = full_catalog.iter().map(|t| t.name.as_ref()).collect();
        assert!(names.contains(&"tool_a"));
        assert!(names.contains(&"tool_b"));

        // Also verify discovery_meta_tools returns a different (smaller) set.
        let meta = SubsystemMcpServer::<MockSubsystem>::discovery_meta_tools();
        assert_eq!(meta.len(), 2);
        assert!(meta
            .iter()
            .all(|t| t.name.as_ref() != "tool_a" && t.name.as_ref() != "tool_b"));
    }

    /// Verify that select_tools_for_mode returns the correct tool set for each
    /// of the three NegotiatedMode arms. Covers the Pending arm that was
    /// previously only reachable through the full rmcp list_tools path.
    #[test]
    fn select_tools_for_mode_all_arms() {
        let tools = vec![
            ToolDescriptor::new(
                "alpha",
                "First tool",
                json!({"type": "object", "properties": {}}),
            ),
            ToolDescriptor::new(
                "beta",
                "Second tool",
                json!({"type": "object", "properties": {}}),
            ),
        ];
        let subsystem = MockSubsystem::with_tools("mode-test", tools);
        let server = SubsystemMcpServer::new(subsystem, WireFormat::default());

        // Full mode: returns the subsystem's full catalog.
        let full = server.select_tools_for_mode(NegotiatedMode::Full);
        assert_eq!(full.len(), 2, "Full mode should return all subsystem tools");
        let full_names: Vec<&str> = full.iter().map(|t| t.name.as_ref()).collect();
        assert!(full_names.contains(&"alpha") && full_names.contains(&"beta"));

        // Discovery mode: returns only the two meta-tools.
        let discovery = server.select_tools_for_mode(NegotiatedMode::Discovery);
        assert_eq!(
            discovery.len(),
            2,
            "Discovery mode should return exactly two meta-tools"
        );
        let disc_names: Vec<&str> = discovery.iter().map(|t| t.name.as_ref()).collect();
        assert!(disc_names.contains(&DISCOVERY_TOOL_SEARCH));
        assert!(disc_names.contains(&DISCOVERY_TOOL_DESCRIBE));

        // Pending mode: falls back to full catalog (same as Full arm).
        // The warn! fires but requires a tracing subscriber to observe; we
        // assert only the returned set here.
        let pending = server.select_tools_for_mode(NegotiatedMode::Pending);
        assert_eq!(
            pending.len(),
            2,
            "Pending mode should fall back to full catalog"
        );
        let pend_names: Vec<&str> = pending.iter().map(|t| t.name.as_ref()).collect();
        assert!(pend_names.contains(&"alpha") && pend_names.contains(&"beta"));
    }

    /// Verify that get_info() returns a ServerInfo with the subsystem's name.
    #[test]
    fn server_info_reflects_subsystem_name() {
        let subsystem = MockSubsystem::with_tools("my-subsystem", vec![]);
        let server = SubsystemMcpServer::new(subsystem, WireFormat::default());
        let info = server.get_info();
        assert_eq!(info.server_info.name, "my-subsystem");
    }
}