codex/codex-rs/core/src/session/mcp.rs

use super::*;
use crate::McpRefreshReason;
use crate::environment_selection::TurnEnvironmentSnapshot;
use crate::environment_selection::combine_selected_capability_roots;
use crate::mcp_tool_call::McpToolApprovalContext;
use anyhow::Context;
use codex_exec_server::DiscoverV2CapabilitiesRequest;
use codex_exec_server::ExecutorCapabilityDiscoveryCache;
use codex_exec_server::ExecutorCapabilityDiscoverySnapshot;
use codex_exec_server::ExecutorCapabilityDiscoveryV2Cache;
use codex_exec_server::ExecutorCapabilityDiscoveryV2Snapshot;
use codex_exec_server::MAX_SELECTED_CAPABILITY_ROOTS;
use codex_exec_server::ResolvedSelectedCapabilityRoot;
use codex_exec_server::supports_capability_discovery_v2;
use codex_guardian_reviewer::GuardianElicitationReview;
use codex_guardian_reviewer::guardian_elicitation_review_request;
use codex_guardian_reviewer::mcp_elicitation_decline_without_message;
use codex_guardian_reviewer::mcp_elicitation_response_from_guardian_decision;
use codex_mcp::CODEX_APPS_MCP_SERVER_NAME;
use codex_mcp::ElicitationReviewRequest;
use codex_mcp::ElicitationReviewer;
use codex_mcp::ElicitationReviewerHandle;
use codex_mcp::MCP_TOOL_CODEX_APPS_META_KEY;
use codex_otel::auth_storage::AuthStorageOriginator;
use codex_protocol::capabilities::CapabilityRootLocation;
use codex_protocol::capabilities::SelectedCapabilityRoot;
use codex_protocol::config_types::ApprovalsReviewer;
use codex_protocol::mcp::is_node_repl_backed_connector;
use codex_protocol::mcp::is_node_repl_backed_server;
use codex_protocol::mcp_approval_meta::APPROVAL_KIND_KEY as MCP_ELICITATION_APPROVAL_KIND_KEY;
use codex_protocol::mcp_approval_meta::APPROVAL_KIND_MCP_TOOL_CALL as MCP_ELICITATION_APPROVAL_KIND_MCP_TOOL_CALL;
use codex_protocol::mcp_approval_meta::APPROVAL_KIND_TOOL_SUGGESTION as MCP_ELICITATION_APPROVAL_KIND_TOOL_SUGGESTION;
use codex_protocol::mcp_approval_meta::CONNECTOR_ID_KEY as MCP_ELICITATION_CONNECTOR_ID_KEY;
use codex_protocol::mcp_approval_meta::PERSIST_KEY as MCP_ELICITATION_PERSIST_KEY;
use codex_protocol::mcp_approval_meta::SENSITIVE_ACTION_KEY as MCP_ELICITATION_SENSITIVE_ACTION_KEY;
use codex_protocol::mcp_approval_meta::STRICT_AUTO_REVIEW_KEY as MCP_ELICITATION_STRICT_AUTO_REVIEW_KEY;
use codex_protocol::mcp_approval_meta::TOOL_NAME_KEY as MCP_ELICITATION_TOOL_NAME_KEY;
use codex_protocol::openai_models::GuardianScope;
use codex_rmcp_client::Elicitation;
use rmcp::model::ElicitationAction;
use serde_json::Map;
use tokio::sync::SemaphorePermit;

const TOOL_SUGGESTION_ACTION_INSTALL: &str = "install";
const TOOL_SUGGESTION_ACTION_KEY: &str = "suggest_type";
const TOOL_SUGGESTION_TOOL_ID_KEY: &str = "tool_id";
const TOOL_SUGGESTION_TOOL_TYPE_KEY: &str = "tool_type";

struct GuardianMcpElicitationReviewer {
    session: std::sync::Weak<Session>,
}

#[derive(Debug, PartialEq, Eq)]
struct PluginInstallElicitationTelemetryMetadata {
    tool_type: String,
    tool_id: String,
    tool_name: String,
}

impl GuardianMcpElicitationReviewer {
    fn new(session: &Arc<Session>) -> Self {
        Self {
            session: Arc::downgrade(session),
        }
    }
}

impl ElicitationReviewer for GuardianMcpElicitationReviewer {
    fn review(
        &self,
        request: ElicitationReviewRequest,
    ) -> BoxFuture<'static, anyhow::Result<Option<ElicitationResponse>>> {
        let session = self.session.clone();
        Box::pin(async move {
            let Some(session) = session.upgrade() else {
                return Ok(None);
            };
            review_guardian_mcp_elicitation(session, request).await
        })
    }
}

impl Session {
    pub(crate) async fn runtime_mcp_config(&self, config: &Config) -> McpConfig {
        self.runtime_mcp_config_and_context(config).await.0
    }

    pub(crate) async fn runtime_mcp_config_and_context(
        &self,
        config: &Config,
    ) -> (McpConfig, McpRuntimeContext) {
        let originator = self.originator().await;
        let (session_source, host_fallback_cwd, disabled_plugin_ids) = {
            let state = self.state.lock().await;
            (
                state.session_configuration.session_source.clone(),
                state.session_configuration.cwd().clone(),
                state.active_disabled_plugin_ids.clone(),
            )
        };
        let environments = self.services.turn_environments.snapshot().await;
        let environment_selections = environments.all_selections();
        let selected_capability_roots = self
            .resolve_selected_capability_roots_for_step(&environments)
            .await;
        let ready_selected_capability_roots =
            Self::ready_selected_capability_roots(&selected_capability_roots);
        let (executor_capability_discovery_v2, inventory_changed) = self
            .executor_capability_discovery_for_step_v2(
                config,
                &environments,
                /*skill_turn*/ None,
            )
            .await;
        let (executor_capability_discovery, discovery_recovered) = self
            .executor_capability_discovery_for_step(
                config,
                &selected_capability_roots,
                &environments,
                executor_capability_discovery_v2.as_deref(),
                /*skill_turn*/ None,
            )
            .await;
        if inventory_changed {
            self.mark_mcp_runtime_dirty(McpRefreshReason::ExecutorCapabilityInventoryChanged);
        }
        if discovery_recovered {
            self.mark_mcp_runtime_dirty(McpRefreshReason::ExecutorCapabilityDiscoveryRecovered);
        }
        let mcp_projection =
            self.services
                .mcp_manager
                .runtime_config_for_step(
                    config,
                    &self.services.mcp_thread_init,
                    &self.services.thread_extension_data,
                    McpThreadIdentity {
                        auth_changed: !self.services.mcp_runtime.current_auth_matches(
                            self.services.auth_manager.auth_cached().as_ref(),
                        ),
                        session_source: &session_source,
                        originator: &originator,
                        disabled_plugin_ids: &disabled_plugin_ids,
                        environments: McpEnvironmentScope::Selected(&environment_selections),
                    },
                    &ready_selected_capability_roots,
                    executor_capability_discovery.as_deref(),
                    executor_capability_discovery_v2.as_deref(),
                )
                .await;
        let mcp_config = self
            .project_selected_environment_mcp_servers(config, &environments, mcp_projection)
            .await
            .config;
        let runtime_context = self.mcp_runtime_context(&environments, &host_fallback_cwd);
        (mcp_config, runtime_context)
    }

    pub(crate) async fn current_mcp_runtime_context(&self) -> McpRuntimeContext {
        let host_fallback_cwd = self.state.lock().await.session_configuration.cwd().clone();
        let environments = self.services.turn_environments.snapshot().await;
        self.mcp_runtime_context(&environments, &host_fallback_cwd)
    }

    pub(super) fn mcp_runtime_context(
        &self,
        environments: &TurnEnvironmentSnapshot,
        host_fallback_cwd: &std::path::Path,
    ) -> McpRuntimeContext {
        let local_process_cwd = environments
            .local_environment_cwd()
            .map(|cwd| cwd.to_path_buf())
            .unwrap_or_else(|| host_fallback_cwd.to_path_buf());
        McpRuntimeContext::new(
            self.services.turn_environments.environment_manager(),
            local_process_cwd,
        )
        .with_connection_pool(self.services.mcp_manager.connection_pool())
        .with_selected_environments(
            environments.all_selections().into(),
            environments.ready_environment_handles(),
        )
    }

    pub(crate) async fn runtime_mcp_servers(
        &self,
        config: &Config,
    ) -> HashMap<String, McpServerConfig> {
        codex_mcp::configured_mcp_servers(&self.runtime_mcp_config(config).await)
    }

    /// Publishes changed MCP state, waiting for any refresh already in progress.
    #[tracing::instrument(name = "mcp.runtime.refresh_if_dirty", skip_all)]
    pub(crate) async fn refresh_mcp_if_dirty(self: &Arc<Self>) {
        let Ok(_refresh) = self.mcp_refresh.acquire().await else {
            error!("MCP runtime refresh semaphore closed");
            return;
        };
        self.reconcile_mcp_runtime().await;
    }

    /// Reconcile pending state under the session permit.
    async fn reconcile_mcp_runtime(self: &Arc<Self>) {
        loop {
            // Compare and rebuild from current environments, not choices saved for a future turn.
            let environments = self.services.turn_environments.snapshot().await;
            let environment_selections = environments.all_selections();
            if !self.services.mcp_runtime.current_environments_match(
                &environment_selections,
                &environments.ready_environment_handles(),
            ) {
                self.mark_mcp_runtime_dirty(McpRefreshReason::EnvironmentBindingsChanged);
            }
            let auth = self.services.auth_manager.auth_cached();
            if !self
                .services
                .mcp_runtime
                .current_auth_matches(auth.as_ref())
            {
                self.mark_mcp_runtime_dirty(McpRefreshReason::RuntimeAuthChanged);
            }

            let Some(mut refresh_invalidation) = self.mcp_refresh.claim(self.thread_id) else {
                return;
            };
            let auth = self.services.auth_manager.auth().await;
            // Capture after claiming the refresh so a new invalidation cannot be consumed
            // with the environment snapshot taken before its permissions changed.
            let environments = self.services.turn_environments.snapshot().await;
            let environment_selections = environments.all_selections();
            let desired = self.latest_mcp_desired_state(auth, environments).await;
            let selected_capability_roots = self
                .resolve_selected_capability_roots_for_step(&desired.environments)
                .await;
            let ready_selected_capability_roots =
                Self::ready_selected_capability_roots(&selected_capability_roots);
            let (executor_capability_discovery_v2, _) = self
                .executor_capability_discovery_for_step_v2(
                    &desired.config,
                    &desired.environments,
                    /*skill_turn*/ None,
                )
                .await;
            let (executor_capability_discovery, _) = self
                .executor_capability_discovery_for_step(
                    &desired.config,
                    &selected_capability_roots,
                    &desired.environments,
                    executor_capability_discovery_v2.as_deref(),
                    /*skill_turn*/ None,
                )
                .await;
            let mcp_projection = self
                .services
                .mcp_manager
                .runtime_config_for_step(
                    &desired.config,
                    &self.services.mcp_thread_init,
                    &self.services.thread_extension_data,
                    McpThreadIdentity {
                        auth_changed: !self
                            .services
                            .mcp_runtime
                            .current_auth_matches(desired.auth.as_ref()),
                        session_source: &desired.session_source,
                        originator: &desired.originator,
                        disabled_plugin_ids: &desired.disabled_plugin_ids,
                        environments: McpEnvironmentScope::Selected(&environment_selections),
                    },
                    &ready_selected_capability_roots,
                    executor_capability_discovery.as_deref(),
                    executor_capability_discovery_v2.as_deref(),
                )
                .await;
            self.publish_mcp_runtime(
                &desired,
                mcp_projection,
                &ready_selected_capability_roots,
                Some(self.mcp_elicitation_reviewer()),
            )
            .await;
            refresh_invalidation.published = true;
        }
    }

    /// Refreshes Apps tools on the published thread runtime and returns that client's snapshot.
    pub(crate) async fn refresh_codex_apps_tools(
        self: &Arc<Self>,
    ) -> anyhow::Result<codex_mcp::CodexAppsToolSnapshot> {
        let permit = self
            .mcp_refresh
            .acquire()
            .await
            .context("MCP runtime refresh semaphore closed")?;
        let _in_flight = self.mcp_refresh.apps_refreshes.token();
        // Reconcile once per batch so a failed or closed client can be replaced.
        if self.mcp_refresh.apps_refreshes.len() == 1 {
            self.mark_mcp_runtime_dirty(McpRefreshReason::CodexAppsToolRefresh);
        }
        if let Some(snapshot) = self.fetch_apps_catalog(permit).await? {
            return Ok(snapshot);
        }
        let permit = self
            .mcp_refresh
            .acquire()
            .await
            .context("MCP runtime refresh semaphore closed")?;
        self.fetch_apps_catalog(permit)
            .await?
            .context("MCP runtime changed during both Apps tool refresh attempts")
    }

    async fn fetch_apps_catalog(
        self: &Arc<Self>,
        permit: SemaphorePermit<'_>,
    ) -> anyhow::Result<Option<codex_mcp::CodexAppsToolSnapshot>> {
        self.reconcile_mcp_runtime().await;
        let refresh = self.services.mcp_runtime.capture_apps_tool_refresh();
        // An invalidation may arrive after reconciliation but before capture.
        if self.mcp_refresh.is_pending() {
            return Ok(None);
        }
        drop(permit);
        refresh.await
    }

    /// Wait outside the session permit so a superseded refresh can retry.
    #[tracing::instrument(name = "mcp.runtime.apps_catalog_wait", skip_all)]
    pub(crate) async fn wait_for_apps_tool_refresh(&self) {
        self.mcp_refresh.apps_refreshes.wait().await;
    }

    /// Reconnects the runtime so refreshed Apps tools belong to their new exact client.
    pub(crate) async fn hard_refresh_latest_codex_apps_tools(
        self: &Arc<Self>,
    ) -> anyhow::Result<Vec<codex_mcp::ToolInfo>> {
        let _refresh = self
            .mcp_refresh
            .acquire()
            .await
            .map_err(|_| anyhow::anyhow!("MCP runtime refresh semaphore closed"))?;
        // This is forced work even when there was no pending invalidation.
        let mut refresh_invalidation = self
            .mcp_refresh
            .claim_forced(self.thread_id, McpRefreshReason::HardRefreshCodexAppsTools);
        let auth = self.services.auth_manager.auth().await;
        let environments = self.services.turn_environments.snapshot().await;
        let environment_selections = environments.all_selections();
        let desired = self.latest_mcp_desired_state(auth, environments).await;
        let selected_capability_roots = self
            .resolve_selected_capability_roots_for_step(&desired.environments)
            .await;
        let ready_selected_capability_roots =
            Self::ready_selected_capability_roots(&selected_capability_roots);
        let (executor_capability_discovery_v2, _) = self
            .executor_capability_discovery_for_step_v2(
                &desired.config,
                &desired.environments,
                /*skill_turn*/ None,
            )
            .await;
        let (executor_capability_discovery, _) = self
            .executor_capability_discovery_for_step(
                &desired.config,
                &selected_capability_roots,
                &desired.environments,
                executor_capability_discovery_v2.as_deref(),
                /*skill_turn*/ None,
            )
            .await;
        let mcp_projection = self
            .services
            .mcp_manager
            .runtime_config_for_step(
                &desired.config,
                &self.services.mcp_thread_init,
                &self.services.thread_extension_data,
                McpThreadIdentity {
                    auth_changed: !self
                        .services
                        .mcp_runtime
                        .current_auth_matches(desired.auth.as_ref()),
                    session_source: &desired.session_source,
                    originator: &desired.originator,
                    disabled_plugin_ids: &desired.disabled_plugin_ids,
                    environments: McpEnvironmentScope::Selected(&environment_selections),
                },
                &ready_selected_capability_roots,
                executor_capability_discovery.as_deref(),
                executor_capability_discovery_v2.as_deref(),
            )
            .await;
        let mcp_projection = self
            .project_selected_environment_mcp_servers(
                &desired.config,
                &desired.environments,
                mcp_projection,
            )
            .await;
        let selected_plugins = mcp_projection.selected_plugins.clone();
        let input = self.build_mcp_runtime_input(
            &desired,
            mcp_projection,
            &ready_selected_capability_roots,
            Some(self.mcp_elicitation_reviewer()),
        );
        anyhow::ensure!(
            input.mcp_servers.contains_key(CODEX_APPS_MCP_SERVER_NAME),
            "unknown MCP server '{CODEX_APPS_MCP_SERVER_NAME}'"
        );
        let refreshed = AuthStorageOriginator::from_client_name(&desired.originator)
            .scope(self.services.mcp_runtime.replace_fresh(input))
            .await;
        self.services.thread_extension_data.insert(selected_plugins);
        refresh_invalidation.published = refreshed.is_ok();
        refreshed
    }

    pub(super) fn mark_mcp_runtime_dirty(&self, reason: McpRefreshReason) {
        self.mcp_refresh.invalidate(reason);
        self.services.mcp_runtime.invalidate_apps_tool_refresh();
    }

    #[tracing::instrument(
        name = "mcp.runtime.resolve_for_step",
        skip_all,
        fields(turn.id = %turn_context.sub_id)
    )]
    pub(crate) async fn mcp_runtime_for_step(
        self: &Arc<Self>,
        turn_context: &TurnContext,
        selected_capability_roots: &[ResolvedSelectedCapabilityRoot],
        required_servers: &[String],
        required_plugins: &HashSet<String>,
    ) -> Arc<codex_mcp::McpBinding> {
        let ready_selected_capability_roots =
            Self::ready_selected_capability_roots(selected_capability_roots);
        if self
            .services
            .mcp_runtime
            .current_ready_selected_capability_roots()
            != ready_selected_capability_roots
        {
            self.mark_mcp_runtime_dirty(McpRefreshReason::SelectedCapabilityRootsChanged);
        }

        let recovered_oauth_servers =
            AuthStorageOriginator::from_client_name(&turn_context.originator)
                .scope(
                    self.services
                        .mcp_runtime
                        .updated_oauth_credentials_after_auth_failure(),
                )
                .await;
        if !recovered_oauth_servers.is_empty()
            && let Ok(_refresh) = self.mcp_refresh.acquire().await
            && self
                .services
                .mcp_runtime
                .has_authentication_failed_servers(&recovered_oauth_servers)
                .await
        {
            self.mark_mcp_runtime_dirty(McpRefreshReason::OAuthCredentialsRecovered);
        }
        self.refresh_mcp_if_dirty().await;
        // Prewarm and other context-only captures must not start cloud discovery.
        let is_active_regular_turn = {
            let active_turn = self.active_turn.lock().await;
            active_turn
                .as_ref()
                .and_then(|turn| turn.task.as_ref())
                .is_some_and(|task| {
                    task.kind == TaskKind::Regular
                        && task.turn_context.sub_id == turn_context.sub_id
                })
        };
        if is_active_regular_turn {
            // Refresh cloud catalogs without holding the MCP permit or active-turn lock.
            for contributor in self.services.extensions.turn_lifecycle_contributors() {
                if contributor
                    .refresh_catalog_for_step(codex_extension_api::CatalogRefreshInput {
                        turn_id: &turn_context.sub_id,
                        session_store: &self.services.session_extension_data,
                        thread_store: &self.services.thread_extension_data,
                        turn_store: turn_context.extension_data.as_ref(),
                    })
                    .await
                {
                    // Preserve MCP updates if a later refresh is cancelled.
                    self.mark_mcp_runtime_dirty(McpRefreshReason::ExtensionCatalogChanged);
                }
            }
            self.refresh_mcp_if_dirty().await;
        }
        let required_servers = required_servers
            .iter()
            .chain(&recovered_oauth_servers)
            .cloned()
            .collect::<Vec<_>>();
        self.wait_for_apps_tool_refresh().await;
        if let Some(binding) = self
            .services
            .mcp_runtime
            .current_binding_with_requirements(&required_servers, required_plugins)
            .await
        {
            return binding;
        }
        let config = Arc::new(self.runtime_mcp_config(&turn_context.config).await);
        Arc::new(codex_mcp::McpBinding::empty(config))
    }

    fn discovery_sandbox(
        &self,
        environment: &super::turn_context::TurnEnvironment,
        skill_turn: Option<&TurnContext>,
    ) -> codex_exec_server::FileSystemSandboxContext {
        let id = &environment.selection.environment_id;
        let grants = skill_turn.map_or_else(
            || self.services.granted_permissions(id),
            |turn| turn.granted_permissions(id),
        );
        environment.sandbox_context(grants)
    }

    #[tracing::instrument(
        name = "capability_roots.snapshot_for_step",
        skip_all,
        fields(root_count = resolved_roots.len())
    )]
    pub(crate) async fn executor_capability_discovery_for_step(
        &self,
        config: &Config,
        resolved_roots: &[ResolvedSelectedCapabilityRoot],
        environments: &TurnEnvironmentSnapshot,
        v2_snapshots: Option<&[ExecutorCapabilityDiscoveryV2Snapshot]>,
        skill_turn: Option<&TurnContext>,
    ) -> (Option<Arc<ExecutorCapabilityDiscoverySnapshot>>, bool) {
        if v2_snapshots.is_some() {
            return (None, false);
        }
        let restricted_file_system = environments.primary().map_or_else(
            || {
                !config
                    .permissions
                    .file_system_sandbox_policy()
                    .has_full_disk_read_access()
            },
            |_| {
                environments.turn_environments().any(|environment| {
                    !environment
                        .permission_profile()
                        .file_system_sandbox_policy()
                        .has_full_disk_read_access()
                })
            },
        );
        if !restricted_file_system
            && !config
                .features
                .enabled(Feature::ExecutorCapabilityDiscovery)
        {
            return (None, false);
        }
        let sandbox_contexts = if restricted_file_system {
            environments
                .turn_environments()
                .map(|environment| {
                    (
                        environment.selection.environment_id.clone(),
                        self.discovery_sandbox(environment, skill_turn),
                    )
                })
                .collect::<HashMap<_, _>>()
        } else {
            HashMap::new()
        };
        // Each scope retains one result per root; temporary grants must not evict the
        // session inventory on every model step. The turn cache expires with its grants.
        let cache = skill_turn
            .map_or(&self.services.thread_extension_data, |turn| {
                turn.extension_data.as_ref()
            })
            .get_or_init(ExecutorCapabilityDiscoveryCache::default);
        let selected_capability_roots = resolved_roots
            .iter()
            .filter(|root| {
                let selected_root = root.selected_root();
                if !restricted_file_system {
                    return true;
                }
                let CapabilityRootLocation::Environment { environment_id, .. } =
                    &selected_root.location;
                if sandbox_contexts.contains_key(environment_id) {
                    return true;
                }
                warn!(
                    selected_root = selected_root.id,
                    environment_id, "skipping capability root without a filesystem sandbox context"
                );
                false
            })
            .cloned()
            .collect::<Vec<_>>();
        let discovery = cache
            .snapshot(&selected_capability_roots, &sandbox_contexts)
            .await;
        (
            Some(Arc::new(discovery)),
            skill_turn.is_none() && cache.take_recovered_discovery(),
        )
    }

    /// Acquires V2 inventories and reports changes without scheduling an MCP refresh.
    pub(super) async fn executor_capability_discovery_for_step_v2(
        &self,
        config: &Config,
        environments: &TurnEnvironmentSnapshot,
        skill_turn: Option<&TurnContext>,
    ) -> (Option<Vec<ExecutorCapabilityDiscoveryV2Snapshot>>, bool) {
        if !config
            .features
            .enabled(Feature::ExecutorCapabilityDiscoveryV2)
        {
            return (None, false);
        }
        // Retain this cache if a new turn replaces the thread's current cache.
        let cache = self
            .services
            .thread_extension_data
            .get_or_init(ExecutorCapabilityDiscoveryV2Cache::default);
        let remote_environment_count = environments
            .turn_environments()
            .filter(|selected| selected.environment().is_remote())
            .count();
        cache.ensure_capacity(remote_environment_count);
        let mut snapshots = Vec::new();
        let mut skipped_incompatible_executor = false;
        for selected in environments.turn_environments() {
            if !selected.environment().is_remote() {
                continue;
            }
            let environment_id = &selected.selection.environment_id;
            let info = match selected.environment().info().await {
                Ok(info) => info,
                Err(_) => {
                    tracing::warn!(
                        environment_id,
                        "could not read executor discovery V2 support"
                    );
                    continue;
                }
            };
            if !supports_capability_discovery_v2(&info) {
                skipped_incompatible_executor = true;
                continue;
            }
            let request = DiscoverV2CapabilitiesRequest {
                cwd: selected.cwd().clone(),
                sandbox: Some(self.discovery_sandbox(selected, skill_turn)),
            };
            let result = cache
                .discover(environment_id, selected.environment(), request)
                .await;
            snapshots.push(ExecutorCapabilityDiscoveryV2Snapshot {
                environment_id: environment_id.clone(),
                result,
            });
        }
        let incompatible = snapshots.is_empty() && skipped_incompatible_executor;
        if skill_turn.is_some() {
            // Only session discovery updates the inventory used to invalidate MCP.
            return ((!incompatible).then_some(snapshots), false);
        }
        if incompatible {
            let changed = self
                .services
                .thread_extension_data
                .remove::<Vec<ExecutorCapabilityDiscoveryV2Snapshot>>()
                .is_some();
            return (None, changed);
        }
        let previous = self
            .services
            .thread_extension_data
            .insert(snapshots.clone());
        let inventory_changed = previous.as_ref().map_or(!snapshots.is_empty(), |previous| {
            previous.len() != snapshots.len()
                || previous
                    .iter()
                    .zip(&snapshots)
                    .any(|(previous, current)| !previous.same_discovery(current))
        });
        let inventory_changed = cache.take_recovered_discovery() || inventory_changed;
        (Some(snapshots), inventory_changed)
    }

    pub(crate) async fn resolve_selected_capability_roots_for_step(
        &self,
        environments: &TurnEnvironmentSnapshot,
    ) -> Vec<ResolvedSelectedCapabilityRoot> {
        let captured_environments = environments.captured_environments();
        let thread_roots = environments.selected_capability_roots();
        let thread_root_count = thread_roots.len();
        let mut root_locations_by_id = HashMap::new();
        let mut selected_capability_roots = Vec::new();
        let mut ready_environment_root_count = 0;
        let combined_roots = combine_selected_capability_roots(
            &thread_roots,
            environments.turn_environments().map(|environment| {
                environment
                    .config_origin
                    .selected_capability_roots(environment.environment(), environment.config())
            }),
        );
        for (index, root) in combined_roots.into_iter().enumerate() {
            let CapabilityRootLocation::Environment { environment_id, .. } = &root.location;
            // Filter before readiness resolution, which can reconnect a historical executor.
            if !captured_environments.contains_key(environment_id) {
                tracing::warn!(
                    root_id = root.id,
                    environment_id,
                    "ignoring capability root without a captured turn environment"
                );
                continue;
            }
            if let Some(kept_location) = root_locations_by_id.get(&root.id) {
                if kept_location != &root.location {
                    tracing::warn!(
                        root_id = root.id,
                        ?kept_location,
                        ignored_location = ?root.location,
                        "ignoring selected capability root with conflicting location"
                    );
                }
                continue;
            }
            if index >= thread_root_count {
                if ready_environment_root_count == MAX_SELECTED_CAPABILITY_ROOTS {
                    tracing::warn!(
                        max_root_count = MAX_SELECTED_CAPABILITY_ROOTS,
                        "ignoring excess selected capability roots from ready environments"
                    );
                    break;
                }
                ready_environment_root_count += 1;
            }
            root_locations_by_id.insert(root.id.clone(), root.location.clone());
            selected_capability_roots.push(root);
        }
        self.services
            .turn_environments
            .environment_manager()
            .resolve_selected_capability_roots(&selected_capability_roots, &captured_environments)
            .await
    }

    pub(crate) fn mcp_elicitation_reviewer(self: &Arc<Self>) -> ElicitationReviewerHandle {
        Arc::clone(
            self.mcp_elicitation_reviewer_handle
                .get_or_init(|| Arc::new(GuardianMcpElicitationReviewer::new(self))),
        )
    }

    pub(crate) fn mcp_elicitation_lifecycle(&self) -> codex_mcp::ElicitationLifecycle {
        self.mcp_elicitation_lifecycle_handle
            .get_or_init(|| {
                let elicitations = self.services.elicitations.clone();
                codex_mcp::ElicitationLifecycle::new(move || elicitations.register())
            })
            .clone()
    }

    #[expect(
        clippy::await_holding_invalid_type,
        reason = "active turn checks and turn state updates must remain atomic"
    )]
    pub async fn request_mcp_server_elicitation(
        &self,
        turn_context: &TurnContext,
        server_name: String,
        request_id: RequestId,
        request: ElicitationRequest,
    ) -> Option<ElicitationResponse> {
        // This callback carries a request ID, not a reliable tool invocation owner.
        // Do not report its human wait as ordinary tool execution.
        if turn_context
            .config
            .features
            .enabled(Feature::CriticalPathTiming)
        {
            turn_context.turn_timing_state.invalidate_critical_path();
        }
        let _elicitation = self.services.elicitations.register();
        let (tx_response, rx_response) = oneshot::channel();
        let prev_entry = {
            let mut active = self.active_turn.lock().await;
            match active.as_mut() {
                Some(at) => {
                    let mut ts = at.turn_state.lock().await;
                    ts.insert_pending_elicitation(
                        server_name.clone(),
                        request_id.clone(),
                        tx_response,
                    )
                }
                None => None,
            }
        };
        if prev_entry.is_some() {
            warn!(
                "Overwriting existing pending elicitation for server_name: {server_name}, request_id: {request_id}"
            );
        }
        let id = match request_id {
            rmcp::model::NumberOrString::String(value) => {
                codex_protocol::mcp::RequestId::String(value.to_string())
            }
            rmcp::model::NumberOrString::Number(value) => {
                codex_protocol::mcp::RequestId::Integer(value)
            }
        };
        let event = EventMsg::ElicitationRequest(ElicitationRequestEvent {
            turn_id: Some(turn_context.sub_id.clone()),
            server_name,
            id,
            request,
        });
        let plugin_install_telemetry = plugin_install_elicitation_telemetry_metadata(&event);
        turn_context
            .turn_metadata_state
            .mark_user_input_requested_during_turn();
        self.send_event(turn_context, event).await;
        if let Some(plugin_install_telemetry) = plugin_install_telemetry {
            turn_context
                .session_telemetry
                .record_plugin_install_elicitation_sent(
                    plugin_install_telemetry.tool_type.as_str(),
                    plugin_install_telemetry.tool_id.as_str(),
                    plugin_install_telemetry.tool_name.as_str(),
                );
        }
        rx_response.await.ok()
    }

    #[expect(
        clippy::await_holding_invalid_type,
        reason = "active turn checks and manager fallback must stay serialized"
    )]
    pub async fn resolve_elicitation(
        &self,
        server_name: String,
        id: RequestId,
        response: ElicitationResponse,
    ) -> anyhow::Result<()> {
        let entry = {
            let mut active = self.active_turn.lock().await;
            match active.as_mut() {
                Some(at) => {
                    let mut ts = at.turn_state.lock().await;
                    ts.remove_pending_elicitation(&server_name, &id)
                }
                None => None,
            }
        };
        if let Some(tx_response) = entry {
            tx_response
                .send(response)
                .map_err(|e| anyhow::anyhow!("failed to send elicitation response: {e:?}"))?;
            return Ok(());
        }

        self.services
            .mcp_runtime
            .resolve_elicitation(server_name, id, response)
            .await
    }

    pub(crate) async fn refresh_mcp_servers_now(
        &self,
        added_servers: HashMap<String, McpServerConfig>,
        elicitation_reviewer: Option<ElicitationReviewerHandle>,
    ) {
        let Ok(_refresh) = self.mcp_refresh.acquire().await else {
            error!("MCP runtime refresh semaphore closed");
            return;
        };
        {
            let mut state = self.state.lock().await;
            let mut config = (*state.session_configuration.original_config_do_not_use).clone();
            let mut servers = config.mcp_servers.get().clone();
            for (name, server) in added_servers {
                servers.entry(name).or_insert(server);
            }
            if let Err(err) = config.mcp_servers.set(servers) {
                warn!("failed to refresh MCP dependencies for mentioned skills: {err}");
                return;
            }
            state.session_configuration.original_config_do_not_use = Arc::new(config);
        }
        // Claim after config mutation, before capturing the state this forced refresh handles.
        let mut refresh_invalidation = self
            .mcp_refresh
            .claim_forced(self.thread_id, McpRefreshReason::SkillMcpDependenciesAdded);
        let auth = self.services.auth_manager.auth().await;
        let ready_selected_capability_roots = self
            .services
            .mcp_runtime
            .current_ready_selected_capability_roots();
        let environments = self.services.turn_environments.snapshot().await;
        // Adding MCP dependencies preserves the published root selection.
        let mut captured_environments = environments.captured_environments();
        for root in &ready_selected_capability_roots {
            let CapabilityRootLocation::Environment { environment_id, .. } = &root.location;
            // A previously published root must not reconnect an environment absent from this capture.
            captured_environments
                .entry(environment_id.clone())
                .or_insert(None);
        }
        let selected_capability_roots = self
            .services
            .turn_environments
            .environment_manager()
            .resolve_selected_capability_roots(
                &ready_selected_capability_roots,
                &captured_environments,
            )
            .await;
        let environment_selections = environments.all_selections();
        let desired = self.latest_mcp_desired_state(auth, environments).await;
        let (executor_capability_discovery_v2, _) = self
            .executor_capability_discovery_for_step_v2(
                &desired.config,
                &desired.environments,
                /*skill_turn*/ None,
            )
            .await;
        let (executor_capability_discovery, _) = self
            .executor_capability_discovery_for_step(
                &desired.config,
                &selected_capability_roots,
                &desired.environments,
                executor_capability_discovery_v2.as_deref(),
                /*skill_turn*/ None,
            )
            .await;
        let mcp_projection = self
            .services
            .mcp_manager
            .runtime_config_for_step(
                &desired.config,
                &self.services.mcp_thread_init,
                &self.services.thread_extension_data,
                McpThreadIdentity {
                    auth_changed: !self
                        .services
                        .mcp_runtime
                        .current_auth_matches(desired.auth.as_ref()),
                    session_source: &desired.session_source,
                    originator: &desired.originator,
                    disabled_plugin_ids: &desired.disabled_plugin_ids,
                    environments: McpEnvironmentScope::Selected(&environment_selections),
                },
                &ready_selected_capability_roots,
                executor_capability_discovery.as_deref(),
                executor_capability_discovery_v2.as_deref(),
            )
            .await;
        self.publish_mcp_runtime(
            &desired,
            mcp_projection,
            &ready_selected_capability_roots,
            elicitation_reviewer,
        )
        .await;
        refresh_invalidation.published = true;
    }

    pub(crate) fn ready_selected_capability_roots(
        selected_capability_roots: &[ResolvedSelectedCapabilityRoot],
    ) -> Vec<SelectedCapabilityRoot> {
        selected_capability_roots
            .iter()
            .map(|root| root.selected_root().clone())
            .collect()
    }

    pub(crate) fn cancel_mcp_startup(&self) {
        self.services.mcp_runtime.cancel_startup();
    }
}

async fn review_guardian_mcp_elicitation(
    session: Arc<Session>,
    request: ElicitationReviewRequest,
) -> anyhow::Result<Option<ElicitationResponse>> {
    let Some((turn_context, cancellation_token)) =
        session.active_turn_context_and_cancellation_token().await
    else {
        return Ok(None);
    };

    let Some(mcp_config) = session.services.mcp_runtime.current_config() else {
        return Ok(None);
    };
    let step_settings = turn_context.next_step_settings.load_full();

    // User approval skips ordinary CUA checks, not separate sensitive requests.
    let user_cua_execution = step_settings.approvals_reviewer() == ApprovalsReviewer::User
        && is_node_repl_backed_server(&request.server_name)
        && request.elicitation.meta().is_some_and(|meta| {
            metadata_str(meta, MCP_ELICITATION_TOOL_NAME_KEY) == Some("js")
                && metadata_str(meta, MCP_ELICITATION_CONNECTOR_ID_KEY) == Some("node_repl")
                && metadata_str(meta, MCP_ELICITATION_APPROVAL_KIND_KEY)
                    == Some(MCP_ELICITATION_APPROVAL_KIND_MCP_TOOL_CALL)
                && meta.get(MCP_ELICITATION_SENSITIVE_ACTION_KEY) != Some(&Value::Bool(true))
                && meta.get("codex_requires_user_input") != Some(&Value::Bool(true))
        });

    // Full Access skips inference, not the active-turn and cancellation checks.
    if (user_cua_execution
        || turn_context.initial_environments.has_full_access(
            turn_context.approval_policy(),
            &turn_context
                .config
                .permissions
                .effective_permission_profile(),
        ))
        && matches!(
            &request.elicitation,
            Elicitation::Mcp(rmcp::model::ElicitRequestParams::FormElicitationParams {
                requested_schema, ..
            }) if requested_schema.properties.is_empty()
        )
    {
        let decision = if cancellation_token.is_cancelled() {
            ReviewDecision::Abort
        } else {
            ReviewDecision::Approved
        };
        return Ok(Some(mcp_elicitation_response_from_guardian_decision(
            decision,
            turn_context.model_info(),
        )));
    }

    // The invocation identifies the tool event, but a nested elicitation can
    // review a different action and connector than the enclosing JavaScript.
    let call_id = request
        .elicitation
        .meta()
        .and_then(|meta| match request.server_name.as_str() {
            CODEX_APPS_MCP_SERVER_NAME => meta.get(MCP_TOOL_CODEX_APPS_META_KEY)?.get("call_id"),
            _ => meta.get("callId"),
        })
        .and_then(Value::as_str);
    let approval_context = call_id
        .and_then(|call_id| session.mcp_tool_approval_metadata(&request.server_name, call_id));
    let (originating_call_id, guardian_scope) = if let Some(call_id) = call_id
        && let Some(McpToolApprovalContext {
            invocation: Some(invocation),
            metadata,
            ..
        }) = approval_context.as_ref()
        && invocation.server == request.server_name
        && is_node_repl_backed_connector(&invocation.server, metadata.connector_id.as_deref())
    {
        (Some(call_id), GuardianScope::ComputerUse)
    } else {
        (None, GuardianScope::for_mcp_server(&request.server_name))
    };

    let require_synchronous_review = matches!(
        request
            .elicitation
            .meta()
            .and_then(|meta| meta.get(MCP_ELICITATION_SENSITIVE_ACTION_KEY)),
        Some(Value::Bool(true))
    );

    // Bind the review to the issuing step, which may include environments that
    // became ready after the turn started. Do not refresh an in-flight action.
    let mut review_context = approval_context
        .as_ref()
        .map(|context| context.review_context.clone())
        .unwrap_or_else(|| crate::guardian::GuardianReviewContext::from(&turn_context));
    let strict_auto_review = matches!(
        request
            .elicitation
            .meta()
            .and_then(|meta| meta.get(MCP_ELICITATION_STRICT_AUTO_REVIEW_KEY)),
        Some(Value::Bool(true))
    );
    let mut guardian_request: crate::guardian::ReviewAction = if strict_auto_review {
        let connector_id = elicitation_connector_id(&request.elicitation);
        // A live Browser invocation can review a nested action with its own identity.
        // Other hosted connectors must still review their registered outer invocation.
        let review_outer_invocation =
            request.server_name == CODEX_APPS_MCP_SERVER_NAME && originating_call_id.is_none();
        let trusted_guardian_request = if review_outer_invocation {
            let Some(call_id) = call_id else {
                return Ok(None);
            };
            let Some(McpToolApprovalContext {
                invocation: Some(invocation),
                metadata,
                ..
            }) = approval_context.as_ref()
            else {
                return Ok(None);
            };
            if invocation.server != request.server_name
                || connector_id != metadata.connector_id.as_deref()
                || request
                    .elicitation
                    .meta()
                    .and_then(|meta| metadata_str(meta, MCP_ELICITATION_TOOL_NAME_KEY))
                    != Some(invocation.tool.as_str())
            {
                return Ok(None);
            }
            Some(
                crate::mcp_tool_call::build_guardian_mcp_tool_review_request(
                    call_id,
                    invocation,
                    Some(metadata),
                ),
            )
        } else {
            None
        };
        if !turn_context
            .config
            .features
            .enabled(Feature::GuardianApproval)
            || mcp_config
                .config_layer_stack
                .requirements()
                .approvals_reviewer
                .can_set(&ApprovalsReviewer::AutoReview)
                .is_err()
            || crate::connectors::mcp_approvals_reviewer_from_layers(
                &mcp_config.config_layer_stack,
                ApprovalsReviewer::AutoReview,
                Some(turn_context.model_info().slug.as_str()),
                request.server_name.as_str(),
                connector_id,
                /*link_id*/ None,
            ) != ApprovalsReviewer::AutoReview
            || request
                .elicitation
                .meta()
                .is_some_and(|meta| meta.contains_key(MCP_ELICITATION_PERSIST_KEY))
        {
            return Ok(None);
        }

        let GuardianElicitationReview::ApprovalRequest {
            request: guardian_request,
            tool_call_id,
        } = guardian_elicitation_review_request(&request, originating_call_id)
        else {
            return Ok(None);
        };
        match trusted_guardian_request {
            Some(request) => request.into(),
            None => crate::guardian::ReviewAction {
                tool_call_id,
                ..(*guardian_request).into()
            },
        }
    } else {
        let approval_policy = mcp_config.approval_policy.value();
        match approval_policy {
            AskForApproval::Never => {
                let Some(permission_profile) =
                    mcp_config.permission_profile_for_server(&request.server_name)
                else {
                    return Ok(Some(mcp_elicitation_decline_without_message()));
                };
                if codex_mcp::mcp_permission_prompt_is_auto_approved(
                    approval_policy,
                    permission_profile,
                    codex_mcp::McpPermissionPromptAutoApproveContext::default(),
                ) && matches!(
                    &request.elicitation,
                    Elicitation::Mcp(
                        rmcp::model::ElicitRequestParams::FormElicitationParams {
                            requested_schema,
                            ..
                        }
                    ) if requested_schema.properties.is_empty()
                ) {
                    return Ok(Some(ElicitationResponse {
                        action: ElicitationAction::Accept,
                        content: Some(serde_json::json!({})),
                        meta: None,
                    }));
                }
                return Ok(Some(mcp_elicitation_decline_without_message()));
            }
            AskForApproval::Granular(config) if !config.allows_mcp_elicitations() => {
                return Ok(Some(mcp_elicitation_decline_without_message()));
            }
            AskForApproval::OnRequest
            | AskForApproval::UnlessTrusted
            | AskForApproval::Granular(_) => {}
        }

        let approvals_reviewer = crate::connectors::mcp_approvals_reviewer_from_layers(
            &mcp_config.config_layer_stack,
            step_settings
                .mcp_approvals_reviewer_override
                .unwrap_or(mcp_config.approvals_reviewer),
            Some(turn_context.model_info().slug.as_str()),
            request.server_name.as_str(),
            elicitation_connector_id(&request.elicitation),
            /*link_id*/ None,
        );
        review_context.approval_policy = approval_policy;
        review_context.approvals_reviewer = approvals_reviewer;
        match guardian_elicitation_review_request(&request, originating_call_id) {
            GuardianElicitationReview::NotRequested => return Ok(None),
            GuardianElicitationReview::Decline(reason) => {
                let Elicitation::Mcp(elicitation) = &request.elicitation else {
                    return Ok(None);
                };
                crate::guardian::ReviewAction {
                    action: serde_json::to_value(elicitation)
                        .map(|elicitation| {
                            serde_json::json!({
                                "tool": "mcp_elicitation",
                                "server": request.server_name,
                                "request": elicitation,
                            })
                        })
                        .map_err(|error| error.to_string()),
                    category: GuardianScope::for_mcp_server(&request.server_name),
                    tool_call_id: None,
                    request: Err(reason.to_owned()),
                }
            }
            GuardianElicitationReview::ApprovalRequest {
                request: guardian_request,
                tool_call_id,
            } => crate::guardian::ReviewAction {
                tool_call_id,
                ..(*guardian_request).into()
            },
        }
    };
    guardian_request.category = guardian_scope;
    // Keep the action the server asked about, but use its owner's rules from the original call.
    if let Ok(crate::guardian::GuardianApprovalRequest::McpToolCall {
        prompt_permissions, ..
    }) = &mut guardian_request.request
        && let Some(permissions) =
            approval_context.and_then(|context| context.environment_permissions)
    {
        *prompt_permissions = Some(permissions);
    }
    let declined_reason = guardian_request.request.as_ref().err().cloned();
    let decision = crate::guardian::decide_approval(
        session,
        review_context,
        crate::guardian::new_guardian_review_id(),
        guardian_request,
        crate::tools::sandboxing::ApprovalRequestReasons {
            approval: None,
            retry: None,
        },
        crate::guardian::GuardianReviewOptions {
            require_guardian: strict_auto_review,
            plugin_attribution_override: None,
            approval_request_source: codex_analytics::GuardianApprovalRequestSource::MainTurn,
            external_cancel: Some(cancellation_token),
            require_synchronous_review,
        },
    )
    .await;
    Ok(decision.map(|decision| {
        if let Some(reason) = declined_reason {
            warn!(server_name = %request.server_name,
                request_id = %request.request_id, reason,
                "declining Guardian MCP elicitation before review");
            return mcp_elicitation_decline_without_message();
        }
        mcp_elicitation_response_from_guardian_decision(decision, turn_context.model_info())
    }))
}

fn elicitation_connector_id(elicitation: &Elicitation) -> Option<&str> {
    elicitation
        .meta()
        .and_then(|meta| metadata_str(meta, MCP_ELICITATION_CONNECTOR_ID_KEY))
}

fn metadata_str<'a>(meta: &'a Map<String, Value>, key: &str) -> Option<&'a str> {
    meta.get(key).and_then(Value::as_str)
}

fn metadata_owned_string(meta: &Map<String, Value>, key: &str) -> Option<String> {
    metadata_str(meta, key)
        .map(str::trim)
        .filter(|value| !value.is_empty())
        .map(ToOwned::to_owned)
}

fn plugin_install_elicitation_telemetry_metadata(
    event: &EventMsg,
) -> Option<PluginInstallElicitationTelemetryMetadata> {
    let EventMsg::ElicitationRequest(ElicitationRequestEvent { request, .. }) = event else {
        return None;
    };
    let codex_protocol::approvals::ElicitationRequest::Form {
        meta: Some(Value::Object(meta)),
        ..
    } = request
    else {
        return None;
    };
    if metadata_str(meta, MCP_ELICITATION_APPROVAL_KIND_KEY)
        != Some(MCP_ELICITATION_APPROVAL_KIND_TOOL_SUGGESTION)
        || metadata_str(meta, TOOL_SUGGESTION_ACTION_KEY) != Some(TOOL_SUGGESTION_ACTION_INSTALL)
    {
        return None;
    }

    Some(PluginInstallElicitationTelemetryMetadata {
        tool_type: metadata_owned_string(meta, TOOL_SUGGESTION_TOOL_TYPE_KEY)?,
        tool_id: metadata_owned_string(meta, TOOL_SUGGESTION_TOOL_ID_KEY)?,
        tool_name: metadata_owned_string(meta, MCP_ELICITATION_TOOL_NAME_KEY)?,
    })
}

#[cfg(test)]
#[path = "mcp_tests.rs"]
mod tests;