falsegreen_agent/

mcp.rs

//! MCP client integration backed by the official `rmcp` Rust SDK.
//!
//! This module owns configured MCP connections, negotiated server metadata, tool
//! discovery, deterministic namespacing, bounded calls, and graceful shutdown.
//! It deliberately does not own the agent runtime or FalseGreen authority logic.

use std::collections::{BTreeMap, BTreeSet, HashMap};
use std::env;
use std::fs;
use std::path::{Path, PathBuf};
use std::process::Stdio;
use std::sync::atomic::{AtomicU64, Ordering};
use std::sync::{Mutex, MutexGuard, OnceLock};
use std::time::Duration;

use http::{HeaderName, HeaderValue};
use rmcp::model::{
    CallToolRequest, CallToolRequestParams, ClientCapabilities, ClientInfo, ClientRequest,
    Implementation, ListToolsRequest, PaginatedRequestParams, ServerPeerInfo, ServerResult, Tool,
};
use rmcp::service::{PeerRequestOptions, RunningService};
use rmcp::transport::streamable_http_client::StreamableHttpClientTransportConfig;
use rmcp::transport::{StreamableHttpClientTransport, TokioChildProcess};
use rmcp::{RoleClient, ServiceExt};
use serde::{Deserialize, Serialize};
use serde_json::{Map, Value, json};
use sha2::{Digest, Sha256};
use thiserror::Error;
use tokio::runtime::Runtime;
use url::Url;

use crate::event::EventStore;
use crate::genui::{
    ActionCatalog, GenUiError, Surface, adapt_mcp_schema, mcp_schema_digest, mcp_tool_surface,
    validate_mcp_submission,
};
use crate::inference::ToolCall;
use crate::tools::ToolResult;

const CONFIG_VERSION: u32 = 1;
const MAX_CONFIG_BYTES: usize = 1024 * 1024;
const MAX_SERVERS: usize = 32;
const MAX_TOOLS_PER_SERVER: usize = 1024;
const MAX_LIST_PAGES: usize = 64;
const MAX_REQUEST_BYTES_HARD: usize = 1024 * 1024;
const MAX_RESULT_BYTES_HARD: usize = 4 * 1024 * 1024;
const MAX_SCHEMA_BYTES_HARD: usize = 4 * 1024 * 1024;
const MAX_TIMEOUT_MS: u64 = 15 * 60 * 1000;
const DEFAULT_STARTUP_TIMEOUT_MS: u64 = 15_000;
const DEFAULT_REQUEST_TIMEOUT_MS: u64 = 60_000;
const DEFAULT_SHUTDOWN_TIMEOUT_MS: u64 = 3_000;
const DEFAULT_MAX_REQUEST_BYTES: usize = 256 * 1024;
const DEFAULT_MAX_RESULT_BYTES: usize = 256 * 1024;
const DEFAULT_MAX_SCHEMA_BYTES: usize = 512 * 1024;
const MAX_ERROR_BYTES: usize = 64 * 1024;

static MCP_PUBLICATION_LOCK: OnceLock<Mutex<()>> = OnceLock::new();

pub(crate) type McpPublicationGuard = MutexGuard<'static, ()>;

pub(crate) fn acquire_publication_guard() -> McpPublicationGuard {
    MCP_PUBLICATION_LOCK
        .get_or_init(|| Mutex::new(()))
        .lock()
        .expect("MCP publication lock is not poisoned")
}

#[derive(Debug, Error)]
pub enum McpError {
    #[error("MCP config I/O failed for {path}: {source}")]
    ConfigIo {
        path: String,
        #[source]
        source: std::io::Error,
    },
    #[error("MCP config exceeds the {MAX_CONFIG_BYTES} byte limit")]
    ConfigTooLarge,
    #[error("MCP config JSON is invalid: {0}")]
    ConfigJson(#[from] serde_json::Error),
    #[error("invalid MCP config: {0}")]
    InvalidConfig(String),
    #[error("failed to create the MCP async runtime: {0}")]
    Runtime(#[source] std::io::Error),
    #[error("MCP server {server:?} failed during {phase}: {message}")]
    Server {
        server: String,
        phase: &'static str,
        message: String,
    },
}

#[derive(Debug, Clone, Deserialize)]
#[serde(deny_unknown_fields)]
struct McpConfig {
    version: u32,
    servers: BTreeMap<String, McpServerConfig>,
}

#[derive(Debug, Clone, Deserialize)]
#[serde(tag = "transport", rename_all = "snake_case", deny_unknown_fields)]
enum McpServerConfig {
    Stdio {
        command: String,
        #[serde(default)]
        args: Vec<String>,
        #[serde(default)]
        env: BTreeMap<String, String>,
        #[serde(default)]
        cwd: Option<PathBuf>,
        #[serde(default = "default_startup_timeout_ms")]
        startup_timeout_ms: u64,
        #[serde(default = "default_request_timeout_ms")]
        request_timeout_ms: u64,
        #[serde(default = "default_shutdown_timeout_ms")]
        shutdown_timeout_ms: u64,
        #[serde(default = "default_max_request_bytes")]
        max_request_bytes: usize,
        #[serde(default = "default_max_result_bytes")]
        max_result_bytes: usize,
        #[serde(default = "default_max_schema_bytes")]
        max_schema_bytes: usize,
    },
    StreamableHttp {
        url: String,
        #[serde(default)]
        bearer_token_env: Option<String>,
        #[serde(default)]
        headers: BTreeMap<String, String>,
        #[serde(default)]
        header_env: BTreeMap<String, String>,
        #[serde(default = "default_startup_timeout_ms")]
        startup_timeout_ms: u64,
        #[serde(default = "default_request_timeout_ms")]
        request_timeout_ms: u64,
        #[serde(default = "default_shutdown_timeout_ms")]
        shutdown_timeout_ms: u64,
        #[serde(default = "default_max_request_bytes")]
        max_request_bytes: usize,
        #[serde(default = "default_max_result_bytes")]
        max_result_bytes: usize,
        #[serde(default = "default_max_schema_bytes")]
        max_schema_bytes: usize,
    },
}

impl McpServerConfig {
    const fn limits(&self) -> McpServerLimits {
        match self {
            Self::Stdio {
                startup_timeout_ms,
                request_timeout_ms,
                shutdown_timeout_ms,
                max_request_bytes,
                max_result_bytes,
                max_schema_bytes,
                ..
            }
            | Self::StreamableHttp {
                startup_timeout_ms,
                request_timeout_ms,
                shutdown_timeout_ms,
                max_request_bytes,
                max_result_bytes,
                max_schema_bytes,
                ..
            } => McpServerLimits {
                startup_timeout_ms: *startup_timeout_ms,
                request_timeout_ms: *request_timeout_ms,
                shutdown_timeout_ms: *shutdown_timeout_ms,
                max_request_bytes: *max_request_bytes,
                max_result_bytes: *max_result_bytes,
                max_schema_bytes: *max_schema_bytes,
            },
        }
    }
}

#[derive(Debug, Clone, Copy)]
struct McpServerLimits {
    startup_timeout_ms: u64,
    request_timeout_ms: u64,
    shutdown_timeout_ms: u64,
    max_request_bytes: usize,
    max_result_bytes: usize,
    max_schema_bytes: usize,
}

const fn default_startup_timeout_ms() -> u64 {
    DEFAULT_STARTUP_TIMEOUT_MS
}

const fn default_request_timeout_ms() -> u64 {
    DEFAULT_REQUEST_TIMEOUT_MS
}

const fn default_shutdown_timeout_ms() -> u64 {
    DEFAULT_SHUTDOWN_TIMEOUT_MS
}

const fn default_max_request_bytes() -> usize {
    DEFAULT_MAX_REQUEST_BYTES
}

const fn default_max_result_bytes() -> usize {
    DEFAULT_MAX_RESULT_BYTES
}

const fn default_max_schema_bytes() -> usize {
    DEFAULT_MAX_SCHEMA_BYTES
}

#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize)]
pub enum ToolPolicy {
    ReadOnly,
    Consequential,
}

/// A fresh, read-only projection of the MCP registry state used by the
/// execution path. This is intentionally separate from any GenUI handle so a
/// caller cannot claim that a cached handle is a live source query.
#[derive(Debug, Clone, PartialEq, Eq, Serialize)]
pub struct LiveToolIdentity {
    pub exposed_tool: String,
    pub provider_id: String,
    pub server_id: String,
    pub remote_tool_name: String,
    pub schema_digest: String,
    pub policy_version: u64,
    pub policy_digest: String,
    pub requires_confirmation: bool,
}

impl ToolPolicy {
    #[must_use]
    pub const fn requires_confirmation(self) -> bool {
        matches!(self, Self::Consequential)
    }

    #[must_use]
    pub const fn as_str(self) -> &'static str {
        match self {
            Self::ReadOnly => "read_only",
            Self::Consequential => "consequential",
        }
    }

    #[must_use]
    pub fn digest(self, version: u64) -> String {
        sha256_hex(format!("falsegreen.tool-policy.v1|{}|{version}", self.as_str()).as_bytes())
    }
}

#[derive(Debug, Clone)]
struct ToolRoute {
    connection_index: usize,
    server: String,
    remote_tool: String,
    exposed_tool: String,
    provider_id: String,
    policy: ToolPolicy,
    policy_version: u64,
}

struct McpConnection {
    server: String,
    transport: &'static str,
    request_timeout: Duration,
    shutdown_timeout: Duration,
    max_request_bytes: usize,
    max_result_bytes: usize,
    peer_info: ServerPeerInfo,
    service: RunningService<RoleClient, ClientInfo>,
}

/// A fixed collection of initialized MCP clients and their startup-discovered tools.
pub struct McpClientSet {
    runtime: Runtime,
    connections: Vec<McpConnection>,
    routes: BTreeMap<String, ToolRoute>,
    schemas: Vec<Value>,
    /// Monotonic authority token for route/server/exposed-tool/schema/policy
    /// mutations. Admission captures and re-reads this token in addition to
    /// the content digest at the publication boundary.
    authority_generation: AtomicU64,
}

impl std::fmt::Debug for McpClientSet {
    fn fmt(&self, formatter: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        formatter
            .debug_struct("McpClientSet")
            .field("connections", &self.connections.len())
            .field("routes", &self.routes.keys().collect::<Vec<_>>())
            .finish_non_exhaustive()
    }
}

impl McpClientSet {
    /// Parse and validate configuration without starting servers or exposing configured values.
    pub fn inspect_config_path(path: impl AsRef<Path>) -> Result<Value, McpError> {
        let config = load_config(path.as_ref())?;
        validate_config(&config)?;
        let servers = config
            .servers
            .iter()
            .map(|(name, server)| {
                let (transport, required_environment) = match server {
                    McpServerConfig::Stdio {
                        env: server_env, ..
                    } => (
                        "stdio",
                        server_env
                            .keys()
                            .map(|variable| {
                                json!({"name": variable, "available": env::var_os(variable).is_some()})
                            })
                            .collect::<Vec<_>>(),
                    ),
                    McpServerConfig::StreamableHttp {
                        bearer_token_env,
                        header_env,
                        ..
                    } => {
                        let mut variables = bearer_token_env.iter().collect::<Vec<_>>();
                        variables.extend(header_env.values());
                        variables.sort();
                        variables.dedup();
                        (
                            "streamable_http",
                            variables
                                .into_iter()
                                .map(|variable| {
                                    json!({"name": variable, "available": env::var_os(variable).is_some()})
                                })
                                .collect::<Vec<_>>(),
                        )
                    }
                };
                json!({
                    "name": name,
                    "transport": transport,
                    "required_environment": required_environment
                })
            })
            .collect::<Vec<_>>();
        Ok(json!({"version": config.version, "servers": servers}))
    }

    /// Parse configuration, initialize every server, negotiate capabilities, and discover tools.
    pub fn from_config_path(path: impl AsRef<Path>) -> Result<Self, McpError> {
        let config = load_config(path.as_ref())?;
        Self::from_config(config)
    }

    fn from_config(config: McpConfig) -> Result<Self, McpError> {
        validate_config(&config)?;
        let runtime = tokio::runtime::Builder::new_multi_thread()
            .enable_all()
            .thread_name("falsegreen-agent-mcp")
            .build()
            .map_err(McpError::Runtime)?;
        let mut connections = Vec::with_capacity(config.servers.len());
        let mut routes = BTreeMap::new();
        let mut schemas = Vec::new();

        for (server, server_config) in config.servers {
            let limits = server_config.limits();
            let initialized = runtime.block_on(connect(&server, &server_config))?;
            let connection_index = connections.len();
            let mut server_schema_bytes = 0usize;
            let mut remote_names = BTreeSet::new();
            for tool in initialized.tools {
                let remote_tool = tool.name.to_string();
                if !remote_names.insert(remote_tool.clone()) {
                    return Err(server_error(
                        &server,
                        "tools/list",
                        format!("duplicate remote tool name {remote_tool:?}"),
                    ));
                }
                let exposed_tool = namespaced_tool_name(&server, &remote_tool);
                if routes.contains_key(&exposed_tool) {
                    return Err(server_error(
                        &server,
                        "tools/list",
                        format!("namespaced tool collision at {exposed_tool:?}"),
                    ));
                }
                let schema = openai_schema(&server, &exposed_tool, &tool);
                let schema_bytes = serde_json::to_vec(&schema)?.len();
                server_schema_bytes = server_schema_bytes.saturating_add(schema_bytes);
                if server_schema_bytes > limits.max_schema_bytes {
                    return Err(server_error(
                        &server,
                        "tools/list",
                        format!(
                            "discovered tool schemas exceed the configured {} byte limit",
                            limits.max_schema_bytes
                        ),
                    ));
                }
                // Remote hints are advisory only. Host policy defaults to
                // consequential and can be relaxed by trusted integration code.
                let policy = ToolPolicy::Consequential;
                routes.insert(
                    exposed_tool.clone(),
                    ToolRoute {
                        connection_index,
                        server: server.clone(),
                        remote_tool,
                        exposed_tool,
                        provider_id: provider_identity(&server, &server_config),
                        policy,
                        policy_version: 1,
                    },
                );
                schemas.push(schema);
            }
            connections.push(McpConnection {
                server,
                transport: initialized.transport,
                request_timeout: Duration::from_millis(limits.request_timeout_ms),
                shutdown_timeout: Duration::from_millis(limits.shutdown_timeout_ms),
                max_request_bytes: limits.max_request_bytes,
                max_result_bytes: limits.max_result_bytes,
                peer_info: initialized.peer_info,
                service: initialized.service,
            });
        }

        Ok(Self {
            runtime,
            connections,
            routes,
            schemas,
            authority_generation: AtomicU64::new(1),
        })
    }

    #[must_use]
    pub fn tool_schemas(&self) -> &[Value] {
        &self.schemas
    }

    /// Return the exact OpenAI-compatible schema retained for an exposed MCP
    /// tool. The name is the collision-resistant host identity, not a model
    /// supplied alias.
    #[must_use]
    pub fn tool_schema(&self, tool_name: &str) -> Option<&Value> {
        self.schemas.iter().find(|schema| {
            schema
                .get("function")
                .and_then(|function| function.get("name"))
                .and_then(Value::as_str)
                == Some(tool_name)
        })
    }

    /// Build a trusted GenUI form from the original MCP input schema. This
    /// does not grant permission or invoke the tool.
    pub fn surface_for_tool(
        &self,
        surface_id: impl Into<String>,
        tool_name: &str,
        state_digest: impl Into<String>,
    ) -> Result<Option<Surface>, McpError> {
        let Some(schema) = self.tool_schema(tool_name) else {
            return Ok(None);
        };
        let parameters = schema
            .get("function")
            .and_then(|function| function.get("parameters"))
            .unwrap_or(schema);
        mcp_tool_surface(surface_id, tool_name, parameters, state_digest)
            .map(Some)
            .map_err(genui_error)
    }

    /// Create a host-bound presentation and catalog entry for one discovered
    /// route. The opaque action ID is generated by the host adapter, while the
    /// provider/session/principal/schema identity is retained privately.
    pub fn prepare_genui_surface(
        &self,
        _surface_id: impl Into<String>,
        _session_id: &str,
        _principal: &str,
        _tool_name: &str,
        _state_digest: impl Into<String>,
    ) -> Result<Option<(Surface, ActionCatalog)>, McpError> {
        if !self.routes.contains_key(_tool_name) {
            return Ok(None);
        }
        Err(McpError::InvalidConfig(
            "durable current state is required; use prepare_genui_surface_with_store".to_owned(),
        ))
    }

    /// Create a host-bound GenUI presentation from the durable Agent/session
    /// state. A caller-provided surface digest is never treated as current
    /// authority; the state row is the only source for generation and digest.
    pub fn prepare_genui_surface_with_store(
        &self,
        store: &EventStore,
        surface_id: impl Into<String>,
        session_id: &str,
        principal: &str,
        tool_name: &str,
    ) -> Result<Option<(Surface, ActionCatalog)>, McpError> {
        let Some(route) = self.routes.get(tool_name) else {
            return Ok(None);
        };
        let Some(schema) = self.tool_schema(tool_name) else {
            return Ok(None);
        };
        let parameters = schema
            .get("function")
            .and_then(|function| function.get("parameters"))
            .unwrap_or(schema);
        let (state_generation, state_digest, authorization_context) = store
            .genui_current_state(session_id, principal)
            .map_err(|error| McpError::InvalidConfig(error.to_string()))?
            .ok_or_else(|| {
                McpError::InvalidConfig(
                    "no durable trusted current state exists for this session and principal"
                        .to_owned(),
                )
            })?;
        let mut surface = mcp_tool_surface(surface_id, tool_name, parameters, state_digest.clone())
            .map_err(genui_error)?;
        if route.policy == ToolPolicy::Consequential {
            surface.actions[0].kind = crate::genui::ActionKind::Consequential;
        }
        let digest = surface.digest().map_err(genui_error)?;
        let mut catalog = ActionCatalog::default();
        let action = surface
            .actions
            .first()
            .ok_or_else(|| McpError::InvalidConfig("MCP surface had no action".to_owned()))?;
        catalog
            .bind_mcp_action_with_context(
                action,
                &surface.id,
                &digest,
                session_id,
                principal,
                &authorization_context,
                state_generation,
                &state_digest,
                &route.provider_id,
                &route.server,
                tool_name,
                &mcp_schema_digest(parameters),
                route.policy_version,
                &route.policy.digest(route.policy_version),
                route.policy.requires_confirmation(),
                crate::genui::ActionSourceType::Mcp,
            )
            .map_err(genui_error)?;
        catalog
            .set_remote_tool_name(action.id.as_str(), &route.remote_tool)
            .map_err(genui_error)?;
        catalog
            .restore_current_state(store, session_id, principal)
            .map_err(genui_error)?;
        Ok(Some((surface, catalog)))
    }

    /// Fail closed against the original MCP schema before a GenUI-bound call
    /// is handed to the existing authenticated MCP transport.
    pub fn validate_tool_arguments(
        &self,
        tool_name: &str,
        arguments: &Value,
    ) -> Result<(), McpError> {
        let schema = self
            .tool_schema(tool_name)
            .ok_or_else(|| McpError::InvalidConfig(format!("unknown MCP tool {tool_name:?}")))?;
        let parameters = schema
            .get("function")
            .and_then(|function| function.get("parameters"))
            .unwrap_or(schema);
        validate_mcp_submission(parameters, arguments).map_err(genui_error)
    }

    /// Canonical model-facing MCP gate. Read-only calls are schema-validated
    /// and routed to the private transport seam; consequential calls return a
    /// structured confirmation requirement and never reach the transport.
    #[must_use]
    pub fn execute_validated(&self, call: &ToolCall) -> Option<ToolResult> {
        if self.may_mutate(&call.name) {
            let mut result = ToolResult::rejected(
                &call.name,
                "host confirmation is required for consequential MCP tools",
            );
            result.metadata = json!({
                "authority_state": "confirmation_required",
                "confirmation_required": true,
                "tool": call.name,
            });
            return Some(result);
        }
        self.execute_authorized(call)
    }

    /// Crate-private authority seam used only after the Agent/GenUI host has
    /// revalidated policy, identity, state, confirmation, and lifecycle.
    pub(crate) fn execute_validated_authorized(&self, call: &ToolCall) -> Option<ToolResult> {
        self.execute_authorized(call)
    }

    /// Public compatibility entry point for ordinary read-only MCP calls.
    /// It is an authority gate, never a raw transport executor.
    #[must_use]
    pub fn execute(&self, call: &ToolCall) -> Option<ToolResult> {
        self.execute_validated(call)
    }

    #[must_use]
    pub fn handles(&self, tool_name: &str) -> bool {
        self.routes.contains_key(tool_name)
    }

    #[must_use]
    pub fn may_mutate(&self, tool_name: &str) -> bool {
        self.routes
            .get(tool_name)
            .is_some_and(|route| route.policy == ToolPolicy::Consequential)
    }

    pub fn set_tool_policy(&mut self, tool_name: &str, policy: ToolPolicy) -> Result<(), McpError> {
        let _guard = acquire_publication_guard();
        let route = self
            .routes
            .get_mut(tool_name)
            .ok_or_else(|| McpError::InvalidConfig(format!("unknown MCP tool {tool_name:?}")))?;
        route.policy = policy;
        route.policy_version = route.policy_version.saturating_add(1);
        self.bump_authority_generation();
        Ok(())
    }

    /// Qualification-only confirmation-policy mutation. The execution route
    /// derives confirmation from the live policy, so this exercises the same
    /// authority transition used by the real confirmation gate.
    #[doc(hidden)]
    pub fn set_confirmation_requirement_for_testing(
        &mut self,
        tool_name: &str,
        requires_confirmation: bool,
    ) -> Result<(), McpError> {
        self.set_tool_policy(
            tool_name,
            if requires_confirmation {
                ToolPolicy::Consequential
            } else {
                ToolPolicy::ReadOnly
            },
        )
    }

    #[must_use]
    pub fn tool_identity(&self, tool_name: &str) -> Option<(&str, &str, &str)> {
        self.routes.get(tool_name).map(|route| {
            (
                route.provider_id.as_str(),
                route.server.as_str(),
                route.remote_tool.as_str(),
            )
        })
    }

    pub fn validate_tool_identity(
        &self,
        tool_name: &str,
        provider_id: &str,
        schema_digest: &str,
    ) -> Result<(), McpError> {
        let route = self
            .routes
            .get(tool_name)
            .ok_or_else(|| McpError::InvalidConfig(format!("unknown MCP tool {tool_name:?}")))?;
        self.validate_tool_identity_full(
            tool_name,
            provider_id,
            &route.server,
            schema_digest,
            route.policy_version,
            &route.policy.digest(route.policy_version),
        )
    }

    pub fn validate_tool_identity_full(
        &self,
        tool_name: &str,
        provider_id: &str,
        server_id: &str,
        schema_digest: &str,
        policy_version: u64,
        policy_digest: &str,
    ) -> Result<(), McpError> {
        let route = self
            .routes
            .get(tool_name)
            .ok_or_else(|| McpError::InvalidConfig(format!("unknown MCP tool {tool_name:?}")))?;
        if route.provider_id != provider_id {
            return Err(McpError::InvalidConfig(
                "MCP provider identity changed".to_owned(),
            ));
        }
        if route.server != server_id {
            return Err(McpError::InvalidConfig(
                "MCP server identity changed".to_owned(),
            ));
        }
        if route.policy_version != policy_version
            || route.policy.digest(route.policy_version) != policy_digest
        {
            return Err(McpError::InvalidConfig(
                "MCP tool policy changed".to_owned(),
            ));
        }
        let schema = self
            .tool_schema(tool_name)
            .ok_or_else(|| McpError::InvalidConfig("MCP schema disappeared".to_owned()))?;
        let parameters = schema
            .get("function")
            .and_then(|function| function.get("parameters"))
            .unwrap_or(schema);
        if mcp_schema_digest(parameters) != schema_digest {
            return Err(McpError::InvalidConfig(
                "MCP schema digest changed".to_owned(),
            ));
        }
        Ok(())
    }

    #[must_use]
    pub fn tool_policy_identity(&self, tool_name: &str) -> Option<(u64, String, bool)> {
        self.routes.get(tool_name).map(|route| {
            (
                route.policy_version,
                route.policy.digest(route.policy_version),
                route.policy.requires_confirmation(),
            )
        })
    }

    /// Query the current route, schema, and policy identity from the same
    /// registry used by execution. No cached GenUI/catalog state is consulted.
    #[must_use]
    pub fn live_tool_identity(&self, tool_name: &str) -> Option<LiveToolIdentity> {
        let route = self.routes.get(tool_name)?;
        let schema = self.tool_schema(tool_name)?;
        let parameters = schema
            .get("function")
            .and_then(|function| function.get("parameters"))
            .unwrap_or(schema);
        Some(LiveToolIdentity {
            exposed_tool: route.exposed_tool.clone(),
            provider_id: route.provider_id.clone(),
            server_id: route.server.clone(),
            remote_tool_name: route.remote_tool.clone(),
            schema_digest: mcp_schema_digest(parameters),
            policy_version: route.policy_version,
            policy_digest: route.policy.digest(route.policy_version),
            requires_confirmation: route.policy.requires_confirmation(),
        })
    }

    /// Stable digest of the live MCP registry/source authority.  It covers
    /// every exposed route, remote identity, schema, policy, and confirmation
    /// field used by admission and execution.
    #[must_use]
    pub fn live_registry_digest(&self) -> String {
        let identities = self
            .routes
            .keys()
            .filter_map(|name| self.live_tool_identity(name))
            .collect::<Vec<_>>();
        mcp_schema_digest(&serde_json::to_value(identities).unwrap_or_else(|_| json!([])))
    }

    /// Monotonic live-registry generation used as a publication CAS token.
    /// The digest remains the authoritative content check; this token closes
    /// ABA-style mutation windows where a route could be changed and restored
    /// between two reads.
    #[must_use]
    pub fn live_registry_generation(&self) -> u64 {
        self.authority_generation.load(Ordering::Acquire)
    }

    fn bump_authority_generation(&self) {
        self.authority_generation.fetch_add(1, Ordering::AcqRel);
    }

    /// Qualification-only mutation seam for exercising the production live
    /// admission fence. It changes the in-memory registry that execution and
    /// `live_tool_identity` both read; it is never used by normal runtime code.
    #[doc(hidden)]
    pub fn mutate_live_identity_for_testing(
        &mut self,
        tool_name: &str,
        provider_id: Option<&str>,
        server_id: Option<&str>,
        remote_tool_name: Option<&str>,
        schema_digest: Option<&str>,
    ) -> Result<(), McpError> {
        let _guard = acquire_publication_guard();
        let route = self
            .routes
            .get_mut(tool_name)
            .ok_or_else(|| McpError::InvalidConfig(format!("unknown MCP tool {tool_name:?}")))?;
        if let Some(provider_id) = provider_id {
            route.provider_id = provider_id.to_owned();
        }
        if let Some(server_id) = server_id {
            route.server = server_id.to_owned();
        }
        if let Some(remote_tool_name) = remote_tool_name {
            route.remote_tool = remote_tool_name.to_owned();
        }
        if let Some(schema_digest) = schema_digest {
            // Keep the schema structurally valid while changing its live
            // digest. The caller supplies a hex identity to make the mutation
            // deterministic; the actual schema payload remains host-owned.
            if schema_digest.len() != 64
                || !schema_digest.bytes().all(|byte| byte.is_ascii_hexdigit())
            {
                return Err(McpError::InvalidConfig(
                    "schema digest must be 64 hex characters".to_owned(),
                ));
            }
            let schema = self
                .schemas
                .iter_mut()
                .find(|schema| {
                    schema
                        .get("function")
                        .and_then(|function| function.get("name"))
                        .and_then(Value::as_str)
                        == Some(tool_name)
                })
                .ok_or_else(|| McpError::InvalidConfig("MCP schema disappeared".to_owned()))?;
            let function = schema
                .get_mut("function")
                .and_then(Value::as_object_mut)
                .ok_or_else(|| McpError::InvalidConfig("MCP schema is malformed".to_owned()))?;
            function.insert(
                "parameters".to_owned(),
                json!({"type": "object", "x-live-schema-digest": schema_digest}),
            );
        }
        self.bump_authority_generation();
        Ok(())
    }

    /// Qualification-only exposed-name rebinding. The old route disappears
    /// from the live registry and the schema is renamed, which makes a handle
    /// captured under the old exposed identity fail admission before any call.
    #[doc(hidden)]
    pub fn rebind_exposed_tool_for_testing(
        &mut self,
        old_tool_name: &str,
        new_tool_name: &str,
    ) -> Result<(), McpError> {
        let _guard = acquire_publication_guard();
        if self.routes.contains_key(new_tool_name) {
            return Err(McpError::InvalidConfig(
                "exposed tool name already exists".to_owned(),
            ));
        }
        let mut route = self.routes.remove(old_tool_name).ok_or_else(|| {
            McpError::InvalidConfig(format!("unknown MCP tool {old_tool_name:?}"))
        })?;
        let schema = self
            .schemas
            .iter_mut()
            .find(|schema| {
                schema
                    .get("function")
                    .and_then(|function| function.get("name"))
                    .and_then(Value::as_str)
                    == Some(old_tool_name)
            })
            .ok_or_else(|| McpError::InvalidConfig("MCP schema disappeared".to_owned()))?;
        let function = schema
            .get_mut("function")
            .and_then(Value::as_object_mut)
            .ok_or_else(|| McpError::InvalidConfig("MCP schema is malformed".to_owned()))?;
        function.insert("name".to_owned(), Value::String(new_tool_name.to_owned()));
        route.exposed_tool = new_tool_name.to_owned();
        self.routes.insert(new_tool_name.to_owned(), route);
        self.bump_authority_generation();
        Ok(())
    }

    /// Qualification-only route deletion used to prove missing-route
    /// fail-closed admission.  The old route and its schema both disappear
    /// from the live registry used by transport.
    #[doc(hidden)]
    pub fn remove_live_route_for_testing(&mut self, tool_name: &str) -> Result<(), McpError> {
        let _guard = acquire_publication_guard();
        self.routes
            .remove(tool_name)
            .ok_or_else(|| McpError::InvalidConfig(format!("unknown MCP tool {tool_name:?}")))?;
        self.schemas.retain(|schema| {
            schema
                .get("function")
                .and_then(|function| function.get("name"))
                .and_then(Value::as_str)
                != Some(tool_name)
        });
        self.bump_authority_generation();
        Ok(())
    }

    /// Return non-secret startup provenance suitable for durable event history.
    #[must_use]
    pub fn discovery_metadata(&self) -> Value {
        let servers = self
            .connections
            .iter()
            .enumerate()
            .map(|(connection_index, connection)| {
                let tools = self
                    .routes
                    .values()
                    .filter(|route| route.connection_index == connection_index)
                    .map(|route| {
                        let schema_status = self
                            .tool_schema(&route.exposed_tool)
                            .and_then(|schema| {
                                schema
                                    .get("function")
                                    .and_then(|function| function.get("parameters"))
                                    .or(Some(schema))
                            })
                            .map(|parameters| {
                                let source_digest = mcp_schema_digest(parameters);
                                match adapt_mcp_schema(parameters) {
                                    Ok(ir) => json!({
                                        "supported": true,
                                        "adapter": ir.adapter,
                                        "field_count": ir.fields.len(),
                                        "source_digest": source_digest
                                    }),
                                    Err(error) => json!({
                                        "supported": false,
                                        "source_digest": source_digest,
                                        "reason": error.to_string()
                                    }),
                                }
                            })
                            .unwrap_or_else(|| {
                                json!({
                                    "supported": false,
                                    "reason": "schema_missing"
                                })
                            });
                        json!({
                            "remote_tool": route.remote_tool,
                            "exposed_tool": route.exposed_tool,
                            "policy": route.policy.as_str(),
                            "policy_version": route.policy_version,
                            "provider_id": route.provider_id,
                            "schema_status": schema_status
                        })
                    })
                    .collect::<Vec<_>>();
                json!({
                    "server": connection.server,
                    "transport": connection.transport,
                    "protocol_version": connection.peer_info.protocol_version,
                    "server_info": connection.peer_info.server_info,
                    "capabilities": connection.peer_info.capabilities,
                    "tools": tools
                })
            })
            .collect::<Vec<_>>();
        json!({"sdk": "rmcp", "sdk_version": "3.3.0", "servers": servers})
    }

    fn execute_authorized(&self, call: &ToolCall) -> Option<ToolResult> {
        let route = self.routes.get(&call.name)?;
        if let Err(error) = self.validate_tool_arguments(&call.name, &call.arguments) {
            return Some(ToolResult::rejected(&call.name, &error.to_string()));
        }
        let connection = &self.connections[route.connection_index];
        let argument_bytes = match serde_json::to_vec(&call.arguments) {
            Ok(bytes) => bytes,
            Err(error) => {
                return Some(error_result(
                    route,
                    format!("MCP arguments could not be serialized: {error}"),
                    0,
                    None,
                ));
            }
        };
        let arguments = match call.arguments.as_object() {
            Some(arguments) => arguments.clone(),
            None => {
                return Some(error_result(
                    route,
                    "MCP arguments must be a JSON object".to_owned(),
                    argument_bytes.len(),
                    Some(sha256_hex(&argument_bytes)),
                ));
            }
        };
        if argument_bytes.len() > connection.max_request_bytes {
            return Some(error_result(
                route,
                format!(
                    "MCP arguments exceed the configured {} byte limit",
                    connection.max_request_bytes
                ),
                argument_bytes.len(),
                Some(sha256_hex(&argument_bytes)),
            ));
        }

        let result = self.runtime.block_on(call_tool(
            &connection.service,
            &route.remote_tool,
            arguments,
            connection.request_timeout,
        ));
        Some(match result {
            Ok(result) => {
                bounded_result(route, &argument_bytes, result, connection.max_result_bytes)
            }
            Err(error) => error_result(
                route,
                format!("MCP tools/call failed: {error}"),
                argument_bytes.len(),
                Some(sha256_hex(&argument_bytes)),
            ),
        })
    }
}

fn genui_error(error: GenUiError) -> McpError {
    McpError::InvalidConfig(format!("GenUI schema/action validation failed: {error}"))
}

fn load_config(path: &Path) -> Result<McpConfig, McpError> {
    let bytes = fs::read(path).map_err(|source| McpError::ConfigIo {
        path: path.display().to_string(),
        source,
    })?;
    if bytes.len() > MAX_CONFIG_BYTES {
        return Err(McpError::ConfigTooLarge);
    }
    serde_json::from_slice(&bytes).map_err(McpError::from)
}

impl Drop for McpClientSet {
    fn drop(&mut self) {
        for connection in &mut self.connections {
            let _ = self.runtime.block_on(
                connection
                    .service
                    .close_with_timeout(connection.shutdown_timeout),
            );
        }
    }
}

struct InitializedConnection {
    transport: &'static str,
    peer_info: ServerPeerInfo,
    tools: Vec<Tool>,
    service: RunningService<RoleClient, ClientInfo>,
}

async fn connect(
    server: &str,
    config: &McpServerConfig,
) -> Result<InitializedConnection, McpError> {
    let limits = config.limits();
    let client_info = ClientInfo::new(
        ClientCapabilities::default(),
        Implementation::new("falsegreen-agent", env!("CARGO_PKG_VERSION")),
    );
    let startup_timeout = Duration::from_millis(limits.startup_timeout_ms);
    let (transport, service) = match config {
        McpServerConfig::Stdio {
            command,
            args,
            env,
            cwd,
            ..
        } => {
            let mut command_process = tokio::process::Command::new(command);
            command_process
                .args(args)
                .envs(env)
                .stdin(Stdio::piped())
                .stdout(Stdio::piped())
                .stderr(Stdio::inherit())
                .kill_on_drop(true);
            if let Some(cwd) = cwd {
                command_process.current_dir(cwd);
            }
            let transport = TokioChildProcess::new(command_process)
                .map_err(|error| server_error(server, "spawn", error.to_string()))?;
            let service = tokio::time::timeout(startup_timeout, client_info.serve(transport))
                .await
                .map_err(|_| {
                    server_error(
                        server,
                        "initialize",
                        format!("timed out after {} ms", limits.startup_timeout_ms),
                    )
                })?
                .map_err(|error| server_error(server, "initialize", error.to_string()))?;
            ("stdio", service)
        }
        McpServerConfig::StreamableHttp {
            url,
            bearer_token_env,
            headers,
            header_env,
            ..
        } => {
            let mut custom_headers = HashMap::new();
            for (name, value) in headers {
                insert_header(server, &mut custom_headers, name, value)?;
            }
            for (name, variable) in header_env {
                let value = env::var(variable).map_err(|error| {
                    server_error(
                        server,
                        "configuration",
                        format!("environment variable {variable:?} is unavailable: {error}"),
                    )
                })?;
                insert_header(server, &mut custom_headers, name, &value)?;
            }
            let mut transport_config = StreamableHttpClientTransportConfig::with_uri(url.clone())
                .custom_headers(custom_headers)
                .max_concurrent_requests(1)
                .control_request_timeout(Duration::from_millis(limits.shutdown_timeout_ms))
                .session_recovery_timeout(startup_timeout)
                .max_sse_event_size(limits.max_result_bytes);
            if let Some(variable) = bearer_token_env {
                let token = env::var(variable).map_err(|error| {
                    server_error(
                        server,
                        "configuration",
                        format!("environment variable {variable:?} is unavailable: {error}"),
                    )
                })?;
                transport_config = transport_config.auth_header(token);
            }
            let transport = StreamableHttpClientTransport::from_config(transport_config);
            let service = tokio::time::timeout(startup_timeout, client_info.serve(transport))
                .await
                .map_err(|_| {
                    server_error(
                        server,
                        "initialize",
                        format!("timed out after {} ms", limits.startup_timeout_ms),
                    )
                })?
                .map_err(|error| server_error(server, "initialize", error.to_string()))?;
            ("streamable_http", service)
        }
    };

    let peer_info = service
        .peer_info()
        .ok_or_else(|| server_error(server, "initialize", "server returned no peer information"))?;
    if peer_info.capabilities.tools.is_none() {
        return Err(server_error(
            server,
            "capability negotiation",
            "server did not advertise tools capability",
        ));
    }
    let tools = list_all_tools(
        &service,
        server,
        Duration::from_millis(limits.request_timeout_ms),
    )
    .await?;
    Ok(InitializedConnection {
        transport,
        peer_info: (*peer_info).clone(),
        tools,
        service,
    })
}

async fn list_all_tools(
    service: &RunningService<RoleClient, ClientInfo>,
    server: &str,
    timeout: Duration,
) -> Result<Vec<Tool>, McpError> {
    let mut cursor = None;
    let mut tools = Vec::new();
    for _ in 0..MAX_LIST_PAGES {
        let params = PaginatedRequestParams::default().with_cursor(cursor);
        let request = ClientRequest::ListToolsRequest(ListToolsRequest::with_param(params));
        let response = service
            .send_cancellable_request(request, PeerRequestOptions::with_timeout(timeout))
            .await
            .map_err(|error| server_error(server, "tools/list", error.to_string()))?
            .await_response()
            .await
            .map_err(|error| server_error(server, "tools/list", error.to_string()))?;
        let ServerResult::ListToolsResult(result) = response else {
            return Err(server_error(
                server,
                "tools/list",
                "server returned an unexpected response type",
            ));
        };
        tools.extend(result.tools);
        if tools.len() > MAX_TOOLS_PER_SERVER {
            return Err(server_error(
                server,
                "tools/list",
                format!("server exceeds the {MAX_TOOLS_PER_SERVER} tool limit"),
            ));
        }
        cursor = result.next_cursor;
        if cursor.is_none() {
            return Ok(tools);
        }
    }
    Err(server_error(
        server,
        "tools/list",
        format!("pagination exceeds the {MAX_LIST_PAGES} page limit"),
    ))
}

async fn call_tool(
    service: &RunningService<RoleClient, ClientInfo>,
    remote_tool: &str,
    arguments: Map<String, Value>,
    timeout: Duration,
) -> Result<rmcp::model::CallToolResult, rmcp::ServiceError> {
    let params = CallToolRequestParams::new(remote_tool.to_owned()).with_arguments(arguments);
    let request = ClientRequest::CallToolRequest(CallToolRequest::new(params));
    let response = service
        .send_cancellable_request(request, PeerRequestOptions::with_timeout(timeout))
        .await?
        .await_response()
        .await?;
    match response {
        ServerResult::CallToolResult(result) => Ok(result),
        _ => Err(rmcp::ServiceError::UnexpectedResponse),
    }
}

fn validate_config(config: &McpConfig) -> Result<(), McpError> {
    if config.version != CONFIG_VERSION {
        return Err(McpError::InvalidConfig(format!(
            "version must be {CONFIG_VERSION}, found {}",
            config.version
        )));
    }
    if config.servers.len() > MAX_SERVERS {
        return Err(McpError::InvalidConfig(format!(
            "server count exceeds the {MAX_SERVERS} server limit"
        )));
    }
    for (name, server) in &config.servers {
        let limits = server.limits();
        if name.is_empty() || name.len() > 128 || name.contains('\0') {
            return Err(McpError::InvalidConfig(format!(
                "server name {name:?} must contain 1-128 non-NUL bytes"
            )));
        }
        validate_bounded_value(
            name,
            "startup_timeout_ms",
            limits.startup_timeout_ms,
            MAX_TIMEOUT_MS,
        )?;
        validate_bounded_value(
            name,
            "request_timeout_ms",
            limits.request_timeout_ms,
            MAX_TIMEOUT_MS,
        )?;
        validate_bounded_value(
            name,
            "shutdown_timeout_ms",
            limits.shutdown_timeout_ms,
            MAX_TIMEOUT_MS,
        )?;
        validate_bounded_value(
            name,
            "max_request_bytes",
            limits.max_request_bytes,
            MAX_REQUEST_BYTES_HARD,
        )?;
        validate_bounded_value(
            name,
            "max_result_bytes",
            limits.max_result_bytes,
            MAX_RESULT_BYTES_HARD,
        )?;
        validate_bounded_value(
            name,
            "max_schema_bytes",
            limits.max_schema_bytes,
            MAX_SCHEMA_BYTES_HARD,
        )?;
        match server {
            McpServerConfig::Stdio {
                command,
                args,
                env,
                cwd,
                ..
            } => {
                if command.is_empty() || command.contains('\0') {
                    return Err(McpError::InvalidConfig(format!(
                        "server {name:?} command must be non-empty and contain no NUL"
                    )));
                }
                if args.iter().any(|arg| arg.contains('\0')) {
                    return Err(McpError::InvalidConfig(format!(
                        "server {name:?} arguments must contain no NUL"
                    )));
                }
                if env.iter().any(|(key, value)| {
                    key.is_empty() || key.contains(['=', '\0']) || value.contains('\0')
                }) {
                    return Err(McpError::InvalidConfig(format!(
                        "server {name:?} environment contains an invalid key or value"
                    )));
                }
                if cwd.as_ref().is_some_and(|path| !path.is_dir()) {
                    return Err(McpError::InvalidConfig(format!(
                        "server {name:?} cwd does not name an existing directory"
                    )));
                }
            }
            McpServerConfig::StreamableHttp {
                url,
                bearer_token_env,
                headers,
                header_env,
                ..
            } => {
                validate_http_url(name, url)?;
                if bearer_token_env
                    .as_ref()
                    .is_some_and(|variable| variable.is_empty() || variable.contains(['=', '\0']))
                {
                    return Err(McpError::InvalidConfig(format!(
                        "server {name:?} bearer_token_env is invalid"
                    )));
                }
                for header_name in headers.keys().chain(header_env.keys()) {
                    HeaderName::try_from(header_name).map_err(|error| {
                        McpError::InvalidConfig(format!(
                            "server {name:?} has invalid HTTP header name {header_name:?}: {error}"
                        ))
                    })?;
                }
                if header_env
                    .values()
                    .any(|variable| variable.is_empty() || variable.contains(['=', '\0']))
                {
                    return Err(McpError::InvalidConfig(format!(
                        "server {name:?} header_env contains an invalid environment variable"
                    )));
                }
            }
        }
    }
    Ok(())
}

fn validate_bounded_value<T>(
    server: &str,
    field: &str,
    value: T,
    maximum: T,
) -> Result<(), McpError>
where
    T: Copy + Ord + std::fmt::Display + From<u8>,
{
    if value < T::from(1) || value > maximum {
        return Err(McpError::InvalidConfig(format!(
            "server {server:?} {field} must be between 1 and {maximum}, found {value}"
        )));
    }
    Ok(())
}

fn validate_http_url(server: &str, raw: &str) -> Result<(), McpError> {
    let url = Url::parse(raw).map_err(|error| {
        McpError::InvalidConfig(format!("server {server:?} URL is invalid: {error}"))
    })?;
    if !matches!(url.scheme(), "http" | "https")
        || url.host_str().is_none()
        || !url.username().is_empty()
        || url.password().is_some()
        || url.fragment().is_some()
    {
        return Err(McpError::InvalidConfig(format!(
            "server {server:?} URL must be absolute HTTP(S) without userinfo or fragment"
        )));
    }
    Ok(())
}

fn insert_header(
    server: &str,
    target: &mut HashMap<HeaderName, HeaderValue>,
    name: &str,
    value: &str,
) -> Result<(), McpError> {
    let name = HeaderName::try_from(name).map_err(|error| {
        server_error(
            server,
            "configuration",
            format!("invalid HTTP header name: {error}"),
        )
    })?;
    let value = HeaderValue::try_from(value).map_err(|error| {
        server_error(
            server,
            "configuration",
            format!("invalid value for HTTP header {name}: {error}"),
        )
    })?;
    if target.insert(name.clone(), value).is_some() {
        return Err(server_error(
            server,
            "configuration",
            format!("duplicate HTTP header {name}"),
        ));
    }
    Ok(())
}

fn namespaced_tool_name(server: &str, remote_tool: &str) -> String {
    let server_slug = slug(server, 16);
    let tool_slug = slug(remote_tool, 24);
    let mut digest = Sha256::new();
    digest.update(server.as_bytes());
    digest.update([0]);
    digest.update(remote_tool.as_bytes());
    let suffix = format!("{:x}", digest.finalize());
    format!("mcp__{server_slug}__{tool_slug}__{}", &suffix[..8])
}

fn provider_identity(server: &str, config: &McpServerConfig) -> String {
    let descriptor = match config {
        McpServerConfig::Stdio {
            command, args, cwd, ..
        } => format!("{server}:stdio:{command}:{args:?}:{cwd:?}"),
        McpServerConfig::StreamableHttp { url, .. } => format!("{server}:streamable_http:{url}"),
    };
    format!("provider_{}", sha256_hex(descriptor.as_bytes()))
}

fn slug(value: &str, maximum: usize) -> String {
    let mut output = String::new();
    for character in value.chars() {
        if output.len() >= maximum {
            break;
        }
        let character = if character.is_ascii_alphanumeric() || matches!(character, '_' | '-') {
            character.to_ascii_lowercase()
        } else {
            '_'
        };
        output.push(character);
    }
    if output.is_empty() {
        output.push('_');
    }
    output
}

fn openai_schema(server: &str, exposed_tool: &str, tool: &Tool) -> Value {
    let description = tool.description.as_deref().map_or_else(
        || {
            format!(
                "MCP tool {:?} from configured server {:?}",
                tool.name, server
            )
        },
        |description| format!("MCP server {server:?}: {description}"),
    );
    let description = truncate_schema_description(description, 160);
    // Preserve the exact server-provided schema. Presentation code may derive
    // a projection, but identity and invocation validation must use this value
    // byte-for-byte (under canonical JSON serialization).
    let parameters = Value::Object(tool.input_schema.as_ref().clone());
    json!({
        "type": "function",
        "function": {
            "name": exposed_tool,
            "description": description,
            "parameters": parameters
        }
    })
}

fn truncate_schema_description(description: String, maximum_chars: usize) -> String {
    if description.chars().count() <= maximum_chars {
        return description;
    }
    let mut truncated = description.chars().take(maximum_chars).collect::<String>();
    truncated.push('…');
    truncated
}

/// Legacy presentation helper retained for callers that explicitly request a
/// compact projection. It is deliberately not used for tool identity or
/// invocation validation; those paths retain the exact original schema.
#[allow(dead_code)]
fn remove_nonsemantic_schema_annotations(value: &mut Value) {
    match value {
        Value::Object(object) => {
            object.remove("title");
            object.remove("default");
            object.remove("examples");
            for child in object.values_mut() {
                remove_nonsemantic_schema_annotations(child);
            }
        }
        Value::Array(items) => {
            for item in items {
                remove_nonsemantic_schema_annotations(item);
            }
        }
        _ => {}
    }
}

fn bounded_result(
    route: &ToolRoute,
    argument_bytes: &[u8],
    result: rmcp::model::CallToolResult,
    maximum: usize,
) -> ToolResult {
    let serialized = match serde_json::to_vec(&result) {
        Ok(serialized) => serialized,
        Err(error) => {
            return error_result(
                route,
                format!("MCP result could not be serialized: {error}"),
                argument_bytes.len(),
                Some(sha256_hex(argument_bytes)),
            );
        }
    };
    let result_sha256 = sha256_hex(&serialized);
    let result_bytes = serialized.len();
    let is_error = result.is_error == Some(true);
    if result_bytes > maximum {
        return ToolResult {
            tool: route.exposed_tool.clone(),
            tool_call_id: None,
            ok: false,
            exit_code: None,
            timed_out: false,
            stdout: String::new(),
            stderr: format!("MCP result exceeds the configured {maximum} byte limit"),
            output_truncated: true,
            metadata: provenance(
                route,
                argument_bytes.len(),
                Some(sha256_hex(argument_bytes)),
                Some(result_bytes),
                Some(result_sha256),
                Some(is_error),
                Some("result_too_large"),
            ),
        };
    }
    ToolResult {
        tool: route.exposed_tool.clone(),
        tool_call_id: None,
        ok: !is_error,
        exit_code: None,
        timed_out: false,
        stdout: String::from_utf8(serialized).expect("serde_json emits UTF-8"),
        stderr: if is_error {
            "MCP tool returned isError=true".to_owned()
        } else {
            String::new()
        },
        output_truncated: false,
        metadata: provenance(
            route,
            argument_bytes.len(),
            Some(sha256_hex(argument_bytes)),
            Some(result_bytes),
            Some(result_sha256),
            Some(is_error),
            None,
        ),
    }
}

fn error_result(
    route: &ToolRoute,
    message: String,
    argument_bytes: usize,
    argument_sha256: Option<String>,
) -> ToolResult {
    let timed_out = message.to_ascii_lowercase().contains("timeout")
        || message.to_ascii_lowercase().contains("timed out");
    let (message, output_truncated) = truncate_utf8(message, MAX_ERROR_BYTES);
    ToolResult {
        tool: route.exposed_tool.clone(),
        tool_call_id: None,
        ok: false,
        exit_code: None,
        timed_out,
        stdout: String::new(),
        stderr: message,
        output_truncated,
        metadata: provenance(
            route,
            argument_bytes,
            argument_sha256,
            None,
            None,
            None,
            Some("request_failed"),
        ),
    }
}

fn truncate_utf8(mut value: String, maximum: usize) -> (String, bool) {
    if value.len() <= maximum {
        return (value, false);
    }
    let mut end = maximum;
    while !value.is_char_boundary(end) {
        end -= 1;
    }
    value.truncate(end);
    (value, true)
}

#[allow(clippy::too_many_arguments)]
fn provenance(
    route: &ToolRoute,
    argument_bytes: usize,
    argument_sha256: Option<String>,
    result_bytes: Option<usize>,
    result_sha256: Option<String>,
    is_error: Option<bool>,
    error_kind: Option<&str>,
) -> Value {
    json!({
        "transport_protocol": "mcp",
        "mcp_server": route.server,
        "mcp_remote_tool": route.remote_tool,
        "mcp_exposed_tool": route.exposed_tool,
        "argument_bytes": argument_bytes,
        "argument_sha256": argument_sha256,
        "result_bytes": result_bytes,
        "result_sha256": result_sha256,
        "mcp_is_error": is_error,
        "error_kind": error_kind
    })
}

fn sha256_hex(bytes: &[u8]) -> String {
    format!("{:x}", Sha256::digest(bytes))
}

fn server_error(server: &str, phase: &'static str, message: impl Into<String>) -> McpError {
    McpError::Server {
        server: server.to_owned(),
        phase,
        message: message.into(),
    }
}

#[cfg(test)]
mod tests {
    use std::collections::BTreeMap;

    use serde_json::json;

    use super::{
        CONFIG_VERSION, McpConfig, McpServerConfig, namespaced_tool_name,
        remove_nonsemantic_schema_annotations, truncate_schema_description, validate_config,
    };

    fn stdio_config() -> McpConfig {
        McpConfig {
            version: CONFIG_VERSION,
            servers: BTreeMap::from([(
                "False Green".to_owned(),
                McpServerConfig::Stdio {
                    command: "falsegreen-mcp".to_owned(),
                    args: Vec::new(),
                    env: BTreeMap::new(),
                    cwd: None,
                    startup_timeout_ms: 1_000,
                    request_timeout_ms: 1_000,
                    shutdown_timeout_ms: 1_000,
                    max_request_bytes: 1_024,
                    max_result_bytes: 1_024,
                    max_schema_bytes: 1_024,
                },
            )]),
        }
    }

    #[test]
    fn namespace_is_stable_safe_and_disambiguated() {
        let first = namespaced_tool_name("False Green", "verify/task");
        assert_eq!(first, namespaced_tool_name("False Green", "verify/task"));
        assert!(first.starts_with("mcp__false_green__verify_task__"));
        assert!(
            first
                .bytes()
                .all(|byte| byte.is_ascii_alphanumeric() || byte == b'_')
        );
        assert_ne!(
            namespaced_tool_name("a__b", "c"),
            namespaced_tool_name("a", "b__c")
        );
        assert!(first.len() <= 64);
    }

    #[test]
    fn config_rejects_unbounded_values_and_unknown_fields() {
        let mut config = stdio_config();
        match config.servers.get_mut("False Green").unwrap() {
            McpServerConfig::Stdio {
                request_timeout_ms, ..
            }
            | McpServerConfig::StreamableHttp {
                request_timeout_ms, ..
            } => *request_timeout_ms = 0,
        }
        assert!(validate_config(&config).is_err());

        let unknown = json!({
            "version": 1,
            "servers": {"s": {"transport": "stdio", "command": "x", "surprise": true}}
        });
        assert!(serde_json::from_value::<McpConfig>(unknown).is_err());
    }

    #[test]
    fn config_rejects_credential_bearing_url() {
        let value = json!({
            "version": 1,
            "servers": {
                "remote": {
                    "transport": "streamable_http",
                    "url": "https://user:secret@example.invalid/mcp"
                }
            }
        });
        let config: McpConfig = serde_json::from_value(value).expect("shape");
        assert!(validate_config(&config).is_err());
    }

    #[test]
    fn model_schema_compaction_preserves_validation_shape() {
        let mut schema = json!({
            "type": "object",
            "title": "Arguments",
            "description": "semantics",
            "properties": {
                "task_id": {
                    "type": "string",
                    "title": "Task",
                    "default": "unsafe-default",
                    "description": "The canonical task identifier."
                }
            },
            "required": ["task_id"]
        });
        remove_nonsemantic_schema_annotations(&mut schema);
        assert_eq!(schema["type"], "object");
        assert_eq!(schema["required"], json!(["task_id"]));
        assert_eq!(schema["properties"]["task_id"]["type"], "string");
        assert_eq!(
            schema["properties"]["task_id"]["description"],
            "The canonical task identifier."
        );
        assert!(schema.get("title").is_none());
        assert!(schema["properties"]["task_id"].get("default").is_none());
        assert!(
            truncate_schema_description("x".repeat(300), 160)
                .chars()
                .count()
                <= 161
        );
    }
}