codex/codex-rs/codex-mcp/src/runtime.rs

//! Runtime support for Model Context Protocol (MCP) servers.
//!
//! This module contains the thread-owned MCP runtime and data that describes the
//! environment in which MCP servers execute. Transport startup lives in
//! [`crate::rmcp_client`] and connection-set behavior lives in
//! [`crate::connection_manager`]. Registry removal revokes EMA authority before
//! the thread becomes untracked.

mod status;

#[cfg(test)]
#[path = "runtime_shared_connection_tests.rs"]
mod shared_connection_tests;

use std::collections::HashMap;
use std::collections::HashSet;
use std::path::PathBuf;
use std::sync::Arc;
use std::sync::Mutex;
use std::sync::OnceLock;
use std::sync::Weak;
use std::sync::atomic::AtomicBool;
use std::sync::atomic::AtomicU64;
use std::sync::atomic::Ordering;
use std::time::Duration;

use arc_swap::ArcSwap;
use async_channel::Sender;
use codex_config::types::McpServerDisabledReason;
use codex_connectors::ConnectorRuntimeContextKey;
use codex_connectors::ConnectorRuntimeManager;
use codex_exec_server::Environment;
use codex_exec_server::EnvironmentManager;
use codex_exec_server::HttpClient;
use codex_exec_server::RouteAwareHttpClient;
use codex_login::AuthManager;
use codex_login::CodexAuth;
use codex_protocol::ThreadId;
use codex_protocol::capabilities::SelectedCapabilityRoot;
use codex_protocol::mcp::CallToolResult;
use codex_protocol::mcp::ClientMcpExtensions;
use codex_protocol::mcp::McpResourceOriginCheckpoint;
use codex_protocol::protocol::Event;
use codex_protocol::protocol::EventMsg;
use codex_protocol::protocol::TurnEnvironmentSelection;
use codex_rmcp_client::ElicitationResponse;
use codex_rmcp_client::with_http_headers_helper;
use rmcp::model::ReadResourceRequestParams;
use rmcp::model::ReadResourceResult;
use rmcp::model::RequestId;
use tokio::sync::watch;
use tokio_util::sync::CancellationToken;
use tracing::warn;

use crate::McpConfig;
use crate::binding::McpBinding;
use crate::binding::PreparedMcpCall;
use crate::client_tool_catalog::CodexAppsToolSnapshot;
use crate::connection_manager::BindingCatalogRevision;
use crate::connection_manager::McpConnectionSet;
use crate::connection_manager::PreviousMcpConnections;
use crate::connection_pool::McpConnectionPool;
use crate::elicitation::ElicitationLifecycle;
use crate::elicitation::ElicitationRequestRouter;
use crate::elicitation::ElicitationReviewerHandle;
use crate::event_stream::McpEventStreamOpener;
use crate::mcp::CODEX_APPS_MCP_SERVER_NAME;
use crate::resource_client::McpResourceServerCacheKey;
use crate::resource_origin::ResourceOrigins;
use crate::server::EffectiveMcpServer;
use crate::tool_catalog_cache::McpToolCatalogCache;
use crate::tools::ToolInfo;

/// Controls when one task starts its eligible MCP servers.
#[derive(Clone, Copy, Debug, PartialEq, Eq)]
pub enum McpStartupPolicy {
    /// Start configured servers when their task's MCP runtime is published.
    Eager,
    /// Start servers with cached tool definitions on first use.
    LazyWhenCached,
}

/// A host-requested tool result and its identity in the published MCP catalog.
pub struct McpToolCallResult {
    pub result: CallToolResult,
    /// None when the cached catalog is missing, changed, or has no entry for the tool.
    /// This is not MCP wire metadata.
    pub source_tool_namespace: Option<String>,
}

/// Configuration and owning-thread state needed to materialize an MCP runtime.
pub struct McpRuntimeInput {
    /// Policy source captured with the configuration and connections in this publication.
    pub network_policy_metadata: Option<Arc<dyn crate::NetworkPolicyMetadataProvider>>,
    pub startup_policy: McpStartupPolicy,
    pub config: Arc<McpConfig>,
    pub plugins_available: bool,
    pub ready_selected_capability_roots: Vec<SelectedCapabilityRoot>,
    pub mcp_servers: HashMap<String, EffectiveMcpServer>,
    pub submit_id: String,
    /// Unbounded sources receive abandonment notifications; bounded sources retain request-only behavior.
    pub tx_event: Option<Sender<Event>>,
    pub startup_cancellation_token: CancellationToken,
    pub runtime_context: McpRuntimeContext,
    pub codex_apps_tools_cache: ConnectorRuntimeManager<ToolInfo>,
    pub tool_catalog_cache: McpToolCatalogCache,
    pub codex_apps_tools_cache_key: ConnectorRuntimeContextKey,
    pub client_mcp_extensions: ClientMcpExtensions,
    pub auth: Option<CodexAuth>,
    pub auth_manager: Option<Arc<AuthManager>>,
    pub elicitation_reviewer: Option<ElicitationReviewerHandle>,
    pub elicitation_lifecycle: Option<ElicitationLifecycle>,
}

/// Owns all mutable MCP state for one Codex thread.
///
/// Publication replaces the latest state atomically. Existing bindings retain
/// their exact connections and configuration for as long as they are needed.
pub struct McpRuntime {
    current: ArcSwap<PublishedMcpRuntime>,
    // Old bindings keep leases alive after publication; cancellation alone does not release them.
    superseded_connections: Mutex<Vec<Weak<McpConnectionSet>>>,
    event_stream_cancellation: Mutex<EventStreamCancellation>,
    reconnect_pending: AtomicBool,
    resource_cache_generation: AtomicU64,
    elicitation_router: ElicitationRequestRouter,
    resource_origins: Mutex<ResourceOrigins>,
    ema_auth: OnceLock<crate::ema_auth::EmaAuthContext>,
}

struct EventStreamCancellation {
    event_server_available: bool,
    cancel_event_streams_on_server_removal: watch::Sender<()>,
    retained_subscription_cancellation: Option<watch::Sender<()>>,
}

struct PublishedMcpRuntime {
    connections: Arc<McpConnectionSet>,
    config: Option<Arc<McpConfig>>,
    auth: Option<CodexAuth>,
    auth_token: Option<String>,
    auth_manager: Option<Arc<AuthManager>>,
    auth_owner: Option<AuthOwner>,
    auth_generation: Arc<()>,
    plugins_available: bool,
    ready_selected_capability_roots: Vec<SelectedCapabilityRoot>,
    environment_selections: Arc<[TurnEnvironmentSelection]>,
    ready_environments: HashMap<String, Arc<Environment>>,
    cached_binding: Mutex<Option<CachedMcpBinding>>,
}

/// The account that owns a private helper across credential refreshes.
#[derive(Clone, Copy, PartialEq, Eq)]
struct AuthOwner {
    generation: u64,
    is_fedramp: bool,
}

impl AuthOwner {
    fn capture(auth: Option<&CodexAuth>, manager: Option<&AuthManager>) -> Option<Self> {
        let auth = auth?;
        Some(Self {
            generation: manager?.auth_change_state_for(auth)?.owner_generation,
            is_fedramp: auth.is_fedramp_account(),
        })
    }
}

fn ensure_host_owned_apps_registration(
    current: &PublishedMcpRuntime,
    server: &str,
) -> anyhow::Result<()> {
    if !current
        .config
        .as_ref()
        .and_then(|config| config.mcp_server_catalog.server(server))
        .is_some_and(|registration| {
            registration
                .source()
                .is_host_owned_apps(server, registration.config())
        })
    {
        anyhow::bail!("MCP server '{server}' is not registered by the hosted runtime");
    }
    Ok(())
}

impl PublishedMcpRuntime {
    // Shared by dirty detection and cache-key publication so their auth rules agree.
    fn auth_matches(&self, auth: Option<&CodexAuth>) -> bool {
        match (self.auth.as_ref(), auth) {
            (Some(previous), Some(latest)) => {
                previous == latest
                    && previous.get_account_id() == latest.get_account_id()
                    && previous.get_chatgpt_user_id() == latest.get_chatgpt_user_id()
                    && previous.is_fedramp_account() == latest.is_fedramp_account()
                    && self.auth_token == latest.get_token().ok()
            }
            (None, None) => true,
            (Some(_), None) | (None, Some(_)) => false,
        }
    }
}

struct CachedMcpBinding {
    catalog_revisions: HashMap<String, BindingCatalogRevision>,
    // Reuse a frozen binding while a model step or caller still needs it.
    binding: Weak<McpBinding>,
}

struct McpReconnectGuard<'a> {
    pending: &'a AtomicBool,
    claimed: bool,
}

impl Drop for McpReconnectGuard<'_> {
    fn drop(&mut self) {
        if self.claimed {
            self.pending.store(true, Ordering::Release);
        }
    }
}

#[derive(Clone)]
pub(crate) struct McpPublicationGate {
    published: Option<watch::Receiver<bool>>,
}

impl McpPublicationGate {
    fn pending() -> (watch::Sender<bool>, Self) {
        let (publish, published) = watch::channel(false);
        (
            publish,
            Self {
                published: Some(published),
            },
        )
    }

    pub(crate) fn already_published() -> Self {
        Self { published: None }
    }

    pub(crate) async fn wait(mut self) -> bool {
        let Some(published) = self.published.as_mut() else {
            return true;
        };
        loop {
            if *published.borrow() {
                return true;
            }
            if published.changed().await.is_err() {
                return false;
            }
        }
    }
}

impl McpRuntime {
    /// Creates a runtime with no configured servers.
    ///
    /// This is useful while constructing a thread that must publish a stable
    /// runtime handle before its full MCP inputs are available.
    pub fn empty(prefix_mcp_tool_names: bool) -> Self {
        Self {
            current: ArcSwap::from_pointee(PublishedMcpRuntime {
                connections: Arc::new(McpConnectionSet::empty(prefix_mcp_tool_names)),
                config: None,
                auth: None,
                auth_token: None,
                auth_manager: None,
                auth_owner: None,
                auth_generation: Arc::new(()),
                plugins_available: false,
                ready_selected_capability_roots: Vec::new(),
                environment_selections: Arc::default(),
                ready_environments: HashMap::new(),
                cached_binding: Mutex::new(None),
            }),
            superseded_connections: Mutex::default(),
            event_stream_cancellation: Mutex::new(EventStreamCancellation {
                event_server_available: false,
                cancel_event_streams_on_server_removal: watch::channel(()).0,
                retained_subscription_cancellation: None,
            }),
            reconnect_pending: AtomicBool::new(false),
            resource_cache_generation: AtomicU64::new(0),
            elicitation_router: ElicitationRequestRouter::default(),
            resource_origins: Mutex::default(),
            ema_auth: OnceLock::new(),
        }
    }

    /// Updates this thread's bounded resource provenance from a live or restored event.
    pub fn observe_event(&self, event: &EventMsg) {
        if !matches!(
            event,
            EventMsg::TurnStarted(_)
                | EventMsg::ItemCompleted(_)
                | EventMsg::McpToolCallEnd(_)
                | EventMsg::ThreadRolledBack(_)
        ) {
            return;
        }
        self.resource_origins
            .lock()
            .unwrap_or_else(std::sync::PoisonError::into_inner)
            .observe(event);
    }

    /// Captures bounded widget provenance for the next compaction checkpoint.
    pub fn resource_origin_checkpoint(&self) -> Option<McpResourceOriginCheckpoint> {
        self.resource_origins
            .lock()
            .unwrap_or_else(std::sync::PoisonError::into_inner)
            .checkpoint()
    }

    /// Restores widget provenance retained by a compaction checkpoint.
    pub fn restore_resource_origin_checkpoint(&self, checkpoint: &McpResourceOriginCheckpoint) {
        self.resource_origins
            .lock()
            .unwrap_or_else(std::sync::PoisonError::into_inner)
            .restore_checkpoint(checkpoint);
    }

    /// Reads a widget through the current binding of the app tool that produced it.
    pub async fn read_resource_for_call(
        &self,
        thread_id: ThreadId,
        call_id: &str,
        params: ReadResourceRequestParams,
    ) -> anyhow::Result<ReadResourceResult> {
        let origin = self
            .resource_origins
            .lock()
            .unwrap_or_else(std::sync::PoisonError::into_inner)
            .find(call_id)?;
        let binding = self
            .current_binding_for_call(crate::CODEX_APPS_MCP_SERVER_NAME)
            .await
            .ok_or_else(|| anyhow::anyhow!("codex_apps MCP server is unavailable"))?;

        origin.read(&binding, thread_id, params).await
    }

    pub async fn new(input: McpRuntimeInput) -> Self {
        let runtime = Self::empty(input.config.prefix_mcp_tool_names);
        runtime.replace(input).await;
        runtime
    }

    /// Reconciles configured servers and publishes their immutable runtime snapshot.
    pub async fn replace(&self, input: McpRuntimeInput) {
        let mut reconnect = McpReconnectGuard {
            pending: &self.reconnect_pending,
            claimed: self.reconnect_pending.swap(false, Ordering::AcqRel),
        };
        self.publish(input, /*reuse_previous*/ !reconnect.claimed)
            .await;
        reconnect.claimed = false;
    }

    /// Starts fresh connections and returns their complete, refreshed Apps catalog.
    pub async fn replace_fresh(&self, input: McpRuntimeInput) -> anyhow::Result<Vec<ToolInfo>> {
        self.publish(input, /*reuse_previous*/ false).await;
        self.latest_hard_refresh_codex_apps_tools_cache().await
    }

    /// Revoke a removed thread's EMA access before releasing the registry lock.
    pub fn revoke_ema(&self) {
        if let Some(Ok(auth)) = self.ema_auth.get() {
            auth.authority.revoke();
        }
    }

    /// Drain EMA responses after its configuration binding has been revoked.
    pub async fn quiesce_ema(&self) {
        if let Some(Ok(auth)) = self.ema_auth.get() {
            auth.authority.revoke();
            auth.authority.quiesce().await;
        }
    }

    async fn publish(&self, input: McpRuntimeInput, reuse_previous: bool) {
        let current = self.current.load_full();
        let retain_ema = current.config.is_some();
        let ema_auth = (input.config.xaa_enabled
            && input.mcp_servers.values().any(|server| {
                server.config().enabled
                    && matches!(server.config().auth, codex_config::McpServerAuth::EmaAuth)
            }))
        .then(|| {
            self.ema_auth
                .get_or_init(|| {
                    crate::ema_auth::EmaAuth::new(
                        Arc::clone(&input.config),
                        input.auth.as_ref(),
                        input.auth_manager.as_ref(),
                    )
                    .map_err(Into::into)
                })
                .clone()
        });
        let (publish, publication_gate) = McpPublicationGate::pending();
        let config = Arc::clone(&input.config);
        let auth = input.auth.clone();
        let auth_manager = input.auth_manager.clone();
        let auth_token = auth.as_ref().and_then(|auth| auth.get_token().ok());
        if !reuse_previous {
            current.connections.retire_shared_connections();
        }
        let auth_matches = current.auth_matches(auth.as_ref());
        let auth_owner = AuthOwner::capture(auth.as_ref(), auth_manager.as_deref());
        let same_owner = match (current.auth_owner, auth_owner) {
            (Some(previous), Some(next)) => {
                previous == next
                    && current
                        .auth_manager
                        .as_ref()
                        .zip(auth_manager.as_ref())
                        .is_some_and(|(previous, next)| Arc::ptr_eq(previous, next))
            }
            (None, None) => auth_matches,
            _ => false,
        };
        let auth_generation = if auth_matches {
            Arc::clone(&current.auth_generation)
        } else {
            Arc::new(())
        };
        let plugins_available = input.plugins_available;
        let ready_selected_capability_roots = input.ready_selected_capability_roots.clone();
        let environment_selections = Arc::clone(&input.runtime_context.environment_selections);
        let ready_environments = input.runtime_context.ready_environments.clone();
        // Keep MCP startup state out of the caller's nested session-initialization future.
        let connections = Arc::new(
            Box::pin(McpConnectionSet::new(
                ema_auth,
                retain_ema.then_some(current.connections.as_ref()),
                reuse_previous.then_some(PreviousMcpConnections {
                    connections: &current.connections,
                    same_owner,
                }),
                publication_gate,
                input,
                self.elicitation_router.clone(),
            ))
            .await,
        );
        let hosted_event_server_retained = connections.contains_server(CODEX_APPS_MCP_SERVER_NAME)
            && config
                .mcp_server_catalog
                .server(CODEX_APPS_MCP_SERVER_NAME)
                .is_some_and(|registration| {
                    registration
                        .source()
                        .is_host_owned_apps(CODEX_APPS_MCP_SERVER_NAME, registration.config())
                });
        let mut cancellation = self
            .event_stream_cancellation
            .lock()
            .unwrap_or_else(std::sync::PoisonError::into_inner);
        {
            let mut superseded = self
                .superseded_connections
                .lock()
                .unwrap_or_else(std::sync::PoisonError::into_inner);
            superseded.retain(|connections| connections.strong_count() > 0);
            superseded.push(Arc::downgrade(&current.connections));
        }
        self.current.store(Arc::new(PublishedMcpRuntime {
            connections,
            config: Some(config),
            auth,
            auth_token,
            auth_manager,
            auth_owner,
            auth_generation,
            plugins_available,
            ready_selected_capability_roots,
            environment_selections,
            ready_environments,
            cached_binding: Mutex::new(None),
        }));
        let _ = publish.send(true);
        cancellation.event_server_available = hosted_event_server_retained;
        if !hosted_event_server_retained {
            cancellation
                .cancel_event_streams_on_server_removal
                .send_replace(());
            if let Some(retained) = &cancellation.retained_subscription_cancellation {
                retained.send_replace(());
            }
        }
    }

    /// Ensures the next refresh creates fresh connections for every configured server.
    pub fn reconnect_on_next_refresh(&self) {
        self.reconnect_pending.store(true, Ordering::Release);
    }

    /// Invalidates resource catalogs and contents even when connections can be reused.
    pub fn invalidate_resource_caches(&self) {
        self.resource_cache_generation
            .fetch_add(1, Ordering::AcqRel);
    }

    pub(crate) fn resource_cache_key(&self, server: &str) -> Option<McpResourceServerCacheKey> {
        let generation = self.resource_cache_generation.load(Ordering::Acquire);
        self.latest_connections()
            .resource_cache_key(server, generation)
    }

    /// Captures the latest published configuration and live client handles.
    pub async fn current_binding(&self) -> Option<Arc<McpBinding>> {
        self.current_binding_with_requirements(&[], &HashSet::new())
            .await
    }

    /// Captures one runtime, waiting for explicitly required servers and selected plugins.
    /// Plugin IDs are resolved by the captured connection set, even if a refresh publishes later.
    pub async fn current_binding_with_requirements(
        &self,
        required_servers: &[String],
        required_plugins: &HashSet<String>,
    ) -> Option<Arc<McpBinding>> {
        let current = self.current.load_full();
        current.connections.record_startup_readiness(
            "model_binding",
            required_servers,
            required_plugins,
        );
        Self::binding_from_published_runtime(current, required_servers, required_plugins).await
    }

    async fn binding_from_published_runtime(
        current: Arc<PublishedMcpRuntime>,
        required_servers: &[String],
        required_plugins: &HashSet<String>,
    ) -> Option<Arc<McpBinding>> {
        let config = Arc::clone(current.config.as_ref()?);
        let stable_catalog_revisions = current
            .connections
            .stable_catalog_revisions(required_servers, required_plugins)
            .await;
        if let Some(catalog_revisions) = &stable_catalog_revisions {
            let cached = current
                .cached_binding
                .lock()
                .unwrap_or_else(std::sync::PoisonError::into_inner);
            if let Some(cached) = cached.as_ref()
                && &cached.catalog_revisions == catalog_revisions
                && let Some(binding) = cached.binding.upgrade()
            {
                return Some(binding);
            }
        }

        let binding = Arc::new(
            current
                .connections
                .capture_binding_with_metadata(
                    config,
                    current.plugins_available,
                    required_servers,
                    required_plugins,
                )
                .await,
        );
        if let Some(catalog_revisions) = stable_catalog_revisions
            && current
                .connections
                .stable_catalog_revisions(required_servers, required_plugins)
                .await
                .as_ref()
                == Some(&catalog_revisions)
        {
            let mut cached = current
                .cached_binding
                .lock()
                .unwrap_or_else(std::sync::PoisonError::into_inner);
            if let Some(cached) = cached.as_ref()
                && cached.catalog_revisions == catalog_revisions
                && let Some(binding) = cached.binding.upgrade()
            {
                return Some(binding);
            }
            *cached = Some(CachedMcpBinding {
                catalog_revisions,
                binding: Arc::downgrade(&binding),
            });
        }
        Some(binding)
    }

    /// Returns whether the published snapshot still belongs to the current credentials.
    pub fn current_auth_matches(&self, auth: Option<&CodexAuth>) -> bool {
        self.current.load().auth_matches(auth)
    }

    pub(crate) fn auth_cache_key_for_server(
        &self,
        server: &str,
    ) -> crate::McpResourceClientAuthKey {
        let current = self.current.load();
        crate::McpResourceClientAuthKey {
            generation: Arc::clone(&current.auth_generation),
            server: server.to_string(),
            available: current.connections.contains_server(server),
        }
    }

    /// Detects newly saved credentials for servers whose startup failed authentication.
    pub async fn updated_oauth_credentials_after_auth_failure(&self) -> Vec<String> {
        let current = self.current.load_full();
        let Some(config) = current.config.as_ref() else {
            return Vec::new();
        };
        current
            .connections
            .updated_oauth_credentials_after_auth_failure(config)
            .await
    }

    /// Checks the current generation before retrying servers detected outside the refresh gate.
    pub async fn has_authentication_failed_servers(&self, server_names: &[String]) -> bool {
        self.current
            .load_full()
            .connections
            .authentication_failed_servers()
            .await
            .into_iter()
            .any(|server_name| server_names.contains(&server_name))
    }

    /// Waits for the selected server without capturing an execution binding.
    pub async fn wait_for_server_startup(&self, server: &str) {
        self.current
            .load_full()
            .connections
            .wait_for_server_startup(server)
            .await;
    }

    /// Prepares one advertised tool against this publication's current client and policy.
    /// Its model-visible name stays fixed while execution metadata comes from the live catalog.
    pub async fn prepare_call(&self, advertised_tool: &ToolInfo) -> Option<PreparedMcpCall> {
        let current = self.current.load_full();
        current
            .connections
            .prepare_call_for_tool(Arc::clone(current.config.as_ref()?), advertised_tool)
            .await
    }

    /// Captures the current runtime after its selected server has finished startup.
    pub async fn current_binding_for_call(&self, server: &str) -> Option<Arc<McpBinding>> {
        let current = self.current.load_full();
        current.config.as_ref()?;
        if !current.connections.wait_for_server_startup(server).await {
            return None;
        }
        Self::binding_from_published_runtime(
            current,
            /*required_servers*/ &[],
            /*required_plugins*/ &HashSet::new(),
        )
        .await
    }

    /// Returns the latest published configuration without waiting for clients.
    pub fn current_config(&self) -> Option<Arc<McpConfig>> {
        self.current.load().config.clone()
    }

    pub fn current_ready_selected_capability_roots(&self) -> Vec<SelectedCapabilityRoot> {
        self.current.load().ready_selected_capability_roots.clone()
    }

    /// Whether this publication uses the same selections and currently ready environment handles.
    pub fn current_environments_match(
        &self,
        selections: &[TurnEnvironmentSelection],
        ready_environments: &HashMap<String, Arc<Environment>>,
    ) -> bool {
        let current = self.current.load();
        current.config.is_some()
            && current.environment_selections.as_ref() == selections
            && current.ready_environments.len() == ready_environments.len()
            && ready_environments.iter().all(|(id, environment)| {
                current
                    .ready_environments
                    .get(id)
                    .is_some_and(|published| Arc::ptr_eq(published, environment))
            })
    }

    pub fn elicitations_auto_deny(&self) -> bool {
        self.elicitation_router.auto_deny()
    }

    pub fn set_elicitations_auto_deny(&self, auto_deny: bool) {
        self.elicitation_router.set_auto_deny(auto_deny);
    }

    pub fn enable_full_access_form_input(&self) {
        self.elicitation_router.enable_full_access_form_input();
    }

    pub async fn resolve_elicitation(
        &self,
        server_name: String,
        id: RequestId,
        response: ElicitationResponse,
    ) -> anyhow::Result<()> {
        self.elicitation_router
            .resolve(server_name, id, response)
            .await
    }

    pub async fn latest_hard_refresh_codex_apps_tools_cache(
        &self,
    ) -> anyhow::Result<Vec<ToolInfo>> {
        self.latest_connections()
            .refresh_codex_apps_tools_for_discovery()
            .await
    }

    /// Refreshes the published Apps client and returns its exact inventory and MCP eligibility.
    pub async fn refresh_codex_apps_tools(&self) -> anyhow::Result<CodexAppsToolSnapshot> {
        let current = self.current.load_full();
        let config = current
            .config
            .as_deref()
            .ok_or_else(|| anyhow::anyhow!("MCP runtime is not configured"))?;
        current
            .connections
            .refresh_codex_apps_client_catalog(config)
            .await
    }

    /// Returns tools from this thread’s ready, unshared Apps MCP client.
    /// Returns an empty list if the client is absent, still starting, or uses
    /// a shared catalog. Does not start or wait for discovery.
    pub async fn apps_tools_from_mcp_client(&self) -> Vec<ToolInfo> {
        self.latest_connections().apps_tools_from_mcp_client().await
    }

    /// Lists the latest known tools for non-model discovery surfaces.
    ///
    /// Unlike [`Self::current_binding`], this may return cached tools while their
    /// client reconnects because callers only inspect tool metadata.
    pub async fn latest_list_all_tools(&self) -> Vec<ToolInfo> {
        self.latest_connections().list_all_tools().await
    }

    pub async fn latest_read_resource(
        &self,
        server: &str,
        params: ReadResourceRequestParams,
    ) -> anyhow::Result<ReadResourceResult> {
        self.latest_connections()
            .read_resource(server, params)
            .await
    }

    pub async fn latest_call_tool_without_attribution(
        &self,
        server: &str,
        tool: &str,
        arguments: Option<serde_json::Value>,
        meta: Option<serde_json::Value>,
    ) -> anyhow::Result<CallToolResult> {
        let connections = self.latest_connections();
        connections.validate_tool_call(server, tool, /*environment_id*/ None)?;
        connections.wait_for_server_startup(server).await;
        connections
            .call_tool(
                server, tool, /*environment_id*/ None, arguments, meta,
                /*requested_timeout*/ None, /*wait_for_server*/ true,
            )
            .await
    }

    /// Calls a tool for the host, retaining attribution while its catalogs stay unchanged.
    /// Waiting calls capture a binding after the selected server starts; non-waiting calls
    /// use the cached binding. Missing or changed catalogs leave the namespace unknown.
    #[allow(clippy::too_many_arguments)]
    pub async fn latest_call_tool(
        &self,
        server: &str,
        tool: &str,
        environment_id: Option<&str>,
        arguments: Option<serde_json::Value>,
        meta: Option<serde_json::Value>,
        requested_timeout: Option<Duration>,
        wait_for_server: bool,
    ) -> anyhow::Result<McpToolCallResult> {
        let current = self.current.load_full();
        current
            .connections
            .validate_tool_call(server, tool, environment_id)?;
        let attribution = if wait_for_server
            && current.connections.wait_for_server_startup(server).await
            && let Some(revisions) = current
                .connections
                .stable_catalog_revisions(
                    /*required_servers*/ &[],
                    /*required_plugins*/ &HashSet::new(),
                )
                .await
        {
            Self::binding_from_published_runtime(
                Arc::clone(&current),
                /*required_servers*/ &[],
                /*required_plugins*/ &HashSet::new(),
            )
            .await
            .map(|binding| (revisions, binding))
        } else {
            current
                .cached_binding
                .lock()
                .unwrap_or_else(std::sync::PoisonError::into_inner)
                .as_ref()
                .and_then(|cached| {
                    Some((cached.catalog_revisions.clone(), cached.binding.upgrade()?))
                })
        };
        let result = current
            .connections
            .call_tool(
                server,
                tool,
                environment_id,
                arguments,
                meta,
                requested_timeout,
                wait_for_server,
            )
            .await?;
        // Host calls use live connections, unlike model calls pinned to a binding.
        // Check after dispatch so a refresh before or during the call invalidates attribution.
        // Revisions include client identity and increase on catalog changes. Reading them
        // does not start servers or refresh catalogs; all catalogs matter for name collisions.
        let namespace = if let Some((revisions, binding)) = attribution
            && current
                .connections
                .stable_catalog_revisions(
                    /*required_servers*/ &[],
                    /*required_plugins*/ &HashSet::new(),
                )
                .await
                .as_ref()
                == Some(&revisions)
        {
            binding
                .tool_info(server, tool)
                .map(|info| info.callable_namespace.clone())
        } else {
            None
        };
        Ok(McpToolCallResult {
            result,
            source_tool_namespace: namespace,
        })
    }

    pub async fn latest_wait_for_server_ready(&self, server: &str, timeout: Duration) -> bool {
        self.latest_connections()
            .wait_for_server_ready(server, timeout)
            .await
    }

    pub async fn validate_required_servers(&self) -> anyhow::Result<()> {
        self.latest_connections().validate_required_servers().await
    }

    pub fn cancel_startup(&self) {
        self.current.load().connections.cancel_startup();
    }

    /// Observes matching published registrations without starting or reconnecting servers.
    pub async fn connection_statuses(
        &self,
        config: &McpConfig,
    ) -> std::collections::HashMap<String, codex_protocol::mcp::McpServerConnectionStatus> {
        let current = self.current.load_full();
        let Some(published_config) = current.config.as_ref() else {
            return HashMap::new();
        };
        let mut statuses = current.connections.connection_statuses().await;
        statuses.retain(|name, _| {
            published_config
                .mcp_server_catalog
                .server(name)
                .is_some_and(|server| config.mcp_server_catalog.server(name) == Some(server))
        });
        statuses
    }

    pub(crate) fn latest_connections(&self) -> Arc<McpConnectionSet> {
        Arc::clone(&self.current.load().connections)
    }

    pub(crate) fn latest_host_owned_codex_apps_connections(
        &self,
    ) -> anyhow::Result<Arc<McpConnectionSet>> {
        let current = self.current.load();
        ensure_host_owned_apps_registration(&current, CODEX_APPS_MCP_SERVER_NAME)?;
        Ok(Arc::clone(&current.connections))
    }

    pub(crate) fn latest_connections_for_event_server(
        &self,
        server: &str,
    ) -> anyhow::Result<(Arc<McpConnectionSet>, watch::Receiver<()>)> {
        let cancellation = self
            .event_stream_cancellation
            .lock()
            .unwrap_or_else(std::sync::PoisonError::into_inner);
        let cancel_event_streams_on_server_removal = cancellation
            .cancel_event_streams_on_server_removal
            .subscribe();
        let current = self.current.load();
        if server == CODEX_APPS_MCP_SERVER_NAME {
            ensure_host_owned_apps_registration(&current, server)?;
        }
        Ok((
            Arc::clone(&current.connections),
            cancel_event_streams_on_server_removal,
        ))
    }

    pub(crate) fn event_stream_opener(&self) -> anyhow::Result<McpEventStreamOpener> {
        let cancellation = self
            .event_stream_cancellation
            .lock()
            .unwrap_or_else(std::sync::PoisonError::into_inner);
        let connection = self
            .current
            .load()
            .connections
            .event_stream_connection
            .clone()
            .ok_or_else(|| anyhow::anyhow!("Event subscriptions are unavailable for this task"))?;
        let cancel_event_streams_on_server_removal = cancellation
            .retained_subscription_cancellation
            .as_ref()
            .unwrap_or(&cancellation.cancel_event_streams_on_server_removal)
            .clone();
        Ok(McpEventStreamOpener {
            connection,
            cancellation_receiver: cancel_event_streams_on_server_removal.subscribe(),
            cancel_event_streams_on_server_removal,
        })
    }

    pub(crate) fn forward_event_server_removals_to(&self, retained: watch::Sender<()>) {
        let mut cancellation = self
            .event_stream_cancellation
            .lock()
            .unwrap_or_else(std::sync::PoisonError::into_inner);
        if !cancellation.event_server_available {
            retained.send_replace(());
        }
        cancellation.retained_subscription_cancellation = Some(retained);
    }

    pub async fn shutdown(&self) {
        self.elicitation_router.close().await;
        let superseded = std::mem::take(
            &mut *self
                .superseded_connections
                .lock()
                .unwrap_or_else(std::sync::PoisonError::into_inner),
        );
        let current = self.latest_connections();
        // Keep every publication's cleanup alive if the caller is interrupted.
        let shutdown = tokio::spawn(async move {
            for connections in superseded
                .into_iter()
                .filter_map(|connections| connections.upgrade())
            {
                connections.shutdown_shared_connections().await;
            }
            current.shutdown().await;
        });
        if let Err(error) = shutdown.await {
            warn!("MCP runtime shutdown task failed: {error}");
        }
    }
}

/// Runtime context used when resolving per-server MCP environments.
///
/// `McpConfig` describes what servers exist. This value carries the canonical
/// environment registry, captured thread selections, and the host-local cwd used by MCP processes.
#[derive(Clone)]
pub struct McpRuntimeContext {
    pub(crate) connection_pool: Option<McpConnectionPool>,
    environment_manager: Arc<EnvironmentManager>,
    environment_selections: Arc<[TurnEnvironmentSelection]>,
    ready_environments: HashMap<String, Arc<Environment>>,
    local_process_cwd: PathBuf,
    local_http_client: Arc<dyn HttpClient>,
}

/// Applies the local HTTP headers helper configured for an MCP server.
///
/// Callers retain ownership of selecting the underlying HTTP transport. This
/// function centralizes the helper-specific policy checks and decoration used
/// by both MCP runtime startup and standalone OAuth login.
pub fn apply_http_headers_helper(
    client: Arc<dyn HttpClient>,
    config: &codex_config::McpServerConfig,
    local_process_cwd: PathBuf,
) -> Result<Arc<dyn HttpClient>, String> {
    if matches!(
        config.disabled_reason,
        Some(
            McpServerDisabledReason::Requirements { .. }
                | McpServerDisabledReason::ExplicitAndRequirements { .. }
        )
    ) {
        return Err("the MCP server is disabled by managed requirements".to_string());
    }
    let codex_config::McpServerTransportConfig::StreamableHttp {
        url,
        http_headers_helper: Some(command),
        ..
    } = &config.transport
    else {
        return Ok(client);
    };
    if !config.is_local_environment() {
        return Err("HTTP headers helpers can only run in the local environment".to_string());
    }
    with_http_headers_helper(client, url, command, local_process_cwd)
        .map_err(|error| error.to_string())
}

impl McpRuntimeContext {
    pub fn new(environment_manager: Arc<EnvironmentManager>, local_process_cwd: PathBuf) -> Self {
        let local_http_client = Arc::new(
            RouteAwareHttpClient::new(environment_manager.http_client_factory().clone())
                .with_tls_backend_fallback(),
        );
        Self {
            connection_pool: None,
            environment_manager,
            environment_selections: Arc::default(),
            ready_environments: HashMap::new(),
            local_process_cwd,
            local_http_client,
        }
    }

    /// Uses this process's pool for compatible local stdio servers using MCP 2026-07-28.
    pub fn with_connection_pool(mut self, pool: McpConnectionPool) -> Self {
        self.connection_pool = Some(pool);
        self
    }

    /// Pins all captured selections and the concrete handles for environments already ready.
    pub fn with_selected_environments(
        mut self,
        environment_selections: Arc<[TurnEnvironmentSelection]>,
        ready_environments: HashMap<String, Arc<Environment>>,
    ) -> Self {
        self.environment_selections = environment_selections;
        self.ready_environments = ready_environments;
        self
    }

    pub(crate) fn local_process_cwd(&self) -> PathBuf {
        self.local_process_cwd.clone()
    }

    pub(crate) fn local_http_client(&self) -> Arc<dyn HttpClient> {
        Arc::clone(&self.local_http_client)
    }

    pub(crate) fn resolve_server_environment(
        &self,
        server_name: &str,
        config: &codex_config::McpServerConfig,
    ) -> Result<Option<Arc<Environment>>, String> {
        // Resolve `"local"` through the shared registry when available. Local
        // HTTP is the one current exception: it can use the ambient HTTP client
        // even when no local Environment is configured.
        if let Some(environment) = self
            .ready_environments
            .get(&config.environment_id)
            .cloned()
            .or_else(|| {
                self.environment_manager
                    .get_environment(&config.environment_id)
            })
        {
            return Ok(Some(environment));
        }

        if config.is_local_environment() {
            return match config.transport {
                codex_config::McpServerTransportConfig::Stdio { .. } => Err(format!(
                    "local stdio MCP server `{server_name}` requires a local environment"
                )),
                codex_config::McpServerTransportConfig::StreamableHttp { .. } => Ok(None),
            };
        }

        Err(format!(
            "MCP server `{server_name}` references unknown environment id `{}`",
            config.environment_id
        ))
    }

    /// Resolves local MCP's specialized HTTP capability or the selected remote capability.
    pub fn resolve_http_client(
        &self,
        server_name: &str,
        config: &codex_config::McpServerConfig,
    ) -> Result<Arc<dyn HttpClient>, String> {
        let environment = self.resolve_server_environment(server_name, config)?;
        self.http_client_for_server(config, environment.as_ref())
    }

    pub(crate) fn http_client_for_server(
        &self,
        config: &codex_config::McpServerConfig,
        environment: Option<&Arc<Environment>>,
    ) -> Result<Arc<dyn HttpClient>, String> {
        let client = match environment {
            Some(environment) if environment.is_remote() => environment.get_http_client(),
            Some(_) | None => self.local_http_client(),
        };
        apply_http_headers_helper(client, config, self.local_process_cwd())
    }
}

pub(crate) fn emit_duration(metric: &str, duration: Duration, tags: &[(&str, &str)]) {
    if let Some(metrics) = codex_otel::global() {
        let _ = metrics.record_duration(metric, duration, tags);
    }
}

#[cfg(test)]
mod tests {
    use codex_utils_path_uri::PathUri;
    use std::collections::HashMap;

    use codex_config::DEFAULT_MCP_SERVER_ENVIRONMENT_ID;
    use codex_config::McpServerConfig;
    use codex_config::McpServerTransportConfig;
    use codex_exec_server::EnvironmentManager;
    use codex_exec_server_test_support::environment_manager_without_environments;
    use codex_utils_path_uri::LegacyAppPathString;
    use pretty_assertions::assert_eq;

    use super::*;
    use crate::McpServerPermissions;

    fn stdio_server(environment_id: &str) -> McpServerConfig {
        McpServerConfig {
            auth: Default::default(),
            transport: McpServerTransportConfig::Stdio {
                command: "echo".to_string(),
                args: Vec::new(),
                env: None,
                env_vars: Vec::new(),
                cwd: None,
            },
            environment_id: environment_id.to_string(),
            enabled: true,
            required: false,
            startup_readiness: Default::default(),
            supports_parallel_tool_calls: false,
            tool_input_schema_max_bytes: None,
            omit_tools_from: None,
            disabled_reason: None,
            startup_timeout_sec: None,
            tool_timeout_sec: None,
            default_tools_approval_mode: None,
            enabled_tools: None,
            disabled_tools: None,
            scopes: None,
            oauth: None,
            oauth_resource: None,
            tools: HashMap::new(),
        }
    }

    #[tokio::test]
    async fn publication_gate_opens_only_for_the_winning_candidate() {
        let (publish, gate) = McpPublicationGate::pending();
        let wait = tokio::spawn(gate.wait());
        tokio::task::yield_now().await;
        assert!(!wait.is_finished());

        publish.send(true).expect("publish candidate");
        assert!(wait.await.expect("gate task"));

        let (publish, gate) = McpPublicationGate::pending();
        drop(publish);
        assert!(!gate.wait().await);
    }

    #[tokio::test]
    async fn cached_bindings_follow_the_clients_catalog_revision() -> anyhow::Result<()> {
        let codex_home = tempfile::tempdir()?;
        let cache_context = ConnectorRuntimeManager::<ToolInfo>::default().context(
            codex_home.path().to_path_buf(),
            ConnectorRuntimeContextKey::personal(
                /*account_id*/ None, /*chatgpt_user_id*/ None,
            ),
        );
        let connections =
            crate::connection_manager::tests::create_test_manager_with_ready_apps_client(
                cache_context,
                "search",
                /*list_started*/ None,
                /*release_list*/ None,
            )
            .await?;
        // Complete the fixture's shared startup future before testing stable reuse.
        connections.list_all_tools().await;
        let mut config = crate::mcp::tests::test_mcp_config(codex_home.path().to_path_buf());
        config.server_permissions.insert(
            CODEX_APPS_MCP_SERVER_NAME.to_string(),
            McpServerPermissions::Thread,
        );
        let published = Arc::new(PublishedMcpRuntime {
            connections: Arc::clone(&connections),
            config: Some(Arc::new(config)),
            auth: None,
            auth_token: None,
            auth_manager: None,
            auth_owner: None,
            auth_generation: Arc::new(()),
            plugins_available: false,
            ready_selected_capability_roots: Vec::new(),
            environment_selections: Arc::default(),
            ready_environments: HashMap::new(),
            cached_binding: Mutex::new(None),
        });
        let runtime = McpRuntime::empty(/*prefix_mcp_tool_names*/ true);
        runtime.current.store(Arc::clone(&published));
        runtime
            .latest_call_tool_without_attribution(
                CODEX_APPS_MCP_SERVER_NAME,
                "search",
                /*arguments*/ None,
                /*meta*/ None,
            )
            .await?;
        assert!(published.cached_binding.lock().unwrap().is_none());
        let before = McpRuntime::binding_from_published_runtime(
            Arc::clone(&published),
            /*required_servers*/ &[],
            /*required_plugins*/ &HashSet::new(),
        )
        .await
        .expect("initial binding");
        let repeated = McpRuntime::binding_from_published_runtime(
            Arc::clone(&published),
            /*required_servers*/ &[],
            /*required_plugins*/ &HashSet::new(),
        )
        .await
        .expect("cached initial binding");
        assert!(Arc::ptr_eq(&before, &repeated));

        connections.refresh_codex_apps_tools_for_discovery().await?;

        let refreshed = McpRuntime::binding_from_published_runtime(
            Arc::clone(&published),
            /*required_servers*/ &[],
            /*required_plugins*/ &HashSet::new(),
        )
        .await
        .expect("refreshed binding");
        assert!(!Arc::ptr_eq(&before, &refreshed));
        let call = refreshed
            .prepare_call(CODEX_APPS_MCP_SERVER_NAME, "search")
            .expect("refreshed call");
        let error = call
            .call_with_preparation(/*requested_timeout*/ None, || async {
                Err(anyhow::anyhow!("reached refreshed call preparation"))
            })
            .await
            .expect_err("stop before dispatch");
        assert!(
            error
                .to_string()
                .contains("reached refreshed call preparation")
        );
        let repeated = McpRuntime::binding_from_published_runtime(
            Arc::clone(&published),
            /*required_servers*/ &[],
            /*required_plugins*/ &HashSet::new(),
        )
        .await
        .expect("cached refreshed binding");
        assert!(Arc::ptr_eq(&refreshed, &repeated));
        let released = Arc::downgrade(&refreshed);
        drop(refreshed);
        drop(repeated);
        assert!(
            released.upgrade().is_none(),
            "the runtime cache must not pin an unused binding"
        );
        Ok(())
    }

    #[tokio::test]
    async fn cached_bindings_are_scoped_to_the_published_runtime() {
        let published = Arc::new(PublishedMcpRuntime {
            connections: Arc::new(McpConnectionSet::empty(/*prefix_mcp_tool_names*/ true)),
            config: Some(Arc::new(crate::mcp::tests::test_mcp_config(
                std::env::temp_dir(),
            ))),
            auth: None,
            auth_token: None,
            auth_manager: None,
            auth_owner: None,
            auth_generation: Arc::new(()),
            plugins_available: false,
            ready_selected_capability_roots: Vec::new(),
            environment_selections: Arc::default(),
            ready_environments: HashMap::new(),
            cached_binding: Mutex::new(None),
        });
        let first = McpRuntime::binding_from_published_runtime(
            Arc::clone(&published),
            /*required_servers*/ &[],
            /*required_plugins*/ &HashSet::new(),
        )
        .await
        .expect("first binding");
        let repeated = McpRuntime::binding_from_published_runtime(
            Arc::clone(&published),
            /*required_servers*/ &[],
            /*required_plugins*/ &HashSet::new(),
        )
        .await
        .expect("repeated binding");
        assert!(Arc::ptr_eq(&first, &repeated));

        let previous = Arc::into_inner(published).expect("published runtime has no other owners");
        let republished = Arc::new(PublishedMcpRuntime {
            cached_binding: Mutex::new(None),
            ..previous
        });
        let refreshed = McpRuntime::binding_from_published_runtime(
            republished,
            /*required_servers*/ &[],
            /*required_plugins*/ &HashSet::new(),
        )
        .await
        .expect("republished binding");
        assert!(!Arc::ptr_eq(&first, &refreshed));
    }

    fn http_server(environment_id: &str) -> McpServerConfig {
        McpServerConfig {
            auth: Default::default(),
            transport: McpServerTransportConfig::StreamableHttp {
                url: "http://127.0.0.1:1".to_string(),
                bearer_token_env_var: None,
                http_headers: None,
                env_http_headers: None,
                http_headers_helper: None,
            },
            environment_id: environment_id.to_string(),
            ..stdio_server(environment_id)
        }
    }

    #[test]
    fn local_stdio_requires_local_stdio_availability() {
        let runtime_context = McpRuntimeContext::new(
            Arc::new(environment_manager_without_environments()),
            PathBuf::from("/tmp"),
        );

        let error = match runtime_context
            .resolve_server_environment("stdio", &stdio_server(DEFAULT_MCP_SERVER_ENVIRONMENT_ID))
        {
            Ok(_) => panic!("local stdio MCP should require a local environment"),
            Err(error) => error,
        };
        assert_eq!(
            error,
            "local stdio MCP server `stdio` requires a local environment"
        );
    }

    #[test]
    fn local_http_does_not_require_local_stdio_availability() {
        let runtime_context = McpRuntimeContext::new(
            Arc::new(environment_manager_without_environments()),
            PathBuf::from("/tmp"),
        );

        let resolved_runtime = match runtime_context
            .resolve_server_environment("http", &http_server(DEFAULT_MCP_SERVER_ENVIRONMENT_ID))
        {
            Ok(resolved_runtime) => resolved_runtime,
            Err(error) => panic!("local HTTP MCP should resolve: {error}"),
        };
        assert!(resolved_runtime.is_none());
    }

    #[tokio::test]
    async fn local_http_client_is_shared_across_resolution_and_context_clones() {
        for environment_manager in [
            EnvironmentManager::default_for_tests(),
            environment_manager_without_environments(),
        ] {
            let runtime_context =
                McpRuntimeContext::new(Arc::new(environment_manager), PathBuf::from("/tmp"));
            let config = http_server(DEFAULT_MCP_SERVER_ENVIRONMENT_ID);
            let first_client = runtime_context
                .resolve_http_client("http", &config)
                .expect("first local HTTP capability should resolve");
            let repeated_client = runtime_context
                .resolve_http_client("http", &config)
                .expect("repeated local HTTP capability should resolve");
            let resolved_environment = runtime_context
                .resolve_server_environment("http", &config)
                .expect("local HTTP environment should resolve");
            let startup_client = runtime_context
                .http_client_for_server(&config, resolved_environment.as_ref())
                .expect("startup local HTTP capability should resolve");
            let cloned_client = runtime_context
                .clone()
                .resolve_http_client("http", &config)
                .expect("cloned local HTTP capability should resolve");

            assert!(Arc::ptr_eq(&first_client, &repeated_client));
            assert!(Arc::ptr_eq(&first_client, &startup_client));
            assert!(Arc::ptr_eq(&first_client, &cloned_client));
        }
    }

    #[test]
    fn unknown_explicit_environment_is_rejected() {
        let runtime_context = McpRuntimeContext::new(
            Arc::new(environment_manager_without_environments()),
            PathBuf::from("/tmp"),
        );

        let error =
            match runtime_context.resolve_server_environment("stdio", &stdio_server("remote")) {
                Ok(_) => panic!("unknown MCP environment should fail"),
                Err(error) => error,
            };
        assert_eq!(
            error,
            "MCP server `stdio` references unknown environment id `remote`"
        );
    }

    #[tokio::test]
    async fn explicit_remote_stdio_and_http_accept_named_environment() {
        let runtime_context = McpRuntimeContext::new(
            Arc::new(
                EnvironmentManager::create_for_tests(
                    Some("ws://127.0.0.1:8765".to_string()),
                    /*local_runtime_paths*/ None,
                )
                .await,
            ),
            PathBuf::from("/tmp"),
        );

        let mut remote_stdio = stdio_server("remote");
        let McpServerTransportConfig::Stdio { cwd, .. } = &mut remote_stdio.transport else {
            unreachable!("stdio helper should build stdio transport");
        };
        *cwd = Some(LegacyAppPathString::from_path(&std::env::temp_dir()));
        for resolved_runtime in [
            runtime_context.resolve_server_environment("stdio", &remote_stdio),
            runtime_context.resolve_server_environment("http", &http_server("remote")),
        ] {
            let resolved_runtime = match resolved_runtime {
                Ok(resolved_runtime) => resolved_runtime,
                Err(error) => panic!("remote MCP should resolve: {error}"),
            };
            assert!(resolved_runtime.is_some());
        }

        let mut remote_http_with_helper = http_server("remote");
        let McpServerTransportConfig::StreamableHttp {
            http_headers_helper,
            ..
        } = &mut remote_http_with_helper.transport
        else {
            unreachable!("HTTP helper should build streamable HTTP transport");
        };
        *http_headers_helper = Some("helper-that-must-not-run".to_string());
        let error = match runtime_context.resolve_http_client("http", &remote_http_with_helper) {
            Ok(_) => panic!("remote HTTP helper should be rejected"),
            Err(error) => error,
        };
        assert_eq!(
            error,
            "HTTP headers helpers can only run in the local environment"
        );

        let remote_http = http_server("remote");
        let remote_environment = runtime_context
            .resolve_server_environment("http", &remote_http)
            .expect("remote HTTP MCP should resolve")
            .expect("remote HTTP MCP should have an environment");
        let remote_client = runtime_context
            .resolve_http_client("http", &remote_http)
            .expect("remote HTTP capability should resolve");
        assert!(Arc::ptr_eq(
            &remote_client,
            &remote_environment.get_http_client()
        ));
    }

    #[tokio::test]
    async fn remote_stdio_accepts_foreign_absolute_cwd() {
        let runtime_context = McpRuntimeContext::new(
            Arc::new(
                EnvironmentManager::create_for_tests(
                    Some("ws://127.0.0.1:8765".to_string()),
                    /*local_runtime_paths*/ None,
                )
                .await,
            ),
            PathBuf::from("/tmp"),
        );
        let mut remote_stdio = stdio_server("remote");
        let McpServerTransportConfig::Stdio { cwd, .. } = &mut remote_stdio.transport else {
            unreachable!("stdio helper should build stdio transport");
        };
        *cwd = Some(
            PathUri::parse("file:///C:/plugins/demo")
                .expect("foreign cwd URI")
                .into(),
        );

        let resolved_runtime =
            match runtime_context.resolve_server_environment("stdio", &remote_stdio) {
                Ok(resolved_runtime) => resolved_runtime,
                Err(error) => panic!("foreign cwd should resolve: {error}"),
            };
        assert!(resolved_runtime.is_some());
    }

    #[tokio::test]
    async fn local_stdio_accepts_local_environment_when_available() {
        let runtime_context = McpRuntimeContext::new(
            Arc::new(EnvironmentManager::default_for_tests()),
            PathBuf::from("/tmp"),
        );

        let resolved_runtime = match runtime_context
            .resolve_server_environment("stdio", &stdio_server(DEFAULT_MCP_SERVER_ENVIRONMENT_ID))
        {
            Ok(resolved_runtime) => resolved_runtime,
            Err(error) => panic!("local stdio MCP should resolve: {error}"),
        };
        assert!(resolved_runtime.is_some());
    }
}