nanocodex-tools 0.3.0

//! Background-handshaken MCP tools for Nanocodex Code Mode.

mod catalog;
mod client;
mod config;
mod oauth;
mod stdio;

use std::{
    collections::{BTreeMap, btree_map::Entry},
    sync::{
        Arc,
        atomic::{AtomicBool, Ordering},
    },
    time::Duration,
};

use crate::{DynamicToolProvider, Tool, ToolContext, ToolInput, ToolOutput, ToolResult};
use async_trait::async_trait;
use catalog::{ProviderState, ToolEntry};
use nanocodex_oai_api::tools::ToolDefinition;
use rmcp::model::CallToolRequestParams;
use serde::Deserialize;
use serde_json::{Value, json};
use tracing::{Instrument, info_span};

pub use config::McpServer;
pub use oauth::{McpOAuthCredentials, McpOAuthStore};

fn same_origin_redirect_policy() -> reqwest::redirect::Policy {
    reqwest::redirect::Policy::custom(|attempt| {
        let follows_same_origin = attempt
            .previous()
            .last()
            .is_some_and(|previous| previous.origin() == attempt.url().origin());
        if follows_same_origin && attempt.previous().len() <= 10 {
            attempt.follow()
        } else {
            attempt.stop()
        }
    })
}

/// A configured family of MCP servers installed into [`crate::Tools`].
pub struct Mcp {
    servers: Arc<[NamedServer]>,
    state: Arc<ProviderState>,
    search: Arc<McpSearch>,
    oauth_store: Option<Arc<dyn McpOAuthStore>>,
    oauth_metadata: Arc<oauth::OAuthMetadataCache>,
    started: AtomicBool,
}

struct NamedServer {
    name: String,
    config: McpServer,
}

/// Builder for an MCP provider.
#[derive(Default)]
pub struct McpBuilder {
    servers: BTreeMap<String, McpServer>,
    oauth_store: Option<Arc<dyn McpOAuthStore>>,
    duplicate: Option<String>,
}

/// Cheap control handle for reconnecting and authorizing a running MCP provider.
#[derive(Clone)]
pub struct McpHandle {
    servers: Arc<[NamedServer]>,
    state: Arc<ProviderState>,
    oauth_store: Option<Arc<dyn McpOAuthStore>>,
    oauth_metadata: Arc<oauth::OAuthMetadataCache>,
}

/// An in-progress browser OAuth login.
pub struct McpLogin {
    authorization_url: String,
    completion: tokio::task::JoinHandle<Result<usize, McpControlError>>,
}

/// Failure while controlling an already configured MCP provider.
#[derive(Debug, thiserror::Error)]
pub enum McpControlError {
    /// No configured server has the requested name.
    #[error("unknown MCP server `{0}`")]
    UnknownServer(String),
    /// OAuth login was requested for a stdio server.
    #[error("MCP server `{0}` does not use Streamable HTTP")]
    NotHttp(String),
    /// OAuth login cannot replace an explicit bearer token.
    #[error("MCP server `{0}` has explicit bearer authentication")]
    ExplicitBearer(String),
    /// OAuth login requires caller-owned credential persistence.
    #[error("no MCP OAuth credential store is configured")]
    NoOAuthStore,
    /// OAuth discovery, authorization, callback, or persistence failed.
    #[error("MCP OAuth login failed: {0}")]
    OAuth(String),
    /// A server replacement connection failed.
    #[error("MCP server `{server}` failed to reload: {error}")]
    Reload {
        /// Configured server name.
        server: String,
        /// Complete connection or discovery error.
        error: String,
    },
    /// The spawned login task stopped before returning its result.
    #[error("MCP OAuth login task stopped: {0}")]
    LoginTask(String),
}

/// Invalid MCP provider configuration.
#[derive(Debug, thiserror::Error)]
pub enum McpBuildError {
    /// No servers were configured.
    #[error("at least one MCP server is required")]
    Empty,
    /// A configured server name was empty.
    #[error("MCP server name must not be empty")]
    EmptyName,
    /// The same server name was added more than once.
    #[error("MCP server `{0}` is configured more than once")]
    DuplicateServer(String),
    /// A required transport field was empty.
    #[error("MCP server `{server}` has an empty {field}")]
    EmptyField {
        /// Configured server name.
        server: String,
        /// Invalid field name.
        field: &'static str,
    },
    /// A lifecycle timeout was zero.
    #[error("MCP server `{server}` has a zero {field}")]
    ZeroTimeout {
        /// Configured server name.
        server: String,
        /// Invalid timeout field.
        field: &'static str,
    },
    /// An option was applied to the wrong transport kind.
    #[error("MCP server `{server}` does not support option `{option}` for its transport")]
    UnsupportedOption {
        /// Configured server name.
        server: String,
        /// Unsupported builder option.
        option: &'static str,
    },
}

impl Mcp {
    /// Starts an empty MCP provider builder.
    #[must_use]
    pub fn builder() -> McpBuilder {
        McpBuilder::default()
    }

    /// Returns a cheap cloneable handle for reload and OAuth operations.
    #[must_use]
    pub fn handle(&self) -> McpHandle {
        McpHandle {
            servers: Arc::clone(&self.servers),
            state: Arc::clone(&self.state),
            oauth_store: self.oauth_store.clone(),
            oauth_metadata: Arc::clone(&self.oauth_metadata),
        }
    }
}

impl McpBuilder {
    /// Installs caller-owned persistence for OAuth-capable Streamable HTTP servers.
    #[must_use]
    pub fn oauth_store(mut self, store: Arc<dyn McpOAuthStore>) -> Self {
        self.oauth_store = Some(store);
        self
    }

    /// Adds a named stdio or Streamable HTTP MCP server.
    #[must_use]
    pub fn server(mut self, name: impl Into<String>, server: McpServer) -> Self {
        let name = name.into();
        match self.servers.entry(name) {
            Entry::Vacant(entry) => {
                entry.insert(server);
            }
            Entry::Occupied(entry) => {
                self.duplicate.get_or_insert_with(|| entry.key().clone());
            }
        }
        self
    }

    /// Validates configuration without connecting; handshakes begin with the agent driver.
    ///
    /// # Errors
    ///
    /// Returns an error when no servers are configured, a name is empty or
    /// duplicated, a required transport field is empty, or a timeout is zero.
    pub fn build(self) -> Result<Mcp, McpBuildError> {
        if self.servers.is_empty() {
            return Err(McpBuildError::Empty);
        }
        if let Some(name) = self.duplicate {
            return Err(McpBuildError::DuplicateServer(name));
        }
        let mut discovery_timeout = Duration::ZERO;
        let mut named = Vec::with_capacity(self.servers.len());
        for (name, server) in self.servers {
            validate_server(&name, &server)?;
            discovery_timeout = discovery_timeout.max(server.startup_timeout.saturating_mul(2));
            named.push(NamedServer {
                name,
                config: server,
            });
        }
        let servers: Arc<[NamedServer]> = named.into();
        let state = Arc::new(ProviderState::new(
            servers.iter().map(|server| server.name.clone()),
            discovery_timeout,
        ));
        let search = Arc::new(McpSearch {
            state: Arc::clone(&state),
            description: search_description(&servers),
        });
        Ok(Mcp {
            servers,
            state,
            search,
            oauth_store: self.oauth_store,
            oauth_metadata: Arc::new(oauth::OAuthMetadataCache::default()),
            started: AtomicBool::new(false),
        })
    }
}

impl McpHandle {
    /// Reconnects one configured server and atomically replaces its discovered tools.
    ///
    /// # Errors
    ///
    /// Returns an error when the server name is unknown or the replacement connection cannot
    /// initialize and list its tools.
    pub async fn reload(&self, server_name: &str) -> Result<usize, McpControlError> {
        let span = info_span!(
            target: "nanocodex_tools",
            parent: None,
            "mcp.server_reload",
            otel.kind = "client",
            otel.status_code = tracing::field::Empty,
            mcp.server = server_name,
            status = tracing::field::Empty,
            tool.count = tracing::field::Empty,
        );
        // Do not enter this span while connecting. RMCP's transport task inherits the current
        // tracing context and outlives the reload operation; children use explicit parents.
        let result = self.reload_inner(server_name, &span).await;
        span.record(
            "status",
            if result.is_ok() {
                "completed"
            } else {
                "failed"
            },
        );
        span.record(
            "otel.status_code",
            if result.is_ok() { "OK" } else { "ERROR" },
        );
        if let Ok(tool_count) = &result {
            span.record("tool.count", tool_count);
        }
        result
    }

    async fn reload_inner(
        &self,
        server_name: &str,
        parent: &tracing::Span,
    ) -> Result<usize, McpControlError> {
        let server = self
            .servers
            .iter()
            .find(|server| server.name == server_name)
            .ok_or_else(|| McpControlError::UnknownServer(server_name.to_owned()))?;
        let generation = self.state.begin_server(server_name);
        let result = client::connect(
            server_name,
            &server.config,
            self.oauth_store.clone(),
            Arc::clone(&self.oauth_metadata),
            parent,
        )
        .await;
        match result {
            Ok(connected) => {
                let count = connected.tools.len();
                let entries = connected
                    .tools
                    .into_iter()
                    .map(|tool| {
                        ToolEntry::new(
                            server_name,
                            server.config.description.as_deref(),
                            &tool,
                            Arc::clone(&connected.client),
                            server.config.tool_timeout,
                            server.config.supports_parallel_tool_calls,
                        )
                    })
                    .collect();
                self.state
                    .complete_server(server_name, generation, Ok(entries));
                Ok(count)
            }
            Err(error) => {
                self.state
                    .complete_server(server_name, generation, Err(error.clone()));
                Err(McpControlError::Reload {
                    server: server_name.to_owned(),
                    error,
                })
            }
        }
    }

    /// Starts an OAuth browser login for one server.
    ///
    /// Await [`McpLogin::wait`] after opening [`McpLogin::authorization_url`]. A successful login
    /// persists the credentials and hot-reloads the server before completing.
    ///
    /// # Errors
    ///
    /// Returns an error for an unknown or unsupported server, a missing OAuth store, or when the
    /// authorization flow cannot be initialized.
    pub async fn login(&self, server_name: &str) -> Result<McpLogin, McpControlError> {
        let span = info_span!(
            target: "nanocodex_tools",
            parent: None,
            "mcp.oauth.login",
            otel.kind = "client",
            otel.status_code = tracing::field::Empty,
            mcp.server = server_name,
            status = tracing::field::Empty,
            tool.count = tracing::field::Empty,
        );
        self.login_inner(server_name, span).await
    }

    async fn login_inner(
        &self,
        server_name: &str,
        span: tracing::Span,
    ) -> Result<McpLogin, McpControlError> {
        let server = self
            .servers
            .iter()
            .find(|server| server.name == server_name)
            .ok_or_else(|| McpControlError::UnknownServer(server_name.to_owned()))?;
        let store = self
            .oauth_store
            .clone()
            .ok_or(McpControlError::NoOAuthStore)?;
        let (url, bearer, headers) = match &server.config.transport {
            config::McpTransport::StreamableHttp {
                url,
                bearer,
                headers,
            } => (url.clone(), bearer.is_some(), headers.clone()),
            config::McpTransport::Stdio { .. } => {
                return Err(McpControlError::NotHttp(server_name.to_owned()));
            }
        };
        if bearer {
            return Err(McpControlError::ExplicitBearer(server_name.to_owned()));
        }
        let flow = oauth::begin_login(server_name.to_owned(), url, headers, store)
            .instrument(span.clone())
            .await
            .map_err(|error| {
                span.record("status", "failed");
                span.record("otel.status_code", "ERROR");
                McpControlError::OAuth(error)
            })?;
        let handle = self.clone();
        let name = server_name.to_owned();
        let completion_span = span.clone();
        let completion = tokio::spawn(
            async move {
                let result = async {
                    flow.completion
                        .await
                        .map_err(|error| McpControlError::LoginTask(error.to_string()))?
                        .map_err(McpControlError::OAuth)?;
                    handle.reload(&name).await
                }
                .await;
                completion_span.record(
                    "status",
                    if result.is_ok() {
                        "completed"
                    } else {
                        "failed"
                    },
                );
                completion_span.record(
                    "otel.status_code",
                    if result.is_ok() { "OK" } else { "ERROR" },
                );
                if let Ok(tool_count) = &result {
                    completion_span.record("tool.count", tool_count);
                }
                result
            }
            .instrument(span),
        );
        Ok(McpLogin {
            authorization_url: flow.authorization_url,
            completion,
        })
    }
}

impl McpLogin {
    /// Returns the URL the embedding application should open in a browser.
    #[must_use]
    pub fn authorization_url(&self) -> &str {
        &self.authorization_url
    }

    /// Waits for the browser callback and the automatic server reload.
    ///
    /// # Errors
    ///
    /// Returns an error when authorization, credential persistence, or the subsequent hot reload
    /// fails.
    pub async fn wait(self) -> Result<usize, McpControlError> {
        self.completion
            .await
            .map_err(|error| McpControlError::LoginTask(error.to_string()))?
    }
}

#[async_trait]
impl DynamicToolProvider for Mcp {
    fn start(&self) {
        if self.started.swap(true, Ordering::AcqRel) {
            return;
        }
        for server in &*self.servers {
            let name = server.name.clone();
            let config = server.config.clone();
            let state = Arc::clone(&self.state);
            let oauth_store = self.oauth_store.clone();
            let oauth_metadata = Arc::clone(&self.oauth_metadata);
            let span = info_span!(
                target: "nanocodex_tools",
                parent: None,
                "mcp.server_start",
                otel.kind = "client",
                otel.status_code = tracing::field::Empty,
                mcp.server = %name,
                status = tracing::field::Empty,
                tool.count = tracing::field::Empty,
            );
            drop(tokio::spawn(async move {
                let result = client::connect(&name, &config, oauth_store, oauth_metadata, &span)
                    .await
                    .map(|connected| {
                        connected
                            .tools
                            .into_iter()
                            .map(|tool| {
                                ToolEntry::new(
                                    &name,
                                    config.description.as_deref(),
                                    &tool,
                                    Arc::clone(&connected.client),
                                    config.tool_timeout,
                                    config.supports_parallel_tool_calls,
                                )
                            })
                            .collect::<Vec<_>>()
                    });
                span.record(
                    "status",
                    if result.is_ok() {
                        "completed"
                    } else {
                        "failed"
                    },
                );
                span.record(
                    "otel.status_code",
                    if result.is_ok() { "OK" } else { "ERROR" },
                );
                if let Ok(tools) = &result {
                    span.record("tool.count", tools.len());
                }
                state.complete_server(&name, 0, result);
            }));
        }
    }

    fn direct_tools(&self) -> Vec<Arc<dyn Tool>> {
        vec![Arc::clone(&self.search) as Arc<dyn Tool>]
    }

    fn available_definitions(&self) -> Vec<ToolDefinition> {
        self.state.available_definitions()
    }

    fn contains(&self, name: &str) -> bool {
        self.state.active_entry(name).is_some()
    }

    fn supports_parallel_tool_calls(&self, name: &str) -> bool {
        self.state
            .active_entry(name)
            .is_some_and(|entry| entry.supports_parallel_tool_calls)
    }

    async fn execute(
        &self,
        name: &str,
        input: Value,
        _context: ToolContext<'_>,
    ) -> Option<ToolOutput> {
        let entry = self.state.active_entry(name)?;
        let Value::Object(arguments) = input else {
            return Some(ToolOutput::error(format!(
                "MCP tool {name} requires an object argument"
            )));
        };
        let argument_bytes = serde_json::to_vec(&arguments).map_or(0, |encoded| encoded.len());
        let argument_keys = arguments
            .keys()
            .map(String::as_str)
            .collect::<Vec<_>>()
            .join(",");
        let argument_count = arguments.len();
        let params =
            CallToolRequestParams::new(entry.remote_name.clone()).with_arguments(arguments);
        let span = info_span!(
            target: "nanocodex_tools",
            "mcp.tool_call",
            otel.kind = "client",
            otel.status_code = tracing::field::Empty,
            mcp.server = %entry.server_name,
            mcp.tool = %entry.remote_name,
            mcp.arguments.bytes = argument_bytes,
            mcp.arguments.keys = argument_keys,
            mcp.arguments.count = argument_count,
            status = tracing::field::Empty,
        );
        let result = match tokio::time::timeout(
            entry.timeout,
            entry.client.call_tool(params).instrument(span.clone()),
        )
        .await
        {
            Ok(Ok(result)) => result,
            Ok(Err(error)) => {
                span.record("status", "failed");
                span.record("otel.status_code", "ERROR");
                return Some(ToolOutput::error(format!(
                    "MCP tool {}/{} failed: {error}",
                    entry.server_name, entry.remote_name
                )));
            }
            Err(_) => {
                span.record("status", "timeout");
                span.record("otel.status_code", "ERROR");
                return Some(ToolOutput::error(format!(
                    "MCP tool {}/{} exceeded {:.1} seconds",
                    entry.server_name,
                    entry.remote_name,
                    entry.timeout.as_secs_f64()
                )));
            }
        };
        let success = !result.is_error.unwrap_or(false);
        span.record("status", if success { "completed" } else { "failed" });
        span.record("otel.status_code", if success { "OK" } else { "ERROR" });
        let value = match serde_json::to_value(result) {
            Ok(value) => value,
            Err(error) => {
                span.record("status", "failed");
                span.record("otel.status_code", "ERROR");
                return Some(ToolOutput::error(format!(
                    "failed to encode MCP tool result: {error}"
                )));
            }
        };
        Some(ToolOutput::from_json(value, success).with_metadata(json!({
            "mcp_server": entry.server_name,
            "mcp_tool": entry.remote_name,
        })))
    }
}

struct McpSearch {
    state: Arc<ProviderState>,
    description: String,
}

#[derive(Deserialize)]
#[serde(deny_unknown_fields)]
struct SearchInput {
    query: String,
    #[serde(default)]
    limit: Option<usize>,
}

#[async_trait]
impl Tool for McpSearch {
    fn definition(&self) -> ToolDefinition {
        ToolDefinition::tool_search(
            "client",
            self.description.clone(),
            json!({
                "type": "object",
                "properties": {
                    "query": {
                        "type": "string",
                        "description": "Search query for deferred tools."
                    },
                    "limit": {
                        "type": "number",
                        "description": "Maximum number of tools to return. Defaults to 8."
                    }
                },
                "required": ["query"],
                "additionalProperties": false
            }),
        )
    }

    fn supports_parallel_tool_calls(&self) -> bool {
        true
    }

    async fn execute(&self, input: ToolInput, _context: ToolContext<'_>) -> ToolResult {
        let input = input.decode_json::<SearchInput>()?;
        let span = info_span!(
            target: "nanocodex_tools",
            "mcp.catalog_search",
            otel.kind = "internal",
            otel.status_code = tracing::field::Empty,
            query.bytes = input.query.len(),
            result.count = tracing::field::Empty,
            pending_servers = tracing::field::Empty,
            status = tracing::field::Empty,
        );
        tracing::info!(
            target: "nanocodex_tools",
            parent: &span,
            query = %input.query,
            limit = input.limit,
            "MCP catalog search"
        );
        let result = self
            .state
            .search(&input.query, input.limit)
            .instrument(span.clone())
            .await;
        span.record(
            "status",
            if result.is_ok() {
                "completed"
            } else {
                "failed"
            },
        );
        span.record(
            "otel.status_code",
            if result.is_ok() { "OK" } else { "ERROR" },
        );
        if let Ok(result) = &result {
            span.record("result.count", result.tool_count());
            span.record("pending_servers", result.pending_server_count());
        }
        Ok(match result {
            Ok(result) => match result.loadable_tools() {
                Ok(tools) => ToolOutput::json(&result).with_code_mode_value(tools),
                Err(error) => {
                    ToolOutput::error(format!("failed to encode MCP tool definitions: {error}"))
                }
            },
            Err(error) => ToolOutput::error(error),
        })
    }
}

fn validate_server(name: &str, server: &McpServer) -> Result<(), McpBuildError> {
    if name.trim().is_empty() {
        return Err(McpBuildError::EmptyName);
    }
    if let Some(option) = server.unsupported_option {
        return Err(McpBuildError::UnsupportedOption {
            server: name.to_owned(),
            option,
        });
    }
    let (field, value) = match &server.transport {
        config::McpTransport::Stdio { command, .. } => ("command", command.as_str()),
        config::McpTransport::StreamableHttp { url, .. } => ("URL", url.as_str()),
    };
    if value.trim().is_empty() {
        return Err(McpBuildError::EmptyField {
            server: name.to_owned(),
            field,
        });
    }
    for (field, timeout) in [
        ("startup timeout", server.startup_timeout),
        ("tool timeout", server.tool_timeout),
    ] {
        if timeout.is_zero() {
            return Err(McpBuildError::ZeroTimeout {
                server: name.to_owned(),
                field,
            });
        }
    }
    Ok(())
}

fn search_description(servers: &[NamedServer]) -> String {
    let sources = servers
        .iter()
        .map(|server| match server.config.description.as_deref() {
            Some(description) => format!("- {}: {}", server.name, description.trim()),
            None => format!("- {}", server.name),
        })
        .collect::<Vec<_>>()
        .join("\n");
    format!(
        "# Tool discovery\n\nSearches over deferred tool metadata with BM25 and exposes matching tools for the next model call.\n\nYou have access to tools from the following sources:\n{sources}\nSome of the tools may not have been provided to you upfront, and you should use this tool (`tool_search`) to search for the required tools. For MCP tool discovery, always use `tool_search` instead of `list_mcp_resources` or `list_mcp_resource_templates`."
    )
}

#[cfg(test)]
mod tests {
    use super::*;
    use crate::{ToolOutputBody, contract::DEFAULT_TOOL_OUTPUT_TOKENS};
    use futures_util::future::join_all;
    use nanocodex_oai_api::MODEL;
    use serde_json::value::to_raw_value;

    fn test_context(session_id: &'static str, call_id: &'static str) -> ToolContext<'static> {
        ToolContext::new(MODEL, session_id, call_id, &[], DEFAULT_TOOL_OUTPUT_TOKENS)
    }

    #[test]
    fn validates_empty_and_duplicate_servers() {
        assert!(matches!(Mcp::builder().build(), Err(McpBuildError::Empty)));
        assert!(matches!(
            Mcp::builder()
                .server("docs", McpServer::http("https://example.test/mcp"))
                .server("docs", McpServer::stdio("node"))
                .build(),
            Err(McpBuildError::DuplicateServer(name)) if name == "docs"
        ));
        assert!(matches!(
            Mcp::builder()
                .server(
                    "local",
                    McpServer::stdio("node").bearer_token("not-applicable")
                )
                .build(),
            Err(McpBuildError::UnsupportedOption {
                server,
                option: "bearer_token"
            }) if server == "local"
        ));
    }

    #[test]
    fn search_definition_uses_the_native_provider_contract() {
        let mcp = Mcp::builder()
            .server(
                "docs",
                McpServer::http("https://example.test/mcp")
                    .description("Search product documentation."),
            )
            .build()
            .unwrap();
        let definition = serde_json::to_value(mcp.search.definition()).unwrap();
        assert_eq!(
            definition,
            json!({
                "type": "tool_search",
                "execution": "client",
                "description": "# Tool discovery\n\nSearches over deferred tool metadata with BM25 and exposes matching tools for the next model call.\n\nYou have access to tools from the following sources:\n- docs: Search product documentation.\nSome of the tools may not have been provided to you upfront, and you should use this tool (`tool_search`) to search for the required tools. For MCP tool discovery, always use `tool_search` instead of `list_mcp_resources` or `list_mcp_resource_templates`.",
                "parameters": {
                    "type": "object",
                    "properties": {
                        "query": {
                            "type": "string",
                            "description": "Search query for deferred tools."
                        },
                        "limit": {
                            "type": "number",
                            "description": "Maximum number of tools to return. Defaults to 8."
                        }
                    },
                    "required": ["query"],
                    "additionalProperties": false
                }
            })
        );
    }

    #[tokio::test]
    async fn stdio_handshake_search_and_call_share_the_background_client() {
        let fixture = std::path::Path::new(env!("CARGO_MANIFEST_DIR"))
            .join("tests/fixtures/mcp-stdio-server.mjs");
        let mcp = Mcp::builder()
            .server(
                "fixture",
                McpServer::stdio("node").arg(fixture.to_string_lossy()),
            )
            .build()
            .unwrap();
        mcp.start();
        let context = test_context("test-session", "search-call");
        let search = mcp
            .search
            .execute(
                ToolInput::Function(to_raw_value(&json!({ "query": "echo message" })).unwrap()),
                context,
            )
            .await
            .unwrap();
        assert!(search.success);
        assert_eq!(
            search.code_mode_value(),
            json!([
                {
                    "type": "namespace",
                    "name": "mcp__fixture__",
                    "description": "Tools in the mcp__fixture__ namespace.",
                    "tools": [
                        {
                            "type": "function",
                            "name": "echo",
                            "description": "Echo deterministic MCP fixture message 0.",
                            "strict": false,
                            "defer_loading": true,
                            "parameters": {
                                "type": "object",
                                "properties": {
                                    "message": { "type": "string" },
                                    "delay_ms": {
                                        "type": "integer",
                                        "minimum": 0,
                                        "maximum": 1000
                                    }
                                },
                                "required": ["message"],
                                "additionalProperties": false
                            }
                        }
                    ]
                }
            ])
        );
        assert!(matches!(
            &search.output,
            ToolOutputBody::Text(output) if output.contains("mcp__fixture__echo")
        ));
        assert!(
            mcp.available_definitions()
                .iter()
                .any(|definition| definition.name() == "mcp__fixture__echo")
        );

        let execution = mcp
            .execute(
                "mcp__fixture__echo",
                json!({ "message": "hello" }),
                test_context("test-session", "tool-call"),
            )
            .await
            .unwrap();
        assert!(execution.success);
        assert!(matches!(
            execution.output,
            ToolOutputBody::Text(output) if output.contains("fixture:hello")
        ));
    }

    #[cfg(unix)]
    #[tokio::test]
    async fn dropping_mcp_terminates_stdio_descendants() {
        let nonce = std::time::SystemTime::now()
            .duration_since(std::time::SystemTime::UNIX_EPOCH)
            .unwrap()
            .as_nanos();
        let directory = std::env::temp_dir().join(format!(
            "nanocodex-mcp-descendant-{}-{nonce}",
            std::process::id()
        ));
        std::fs::create_dir_all(&directory).unwrap();
        let marker = directory.join("survived");
        let fixture = std::path::Path::new(env!("CARGO_MANIFEST_DIR"))
            .join("tests/fixtures/mcp-stdio-server.mjs");
        let mcp = Mcp::builder()
            .server(
                "fixture",
                McpServer::stdio("node")
                    .arg(fixture.to_string_lossy())
                    .env("NANOCODEX_MCP_DESCENDANT_MARKER", marker.to_string_lossy()),
            )
            .build()
            .unwrap();
        mcp.start();
        let search = mcp
            .search
            .execute(
                ToolInput::Function(to_raw_value(&json!({ "query": "echo" })).unwrap()),
                test_context("drop-session", "drop-call"),
            )
            .await
            .unwrap();
        assert!(search.success);

        drop(mcp);
        tokio::time::sleep(Duration::from_secs(1)).await;
        let escaped = marker.exists();
        std::fs::remove_dir_all(directory).unwrap();

        assert!(!escaped, "dropping MCP left a stdio descendant running");
    }

    #[tokio::test]
    async fn reload_replaces_a_live_server_without_restarting_or_deactivating_tools() {
        let fixture = std::path::Path::new(env!("CARGO_MANIFEST_DIR"))
            .join("tests/fixtures/mcp-stdio-server.mjs");
        let mcp = Mcp::builder()
            .server(
                "fixture",
                McpServer::stdio("node").arg(fixture.to_string_lossy()),
            )
            .build()
            .unwrap();
        let handle = mcp.handle();
        mcp.start();
        let context = test_context("reload-session", "search-call");
        let search = mcp
            .search
            .execute(
                ToolInput::Function(to_raw_value(&json!({ "query": "echo message" })).unwrap()),
                context,
            )
            .await
            .unwrap();
        assert!(search.success);

        assert_eq!(handle.reload("fixture").await.unwrap(), 1);
        assert!(
            mcp.available_definitions()
                .iter()
                .any(|definition| definition.name() == "mcp__fixture__echo")
        );
        let execution = mcp
            .execute(
                "mcp__fixture__echo",
                json!({ "message": "after-reload" }),
                test_context("reload-session", "tool-call"),
            )
            .await
            .unwrap();
        assert!(execution.success);
        assert!(matches!(
            execution.output,
            ToolOutputBody::Text(output) if output.contains("fixture:after-reload")
        ));
    }

    #[tokio::test]
    async fn concurrent_server_startup_and_remote_calls_are_bounded_and_reusable() {
        const SERVERS: usize = 8;
        const CALLS: usize = 256;

        let fixture = std::path::Path::new(env!("CARGO_MANIFEST_DIR"))
            .join("tests/fixtures/mcp-stdio-server.mjs");
        let mut builder = Mcp::builder();
        for index in 0..SERVERS {
            builder = builder.server(
                format!("fixture_{index}"),
                McpServer::stdio("node").arg(fixture.to_string_lossy()),
            );
        }
        let mcp = builder.build().unwrap();
        mcp.start();
        let context = test_context("stress-session", "stress-call");
        let search = mcp
            .search
            .execute(
                ToolInput::Function(
                    to_raw_value(&json!({ "query": "echo message", "limit": 32 })).unwrap(),
                ),
                context,
            )
            .await
            .unwrap();
        assert!(search.success);
        let names = mcp
            .available_definitions()
            .into_iter()
            .map(|definition| definition.name().to_owned())
            .collect::<Vec<_>>();
        assert_eq!(names.len(), SERVERS);
        assert!(
            names
                .iter()
                .all(|name| mcp.supports_parallel_tool_calls(name)),
            "the read-only echo fixture must retain its parallel-safety hint"
        );

        let calls = (0..CALLS).map(|index| {
            mcp.execute(
                &names[index % names.len()],
                json!({ "message": index.to_string() }),
                context,
            )
        });
        let results = join_all(calls).await;
        assert!(
            results
                .into_iter()
                .all(|result| { result.is_some_and(|execution| execution.success) })
        );
    }

    #[tokio::test]
    #[ignore = "manual Streamable HTTP MCP handshake and discovery smoke"]
    async fn smoke_http_servers_from_environment() {
        let configured = std::env::var("NANOCODEX_MCP_SMOKE_SERVERS")
            .expect("set NANOCODEX_MCP_SMOKE_SERVERS to comma-separated NAME=URL entries");
        let bearers = std::env::var("NANOCODEX_MCP_SMOKE_BEARERS")
            .ok()
            .map(|configured| {
                configured
                    .split(',')
                    .map(|entry| {
                        let (name, variable) = entry
                            .split_once('=')
                            .expect("each smoke bearer must use NAME=ENV");
                        (name.to_owned(), variable.to_owned())
                    })
                    .collect::<BTreeMap<_, _>>()
            })
            .unwrap_or_default();
        let mut builder = Mcp::builder();
        for entry in configured.split(',') {
            let (name, url) = entry
                .split_once('=')
                .expect("each smoke server must use NAME=URL");
            let mut server = McpServer::http(url).startup_timeout(Duration::from_mins(2));
            if let Some(variable) = bearers.get(name) {
                server = server.bearer_token_env(variable);
            }
            builder = builder.server(name, server);
        }
        let mcp = builder.build().unwrap();
        mcp.start();
        let result = mcp
            .search
            .execute(
                ToolInput::Function(
                    to_raw_value(&json!({
                        "query": "documentation status health search list",
                        "limit": 32
                    }))
                    .unwrap(),
                ),
                test_context("http-smoke-session", "http-smoke-search"),
            )
            .await
            .unwrap();
        assert!(result.success);
        let ToolOutputBody::Text(output) = result.output else {
            panic!("expected JSON text search result");
        };
        let output: Value = serde_json::from_str(&output).unwrap();
        assert_eq!(output["pending_servers"], 0);
        assert_eq!(output["failed_servers"], json!({}));
        let tools = output["tools"].as_array().expect("tools must be an array");
        assert!(!tools.is_empty());
        let names = tools
            .iter()
            .filter_map(|tool| tool["name"].as_str())
            .collect::<Vec<_>>();
        eprintln!("HTTP MCP smoke discovered {} tools: {names:?}", tools.len());
    }
}