Restrict enterprise MCP auth and fail closed on config refresh (#49260)

## Why

Configuration reloads must honor current managed restrictions without granting new enterprise MCP authority to existing sessions or overwriting newer configuration with stale snapshots.

## What changed

- Allow enterprise managed authentication only for configured MCP servers. Remove plugin `ema_auth` overlays from the schema and disable servers that still use them.
- Refresh ordinary MCP settings while preserving session settings and initially admitted enterprise registrations. Disable enterprise MCP when its registration changes, policy revokes access, or configuration loading or resolution fails; require a new session to restore enterprise authority.
- Publish refreshes only when their captured configuration is still current, with bounded retries in app-server reloads. Preserve updated managed restrictions even when a refresh is rejected.
- Finish processing other thread refreshes before `config/mcpServer/reload` reports a rejection, and document that an error can accompany partially applied changes.

## Testing

Add regression coverage for stale refreshes, bounded retries, managed-policy revocation, persistent enterprise disablement, ordinary MCP transport changes, and rejection of plugin enterprise authentication. Verify that denied enterprise servers receive no requests.

GitOrigin-RevId: 34429cd339c1ec152a2ee8697cb29f0cff5d9e5f

codex-rs/app-server/README.md

@@ -344,6 +344,14 @@ and migration remain supported so resuming, reading, and forking those threads
 preserves the surviving history. This disk compatibility does not require restoring
 support for new `thread/rollback` requests.
 
+# MCP configuration reload
+
+`config/mcpServer/reload` returns an error when a loaded thread rejects the
+refreshed enterprise policy. The rejected thread retains its previous configuration
+layers with enterprise MCP disabled. Other planned thread refreshes are processed
+before the rejection is reported, so an error does not imply that no changes were
+applied. Correct the policy before retrying the reload.
+
 # Selected workspace routing
 
 The experimental `account/read.workspaceRouting` response field returns the selected ChatGPT workspace's `chatgptAccountId`, resolved HTTPS `backendOrigin`, and backend-provided `accountRoutingOverride`. The routing value is `us`, `us_cr`, or the explicit `NO_CONSTRAINT` value. API-only and signed-out accounts return `null` and do not need `accounts/check`.

codex-rs/app-server/src/mcp_refresh.rs

@@ -1,5 +1,6 @@
 use crate::config_manager::ConfigManager;
 use codex_core::CodexThread;
+use codex_core::ConfigRefreshOutcome;
 use codex_core::ThreadManager;
 use codex_core::config::Config;
 use std::io;
@@ -10,56 +11,153 @@ pub(crate) async fn reload_mcp_config(
     thread_manager: &Arc<ThreadManager>,
     config_manager: &ConfigManager,
 ) -> io::Result<()> {
-    config_manager
-        .load_latest_config(/*fallback_cwd*/ None)
-        .await?;
-    let mut refreshes = Vec::new();
-    for thread_id in thread_manager.list_thread_ids().await {
-        let thread = thread_manager
-            .get_thread(thread_id)
-            .await
-            .map_err(|err| io::Error::other(format!("failed to load thread {thread_id}: {err}")))?;
-        let config = load_refresh_config(thread.as_ref(), config_manager).await?;
-        refreshes.push((thread, config));
-    }
-    for (thread, config) in refreshes {
-        thread.refresh_mcp_config(config).await;
-    }
-    Ok(())
+    // Keep the reload state machine out of the request dispatcher's stack frame.
+    Box::pin(reload_mcp_config_with_policy(
+        thread_manager,
+        config_manager,
+        ReloadPolicy::Strict,
+    ))
+    .await
 }
 
 pub(crate) async fn reload_mcp_config_best_effort(
     thread_manager: &Arc<ThreadManager>,
     config_manager: &ConfigManager,
 ) {
-    for thread_id in thread_manager.list_thread_ids().await {
-        let thread = match thread_manager.get_thread(thread_id).await {
-            Ok(thread) => thread,
-            Err(err) => {
-                warn!(%thread_id, %err, "failed to load thread for MCP configuration refresh");
-                continue;
-            }
-        };
-        let config = match load_refresh_config(thread.as_ref(), config_manager).await {
-            Ok(config) => config,
-            Err(err) => {
-                warn!(%thread_id, %err, "failed to load thread MCP configuration");
-                continue;
-            }
-        };
-        thread.refresh_mcp_config(config).await;
+    if let Err(err) = Box::pin(reload_mcp_config_with_policy(
+        thread_manager,
+        config_manager,
+        ReloadPolicy::BestEffort,
+    ))
+    .await
+    {
+        warn!(%err, "failed to reload MCP configuration");
     }
 }
 
+#[derive(Clone, Copy)]
+enum ReloadPolicy {
+    Strict,
+    BestEffort,
+}
+
+async fn reload_mcp_config_with_policy(
+    thread_manager: &ThreadManager,
+    config_manager: &ConfigManager,
+    policy: ReloadPolicy,
+) -> io::Result<()> {
+    let mut rejected = false;
+    let mut load_error = None;
+    let mut completed: Vec<Arc<CodexThread>> = Vec::new();
+    // Bound both discovery and stale-owner retries so concurrent churn cannot
+    // keep the reload request alive indefinitely. Completed publications remain valid.
+    for pass in 0..3 {
+        let host_loaded = match policy {
+            ReloadPolicy::BestEffort => true,
+            ReloadPolicy::Strict => match config_manager
+                .load_latest_config(/*fallback_cwd*/ None)
+                .await
+            {
+                Ok(_) => true,
+                Err(err) => {
+                    load_error.get_or_insert(err);
+                    false
+                }
+            },
+        };
+        let mut targets = Vec::new();
+        for thread_id in thread_manager.list_thread_ids().await {
+            match thread_manager.get_thread(thread_id).await {
+                Ok(thread) => {
+                    if !completed.iter().any(|done| Arc::ptr_eq(done, &thread)) {
+                        targets.push((thread_id, thread));
+                    }
+                }
+                Err(err) => {
+                    warn!(%thread_id, %err, "failed to load thread for MCP configuration refresh");
+                }
+            }
+        }
+        if targets.is_empty() {
+            break;
+        }
+        for (thread_id, target) in targets {
+            for attempt in 0..3 {
+                let current_config = target.config().await;
+                let mut target_load_error = None;
+                let next_config = if host_loaded {
+                    match load_refresh_config(&current_config, config_manager).await {
+                        Ok(next_config) => next_config,
+                        Err(err) => {
+                            if matches!(policy, ReloadPolicy::BestEffort) {
+                                warn!(%thread_id, %err, "failed to load session configuration");
+                            }
+                            target_load_error = Some(err);
+                            disabled_enterprise_config(&current_config)
+                        }
+                    }
+                } else {
+                    disabled_enterprise_config(&current_config)
+                };
+                match target.refresh_mcp_config(current_config, next_config).await {
+                    ConfigRefreshOutcome::Published => {}
+                    ConfigRefreshOutcome::Rejected => rejected = true,
+                    ConfigRefreshOutcome::Stale => {
+                        if attempt == 2 {
+                            target.disable_mcp_enterprise_auth().await;
+                            rejected = true;
+                        }
+                        continue;
+                    }
+                }
+                if matches!(policy, ReloadPolicy::Strict)
+                    && let Some(err) = target_load_error
+                {
+                    load_error.get_or_insert(err);
+                }
+                break;
+            }
+            completed.push(target);
+        }
+        if pass == 2 {
+            for thread_id in thread_manager.list_thread_ids().await {
+                if let Ok(thread) = thread_manager.get_thread(thread_id).await
+                    && !completed.iter().any(|done| Arc::ptr_eq(done, &thread))
+                {
+                    thread.disable_mcp_enterprise_auth().await;
+                    rejected = true;
+                }
+            }
+        }
+    }
+    // Finish publishing every session's fail-closed configuration before reporting failure.
+    if matches!(policy, ReloadPolicy::Strict)
+        && let Some(err) = load_error
+    {
+        return Err(err);
+    }
+    if rejected && matches!(policy, ReloadPolicy::Strict) {
+        return Err(io::Error::other(
+            "enterprise MCP configuration was rejected; affected sessions retained their previous configuration with enterprise MCP disabled",
+        ));
+    }
+    Ok(())
+}
+
+fn disabled_enterprise_config(current_config: &Config) -> Config {
+    let mut config = current_config.clone();
+    config.disable_mcp_enterprise_auth();
+    config
+}
+
 async fn load_refresh_config(
-    thread: &CodexThread,
+    current_config: &Config,
     config_manager: &ConfigManager,
 ) -> io::Result<Config> {
-    let thread_config = thread.config().await;
     config_manager
         .load_latest_config_with_session_layers(
-            &thread_config.config_layer_stack,
-            &thread_config.cwd,
+            &current_config.config_layer_stack,
+            &current_config.cwd,
         )
         .await
 }
@@ -96,6 +194,27 @@ mod tests {
     use std::sync::atomic::Ordering;
     use tempfile::TempDir;
 
+    #[tokio::test]
+    async fn stale_refresh_stops_after_bounded_attempts() -> anyhow::Result<()> {
+        let (_temp_dir, thread_manager, config_manager, loader) = refresh_test_state().await?;
+        for thread_id in thread_manager.list_thread_ids().await {
+            let thread = thread_manager.get_thread(thread_id).await?;
+            if thread.config().await.cwd.ends_with("good") {
+                *loader.churn_thread.lock().unwrap() = Some(thread);
+            }
+        }
+        tokio::time::timeout(
+            std::time::Duration::from_secs(10),
+            reload_mcp_config(&thread_manager, &config_manager),
+        )
+        .await
+        .expect("reload must terminate under owner churn")
+        .expect_err("reload failed closed");
+        assert_eq!(loader.good_loads.load(Ordering::Relaxed), 3);
+        assert_eq!(loader.bad_loads.load(Ordering::Relaxed), 1);
+        Ok(())
+    }
+
     #[tokio::test]
     async fn strict_refresh_reports_thread_planning_failures() -> anyhow::Result<()> {
         let (temp_dir, thread_manager, config_manager, _loader) = refresh_test_state().await?;
@@ -110,14 +229,15 @@ mod tests {
 
         assert_eq!(err.to_string(), "failed to load refresh config");
         for thread_id in thread_manager.list_thread_ids().await {
+            let config = thread_manager.get_thread(thread_id).await?.config().await;
+            let failed = config.cwd.ends_with("bad");
             assert_eq!(
-                thread_manager
-                    .get_thread(thread_id)
-                    .await?
-                    .config()
-                    .await
-                    .auth_keyring_backend_kind(),
-                AuthKeyringBackendKind::Direct
+                config.auth_keyring_backend_kind(),
+                if failed {
+                    AuthKeyringBackendKind::Direct
+                } else {
+                    AuthKeyringBackendKind::Secrets
+                }
             );
         }
         Ok(())
@@ -148,6 +268,30 @@ mod tests {
         Ok(())
     }
 
+    #[tokio::test]
+    async fn strict_refresh_preserves_config_when_host_config_cannot_load() -> anyhow::Result<()> {
+        let (temp_dir, thread_manager, config_manager, _loader) = refresh_test_state().await?;
+        std::fs::write(
+            temp_dir.path().join(codex_config::CONFIG_TOML_FILE),
+            "[features",
+        )?;
+
+        assert!(
+            reload_mcp_config(&thread_manager, &config_manager)
+                .await
+                .is_err()
+        );
+
+        for thread_id in thread_manager.list_thread_ids().await {
+            let config = thread_manager.get_thread(thread_id).await?.config().await;
+            assert_eq!(
+                config.auth_keyring_backend_kind(),
+                AuthKeyringBackendKind::Direct
+            );
+        }
+        Ok(())
+    }
+
     #[tokio::test]
     async fn invalidation_does_not_reload_thread_config() -> anyhow::Result<()> {
         let (_temp_dir, thread_manager, _config_manager, loader) = refresh_test_state().await?;
@@ -179,8 +323,11 @@ mod tests {
         let thread = good_thread.expect("good test thread should exist");
         let original_model = thread.config().await.model.clone();
 
-        let refresh_config = load_refresh_config(thread.as_ref(), &config_manager).await?;
-        thread.refresh_mcp_config(refresh_config).await;
+        let current_config = thread.config().await;
+        let refresh_config = load_refresh_config(&current_config, &config_manager).await?;
+        let _ = thread
+            .refresh_mcp_config(current_config, refresh_config)
+            .await;
 
         assert_eq!(
             thread.config().await.auth_keyring_backend_kind(),
@@ -221,7 +368,8 @@ enabled = false
 "#,
         )?;
 
-        let refresh_config = load_refresh_config(thread.as_ref(), &config_manager).await?;
+        let current_config = thread.config().await;
+        let refresh_config = load_refresh_config(&current_config, &config_manager).await?;
         let mut actual = refresh_config.mcp_servers.get().clone();
         actual.remove(codex_mcp::CODEX_APPS_MCP_SERVER_NAME);
         let expected = serde_json::from_value::<HashMap<String, McpServerConfig>>(json!({
@@ -366,6 +514,7 @@ enabled = false
             bad_cwd: AbsolutePathBuf::try_from(bad_cwd)?,
             good_loads: AtomicUsize::new(0),
             bad_loads: AtomicUsize::new(0),
+            churn_thread: std::sync::Mutex::new(None),
         });
         let config_manager = ConfigManager::new(
             temp_dir.path().to_path_buf(),
@@ -385,6 +534,7 @@ enabled = false
         bad_cwd: AbsolutePathBuf,
         good_loads: AtomicUsize,
         bad_loads: AtomicUsize,
+        churn_thread: std::sync::Mutex<Option<Arc<CodexThread>>>,
     }
 
     impl CountingThreadConfigLoader {
@@ -394,6 +544,15 @@ enabled = false
         ) -> Result<Vec<ThreadConfigSource>, ThreadConfigLoadError> {
             if context.cwd.as_ref() == Some(&self.good_cwd) {
                 self.good_loads.fetch_add(1, Ordering::Relaxed);
+                let thread = self.churn_thread.lock().unwrap().clone();
+                if let Some(thread) = thread {
+                    let current = thread.config().await;
+                    let next = current.as_ref().clone();
+                    assert_eq!(
+                        thread.refresh_runtime_config(current, next).await,
+                        ConfigRefreshOutcome::Published
+                    );
+                }
             }
             if context.cwd.as_ref() == Some(&self.bad_cwd) {
                 self.bad_loads.fetch_add(1, Ordering::Relaxed);

codex-rs/app-server/src/request_processors/config_processor.rs

@@ -357,11 +357,21 @@ pub(super) async fn reload_user_config(
     config_manager: &ConfigManager,
     thread_manager: &ThreadManager,
 ) {
+    // Reload owns several config snapshots; do not inline that future into dispatch.
+    Box::pin(reload_user_config_inner(config_manager, thread_manager)).await;
+}
+
+async fn reload_user_config_inner(config_manager: &ConfigManager, thread_manager: &ThreadManager) {
     if let Err(err) = config_manager
         .load_latest_config(/*fallback_cwd*/ None)
         .await
     {
         tracing::warn!("failed to rebuild user config for runtime refresh: {err}");
+        for thread_id in thread_manager.list_thread_ids().await {
+            if let Ok(thread) = thread_manager.get_thread(thread_id).await {
+                thread.disable_mcp_enterprise_auth().await;
+            }
+        }
         return;
     }
     let thread_ids = thread_manager.list_thread_ids().await;
@@ -369,22 +379,53 @@ pub(super) async fn reload_user_config(
         let Ok(thread) = thread_manager.get_thread(thread_id).await else {
             continue;
         };
-        let current_config = thread.config().await;
-        let next_config = match config_manager
-            .load_latest_config_with_session_layers(
-                &current_config.config_layer_stack,
-                &current_config.cwd,
-            )
-            .await
-        {
-            Ok(config) => config,
-            Err(err) => {
-                tracing::warn!(%thread_id, %err, "failed to reload thread configuration");
-                continue;
+        for attempt in 0..4 {
+            let current_config = thread.config().await;
+            let next_config = match config_manager
+                .load_latest_config_with_session_layers(
+                    &current_config.config_layer_stack,
+                    &current_config.cwd,
+                )
+                .await
+            {
+                Ok(config) => config,
+                Err(err) => {
+                    tracing::warn!(%thread_id, %err, "failed to reload thread configuration");
+                    thread.disable_mcp_enterprise_auth().await;
+                    break;
+                }
+            };
+            let current_requirements = current_config.config_layer_stack.requirements();
+            let next_requirements = next_config.config_layer_stack.requirements();
+            let promote_mcp = current_requirements.mcp_servers != next_requirements.mcp_servers
+                || current_requirements.plugins != next_requirements.plugins
+                || current_requirements.feature_requirements
+                    != next_requirements.feature_requirements;
+            let outcome = if promote_mcp {
+                // Keep the loaded MCP map and its requirements under one owner,
+                // then reload the user update from that owner.
+                Box::pin(thread.refresh_mcp_config(current_config, next_config)).await
+            } else {
+                // Keep runtime refresh state off the request dispatcher's stack.
+                Box::pin(thread.refresh_runtime_config(current_config, next_config)).await
+            };
+            match outcome {
+                codex_core::ConfigRefreshOutcome::Published => {
+                    if !promote_mcp {
+                        break;
+                    }
+                }
+                codex_core::ConfigRefreshOutcome::Stale => {}
+                codex_core::ConfigRefreshOutcome::Rejected => {
+                    tracing::warn!(%thread_id, "stopped user configuration reload after rejected refresh");
+                    break;
+                }
             }
-        };
-        // Keep runtime refresh state off the request dispatcher's stack.
-        Box::pin(thread.refresh_runtime_config(next_config)).await;
+            if attempt == 3 {
+                thread.disable_mcp_enterprise_auth().await;
+                tracing::warn!(%thread_id, "configuration kept changing during user reload; enterprise MCP disabled");
+            }
+        }
     }
 }
 
@@ -842,6 +883,8 @@ fn config_write_error(code: ConfigWriteErrorCode, message: impl Into<String>) ->
 #[cfg(test)]
 mod tests {
     use super::map_requirements_to_api;
+    use super::reload_user_config;
+    use crate::config_manager::ConfigManager;
     use codex_app_server_protocol::AllowDenyRequirement;
     use codex_app_server_protocol::AutoReviewRequirements;
     use codex_app_server_protocol::BrowserUseAccessApprovalLifetime;
@@ -858,21 +901,98 @@ mod tests {
     use codex_config::BrowserUseAccessApprovalLifetimeToml;
     use codex_config::BrowserUseOriginPolicyToml;
     use codex_config::BrowserUseRequirementsToml;
+    use codex_config::CloudConfigBundleLoader;
     use codex_config::ComputerUseMacosRequirementsToml;
     use codex_config::ComputerUseRequirementsToml;
     use codex_config::ComputerUseWindowsExeRequirementToml;
     use codex_config::ComputerUseWindowsRequirementsToml;
     use codex_config::ConfigRequirementsToml;
+    use codex_config::LoaderOverrides;
     use codex_config::ModelsRequirementsToml;
     use codex_config::NewThreadModelDefaultsToml;
     use codex_config::WindowsRequirementsToml;
     use codex_config::types::FeedbackConfigToml;
+    use codex_config::types::ToolSuggestDisabledTool;
+    use codex_exec_server::EnvironmentManager;
+    use codex_login::CodexAuth;
     use codex_protocol::config_types::ForcedLoginMethod;
     use codex_protocol::openai_models::ReasoningEffort;
     use codex_utils_absolute_path::AbsolutePathBuf;
     use codex_utils_path_uri::PathUri;
     use pretty_assertions::assert_eq;
     use std::collections::BTreeMap;
+    use std::sync::Arc;
+
+    #[tokio::test]
+    async fn reload_user_config_preserves_promoted_mcp_allowlist_denial() -> anyhow::Result<()> {
+        let home = tempfile::tempdir()?;
+        let managed_config_path = home.path().join("managed_config.toml");
+        std::fs::write(
+            &managed_config_path,
+            r#"
+[features]
+use_xaa = true
+[mcp_enterprise_managed_auth.idp]
+issuer = "https://idp.example"
+client_id = "idp-client"
+[mcp_servers.enterprise]
+url = "https://resource.example/mcp"
+auth = "ema_auth"
+oauth_resource = "https://resource.example/mcp"
+[mcp_servers.enterprise.oauth]
+client_id = "mcp-client"
+"#,
+        )?;
+        let requirements_path = home.path().join("requirements.toml");
+        std::fs::write(
+            &requirements_path,
+            "[mcp_servers.enterprise.identity]\nurl = 'https://resource.example/mcp'\n",
+        )?;
+        let config_manager = ConfigManager::new_for_tests(
+            home.path().to_path_buf(),
+            Vec::new(),
+            LoaderOverrides::with_managed_config_path_for_tests(managed_config_path),
+            CloudConfigBundleLoader::default(),
+        );
+        let initial = config_manager
+            .load_latest_config(Some(home.path().to_path_buf()))
+            .await?;
+        let thread_manager = codex_core::test_support::thread_manager_with_models_provider_and_home(
+            CodexAuth::from_api_key("dummy"),
+            initial.model_provider.clone(),
+            initial.codex_home.to_path_buf(),
+            Arc::new(EnvironmentManager::default_for_tests()),
+        );
+        let thread = thread_manager
+            .start_thread(codex_core::StartThreadOptions::new(initial))
+            .await?
+            .thread;
+        assert!(thread.config().await.mcp_servers.get()["enterprise"].enabled);
+
+        std::fs::write(&requirements_path, "[mcp_servers]\n")?;
+        std::fs::write(
+            home.path().join(codex_config::CONFIG_TOML_FILE),
+            "[tool_suggest]\ndisabled_tools = [{ type = 'connector', id = 'calendar' }]\n",
+        )?;
+        reload_user_config(&config_manager, &thread_manager).await;
+
+        let refreshed = thread.config().await;
+        assert_eq!(
+            refreshed
+                .config_layer_stack
+                .requirements()
+                .mcp_servers
+                .as_ref()
+                .map(|requirements| &requirements.value),
+            Some(&BTreeMap::new())
+        );
+        assert!(!refreshed.mcp_servers.get()["enterprise"].enabled);
+        assert_eq!(
+            refreshed.tool_suggest.disabled_tools,
+            vec![ToolSuggestDisabledTool::connector("calendar")]
+        );
+        Ok(())
+    }
 
     fn map_test_requirements(
         requirements: ConfigRequirementsToml,

codex-rs/app-server/src/request_processors/mcp_processor.rs

@@ -3,6 +3,7 @@ use super::*;
 use codex_mcp::McpServerSource;
 use codex_mcp::ReadResourceRequestParams;
 use codex_mcp::resolve_oauth_callback;
+use std::future::Future;
 
 use crate::thread_state::ThreadStateManager;
 
@@ -43,13 +44,17 @@ impl McpRequestProcessor {
             .map(|response| Some(response.into()))
     }
 
-    pub(crate) async fn mcp_server_refresh(
+    pub(crate) fn mcp_server_refresh(
         &self,
         params: Option<()>,
-    ) -> Result<Option<ClientResponsePayload>, JSONRPCErrorError> {
-        self.mcp_server_refresh_response(params)
-            .await
-            .map(|response| Some(response.into()))
+    ) -> impl Future<Output = Result<Option<ClientResponsePayload>, JSONRPCErrorError>> + Send + '_
+    {
+        // Keep refresh state behind a pointer before returning to the shared request handler.
+        Box::pin(async move {
+            self.mcp_server_refresh_response(params)
+                .await
+                .map(|response| Some(response.into()))
+        })
     }
 
     pub(crate) async fn mcp_server_status_list(

codex-rs/app-server/tests/suite/v2/plugin_install.rs

@@ -1495,19 +1495,10 @@ url = "https://example.com/allowed-mcp"
     Ok(())
 }
 
-#[test_case("disabled")]
-#[test_case("enterprise")]
 #[tokio::test]
-async fn plugin_install_skips_mcp_oauth_managed_by_plugin_config(case: &str) -> Result<()> {
+async fn plugin_install_skips_mcp_oauth_for_disabled_server() -> Result<()> {
     let oauth_server = MockServer::start().await;
-    let endpoint = format!("{}/mcp", oauth_server.uri());
-    let mcp_settings = match case {
-        "disabled" => "enabled = false".to_string(),
-        "enterprise" => format!(
-            "ema_auth = {{ url = '{endpoint}', resource = '{endpoint}', client_id = 'resource-client', authorization_server_issuer = 'https://as.example' }}"
-        ),
-        _ => unreachable!("unknown test case"),
-    };
+    let mcp_settings = "enabled = false";
     let codex_home = TempDir::new()?;
     std::fs::write(
         codex_home.path().join("config.toml"),
@@ -1566,7 +1557,7 @@ plugins = true
             .and_then(|plugins| plugins.get("sample-plugin@debug"))
             .and_then(|plugin| plugin.get("mcp_servers"))
             .and_then(|servers| servers.get("sample-mcp")),
-        Some(&toml::from_str::<toml::Value>(&mcp_settings)?)
+        Some(&toml::from_str::<toml::Value>(mcp_settings)?)
     );
     Ok(())
 }

codex-rs/codex-mcp/src/catalog.rs

@@ -522,10 +522,13 @@ impl McpCatalogBuilder {
                     let mut registration = *registration;
                     let persist_disabled_name =
                         registration.source.disabled_registration_is_name_veto()
-                            && !matches!(
-                                registration.config.disabled_reason,
-                                Some(McpServerDisabledReason::EmaRegistration)
-                            );
+                            && !(matches!(
+                                registration.source,
+                                McpServerSource::Plugin(_) | McpServerSource::SelectedPlugin(_)
+                            ) && matches!(
+                                registration.config.auth,
+                                codex_config::McpServerAuth::EmaAuth
+                            ));
                     if !registration.config.enabled || disabled_server_names.contains(&name) {
                         registration.config.enabled = false;
                         if persist_disabled_name {
@@ -537,9 +540,10 @@ impl McpCatalogBuilder {
                         registration.config.auth,
                         codex_config::McpServerAuth::EmaAuth
                     ) {
-                        let allowed = ema_idp.as_ref().is_some_and(|idp| {
-                            registration.config.resolve_ema_registration(idp).is_ok()
-                        });
+                        let allowed = matches!(&registration.source, McpServerSource::Config)
+                            && ema_idp.as_ref().is_some_and(|idp| {
+                                registration.config.resolve_ema_registration(idp).is_ok()
+                            });
                         // EMA denial must not become a persistent name veto.
                         registration.config.enabled &= allowed;
                     }

codex-rs/codex-mcp/src/catalog_tests.rs

@@ -6,11 +6,9 @@ use codex_config::AppToolApproval;
 use codex_config::DEFAULT_MCP_SERVER_ENVIRONMENT_ID;
 use codex_config::McpServerAuth;
 use codex_config::McpServerConfig;
-use codex_config::McpServerDisabledReason;
 use codex_config::McpServerIdpOAuthConfig;
 use codex_config::McpServerToolConfig;
 use codex_config::McpServerTransportConfig;
-use codex_config::types::PluginMcpServerEmaAuthConfig;
 use codex_protocol::mcp_policy::EnvironmentMcpPolicy;
 use codex_protocol::mcp_policy::PluginMcpRequirements;
 use codex_utils_path_uri::PathUri;
@@ -283,77 +281,40 @@ fn ema_policy_survives_rebuilds_and_rebinds_materialized_servers() {
 
 #[test]
 fn rejected_plugin_ema_registration_does_not_veto_hosted_apps() {
-    let idp = McpServerIdpOAuthConfig {
-        issuer: "https://idp.example".to_string(),
-        client_id: "enterprise-client".to_string(),
-    };
-    for (url, resource) in [
-        ("https://other.example/mcp", "https://resource.example"),
-        ("https://resource.example/mcp", " "),
-    ] {
-        for initially_enabled in [true, false] {
-            let mut rejected = server(url);
-            rejected.enabled = initially_enabled;
-            let policy = PluginMcpServerEmaAuthConfig {
-                url: "https://resource.example/mcp".to_string(),
-                client_id: "resource-client".to_string(),
-                authorization_server_issuer: "https://as.example".to_string(),
-                scopes: Vec::new(),
-                resource: resource.to_string(),
-            };
-            policy.apply(&mut rejected);
-            assert!(rejected.resolve_ema_registration(&idp).is_err());
-            assert_eq!(
-                (
-                    rejected.enabled,
-                    rejected.auth.clone(),
-                    rejected.ema_registration()
-                ),
-                (false, McpServerAuth::EmaAuth, None),
-            );
-            assert_eq!(
-                rejected.disabled_reason,
-                initially_enabled.then_some(McpServerDisabledReason::EmaRegistration),
-            );
-
-            let mut builder = ResolvedMcpCatalog::builder();
-            builder.enable_ema(idp.clone());
-            builder.register(McpServerRegistration::from_plugin(
-                CODEX_APPS_MCP_SERVER_NAME.to_string(),
-                plugin("plugin@test"),
-                /*plugin_order*/ 0,
-                rejected.clone(),
-            ));
-            let catalog = builder.build();
-            assert_eq!(
-                catalog.server(CODEX_APPS_MCP_SERVER_NAME).unwrap().config(),
-                &rejected,
-            );
-
-            let materialized = catalog.with_materialized_servers(catalog.configured_servers());
-            for catalog in [catalog, materialized] {
-                let mut builder = catalog.to_builder();
-                let mut expected = server("https://chatgpt.com/mcp");
-                builder.register(McpServerRegistration::from_hosted_apps(
-                    "apps",
-                    /*contribution_order*/ 0,
-                    expected.clone(),
-                ));
-                expected.enabled = initially_enabled;
-                assert_eq!(
-                    builder.build().server(CODEX_APPS_MCP_SERVER_NAME),
-                    Some(&ResolvedMcpServer {
-                        source: McpServerSource::Extension {
-                            id: "apps".to_string(),
-                            host_owned_apps: true,
-                        },
-                        config: expected,
-                        credential_policy: McpCredentialPolicy::HostFallbackAllowed,
-                        protocol_mode: None,
-                    }),
-                );
-            }
-        }
+    let mut rejected = server("https://plugin.example/mcp");
+    rejected.auth = McpServerAuth::EmaAuth;
+    let mut builder = ResolvedMcpCatalog::builder();
+    builder.register(McpServerRegistration::from_plugin(
+        CODEX_APPS_MCP_SERVER_NAME.to_string(),
+        plugin("plugin@test"),
+        /*plugin_order*/ 0,
+        rejected,
+    ));
+    let catalog = builder.build();
+    assert!(
+        !catalog
+            .server(CODEX_APPS_MCP_SERVER_NAME)
+            .unwrap()
+            .config()
+            .enabled
+    );
+    let materialized = catalog.with_materialized_servers(catalog.configured_servers());
+    for catalog in [catalog, materialized] {
+        let mut builder = catalog.to_builder();
+        let expected = server("https://chatgpt.com/mcp");
+        builder.register(McpServerRegistration::from_hosted_apps(
+            "apps",
+            /*contribution_order*/ 0,
+            expected.clone(),
+        ));
+        assert_eq!(
+            builder
+                .build()
+                .server(CODEX_APPS_MCP_SERVER_NAME)
+                .unwrap()
+                .config(),
+            &expected,
+        );
     }
 }
 
@@ -837,3 +798,28 @@ fn environment_policy_preserves_selected_plugin_and_empty_server_allowlist_seman
         );
     }
 }
+
+#[test]
+fn retired_plugin_ema_does_not_veto_a_later_runtime_registration() {
+    let mut retired = server("https://plugin.example/mcp");
+    retired.auth = McpServerAuth::EmaAuth;
+    retired.enabled = false;
+    let mut builder = ResolvedMcpCatalog::builder();
+    builder.register(McpServerRegistration::from_plugin(
+        "docs".to_string(),
+        plugin("plugin@test"),
+        /*plugin_order*/ 0,
+        retired,
+    ));
+    let catalog = builder.build();
+    assert!(!catalog.server("docs").unwrap().config().enabled);
+    let mut builder = catalog.to_builder();
+    let expected = server("https://extension.example/mcp");
+    builder.register(McpServerRegistration::from_extension(
+        "docs".to_string(),
+        "hosted",
+        /*contribution_order*/ 0,
+        expected.clone(),
+    ));
+    assert_eq!(builder.build().server("docs").unwrap().config(), &expected);
+}

codex-rs/codex-mcp/src/mcp/mod_tests.rs

@@ -132,7 +132,7 @@ pub(crate) fn test_elicitation_config(
 }
 
 #[test]
-fn ema_catalog_supports_configured_installed_and_selected_plugins_without_widening_policy() {
+fn ema_catalog_allows_only_configured_servers_without_widening_policy() {
     let server: McpServerConfig = serde_json::from_value(serde_json::json!({
         "url": "https://resource.example/mcp", "auth": "ema_auth",
         "oauth": { "client_id": "resource-client", "authorization_server_issuer": "https://as.example" }
@@ -179,11 +179,9 @@ fn ema_catalog_supports_configured_installed_and_selected_plugins_without_wideni
         });
         let servers = effective_mcp_servers(&config, /*auth*/ None);
         for name in ["configured", "installed", "selected"] {
-            assert_eq!(servers[name].enabled(), xaa_enabled && !denied, "{name}");
-            assert_eq!(
-                servers[name].config().oauth_idp(),
-                xaa_enabled.then_some(&idp)
-            );
+            let eligible = name == "configured" && xaa_enabled;
+            assert_eq!(servers[name].enabled(), eligible && !denied, "{name}");
+            assert_eq!(servers[name].config().oauth_idp(), eligible.then_some(&idp));
             assert_eq!(servers[name].config().auth, McpServerAuth::EmaAuth);
         }
     }

codex-rs/config/src/mcp_ema.rs

@@ -15,7 +15,6 @@ use crate::ConfigLayerSource;
 use crate::ConfigLayerStack;
 use crate::McpServerConfig;
 use crate::McpServerTransportConfig;
-use crate::types::PluginMcpServerEmaAuthConfig;
 
 #[derive(Serialize, Deserialize, Debug, Clone, PartialEq, Eq, JsonSchema)]
 #[serde(deny_unknown_fields)]
@@ -73,7 +72,6 @@ impl McpEmaRegistration {
 
 impl McpEnterpriseManagedAuthConfig {
     /// Resolves the trusted profile and checks provenance at a config load boundary.
-    /// Plugin endpoints are checked later, when their declarations are materialized.
     pub fn resolve(
         stack: &ConfigLayerStack,
         fallback: Option<&Self>,
@@ -139,17 +137,13 @@ impl McpServerConfig {
         self.oauth.as_ref()?.ema_registration.as_ref()
     }
 
-    /// Called once on a materialized catalog, after configuration provenance and
-    /// plugin endpoint policy have been checked. Runtime consumers use the result.
+    /// Called once on a materialized catalog after configuration provenance is checked.
     pub fn resolve_ema_registration(
         &mut self,
         idp: &McpServerIdpOAuthConfig,
     ) -> Result<(), &'static str> {
         if let Some(oauth) = &mut self.oauth {
             oauth.ema_registration = None;
-            if let Some(error) = oauth.ema_registration_error {
-                return Err(error);
-            }
         }
         if !matches!(self.auth, crate::McpServerAuth::EmaAuth) {
             return Err("enterprise registration requires ema_auth");
@@ -203,22 +197,6 @@ impl McpServerConfig {
     }
 }
 
-fn plugin_ema_registration(
-    config: &toml::Value,
-    plugin_name: &str,
-    server_name: &str,
-) -> Option<PluginMcpServerEmaAuthConfig> {
-    config
-        .get("plugins")?
-        .get(plugin_name)?
-        .get("mcp_servers")?
-        .get(server_name)?
-        .get("ema_auth")?
-        .clone()
-        .try_into()
-        .ok()
-}
-
 fn has_trusted_atomic_registration(
     stack: &ConfigLayerStack,
     non_project: &toml::Value,
@@ -299,54 +277,6 @@ fn validate_ema_auth_sources(
         }
     }
 
-    let effective_config = stack.effective_config();
-    let Some(plugins) = effective_config
-        .get("plugins")
-        .and_then(toml::Value::as_table)
-    else {
-        return Ok(());
-    };
-    for (plugin_name, plugin) in plugins {
-        let Some(plugin_servers) = plugin.get("mcp_servers").and_then(toml::Value::as_table) else {
-            continue;
-        };
-        for (server_name, server) in plugin_servers {
-            let Some(ema_auth) = server.get("ema_auth") else {
-                continue;
-            };
-            let non_project_server = non_project
-                .get("plugins")
-                .and_then(|plugins| plugins.get(plugin_name))
-                .and_then(|plugin| plugin.get("mcp_servers"))
-                .and_then(|servers| servers.get(server_name));
-            if reenabled_over_non_project_denial(
-                server.get("enabled").and_then(toml::Value::as_bool) != Some(false),
-                non_project_server,
-            ) {
-                return Err(io::Error::new(
-                    io::ErrorKind::InvalidInput,
-                    format!(
-                        "project configuration cannot re-enable enterprise MCP plugin `{plugin_name}` server `{server_name}`"
-                    ),
-                ));
-            }
-            let effective = ema_auth
-                .clone()
-                .try_into::<PluginMcpServerEmaAuthConfig>()
-                .map_err(|error| io::Error::new(io::ErrorKind::InvalidInput, error))?;
-            if !has_trusted_atomic_registration(stack, &non_project, |config| {
-                plugin_ema_registration(config, plugin_name, server_name)
-                    .is_some_and(|registration| registration == effective)
-            }) {
-                return Err(io::Error::new(
-                    io::ErrorKind::InvalidInput,
-                    format!(
-                        "enterprise MCP registration for plugin `{plugin_name}` server `{server_name}` must be defined in one non-project config layer"
-                    ),
-                ));
-            }
-        }
-    }
     Ok(())
 }
 

codex-rs/config/src/mcp_ema_tests.rs

@@ -109,97 +109,51 @@ fn ema_profiles_and_auth_modes_preserve_non_project_authority() {
 #[test]
 fn ema_registrations_require_one_atomic_non_project_source() {
     let (system, user, project) = local_sources("ema-registration-sources");
-    for (section, authorization, protected_changes) in [
-        (
-            "mcp_servers.enterprise",
-            r#"
-            auth='ema_auth'
-            oauth_resource='https://resource.example'
-            oauth.client_id='resource-client'
-            oauth.authorization_server_issuer='https://as.example'
-        "#,
-            &[
-                "oauth_resource='https://other.example'",
-                "oauth.client_id='other-client'",
-                "oauth.authorization_server_issuer='https://other-as.example'",
-            ] as &[&str],
-        ),
-        (
-            "plugins.\"sample@test\".mcp_servers.enterprise.ema_auth",
-            r#"
-            resource='https://resource.example'
-            client_id='resource-client'
-            authorization_server_issuer='https://as.example'
-        "#,
-            &[
-                "resource='https://other.example'",
-                "client_id='other-client'",
-                "authorization_server_issuer='https://other-as.example'",
-            ],
-        ),
+    let registration = r#"[mcp_servers.enterprise]
+url='https://resource.example/mcp'
+auth='ema_auth'
+scopes=['tools']
+oauth_resource='https://resource.example'
+oauth.client_id='resource-client'
+oauth.authorization_server_issuer='https://as.example'"#;
+    for (source, allowed) in [
+        (system.clone(), true),
+        (user.clone(), true),
+        (project.clone(), false),
     ] {
-        let registration = format!(
-            "[{section}]\nurl='https://resource.example/mcp'\nscopes=['tools']\n{authorization}"
-        );
-        for (source, allowed) in [
-            (system.clone(), true),
-            (user.clone(), true),
-            (project.clone(), false),
-        ] {
-            assert_eq!(
-                valid_ema_layers(vec![(source, registration.as_str())]),
-                allowed,
-                "{section}"
-            );
-        }
-        for change in protected_changes.iter().copied().chain([
-            "url='https://other.example/mcp'",
-            "url='https://resource.example/mcp/other'",
-            "scopes=['admin']",
-        ]) {
-            let overlay = format!("[{section}]\n{change}");
-            assert!(
-                !valid_ema_layers(vec![
-                    (system.clone(), &registration),
-                    (project.clone(), &overlay)
-                ]),
-                "{section}: {change}"
-            );
-        }
-        let higher = registration.replace("tools", "managed-tools");
-        assert!(
-            !valid_ema_layers(vec![
-                (system.clone(), &registration),
-                (user.clone(), &higher),
-                (project.clone(), &registration)
-            ]),
-            "project rollback: {section}"
-        );
-        let policy_section = section.strip_suffix(".ema_auth").unwrap_or(section);
-        let policy = format!(
-            "[{policy_section}]\nenabled=false\n[{policy_section}.tools.read]\napproval_mode='prompt'"
-        );
-        assert!(valid_ema_layers(vec![
-            (system.clone(), &registration),
-            (project.clone(), &policy)
-        ]));
-        if section.ends_with(".ema_auth") {
-            for required in [
-                "url='https://resource.example/mcp'\n",
-                "resource='https://resource.example'\n",
-            ] {
-                assert!(
-                    !valid_ema_layers(vec![(system.clone(), &registration.replace(required, ""))]),
-                    "missing {required}"
-                );
-            }
-        } else {
-            assert!(valid_ema_layers(vec![
-                (system.clone(), &registration),
-                (project.clone(), "[mcp_servers.enterprise]\nauth='ema_auth'")
-            ]));
-        }
+        assert_eq!(valid_ema_layers(vec![(source, registration)]), allowed);
     }
+    for change in [
+        "oauth_resource='https://other.example'",
+        "oauth.client_id='other-client'",
+        "oauth.authorization_server_issuer='https://other-as.example'",
+        "url='https://other.example/mcp'",
+        "url='https://resource.example/mcp/other'",
+        "scopes=['admin']",
+    ] {
+        let overlay = format!("[mcp_servers.enterprise]\n{change}");
+        assert!(!valid_ema_layers(vec![
+            (system.clone(), registration),
+            (project.clone(), &overlay),
+        ]));
+    }
+    let higher = registration.replace("tools", "managed-tools");
+    assert!(!valid_ema_layers(vec![
+        (system.clone(), registration),
+        (user, &higher),
+        (project.clone(), registration),
+    ]));
+    assert!(valid_ema_layers(vec![
+        (system.clone(), registration),
+        (
+            project.clone(),
+            "[mcp_servers.enterprise]\nenabled=false\n[mcp_servers.enterprise.tools.read]\napproval_mode='prompt'"
+        ),
+    ]));
+    assert!(valid_ema_layers(vec![
+        (system, registration),
+        (project, "[mcp_servers.enterprise]\nauth='ema_auth'"),
+    ]));
 }
 
 #[test]
@@ -222,32 +176,6 @@ fn project_cannot_reenable_trusted_disabled_ema_registration() {
     ]));
 }
 
-#[test]
-fn project_cannot_reenable_trusted_disabled_ema_plugin() {
-    let (system, _, project) = local_sources("ema-disabled-plugin");
-    let trusted = r#"
-        [plugins."sample@test".mcp_servers.enterprise]
-        enabled=false
-        [plugins."sample@test".mcp_servers.enterprise.ema_auth]
-        url='https://resource.example/mcp'
-        client_id='resource-client'
-        authorization_server_issuer='https://as.example'
-        resource='https://resource.example'
-    "#;
-    let project_enable = "[plugins.\"sample@test\".mcp_servers.enterprise]\nenabled=true";
-    assert!(!valid_ema_layers(vec![
-        (system.clone(), trusted),
-        (project.clone(), project_enable)
-    ]));
-    assert!(valid_ema_layers(vec![
-        (system, trusted),
-        (
-            project,
-            "[plugins.\"sample@test\".mcp_servers.enterprise]\nenabled=false"
-        )
-    ]));
-}
-
 #[test]
 fn non_project_auth_changes_and_ordinary_oauth_remain_allowed() {
     let (system, user, project) = local_sources("ema-auth-downgrade");

codex-rs/config/src/mcp_types.rs

@@ -61,8 +61,6 @@ pub enum McpServerDisabledReason {
     Unknown,
     /// The server was disabled by config requirements from the given source.
     Requirements { source: RequirementSource },
-    /// Enterprise authorization was rejected for this registration, not its name.
-    EmaRegistration,
 }
 
 impl fmt::Display for McpServerDisabledReason {
@@ -72,9 +70,6 @@ impl fmt::Display for McpServerDisabledReason {
             McpServerDisabledReason::Requirements { source } => {
                 write!(f, "requirements ({source})")
             }
-            McpServerDisabledReason::EmaRegistration => {
-                write!(f, "invalid enterprise registration")
-            }
         }
     }
 }
@@ -185,11 +180,6 @@ pub struct McpServerOAuthConfig {
     #[serde(skip)]
     #[schemars(skip)]
     pub ema_registration: Option<McpEmaRegistration>,
-
-    /// Host-policy rejection retained until catalog finalization; never deserialized.
-    #[serde(skip)]
-    #[schemars(skip)]
-    pub ema_registration_error: Option<&'static str>,
 }
 
 /// Authentication flow for an HTTP MCP server. Explicit credentials take

codex-rs/config/src/state.rs

@@ -307,6 +307,21 @@ impl ConfigLayerStack {
         self.ignore_user_and_project_exec_policy_rules
     }
 
+    /// Retains session layers while adopting current MCP, plugin, and feature restrictions.
+    /// Rejected refreshes must not restore an earlier policy on the next user reload.
+    pub fn with_mcp_requirements_from(&self, incoming: &Self) -> Self {
+        let mut stack = self.clone();
+        stack.requirements.mcp_servers = incoming.requirements.mcp_servers.clone();
+        stack.requirements.plugins = incoming.requirements.plugins.clone();
+        stack.requirements.feature_requirements =
+            incoming.requirements.feature_requirements.clone();
+        stack.requirements_toml.mcp_servers = incoming.requirements_toml.mcp_servers.clone();
+        stack.requirements_toml.plugins = incoming.requirements_toml.plugins.clone();
+        stack.requirements_toml.feature_requirements =
+            incoming.requirements_toml.feature_requirements.clone();
+        stack
+    }
+
     pub(crate) fn with_startup_warnings(mut self, startup_warnings: Vec<String>) -> Self {
         self.startup_warnings = Some(startup_warnings);
         self

codex-rs/config/src/types.rs

@@ -1021,9 +1021,15 @@ pub struct PluginMcpServerConfig {
     #[serde(default = "default_enabled")]
     pub enabled: bool,
 
-    /// Host-configured EMA registration; the plugin still owns its endpoint.
-    #[serde(default, skip_serializing_if = "Option::is_none")]
-    pub ema_auth: Option<PluginMcpServerEmaAuthConfig>,
+    /// Retired EMA overlays must disable the server instead of losing their auth policy.
+    #[serde(
+        default,
+        rename = "ema_auth",
+        deserialize_with = "unsupported_plugin_ema_auth",
+        skip_serializing_if = "std::ops::Not::not"
+    )]
+    #[schemars(skip)]
+    pub has_unsupported_ema_auth: bool,
 
     /// Approval mode for tools in this server unless a tool override exists.
     #[serde(default, skip_serializing_if = "Option::is_none")]
@@ -1046,7 +1052,7 @@ impl Default for PluginMcpServerConfig {
     fn default() -> Self {
         Self {
             enabled: true,
-            ema_auth: None,
+            has_unsupported_ema_auth: false,
             default_tools_approval_mode: None,
             enabled_tools: None,
             disabled_tools: None,
@@ -1055,46 +1061,11 @@ impl Default for PluginMcpServerConfig {
     }
 }
 
-/// Resource registration applied through an existing per-plugin policy overlay.
-/// The enterprise IdP is selected separately by trusted host configuration.
-#[derive(Serialize, Deserialize, Debug, Clone, PartialEq, Eq, JsonSchema)]
-#[serde(deny_unknown_fields)]
-pub struct PluginMcpServerEmaAuthConfig {
-    /// Exact plugin endpoint approved by the host; never overrides the declaration.
-    pub url: String,
-    pub client_id: String,
-    pub authorization_server_issuer: String,
-    #[serde(default)]
-    pub scopes: Vec<String>,
-    pub resource: String,
-}
-
-impl PluginMcpServerEmaAuthConfig {
-    pub fn apply(&self, server: &mut McpServerConfig) {
-        let registration_error = if self.resource.trim().is_empty() {
-            Some("plugin EMA registration requires a resource")
-        } else if !server.matches_requirement(&crate::McpServerRequirement::Identity {
-            identity: crate::McpServerIdentity::Url {
-                url: self.url.clone(),
-            },
-        }) {
-            Some("plugin endpoint does not match its EMA registration")
-        } else {
-            None
-        };
-        if registration_error.is_some() && server.enabled {
-            server.enabled = false;
-            server.disabled_reason = Some(crate::McpServerDisabledReason::EmaRegistration);
-        }
-        server.auth = McpServerAuth::EmaAuth;
-        let oauth = server.oauth.get_or_insert_default();
-        oauth.client_id = Some(self.client_id.clone());
-        oauth.authorization_server_issuer = Some(self.authorization_server_issuer.clone());
-        server.scopes = Some(self.scopes.clone());
-        oauth.ema_registration = None;
-        oauth.ema_registration_error = registration_error;
-        server.oauth_resource = Some(self.resource.clone());
-    }
+fn unsupported_plugin_ema_auth<'de, D>(deserializer: D) -> Result<bool, D::Error>
+where
+    D: serde::Deserializer<'de>,
+{
+    serde::de::IgnoredAny::deserialize(deserializer).map(|_| true)
 }
 
 #[derive(Serialize, Deserialize, Debug, Clone, PartialEq, Eq, Default, JsonSchema)]

codex-rs/core-plugins/src/loader.rs

@@ -966,9 +966,9 @@ async fn load_plugin(
 }
 
 fn apply_plugin_mcp_server_policy(config: &mut McpServerConfig, policy: &PluginMcpServerConfig) {
-    config.enabled = policy.enabled;
-    if let Some(ema) = &policy.ema_auth {
-        ema.apply(config);
+    config.enabled = policy.enabled && !policy.has_unsupported_ema_auth;
+    if policy.has_unsupported_ema_auth {
+        config.auth = codex_config::McpServerAuth::EmaAuth;
     }
     if let Some(approval_mode) = policy.default_tools_approval_mode {
         config.default_tools_approval_mode = Some(approval_mode);
@@ -1437,9 +1437,9 @@ pub fn apply_configured_plugin_mcp_server_policies(
 ) {
     for (name, server) in servers {
         if let Some(policy) = policies.get(name) {
-            server.enabled &= policy.enabled;
-            if let Some(ema) = &policy.ema_auth {
-                ema.apply(server);
+            server.enabled &= policy.enabled && !policy.has_unsupported_ema_auth;
+            if policy.has_unsupported_ema_auth {
+                server.auth = codex_config::McpServerAuth::EmaAuth;
             }
             let declared_approval_mode = server.default_tools_approval_mode.unwrap_or_default();
 

codex-rs/core-plugins/src/loader_tests.rs

@@ -27,102 +27,6 @@ fn user_layer(path: AbsolutePathBuf, config: &str) -> ConfigLayerEntry {
     )
 }
 
-#[tokio::test]
-async fn ema_policy_overlays_native_and_agent_plugins_without_changing_endpoints() {
-    let temp_dir = TempDir::new().expect("tempdir");
-    let policy_config = r#"
-        [plugins."sample@test".mcp_servers.example]
-        enabled = true
-        [plugins."sample@test".mcp_servers.example.ema_auth]
-        url = "https://resource.example/mcp"
-        client_id = "resource-client"
-        authorization_server_issuer = "https://as.example"
-        resource = "https://resource.example"
-        scopes = ["tools"]
-    "#;
-    let stack = ConfigLayerStack::new(
-        vec![ConfigLayerEntry::new(
-            ConfigLayerSource::EnterpriseManaged {
-                id: "ema-policy".to_string(),
-                name: "EMA policy".to_string(),
-            },
-            toml::from_str(policy_config).expect("managed policy toml"),
-        )],
-        ConfigRequirements::default(),
-        ConfigRequirementsToml::default(),
-    )
-    .expect("valid trusted policy stack");
-    let policies = configured_plugins_from_stack(&stack, temp_dir.path())
-        .remove("sample@test")
-        .expect("configured plugin")
-        .mcp_servers;
-
-    for (name, manifest_path, manifest, mcp_path, mcp) in [
-        (
-            "native",
-            ".codex-plugin/plugin.json",
-            r#"{"name":"native"}"#,
-            ".mcp.json",
-            r#"{"mcpServers":{"example":{"url":"https://resource.example/mcp"}}}"#,
-        ),
-        (
-            "agent",
-            "plugin.json",
-            r#"{"$schema":"https://agent-plugins.org/schemas/1.0.0/plugin.schema.json","name":"agent"}"#,
-            "mcp.json",
-            r#"{"$schema":"https://agent-plugins.org/schemas/1.0.0/mcp.schema.json","mcpServers":{"example":{"type":"streamable-http","url":"https://resource.example/mcp"}}}"#,
-        ),
-    ] {
-        for endpoint in [
-            "https://resource.example/mcp",
-            "https://resource.example/mcp/other",
-            "https://resource.example/mcp?tenant=other",
-            "https://other.example/mcp",
-        ] {
-            let root = temp_dir.path().join(name);
-            write_file(&root.join(manifest_path), manifest);
-            write_file(
-                &root.join(mcp_path),
-                &mcp.replace("https://resource.example/mcp", endpoint),
-            );
-            let declared = load_plugin_mcp_servers(&root, /*auth_mode*/ None).await;
-            let transport = declared["example"].transport.clone();
-            let installed = load_plugin_mcp_servers_with_policy(
-                &root,
-                /*auth_mode*/ None,
-                Some(&policies),
-            )
-            .await;
-            let mut selected = declared;
-            apply_configured_plugin_mcp_server_policies(&policies, &mut selected);
-            assert_eq!(installed, selected);
-            let server = &selected["example"];
-            assert_eq!(server.transport, transport);
-            assert_eq!(server.enabled, endpoint == "https://resource.example/mcp");
-            assert_eq!(server.auth, codex_config::McpServerAuth::EmaAuth);
-            assert_eq!(
-                server.oauth,
-                Some(codex_config::McpServerOAuthConfig {
-                    client_id: Some("resource-client".into()),
-                    authorization_server_issuer: Some("https://as.example".into()),
-                    ema_registration_error: (endpoint != "https://resource.example/mcp")
-                        .then_some("plugin endpoint does not match its EMA registration"),
-                    ..Default::default()
-                })
-            );
-            assert_eq!(
-                server.oauth_resource.as_deref(),
-                Some("https://resource.example")
-            );
-            assert_eq!(
-                server.scopes.as_deref(),
-                Some(["tools".to_string()].as_slice())
-            );
-            assert_eq!(server.oauth_idp(), None);
-        }
-    }
-}
-
 #[tokio::test]
 async fn agent_plugin_overlay_apps_are_not_runtime_active() {
     let temp_dir = TempDir::new().expect("tempdir");
@@ -492,6 +396,54 @@ fn configured_plugins_from_stack_merges_enabled_effective_layers() {
     );
 }
 
+#[tokio::test]
+async fn legacy_ema_policy_disables_installed_and_selected_plugin_servers() {
+    let temp_dir = TempDir::new().expect("tempdir");
+    let plugin_root = temp_dir.path().join("plugin");
+    write_file(
+        &plugin_root.join(".codex-plugin/plugin.json"),
+        r#"{"name":"plugin","mcpServers":"./mcp.json"}"#,
+    );
+    write_file(
+        &plugin_root.join("mcp.json"),
+        r#"{"mcpServers":{"example":{"url":"https://resource.example/mcp"}}}"#,
+    );
+    for endpoint in ["https://resource.example/mcp", "https://other.example/mcp"] {
+        let stack = ConfigLayerStack::new(
+            vec![user_layer(
+                user_config_path(&temp_dir, "config.toml"),
+                &format!(
+                    "[plugins.\"plugin@market\".mcp_servers.example.ema_auth]\nurl = '{endpoint}'"
+                ),
+            )],
+            ConfigRequirements::default(),
+            ConfigRequirementsToml::default(),
+        )
+        .expect("legacy policy stack");
+        let policies = configured_plugin_mcp_server_policies(&stack);
+        let policy = policies.get("plugin@market").expect("plugin policy");
+        let installed = load_plugin_mcp_servers_with_policy(
+            &plugin_root,
+            /*auth_mode*/ None,
+            Some(policy),
+        )
+        .await;
+        let mut selected = load_plugin_mcp_servers(&plugin_root, /*auth_mode*/ None).await;
+        assert!(selected["example"].enabled);
+        apply_configured_plugin_mcp_server_policies(policy, &mut selected);
+        assert!(!installed["example"].enabled);
+        assert_eq!(installed, selected);
+
+        let round_trip: HashMap<String, PluginMcpServerConfig> =
+            toml::from_str(&toml::to_string(policy).expect("serialize policy"))
+                .expect("deserialize policy");
+        assert_eq!(
+            &round_trip, policy,
+            "serialization must preserve the denial"
+        );
+    }
+}
+
 #[tokio::test]
 async fn hooks_only_scope_shares_plugin_resolution_without_loading_other_capabilities() {
     let temp_dir = TempDir::new().expect("tempdir");

codex-rs/core/config.schema.json

@@ -3517,14 +3517,6 @@
           },
           "type": "array"
         },
-        "ema_auth": {
-          "allOf": [
-            {
-              "$ref": "#/definitions/PluginMcpServerEmaAuthConfig"
-            }
-          ],
-          "description": "Host-configured EMA registration; the plugin still owns its endpoint."
-        },
         "enabled": {
           "default": true,
           "description": "When `false`, Codex skips initializing this plugin MCP server.",
@@ -3547,39 +3539,6 @@
       },
       "type": "object"
     },
-    "PluginMcpServerEmaAuthConfig": {
-      "additionalProperties": false,
-      "description": "Resource registration applied through an existing per-plugin policy overlay. The enterprise IdP is selected separately by trusted host configuration.",
-      "properties": {
-        "authorization_server_issuer": {
-          "type": "string"
-        },
-        "client_id": {
-          "type": "string"
-        },
-        "resource": {
-          "type": "string"
-        },
-        "scopes": {
-          "default": [],
-          "items": {
-            "type": "string"
-          },
-          "type": "array"
-        },
-        "url": {
-          "description": "Exact plugin endpoint approved by the host; never overrides the declaration.",
-          "type": "string"
-        }
-      },
-      "required": [
-        "authorization_server_issuer",
-        "client_id",
-        "resource",
-        "url"
-      ],
-      "type": "object"
-    },
     "ProfileTui": {
       "additionalProperties": false,
       "description": "TUI settings supported inside a named profile.",

codex-rs/core/src/codex_thread.rs

@@ -161,6 +161,18 @@ pub struct CodexThreadSettingsOverrides {
     pub disabled_plugin_ids: Option<Vec<String>>,
 }
 
+/// Result of publishing a loaded configuration snapshot for a thread.
+#[derive(Clone, Copy, Debug, Eq, PartialEq)]
+#[must_use]
+pub enum ConfigRefreshOutcome {
+    /// The resolved configuration was published.
+    Published,
+    /// The captured owner changed; the caller must reload before retrying.
+    Stale,
+    /// Resolution failed and a configuration disabling enterprise MCP was published.
+    Rejected,
+}
+
 pub use codex_guardian_context::GuardianRootMessage;
 
 /// Authorization state that changes on genuine user input or history resets.
@@ -985,14 +997,36 @@ impl CodexThread {
 
     /// Refresh the thread's layer-backed user config state from a caller-supplied
     /// config snapshot. Thread-scoped layers and session-static settings remain
-    /// unchanged.
-    pub async fn refresh_runtime_config(&self, next_config: crate::config::Config) {
-        self.session.refresh_runtime_config(next_config).await;
+    /// unchanged. A stale owner requires reloading from the current config before retrying.
+    pub async fn refresh_runtime_config(
+        &self,
+        expected_config: Arc<crate::config::Config>,
+        next_config: crate::config::Config,
+    ) -> ConfigRefreshOutcome {
+        Box::pin(
+            self.session
+                .refresh_runtime_config(expected_config, next_config),
+        )
+        .await
+    }
+
+    /// Revokes enterprise MCP authority from the current owner after a failed reload.
+    pub async fn disable_mcp_enterprise_auth(&self) {
+        self.session.disable_mcp_enterprise_auth().await;
     }
 
     /// Refresh MCP configuration and managed requirements without reloading unrelated settings.
-    pub async fn refresh_mcp_config(&self, next_config: crate::config::Config) {
-        self.session.refresh_mcp_config(next_config).await;
+    /// A stale owner requires reloading from the current config before retrying.
+    pub async fn refresh_mcp_config(
+        &self,
+        expected_config: Arc<crate::config::Config>,
+        next_config: crate::config::Config,
+    ) -> ConfigRefreshOutcome {
+        Box::pin(
+            self.session
+                .refresh_mcp_config(expected_config, next_config),
+        )
+        .await
     }
 
     /// Refreshes this thread's Apps tools before returning their runtime state.

codex-rs/core/src/compact_tests.rs

@@ -143,9 +143,10 @@ async fn local_compaction_respects_tool_metadata_state(
     assert!(metadata_bytes > 2 * 1024 * 1024);
 
     if !metadata_enabled {
-        let mut config = (*session.get_config().await).clone();
+        let current_config = session.get_config().await;
+        let mut config = current_config.as_ref().clone();
         config.features.disable(Feature::ExecutedToolCallMetadata)?;
-        session.refresh_runtime_config(config).await;
+        let _ = session.refresh_runtime_config(current_config, config).await;
     }
 
     let mock = responses::mount_sse_once(

codex-rs/core/src/config/managed_features.rs

@@ -118,6 +118,28 @@ impl ManagedFeatures {
         self.set(next)
     }
 
+    /// Refresh ordinary MCP feature values and pins without changing session-static features.
+    pub(crate) fn refresh_mcp_features(&mut self, next: &Self) -> ConstraintResult<()> {
+        let mut refreshed = self.clone();
+        let mut values = self.get().clone();
+        for feature in [
+            Feature::EnableMcpApps,
+            Feature::SecretAuthStorage,
+            Feature::McpOAuthRefreshCoordination,
+        ] {
+            values.set_enabled(feature, next.enabled(feature));
+            if let Some(enabled) = next.pinned_features.get(&feature) {
+                refreshed.pinned_features.insert(feature, *enabled);
+            } else {
+                refreshed.pinned_features.remove(&feature);
+            }
+        }
+        refreshed.value.source = next.value.source.clone().or(refreshed.value.source);
+        refreshed.set(values)?;
+        *self = refreshed;
+        Ok(())
+    }
+
     pub fn enable(&mut self, feature: Feature) -> ConstraintResult<()> {
         self.set_enabled(feature, /*enabled*/ true)
     }

codex-rs/core/src/config/mod.rs

@@ -176,6 +176,8 @@ mod permission_profile_selection;
 mod permissions;
 mod requirements;
 mod resolved_permission_profile;
+mod runtime_refresh;
+pub(crate) use runtime_refresh::RuntimeConfigRefresh;
 #[cfg(test)]
 mod schema;
 mod token_budget_startup;

codex-rs/core/src/config/runtime_refresh.rs

@@ -0,0 +1,171 @@
+//! Resolves refreshable MCP inputs without rebuilding unrelated session configuration.
+//! Existing sessions refresh ordinary MCP state without gaining enterprise authority.
+
+use crate::config::Config;
+use crate::config::ManagedFeatures;
+use crate::config::constrain_mcp_servers;
+use crate::config::resolve_tool_suggest_config_from_layer_stack;
+use codex_config::McpServerAuth;
+use codex_config::McpServerConfig;
+use codex_features::Feature;
+use codex_features::FeatureConfigSource;
+use codex_features::FeatureOverrides;
+use codex_features::Features;
+use codex_features::FeaturesToml;
+use serde::Deserialize;
+use std::collections::HashMap;
+
+#[cfg(test)]
+#[path = "runtime_refresh_tests.rs"]
+mod tests;
+
+#[derive(Clone, Copy)]
+pub(crate) enum RuntimeConfigRefresh {
+    User,
+    Mcp,
+}
+
+#[derive(Default, Deserialize)]
+#[serde(default)]
+struct McpRefreshToml {
+    features: Option<FeaturesToml>,
+}
+
+impl Config {
+    pub(crate) fn resolve_runtime_refresh(
+        &self,
+        incoming: &Self,
+        scope: RuntimeConfigRefresh,
+    ) -> std::io::Result<Self> {
+        let layers = match scope {
+            RuntimeConfigRefresh::User => self
+                .config_layer_stack
+                .with_user_layer_from(&incoming.config_layer_stack),
+            RuntimeConfigRefresh::Mcp => Self::layer_stack_preserving_session(
+                &self.config_layer_stack,
+                &incoming.config_layer_stack,
+            )?
+            .with_user_layer_from(&self.config_layer_stack),
+        };
+        let cfg: McpRefreshToml = layers
+            .effective_config()
+            .try_into()
+            .map_err(|err| std::io::Error::new(std::io::ErrorKind::InvalidData, err))?;
+        let mut configured_features = Features::from_sources(
+            FeatureConfigSource {
+                features: cfg.features.as_ref(),
+                ..Default::default()
+            },
+            FeatureConfigSource::default(),
+            FeatureOverrides::default(),
+        );
+        // These existing operational controls also accept host runtime overrides.
+        // Enterprise opt-in always comes from trusted layers instead.
+        for feature in [
+            Feature::SecretAuthStorage,
+            Feature::McpOAuthRefreshCoordination,
+        ] {
+            // A publication wait may have allowed a newer host/user override.
+            configured_features.set_enabled(feature, incoming.features.enabled(feature));
+        }
+        let features = ManagedFeatures::from_configured(
+            configured_features,
+            layers.requirements().feature_requirements.clone(),
+        )?;
+        let is_enterprise = |server: &McpServerConfig| {
+            matches!(server.auth, McpServerAuth::EmaAuth) || server.ema_registration().is_some()
+        };
+        // Refresh ordinary servers from the complete host-owned set. Keep the
+        // initially admitted EMA configuration until a fresh session is constructed.
+        let mut servers: HashMap<_, _> = incoming
+            .mcp_servers
+            .get()
+            .iter()
+            .filter(|(name, server)| {
+                !is_enterprise(server)
+                    && !self.mcp_servers.get().get(*name).is_some_and(is_enterprise)
+            })
+            .map(|(name, server)| (name.clone(), server.clone()))
+            .collect();
+        servers.extend(
+            self.mcp_servers
+                .get()
+                .iter()
+                .filter(|(_, server)| is_enterprise(server))
+                .map(|(name, server)| (name.clone(), server.clone())),
+        );
+        let mut config = self.clone();
+        config
+            .features
+            .refresh_mcp_features(&features)
+            .map_err(std::io::Error::other)?;
+        config.mcp_servers =
+            constrain_mcp_servers(servers, layers.requirements().mcp_servers.as_ref())
+                .map_err(std::io::Error::other)?;
+        // A missing profile also carries the host's fail-closed load result.
+        let enterprise_retired = incoming.mcp_enterprise_managed_auth.is_none()
+            || !incoming.features.enabled(Feature::UseXaa)
+            || (self
+                .mcp_servers
+                .get()
+                .values()
+                .any(|server| is_enterprise(server) && server.enabled)
+                && incoming.mcp_enterprise_managed_auth != self.mcp_enterprise_managed_auth)
+            || self.mcp_servers.get().iter().any(|(name, server)| {
+                is_enterprise(server)
+                    && server.enabled
+                    && !incoming
+                        .mcp_servers
+                        .get()
+                        .get(name)
+                        .is_some_and(|incoming| {
+                            incoming.enabled
+                                && incoming.auth == server.auth
+                                && incoming.transport == server.transport
+                                && incoming.oauth_resource == server.oauth_resource
+                                && incoming.oauth == server.oauth
+                                && incoming.scopes == server.scopes
+                                && incoming.environment_id == server.environment_id
+                                && incoming.ema_registration() == server.ema_registration()
+                        })
+            })
+            || self.mcp_servers.get().iter().any(|(name, server)| {
+                is_enterprise(server)
+                    && server.enabled
+                    && !config
+                        .mcp_servers
+                        .get()
+                        .get(name)
+                        .is_some_and(|server| server.enabled)
+            });
+        config.mcp_optional_startup_grace = incoming.mcp_optional_startup_grace;
+        config.mcp_oauth_credentials_store_mode = incoming.mcp_oauth_credentials_store_mode;
+        if !matches!(scope, RuntimeConfigRefresh::Mcp) {
+            config.active_project = incoming.active_project.clone();
+            config.tool_suggest = resolve_tool_suggest_config_from_layer_stack(&layers);
+        }
+        config.config_layer_stack = layers;
+        if enterprise_retired {
+            config.disable_mcp_enterprise_auth();
+        }
+        Ok(config)
+    }
+
+    /// Disables enterprise MCP authorization while preserving unrelated MCP servers.
+    pub fn disable_mcp_enterprise_auth(&mut self) {
+        self.mcp_enterprise_managed_auth = None;
+        let mut servers = self.mcp_servers.get().clone();
+        for server in servers.values_mut() {
+            if matches!(server.auth, McpServerAuth::EmaAuth) || server.ema_registration().is_some()
+            {
+                server.enabled = false;
+                if let Some(oauth) = &mut server.oauth {
+                    oauth.ema_registration = None;
+                }
+            }
+        }
+        if self.mcp_servers.set(servers).is_err() {
+            tracing::warn!("failed to clear disabled enterprise registrations");
+        }
+    }
+}

codex-rs/core/src/config/runtime_refresh_tests.rs

@@ -0,0 +1,237 @@
+//! Ordinary refresh resolution and enterprise revocation without Session construction.
+
+use super::*;
+use crate::config::ConfigBuilder;
+use codex_config::CONFIG_TOML_FILE;
+use codex_config::ConfigLayerSource;
+use codex_config::test_support::CloudConfigBundleFixture;
+use codex_core_plugins::PluginLoadOutcome;
+use pretty_assertions::assert_eq;
+
+async fn base_config() -> (tempfile::TempDir, Config) {
+    let home = tempfile::tempdir().unwrap();
+    let config = ConfigBuilder::without_managed_config_for_tests()
+        .codex_home(home.path().to_path_buf())
+        .fallback_cwd(Some(home.path().to_path_buf()))
+        .build()
+        .await
+        .unwrap();
+    (home, config)
+}
+
+async fn layered_config(
+    base: &Config,
+    user: &str,
+    managed: &str,
+    requirements: &str,
+    session: &str,
+) -> Config {
+    std::fs::write(base.codex_home.join(CONFIG_TOML_FILE), user).unwrap();
+    let overrides: toml::Table = toml::from_str(session).unwrap();
+    ConfigBuilder::without_managed_config_for_tests()
+        .codex_home(base.codex_home.to_path_buf())
+        .fallback_cwd(Some(base.cwd.to_path_buf()))
+        .cli_overrides(overrides.into_iter().collect())
+        .cloud_config_bundle(
+            CloudConfigBundleFixture::enterprise_config(managed)
+                .add_enterprise_requirement(format!("[features]\n{requirements}"))
+                .into_loader(),
+        )
+        .build()
+        .await
+        .unwrap()
+}
+
+fn enterprise_config(issuer: &str, resource: &str) -> String {
+    format!(
+        "[features]\nuse_xaa = true\napps = false\nsecret_auth_storage = false\n\
+         [mcp_servers.enterprise]\nurl = 'https://resource.example/mcp'\nauth = 'ema_auth'\n\
+         oauth_resource = '{resource}'\n\
+         [mcp_enterprise_managed_auth.idp]\nissuer = '{issuer}'\nclient_id = '{issuer}-client'\n"
+    )
+}
+
+fn runtime_servers(config: &Config) -> HashMap<String, McpServerConfig> {
+    codex_mcp::configured_mcp_servers(
+        &config.to_mcp_config_with_loaded_plugins(&PluginLoadOutcome::default(), []),
+    )
+}
+
+#[test_case::test_case(RuntimeConfigRefresh::User; "user refresh")]
+#[test_case::test_case(RuntimeConfigRefresh::Mcp; "MCP refresh")]
+#[tokio::test]
+async fn refresh_honors_fresh_managed_denial(scope: RuntimeConfigRefresh) {
+    let (_home, base) = base_config().await;
+    let enterprise = enterprise_config("https://idp.example", "resource");
+    let current = layered_config(&base, &enterprise, "", "use_xaa = true", "").await;
+    let next = layered_config(&base, &enterprise, "", "use_xaa = false", "").await;
+    let refreshed = current.resolve_runtime_refresh(&next, scope).unwrap();
+    assert!(!runtime_servers(&refreshed)["enterprise"].enabled);
+    assert!(refreshed.mcp_enterprise_managed_auth.is_none());
+    assert_eq!(refreshed.model, current.model);
+}
+
+#[tokio::test]
+async fn refresh_honors_disabled_enterprise_host_load_result() {
+    let (_home, base) = base_config().await;
+    let current = layered_config(
+        &base,
+        &enterprise_config("https://idp.example", "resource"),
+        "",
+        "",
+        "",
+    )
+    .await;
+    let mut failed = current.clone();
+    failed.disable_mcp_enterprise_auth();
+    let refreshed = current
+        .resolve_runtime_refresh(&failed, RuntimeConfigRefresh::Mcp)
+        .unwrap();
+    assert!(refreshed.mcp_enterprise_managed_auth.is_none());
+    assert!(!runtime_servers(&refreshed)["enterprise"].enabled);
+}
+
+#[tokio::test]
+async fn ordinary_transport_transitions_use_the_resolved_incoming_map() {
+    let (_home, base) = base_config().await;
+    let command = "[mcp_servers.ordinary]\ncommand = 'ordinary-server'";
+    let url = "[mcp_servers.ordinary]\nurl = 'https://ordinary.example/mcp'";
+    for scope in [RuntimeConfigRefresh::User, RuntimeConfigRefresh::Mcp] {
+        let (current, incoming) = match scope {
+            RuntimeConfigRefresh::User => (
+                layered_config(&base, "", "", "", url).await,
+                layered_config(&base, command, "", "", "").await,
+            ),
+            RuntimeConfigRefresh::Mcp => (
+                layered_config(&base, command, "", "", "").await,
+                layered_config(&base, "", url, "", "").await,
+            ),
+        };
+        let refreshed = current.resolve_runtime_refresh(&incoming, scope).unwrap();
+        assert_eq!(refreshed.mcp_servers.get(), incoming.mcp_servers.get());
+    }
+}
+
+#[tokio::test]
+async fn mcp_refresh_preserves_current_user_and_session_layers_without_reopening_ema() {
+    let (_home, base) = base_config().await;
+    let current = layered_config(
+        &base,
+        &enterprise_config("https://current-idp.example", "current-resource"),
+        "",
+        "",
+        "[mcp_servers.enterprise]\nenabled = false",
+    )
+    .await;
+    let next = layered_config(
+        &base,
+        &enterprise_config("https://old-idp.example", "old-resource"),
+        "",
+        "",
+        "",
+    )
+    .await;
+    let refreshed = current
+        .resolve_runtime_refresh(&next, RuntimeConfigRefresh::Mcp)
+        .unwrap();
+    let preserved_layers = |config: &Config| {
+        config
+            .config_layer_stack
+            .all_layers_low_to_high()
+            .filter(|layer| {
+                matches!(
+                    layer.name,
+                    ConfigLayerSource::User { .. } | ConfigLayerSource::SessionFlags
+                )
+            })
+            .cloned()
+            .collect::<Vec<_>>()
+    };
+    assert_eq!(preserved_layers(&refreshed), preserved_layers(&current));
+    let server = &runtime_servers(&refreshed)["enterprise"];
+    assert!(!server.enabled);
+    assert_eq!(
+        server.ema_registration(),
+        runtime_servers(&current)["enterprise"].ema_registration()
+    );
+}
+
+#[test_case::test_case("removed"; "removed")]
+#[test_case::test_case("disabled"; "disabled")]
+#[test_case::test_case("resource"; "resource changed")]
+#[test_case::test_case("transport"; "transport changed")]
+#[test_case::test_case("scopes"; "scopes changed")]
+#[test_case::test_case("oauth"; "OAuth client changed")]
+#[test_case::test_case("idp"; "IdP changed")]
+#[tokio::test]
+async fn refresh_revokes_changed_enterprise_admission(change: &str) {
+    let (_home, base) = base_config().await;
+    let current = layered_config(
+        &base,
+        &enterprise_config("https://idp.example", "resource"),
+        "",
+        "",
+        "",
+    )
+    .await;
+    let mut incoming = current.clone();
+    let mut servers = incoming.mcp_servers.get().clone();
+    match change {
+        "removed" => {
+            servers.remove("enterprise");
+        }
+        "disabled" => servers.get_mut("enterprise").unwrap().enabled = false,
+        "resource" => servers.get_mut("enterprise").unwrap().oauth_resource = Some("other".into()),
+        "scopes" => servers.get_mut("enterprise").unwrap().scopes = Some(vec!["other".into()]),
+        "oauth" => {
+            servers
+                .get_mut("enterprise")
+                .unwrap()
+                .oauth
+                .get_or_insert_default()
+                .client_id = Some("other-client".into())
+        }
+        "idp" => {
+            incoming
+                .mcp_enterprise_managed_auth
+                .as_mut()
+                .unwrap()
+                .idp
+                .client_id = "other-client".into()
+        }
+        "transport" => {
+            let codex_config::McpServerTransportConfig::StreamableHttp { url, .. } =
+                &mut servers.get_mut("enterprise").unwrap().transport
+            else {
+                unreachable!()
+            };
+            *url = "https://other.example/mcp".into();
+        }
+        _ => unreachable!(),
+    }
+    incoming.mcp_servers.set(servers).unwrap();
+    for scope in [RuntimeConfigRefresh::User, RuntimeConfigRefresh::Mcp] {
+        let revoked = current.resolve_runtime_refresh(&incoming, scope).unwrap();
+        assert!(revoked.mcp_enterprise_managed_auth.is_none());
+        assert!(!runtime_servers(&revoked)["enterprise"].enabled);
+        let restored = revoked.resolve_runtime_refresh(&current, scope).unwrap();
+        assert!(restored.mcp_enterprise_managed_auth.is_none());
+        assert!(!runtime_servers(&restored)["enterprise"].enabled);
+    }
+}
+
+#[tokio::test]
+async fn mcp_refresh_preserves_apps_until_new_session() {
+    let (_home, base) = base_config().await;
+    let current = layered_config(&base, "[features]\napps = true", "", "", "").await;
+    let next = layered_config(&base, "[features]\napps = true", "", "apps = false", "").await;
+    let refreshed = current
+        .resolve_runtime_refresh(&next, RuntimeConfigRefresh::Mcp)
+        .unwrap();
+    assert!(refreshed.features.enabled(Feature::Apps));
+    let reloaded = refreshed
+        .resolve_runtime_refresh(&next, RuntimeConfigRefresh::User)
+        .unwrap();
+    assert!(reloaded.features.enabled(Feature::Apps));
+    assert!(!next.features.enabled(Feature::Apps));
+}

codex-rs/core/src/guardian/tests.rs

@@ -3449,7 +3449,8 @@ async fn guardian_review_routes_required_actions(
         | RequiredGuardianReview::LiveManagedModelWithGuardianV2 => {
             Arc::make_mut(&mut context.model_info).slug = "required-action-model".to_string();
             // The admitted turn has neither this model nor the new requirement.
-            let mut config = session.get_config().await.as_ref().clone();
+            let current_config = session.get_config().await;
+            let mut config = current_config.as_ref().clone();
             let requirements = codex_config::ConfigRequirements {
                 auto_review_required_models: Some(Sourced::new(
                     std::collections::BTreeSet::from([context.model_info.slug.clone()]),
@@ -3466,7 +3467,7 @@ async fn guardian_review_routes_required_actions(
                 requirements,
                 config.config_layer_stack.requirements_toml().clone(),
             )?;
-            session.refresh_mcp_config(config).await;
+            let _ = session.refresh_mcp_config(current_config, config).await;
             (
                 guardian_exec_command_request("shell-live-managed-model"),
                 ApprovalRequestReasons::default(),

codex-rs/core/src/lib.rs

@@ -47,6 +47,7 @@ pub use codex_protocol::protocol::EnvironmentConfig;
 pub use codex_thread::BackgroundTerminalInfo;
 pub use codex_thread::CodexThread;
 pub use codex_thread::CodexThreadSettingsOverrides;
+pub use codex_thread::ConfigRefreshOutcome;
 pub use codex_thread::GuardianAuthorizationVersion;
 pub use codex_thread::GuardianRootMessage;
 pub use codex_thread::GuardianRootSnapshot;

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

@@ -0,0 +1,147 @@
+//! Publishes resolved MCP configuration while preserving session-static settings.
+
+use std::sync::Arc;
+
+use crate::ConfigRefreshOutcome;
+use crate::config::Config;
+use crate::config::Constrained;
+use crate::config::RuntimeConfigRefresh;
+use crate::session::Session;
+use codex_config::McpServerAuth;
+use codex_features::Feature;
+use tracing::warn;
+
+impl Session {
+    pub(crate) async fn disable_mcp_enterprise_auth(&self) {
+        let mut state = self.state.lock().await;
+        let mut config = state
+            .session_configuration
+            .original_config_do_not_use
+            .as_ref()
+            .clone();
+        config.disable_mcp_enterprise_auth();
+        state.session_configuration.original_config_do_not_use = Arc::new(config);
+        self.services.mcp_runtime.invalidate_resource_caches();
+        self.mark_mcp_runtime_dirty();
+        drop(state);
+        self.schedule_mcp_prewarm();
+    }
+
+    pub(crate) async fn refresh_runtime_config(
+        &self,
+        expected_config: Arc<Config>,
+        next_config: Config,
+    ) -> ConfigRefreshOutcome {
+        self.refresh_config(expected_config, next_config, RuntimeConfigRefresh::User)
+            .await
+    }
+
+    pub(crate) async fn refresh_mcp_config(
+        &self,
+        expected_config: Arc<Config>,
+        next_config: Config,
+    ) -> ConfigRefreshOutcome {
+        self.refresh_config(expected_config, next_config, RuntimeConfigRefresh::Mcp)
+            .await
+    }
+
+    pub(super) async fn refresh_config(
+        &self,
+        expected_config: Arc<Config>,
+        next_config: Config,
+        scope: RuntimeConfigRefresh,
+    ) -> ConfigRefreshOutcome {
+        let (mut config, outcome, rejection_kind) =
+            match expected_config.resolve_runtime_refresh(&next_config, scope) {
+                Ok(config) => (config, ConfigRefreshOutcome::Published, None),
+                Err(error) => {
+                    let mut config = expected_config.as_ref().clone();
+                    if matches!(scope, RuntimeConfigRefresh::Mcp) {
+                        // The host-resolved map already carries fresh requirements and
+                        // preserved session overrides. Keep ordinary-server revocations even
+                        // when retained layers make the enterprise merge invalid.
+                        let mut refreshed_servers = next_config.mcp_servers.clone();
+                        let mut servers = refreshed_servers.get().clone();
+                        // Retain enterprise identities so a later corrected refresh cannot
+                        // admit an ordinary-auth replacement under a retired server's name.
+                        for (name, server) in expected_config.mcp_servers.get() {
+                            if matches!(server.auth, McpServerAuth::EmaAuth)
+                                || server.ema_registration().is_some()
+                            {
+                                let mut retired = server.clone();
+                                retired.enabled = false;
+                                servers.insert(name.clone(), retired);
+                            }
+                        }
+                        config.mcp_servers = if refreshed_servers.set(servers).is_ok() {
+                            refreshed_servers
+                        } else {
+                            Constrained::allow_any(Default::default())
+                        };
+                    }
+                    config.config_layer_stack = config
+                        .config_layer_stack
+                        .with_mcp_requirements_from(&next_config.config_layer_stack);
+                    if let Err(err) = config.features.refresh_mcp_features(&next_config.features) {
+                        warn!(%err, "failed to apply managed MCP feature restrictions");
+                        config.mcp_servers = Constrained::allow_any(Default::default());
+                    }
+                    config.disable_mcp_enterprise_auth();
+                    (config, ConfigRefreshOutcome::Rejected, Some(error.kind()))
+                }
+            };
+        for (feature, description) in [
+            (Feature::Mcp20260728, "MCP protocol"),
+            (Feature::CodexAppsMcp20260728, "Codex Apps MCP protocol"),
+        ] {
+            if let Err(err) = config
+                .features
+                .set_enabled(feature, next_config.features.enabled(feature))
+            {
+                warn!("failed to refresh {description} config: {err}");
+            }
+        }
+        let config = Arc::new(config);
+        let notify_contributors = !matches!(scope, RuntimeConfigRefresh::Mcp)
+            && !self.services.extensions.config_contributors().is_empty();
+        let (previous_config, new_config) = {
+            let mut state = self.state.lock().await;
+            // Loaded snapshots can publish only while their captured owner is current.
+            if !Arc::ptr_eq(
+                &state.session_configuration.original_config_do_not_use,
+                &expected_config,
+            ) {
+                return ConfigRefreshOutcome::Stale;
+            }
+            if let Some(rejection_kind) = rejection_kind {
+                warn!(
+                    ?rejection_kind,
+                    "disabling enterprise MCP after rejected configuration refresh"
+                );
+            }
+            // A host refresh carries current rollout settings for recording. A
+            // legacy file reload uses its original snapshot and bypasses this path.
+            if matches!(scope, RuntimeConfigRefresh::User) {
+                self.services
+                    .executed_tool_calls
+                    .refresh(&next_config.features);
+            }
+            let previous_config = notify_contributors
+                .then(|| self.build_effective_session_config(&state.session_configuration));
+            state.session_configuration.original_config_do_not_use = Arc::clone(&config);
+            if matches!(scope, RuntimeConfigRefresh::Mcp) {
+                self.services.mcp_runtime.invalidate_resource_caches();
+            }
+            self.mark_mcp_runtime_dirty();
+            let new_config = notify_contributors
+                .then(|| self.build_effective_session_config(&state.session_configuration));
+            (previous_config, new_config)
+        };
+        self.emit_config_changed_contributors(previous_config.as_ref(), new_config.as_ref());
+        self.schedule_mcp_prewarm();
+        if !matches!(scope, RuntimeConfigRefresh::Mcp) {
+            self.refresh_hooks(config).await;
+        }
+        outcome
+    }
+}

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

@@ -238,7 +238,7 @@ pub fn refresh_mcp_servers(sess: &Session) {
 }
 
 pub async fn reload_user_config(sess: &Arc<Session>) {
-    sess.reload_user_config_layer().await;
+    Box::pin(sess.reload_user_config_layer()).await;
 }
 
 pub async fn compact(sess: &Arc<Session>, sub_id: String) {

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

@@ -228,6 +228,7 @@ use codex_protocol::error::Result as CodexResult;
 use codex_protocol::exec_output::StreamOutput;
 
 mod code_mode_warning;
+mod config_refresh;
 pub(crate) mod context_window;
 mod daemon_recovery;
 mod environment;
@@ -2046,18 +2047,25 @@ impl Session {
         state.session_configuration.provider.info().clone()
     }
 
-    pub(crate) async fn refresh_runtime_config(&self, next_config: Config) {
-        self.refresh_runtime_config_inner(next_config, /*refresh_recording*/ true)
-            .await;
-    }
-
-    async fn refresh_runtime_config_inner(&self, next_config: Config, refresh_recording: bool) {
+    async fn refresh_file_config(
+        &self,
+        expected_config: Arc<Config>,
+        next_config: Config,
+    ) -> crate::ConfigRefreshOutcome {
         // Refresh only the user layer from the incoming snapshot. Preserve thread-local
         // layers such as request/session overrides that were present when this session
         // was created.
         let notify_config_contributors = !self.services.extensions.config_contributors().is_empty();
         let (previous_config, new_config, config) = {
             let mut state = self.state.lock().await;
+            if !Arc::ptr_eq(
+                &state.session_configuration.original_config_do_not_use,
+                &expected_config,
+            ) {
+                return crate::ConfigRefreshOutcome::Stale;
+            }
+            self.services.skills_service.clear_cache();
+            self.services.plugins_manager.clear_cache();
             let previous_config = notify_config_contributors
                 .then(|| self.build_effective_session_config(&state.session_configuration));
             let mut config = (*state.session_configuration.original_config_do_not_use).clone();
@@ -2070,13 +2078,6 @@ impl Session {
             config.mcp_servers = next_config.mcp_servers.clone();
             config.mcp_optional_startup_grace = next_config.mcp_optional_startup_grace;
             config.mcp_oauth_credentials_store_mode = next_config.mcp_oauth_credentials_store_mode;
-            // Recording can follow rollout changes without changing the session's
-            // execution features (including Code Mode's dispatch gate).
-            if refresh_recording {
-                self.services
-                    .executed_tool_calls
-                    .refresh(&next_config.features);
-            }
             if let Err(err) = config.features.set_enabled(
                 Feature::Mcp20260728,
                 next_config.features.enabled(Feature::Mcp20260728),
@@ -2113,6 +2114,7 @@ impl Session {
         self.emit_config_changed_contributors(previous_config.as_ref(), new_config.as_ref());
         self.schedule_mcp_prewarm();
         self.refresh_hooks(config).await;
+        crate::ConfigRefreshOutcome::Published
     }
 
     pub(crate) async fn refresh_hooks(&self, config: Arc<Config>) {
@@ -2139,48 +2141,6 @@ impl Session {
         }
     }
 
-    pub(crate) async fn refresh_mcp_config(&self, next_config: Config) {
-        let mut state = self.state.lock().await;
-        let mut config = (*state.session_configuration.original_config_do_not_use).clone();
-        config.config_layer_stack = next_config
-            .config_layer_stack
-            .with_user_layer_from(&config.config_layer_stack);
-        config.mcp_servers = next_config.mcp_servers;
-        config.mcp_optional_startup_grace = next_config.mcp_optional_startup_grace;
-        config.mcp_oauth_credentials_store_mode = next_config.mcp_oauth_credentials_store_mode;
-        if let Err(err) = config.features.set_enabled(
-            Feature::Mcp20260728,
-            next_config.features.enabled(Feature::Mcp20260728),
-        ) {
-            warn!("failed to refresh MCP protocol config: {err}");
-        }
-        if let Err(err) = config.features.set_enabled(
-            Feature::CodexAppsMcp20260728,
-            next_config.features.enabled(Feature::CodexAppsMcp20260728),
-        ) {
-            warn!("failed to refresh Codex Apps MCP protocol config: {err}");
-        }
-        if let Err(err) = config.features.set_enabled(
-            Feature::SecretAuthStorage,
-            next_config.features.enabled(Feature::SecretAuthStorage),
-        ) {
-            warn!("failed to refresh MCP auth storage config: {err}");
-        }
-        if let Err(err) = config.features.set_enabled(
-            Feature::McpOAuthRefreshCoordination,
-            next_config
-                .features
-                .enabled(Feature::McpOAuthRefreshCoordination),
-        ) {
-            warn!("failed to refresh MCP OAuth coordination config: {err}");
-        }
-        state.session_configuration.original_config_do_not_use = Arc::new(config);
-        self.services.mcp_runtime.invalidate_resource_caches();
-        self.mark_mcp_runtime_dirty();
-        drop(state);
-        self.schedule_mcp_prewarm();
-    }
-
     fn emit_config_changed_contributors(
         &self,
         previous_config: Option<&Config>,
@@ -2209,55 +2169,55 @@ impl Session {
         //
         // Prefer `refresh_runtime_config()` when the host can already provide a materialized
         // config snapshot. This file-based path exists for legacy local reload flows.
-        let config_toml_paths = {
-            let state = self.state.lock().await;
-            let config = &state.session_configuration.original_config_do_not_use;
-            let user_config_paths = config
-                .config_layer_stack
-                .all_layers_low_to_high()
-                .filter_map(|layer| match &layer.name {
-                    ConfigLayerSource::User { file, .. } => Some(file.clone()),
-                    _ => None,
-                })
-                .collect::<Vec<_>>();
-            if user_config_paths.is_empty() {
-                vec![
-                    state
-                        .session_configuration
-                        .codex_home
-                        .join(CONFIG_TOML_FILE),
-                ]
-            } else {
-                user_config_paths
-            }
-        };
+        loop {
+            let (expected_config, config_toml_paths) = {
+                let state = self.state.lock().await;
+                let config = Arc::clone(&state.session_configuration.original_config_do_not_use);
+                let user_config_paths = config
+                    .config_layer_stack
+                    .all_layers_low_to_high()
+                    .filter_map(|layer| match &layer.name {
+                        ConfigLayerSource::User { file, .. } => Some(file.clone()),
+                        _ => None,
+                    })
+                    .collect::<Vec<_>>();
+                let paths = if user_config_paths.is_empty() {
+                    vec![
+                        state
+                            .session_configuration
+                            .codex_home
+                            .join(CONFIG_TOML_FILE),
+                    ]
+                } else {
+                    user_config_paths
+                };
+                (config, paths)
+            };
 
-        let mut reloaded_user_configs = Vec::with_capacity(config_toml_paths.len());
-        for config_toml_path in config_toml_paths {
-            let user_config = match std::fs::read_to_string(&config_toml_path) {
-                Ok(contents) => match toml::from_str::<toml::Value>(&contents) {
-                    Ok(config) => config,
+            let mut reloaded_user_configs = Vec::with_capacity(config_toml_paths.len());
+            for config_toml_path in config_toml_paths {
+                let user_config = match std::fs::read_to_string(&config_toml_path) {
+                    Ok(contents) => match toml::from_str::<toml::Value>(&contents) {
+                        Ok(config) => config,
+                        Err(err) => {
+                            warn!("failed to parse user config while reloading layer: {err}");
+                            return;
+                        }
+                    },
+                    Err(err) if err.kind() == std::io::ErrorKind::NotFound => {
+                        toml::Value::Table(Default::default())
+                    }
                     Err(err) => {
-                        warn!("failed to parse user config while reloading layer: {err}");
+                        warn!("failed to read user config while reloading layer: {err}");
                         return;
                     }
-                },
-                Err(err) if err.kind() == std::io::ErrorKind::NotFound => {
-                    toml::Value::Table(Default::default())
-                }
-                Err(err) => {
-                    warn!("failed to read user config while reloading layer: {err}");
-                    return;
-                }
-            };
-            reloaded_user_configs.push((config_toml_path, user_config));
-        }
+                };
+                reloaded_user_configs.push((config_toml_path, user_config));
+            }
 
-        let next_config = {
-            let state = self.state.lock().await;
-            let mut config = (*state.session_configuration.original_config_do_not_use).clone();
+            let mut next_config = expected_config.as_ref().clone();
             for (config_toml_path, user_config) in reloaded_user_configs {
-                let config_layer_stack = match config
+                let config_layer_stack = match next_config
                     .config_layer_stack
                     .with_user_config(&config_toml_path, user_config)
                 {
@@ -2267,18 +2227,16 @@ impl Session {
                         return;
                     }
                 };
-                config.config_layer_stack = config_layer_stack;
+                next_config.config_layer_stack = config_layer_stack;
             }
-            config.tool_suggest =
-                resolve_tool_suggest_config_from_layer_stack(&config.config_layer_stack);
-            config
-        };
-        self.services.skills_service.clear_cache();
-        self.services.plugins_manager.clear_cache();
-        // This legacy snapshot still has the original execution features, not
-        // the host's latest rollout settings. Leave the live recorder alone.
-        self.refresh_runtime_config_inner(next_config, /*refresh_recording*/ false)
-            .await;
+            next_config.tool_suggest =
+                resolve_tool_suggest_config_from_layer_stack(&next_config.config_layer_stack);
+            if self.refresh_file_config(expected_config, next_config).await
+                != crate::ConfigRefreshOutcome::Stale
+            {
+                return;
+            }
+        }
     }
 
     /// Record a terminal CodexErr before the app-server completion notification is reduced.

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

@@ -16,6 +16,7 @@ use crate::agents_md_manager::SessionInstructions;
 use crate::compact::InitialContextInjection;
 use crate::config::ConfigBuilder;
 use crate::config::ConfigOverrides;
+use crate::config::RuntimeConfigRefresh;
 use crate::context::ContextualUserFragment;
 use crate::context::DeveloperInstructions;
 use crate::context::TurnAborted;
@@ -2028,8 +2029,11 @@ async fn refresh_runtime_config_refreshes_hooks() -> anyhow::Result<()> {
     };
     assert!(session.hooks().preview_session_start(&request).is_empty());
 
+    let current_config = session.get_config().await;
     let next_config = load_latest_config_for_session(&session).await;
-    session.refresh_runtime_config(next_config).await;
+    let _ = session
+        .refresh_runtime_config(current_config, next_config)
+        .await;
 
     assert_eq!(session.hooks().preview_session_start(&request).len(), 1);
     Ok(())
@@ -2089,8 +2093,20 @@ disabled_tools = [
     let mut next_config = load_latest_config_for_session(&session).await;
     next_config.model = Some("gpt-5.4".to_string());
     next_config.notify = Some(vec!["echo".to_string()]);
+    let mcp_protocol_enabled = !original.features.enabled(Feature::Mcp20260728);
+    let apps_protocol_enabled = !original.features.enabled(Feature::CodexAppsMcp20260728);
+    next_config
+        .features
+        .set_enabled(Feature::Mcp20260728, mcp_protocol_enabled)
+        .expect("set MCP protocol rollout");
+    next_config
+        .features
+        .set_enabled(Feature::CodexAppsMcp20260728, apps_protocol_enabled)
+        .expect("set Codex Apps MCP protocol rollout");
 
-    session.refresh_runtime_config(next_config).await;
+    let _ = session
+        .refresh_runtime_config(Arc::clone(&original), next_config)
+        .await;
 
     let config = session.get_config().await;
     let apps_toml = config
@@ -2111,6 +2127,14 @@ disabled_tools = [
     assert_eq!(app.destructive_enabled, Some(false));
     assert_eq!(config.model, original.model);
     assert_eq!(config.notify, original.notify);
+    assert_eq!(
+        config.features.enabled(Feature::Mcp20260728),
+        mcp_protocol_enabled
+    );
+    assert_eq!(
+        config.features.enabled(Feature::CodexAppsMcp20260728),
+        apps_protocol_enabled
+    );
     assert_eq!(
         config.tool_suggest.disabled_tools,
         vec![
@@ -2120,9 +2144,38 @@ disabled_tools = [
     );
 }
 
+#[test_case::test_case(false; "published")]
+#[test_case::test_case(true; "rejected")]
 #[tokio::test]
-async fn refresh_mcp_config_replaces_managed_server_and_plugin_requirements() {
+async fn refresh_mcp_config_replaces_managed_server_and_plugin_requirements(reject: bool) {
     let (session, _turn_context) = make_session_and_context().await;
+    let original = session.get_config().await;
+    if reject {
+        let mut config = original.as_ref().clone();
+        let mut layers: Vec<_> = config
+            .config_layer_stack
+            .all_layers_low_to_high()
+            .filter(|layer| !matches!(layer.name, codex_config::ConfigLayerSource::SessionFlags))
+            .cloned()
+            .collect();
+        layers.push(codex_config::ConfigLayerEntry::new(
+            codex_config::ConfigLayerSource::SessionFlags,
+            toml::from_str("features = 'invalid retained features'").unwrap(),
+        ));
+        config.config_layer_stack = ConfigLayerStack::new(
+            layers,
+            config.config_layer_stack.requirements().clone(),
+            config.config_layer_stack.requirements_toml().clone(),
+        )
+        .unwrap();
+        session
+            .state
+            .lock()
+            .await
+            .session_configuration
+            .original_config_do_not_use = Arc::new(config);
+    }
+    let original = session.get_config().await;
     let server = serde_json::from_value::<McpServerConfig>(json!({
         "url": "https://example.com/mcp",
         "enabled": true
@@ -2137,12 +2190,22 @@ async fn refresh_mcp_config_replaces_managed_server_and_plugin_requirements() {
         codex_config::PluginRequirementsToml {
             mcp_servers: Some(std::collections::BTreeMap::from([(
                 "beta".to_string(),
-                requirement,
+                requirement.clone(),
             )])),
         },
     )]);
 
-    let mut next_config = session.get_config().await.as_ref().clone();
+    let mut next_config = original.as_ref().clone();
+    let mcp_protocol_enabled = !original.features.enabled(Feature::Mcp20260728);
+    let apps_protocol_enabled = !original.features.enabled(Feature::CodexAppsMcp20260728);
+    next_config
+        .features
+        .set_enabled(Feature::Mcp20260728, mcp_protocol_enabled)
+        .expect("set MCP protocol rollout");
+    next_config
+        .features
+        .set_enabled(Feature::CodexAppsMcp20260728, apps_protocol_enabled)
+        .expect("set Codex Apps MCP protocol rollout");
     next_config.mcp_servers = codex_config::Constrained::normalized(
         HashMap::from([("beta".to_string(), server.clone())]),
         |mut servers: HashMap<String, McpServerConfig>| {
@@ -2152,11 +2215,29 @@ async fn refresh_mcp_config_replaces_managed_server_and_plugin_requirements() {
     )
     .expect("valid refreshed MCP constraints");
     let mut requirements = next_config.config_layer_stack.requirements().clone();
+    let mcp_requirements = BTreeMap::from([("beta".to_string(), requirement)]);
+    requirements.mcp_servers = Some(Sourced::new(
+        mcp_requirements.clone(),
+        RequirementSource::LegacyManagedConfigTomlFromMdm,
+    ));
     requirements.plugins = Some(Sourced::new(
         plugin_requirements.clone(),
         RequirementSource::LegacyManagedConfigTomlFromMdm,
     ));
+    let feature_requirements: codex_config::FeatureRequirementsToml =
+        serde_json::from_value(json!({"secret_auth_storage": false})).unwrap();
+    requirements.feature_requirements = Some(Sourced::new(
+        feature_requirements.clone(),
+        RequirementSource::LegacyManagedConfigTomlFromMdm,
+    ));
+    next_config.features = crate::config::ManagedFeatures::from_configured(
+        next_config.features.get().clone(),
+        requirements.feature_requirements.clone(),
+    )
+    .unwrap();
     let mut requirements_toml = next_config.config_layer_stack.requirements_toml().clone();
+    requirements_toml.feature_requirements = Some(feature_requirements);
+    requirements_toml.mcp_servers = Some(mcp_requirements);
     requirements_toml.plugins = Some(plugin_requirements.clone());
     let layers = next_config
         .config_layer_stack
@@ -2166,9 +2247,35 @@ async fn refresh_mcp_config_replaces_managed_server_and_plugin_requirements() {
     next_config.config_layer_stack = ConfigLayerStack::new(layers, requirements, requirements_toml)
         .expect("managed MCP and plugin requirements");
 
-    session.refresh_mcp_config(next_config).await;
+    let outcome = session
+        .refresh_mcp_config(Arc::clone(&original), next_config)
+        .await;
+    assert_eq!(
+        outcome,
+        if reject {
+            crate::ConfigRefreshOutcome::Rejected
+        } else {
+            crate::ConfigRefreshOutcome::Published
+        }
+    );
 
     let config = session.get_config().await;
+    assert!(!config.features.enabled(Feature::SecretAuthStorage));
+    assert!(
+        config
+            .config_layer_stack
+            .requirements()
+            .feature_requirements
+            .is_some()
+    );
+    assert_eq!(
+        config.features.enabled(Feature::Mcp20260728),
+        mcp_protocol_enabled
+    );
+    assert_eq!(
+        config.features.enabled(Feature::CodexAppsMcp20260728),
+        apps_protocol_enabled
+    );
     let mut managed_servers = config.mcp_servers.clone();
     managed_servers
         .set(HashMap::from([
@@ -2176,10 +2283,13 @@ async fn refresh_mcp_config_replaces_managed_server_and_plugin_requirements() {
             ("beta".to_string(), server.clone()),
         ]))
         .expect("apply refreshed managed MCP constraints");
-    assert_eq!(
-        managed_servers.get(),
-        &HashMap::from([("beta".to_string(), server.clone())])
+    assert!(
+        !managed_servers
+            .get()
+            .get("alpha")
+            .is_some_and(|server| server.enabled)
     );
+    assert_eq!(managed_servers.get()["beta"], server);
     assert_eq!(
         config
             .config_layer_stack
@@ -2197,6 +2307,73 @@ async fn refresh_mcp_config_replaces_managed_server_and_plugin_requirements() {
     config.apply_plugin_mcp_server_requirements("example-plugin", &mut plugin_servers);
     assert!(!plugin_servers["alpha"].enabled);
     assert!(plugin_servers["beta"].enabled);
+    let reloaded = config.resolve_runtime_refresh(&config, RuntimeConfigRefresh::User);
+    if let Ok(reloaded) = reloaded {
+        assert_eq!(
+            reloaded.config_layer_stack.requirements().plugins,
+            config.config_layer_stack.requirements().plugins
+        );
+    } else {
+        assert!(reject);
+    }
+}
+
+#[test_case::test_case(RuntimeConfigRefresh::User; "user refresh")]
+#[test_case::test_case(RuntimeConfigRefresh::Mcp; "MCP refresh")]
+#[tokio::test]
+async fn refresh_config_rejects_stale_owner_without_effects(scope: RuntimeConfigRefresh) {
+    let (session, _turn_context) = make_session_and_context().await;
+    let stale_owner = session.get_config().await;
+    let mut stale_config = stale_owner.as_ref().clone();
+    stale_config
+        .features
+        .set_enabled(
+            Feature::Mcp20260728,
+            !stale_owner.features.enabled(Feature::Mcp20260728),
+        )
+        .expect("set MCP protocol rollout");
+    assert_eq!(
+        session
+            .refresh_runtime_config(Arc::clone(&stale_owner), stale_owner.as_ref().clone())
+            .await,
+        crate::ConfigRefreshOutcome::Published
+    );
+    let current_owner = session.get_config().await;
+    assert!(session.mcp_refresh.claim());
+
+    assert_eq!(
+        session
+            .refresh_config(stale_owner, stale_config, scope)
+            .await,
+        crate::ConfigRefreshOutcome::Stale
+    );
+
+    assert!(Arc::ptr_eq(&current_owner, &session.get_config().await));
+    assert!(!session.mcp_refresh.is_pending());
+}
+
+#[tokio::test]
+async fn refresh_file_config_rejects_stale_owner_without_effects() {
+    let (session, _turn_context) = make_session_and_context().await;
+    let stale_owner = session.get_config().await;
+    assert_eq!(
+        session
+            .refresh_runtime_config(Arc::clone(&stale_owner), stale_owner.as_ref().clone())
+            .await,
+        crate::ConfigRefreshOutcome::Published
+    );
+    let current_owner = session.get_config().await;
+    assert!(session.mcp_refresh.claim());
+
+    assert_eq!(
+        session
+            .refresh_file_config(Arc::clone(&stale_owner), stale_owner.as_ref().clone())
+            .await,
+        crate::ConfigRefreshOutcome::Stale
+    );
+
+    assert!(Arc::ptr_eq(&current_owner, &session.get_config().await));
+    assert!(!session.mcp_refresh.is_pending());
 }
 
 #[test]
@@ -3771,8 +3948,11 @@ disabled_tools = [
 "#,
     )
     .expect("write user config");
+    let current_config = session.get_config().await;
     let next_config = load_latest_config_for_session(&session).await;
-    session.refresh_runtime_config(next_config).await;
+    let _ = session
+        .refresh_runtime_config(current_config, next_config)
+        .await;
 
     let expected_disabled_tools = vec![
         ToolSuggestDisabledTool::connector("calendar"),
@@ -13096,3 +13276,65 @@ async fn session_start_hooks_require_project_trust_without_config_toml() -> std:
 
     Ok(())
 }
+
+#[tokio::test]
+async fn rejected_mcp_refresh_then_corrected_user_config_blocks_ordinary_replacement() {
+    let (session, _turn_context) = make_session_and_context().await;
+    let base = session.get_config().await;
+    let enterprise: McpServerConfig = serde_json::from_value(json!({
+        "url": "https://enterprise.example/mcp", "auth": "ema_auth", "enabled": true
+    }))
+    .unwrap();
+    let ordinary: McpServerConfig = serde_json::from_value(json!({
+        "url": "https://ordinary.example/mcp", "enabled": true
+    }))
+    .unwrap();
+    let mut initial = base.as_ref().clone();
+    initial.mcp_servers = crate::config::Constrained::allow_any(HashMap::from([(
+        "enterprise".to_string(),
+        enterprise,
+    )]));
+    initial.config_layer_stack = ConfigLayerStack::new(
+        vec![codex_config::ConfigLayerEntry::new(
+            codex_config::ConfigLayerSource::User {
+                file: initial.codex_home.join(CONFIG_TOML_FILE),
+                profile: None,
+            },
+            toml::from_str("features = 'invalid retained features'").unwrap(),
+        )],
+        base.config_layer_stack.requirements().clone(),
+        base.config_layer_stack.requirements_toml().clone(),
+    )
+    .unwrap();
+    session
+        .state
+        .lock()
+        .await
+        .session_configuration
+        .original_config_do_not_use = Arc::new(initial);
+    let current = session.get_config().await;
+    let mut corrected = base.as_ref().clone();
+    corrected.mcp_servers = crate::config::Constrained::allow_any(HashMap::from([(
+        "enterprise".to_string(),
+        ordinary,
+    )]));
+    assert_eq!(
+        session.refresh_mcp_config(current, corrected.clone()).await,
+        crate::ConfigRefreshOutcome::Rejected
+    );
+    let rejected = session.get_config().await;
+    assert!(rejected.mcp_enterprise_managed_auth.is_none());
+    assert_eq!(
+        session.refresh_runtime_config(rejected, corrected).await,
+        crate::ConfigRefreshOutcome::Published
+    );
+    let final_config = session.get_config().await;
+    assert!(
+        !final_config
+            .mcp_servers
+            .get()
+            .get("enterprise")
+            .is_some_and(|server| server.enabled),
+        "rejection must not permit same-name ordinary-auth replacement in this session"
+    );
+}

codex-rs/core/src/stream_events_utils_tests.rs

@@ -361,9 +361,12 @@ async fn direct_results_keep_their_own_records_when_call_ids_repeat() {
     }
 
     // Reapplying an enabled config keeps the same recording lifetime.
-    session
-        .refresh_runtime_config((*session.get_config().await).clone())
-        .await;
+    let current_config = session.get_config().await;
+    let config = current_config.as_ref().clone();
+    assert_eq!(
+        session.refresh_runtime_config(current_config, config).await,
+        crate::ConfigRefreshOutcome::Published
+    );
     // Await in reverse order: each result must already own its record before history attachment.
     let mut outputs = Vec::new();
     for (arguments, future) in pending.into_iter().rev() {
@@ -413,12 +416,15 @@ async fn direct_results_keep_their_own_records_when_call_ids_repeat() {
                 .expect("dispatched direct call"),
         );
     }
-    let mut config = (*session.get_config().await).clone();
+    let current_config = session.get_config().await;
+    let mut config = current_config.as_ref().clone();
     config
         .features
         .disable(Feature::ExecutedToolCallMetadata)
         .expect("disable metadata");
-    session.refresh_runtime_config(config.clone()).await;
+    let _ = session
+        .refresh_runtime_config(current_config, config.clone())
+        .await;
     let result = pending
         .remove(0)
         .await
@@ -438,7 +444,8 @@ async fn direct_results_keep_their_own_records_when_call_ids_repeat() {
         .features
         .enable(Feature::ExecutedToolCallMetadata)
         .expect("re-enable metadata");
-    session.refresh_runtime_config(config).await;
+    let current_config = session.get_config().await;
+    let _ = session.refresh_runtime_config(current_config, config).await;
     let result = pending
         .pop()
         .expect("call prepared before disable")

codex-rs/core/src/tools/executed_tool_calls/request_metadata_tests.rs

@@ -426,6 +426,33 @@ fn direct_retained_metadata_budget_sheds_results_then_calls_across_refresh() {
     );
 }
 
+#[tokio::test]
+async fn stale_runtime_refresh_preserves_recording_state() {
+    let (session, _turn) = crate::session::tests::make_session_and_context().await;
+    let calls = session.services.executed_tool_calls.clone();
+    assert!(calls.lock_state().is_none());
+    let stale_owner = session.get_config().await;
+    let mut stale_config = stale_owner.as_ref().clone();
+    stale_config
+        .features
+        .enable(Feature::ExecutedToolCallMetadata)
+        .expect("enable executed tool call metadata");
+    assert_eq!(
+        session
+            .refresh_mcp_config(Arc::clone(&stale_owner), stale_owner.as_ref().clone())
+            .await,
+        crate::ConfigRefreshOutcome::Published
+    );
+
+    assert_eq!(
+        session
+            .refresh_runtime_config(stale_owner, stale_config)
+            .await,
+        crate::ConfigRefreshOutcome::Stale
+    );
+    assert!(calls.lock_state().is_none());
+}
+
 #[tokio::test]
 async fn recorder_refreshes_without_changing_execution_features_or_claiming_missing_history() {
     struct MetadataLookup<'a>(std::cell::Cell<usize>, JsonValue, &'a ExecutedToolCalls);
@@ -462,11 +489,14 @@ async fn recorder_refreshes_without_changing_execution_features_or_claiming_miss
     let mut retry_cache = HashMap::new();
     let mut prior_generation = None;
     for (index, enabled) in [false, true, false, true].into_iter().enumerate() {
+        let current_config = session.get_config().await;
         next_config
             .features
             .set_enabled(Feature::ExecutedToolCallMetadata, enabled)
             .expect("test feature must be configurable");
-        session.refresh_runtime_config(next_config.clone()).await;
+        let _ = session
+            .refresh_runtime_config(current_config, next_config.clone())
+            .await;
         assert_eq!(calls.lock_state().is_some(), enabled);
         // A legacy file reload has no new rollout snapshot and must keep this setting.
         session.reload_user_config_layer().await;
@@ -509,6 +539,21 @@ async fn recorder_refreshes_without_changing_execution_features_or_claiming_miss
         );
         assert_eq!(tool_calls_complete(&direct_output), enabled.then_some(true));
 
+        if enabled {
+            // Reapplying an enabled config must preserve already-recorded calls.
+            let prepared = calls.prepare_direct_call(&call, &ToolCallSource::Direct, &step);
+            let current_config = session.get_config().await;
+            assert_eq!(
+                session
+                    .refresh_runtime_config(current_config, next_config.clone())
+                    .await,
+                crate::ConfigRefreshOutcome::Published
+            );
+            let mut replay = output(&call.call_id);
+            calls.attach_direct_call_to_output(&mut replay, prepared);
+            assert_eq!(replay, direct_output);
+        }
+
         let cell = CellId::new(format!("cell-{index}"));
         let origin = format!("exec-{index}");
         calls.start_cell(&cell, &origin);
@@ -523,7 +568,13 @@ async fn recorder_refreshes_without_changing_execution_features_or_claiming_miss
         nested.set_tool_result_metadata(ToolResultMetadata::new(&json!({})));
         if enabled {
             // Reapplying an enabled config must preserve pending calls and their metadata.
-            session.refresh_runtime_config(next_config.clone()).await;
+            let current_config = session.get_config().await;
+            assert_eq!(
+                session
+                    .refresh_runtime_config(current_config, next_config.clone())
+                    .await,
+                crate::ConfigRefreshOutcome::Published
+            );
         }
         calls.finish_cell_recording(&cell);
         let mut prompt = vec![exec_input(&origin), exec_output(&origin)];

codex-rs/core/tests/suite/agents_md.rs

@@ -729,18 +729,26 @@ async fn runtime_trust_reload_refreshes_project_instructions() -> Result<()> {
         vec![global_agents.clone(), project_agents.clone()]
     );
 
-    let mut untrusted_config = (*test.codex.config().await).clone();
+    let current_config = test.codex.config().await;
+    let mut untrusted_config = current_config.as_ref().clone();
     untrusted_config.active_project.trust_level = Some(TrustLevel::Untrusted);
-    test.codex.refresh_runtime_config(untrusted_config).await;
+    let _ = test
+        .codex
+        .refresh_runtime_config(current_config, untrusted_config)
+        .await;
     test.submit_turn("untrusted project").await?;
     assert_eq!(
         test.codex.instruction_sources().await,
         vec![global_agents.clone()]
     );
 
-    let mut trusted_config = (*test.codex.config().await).clone();
+    let current_config = test.codex.config().await;
+    let mut trusted_config = current_config.as_ref().clone();
     trusted_config.active_project.trust_level = Some(TrustLevel::Trusted);
-    test.codex.refresh_runtime_config(trusted_config).await;
+    let _ = test
+        .codex
+        .refresh_runtime_config(current_config, trusted_config)
+        .await;
     test.submit_turn("trusted again").await?;
     assert_eq!(
         test.codex.instruction_sources().await,

codex-rs/core/tests/suite/code_mode.rs

@@ -2742,12 +2742,16 @@ async fn code_mode_result_metadata_follows_runtime_recording_enablement() -> Res
     );
     for (call_id, enabled) in [("call-off", false), ("call-on", true)] {
         if enabled {
+            let current_config = test.codex.config().await;
             let mut config = test.config.clone();
             config
                 .features
                 .enable(Feature::ExecutedToolCallMetadata)
                 .unwrap();
-            test.codex.refresh_runtime_config(config).await;
+            let _ = test
+                .codex
+                .refresh_runtime_config(current_config, config)
+                .await;
             // Runtime recording changes without updating the session's execution features.
             assert!(
                 !test
@@ -2910,7 +2914,11 @@ async fn code_mode_result_metadata_keeps_prepared_call_binding_across_runtime_re
             /*originator*/ None,
         )),
     };
-    test.codex.refresh_runtime_config(test.config.clone()).await;
+    let current_config = test.codex.config().await;
+    let _ = test
+        .codex
+        .refresh_runtime_config(current_config, test.config.clone())
+        .await;
     release_tx.send(()).unwrap();
     let wait = responses::mount_function_call_agent_response(
         &server,

codex-rs/core/tests/suite/direct_tool_metadata.rs

@@ -389,9 +389,13 @@ async fn direct_call_metadata_during_compaction_respects_provider_support(
         })
         .expect("source user message")["content"];
     if !metadata_enabled {
+        let current_config = test.codex.config().await;
         let mut config = test.config.clone();
         config.features.disable(Feature::ExecutedToolCallMetadata)?;
-        test.codex.refresh_runtime_config(config).await;
+        let _ = test
+            .codex
+            .refresh_runtime_config(current_config, config)
+            .await;
     }
     test.codex.submit(Op::Compact).await?;
     wait_for_event(&test.codex, |event| {

codex-rs/core/tests/suite/hooks_executor.rs

@@ -383,6 +383,7 @@ async fn executor_stop_hook_rejects_mismatched_environment() -> Result<()> {
         .await?;
     fixture.wait_for_hook_call().await?;
 
+    let current_config = fixture.test.codex.config().await;
     let mut mismatched_config = fixture.test.config.clone();
     let mut node_repl = mismatched_config
         .mcp_servers
@@ -396,10 +397,10 @@ async fn executor_stop_hook_rejects_mismatched_environment() -> Result<()> {
             .into_iter()
             .collect(),
     )?;
-    fixture
+    let _ = fixture
         .test
         .codex
-        .refresh_mcp_config(mismatched_config)
+        .refresh_mcp_config(current_config, mismatched_config)
         .await;
     fixture
         .test

codex-rs/core/tests/suite/mcp_ema_config.rs

@@ -6,6 +6,7 @@ use anyhow::Result;
 use codex_config::LoaderOverrides;
 use codex_config::McpServerAuth;
 use codex_config::test_support::CloudConfigBundleFixture;
+use codex_core::ConfigRefreshOutcome;
 use codex_core::config::ConfigBuilder;
 use codex_core::config::set_project_trust_level;
 use codex_login::CodexAuth;
@@ -16,27 +17,83 @@ use core_test_support::skip_if_no_network;
 use core_test_support::test_codex::test_codex;
 use core_test_support::wait_for_event_match;
 use pretty_assertions::assert_eq;
-use serde_json::json;
 use tempfile::tempdir;
 use test_case::test_case;
 use wiremock::MockServer;
 
+#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
+async fn refresh_ema_denial_clears_authority_for_configured_servers() -> Result<()> {
+    skip_if_no_network!(Ok(()));
+    let endpoint = MockServer::start().await;
+    let responses_server = responses::start_mock_server().await;
+    let home = Arc::new(tempdir()?);
+    let url = format!("{}/mcp", endpoint.uri());
+    let policy = |revision| {
+        format!(
+            "[features]\nuse_xaa = true\napps = false\nsecret_auth_storage = false\n\
+             [mcp_enterprise_managed_auth.idp]\nissuer = '{}/idp-{revision}'\nclient_id = 'client-{revision}'\n\
+             [mcp_servers.enterprise]\nurl = '{url}'\nauth = 'ema_auth'\noauth_resource = 'resource-{revision}'\n",
+            endpoint.uri(),
+        )
+    };
+    let fixture = test_codex()
+        .with_home(Arc::clone(&home))
+        .with_auth(CodexAuth::from_api_key("test-api-key"))
+        .with_cloud_config_bundle(CloudConfigBundleFixture::loader_with_enterprise_config(
+            policy(1),
+        ))
+        .build_with_auto_env(&responses_server)
+        .await?;
+    let (before, _) = fixture.codex.current_mcp_config_and_runtime_context().await;
+    let current_config = fixture.codex.config().await;
+    let refreshed = ConfigBuilder::default()
+        .loader_overrides(LoaderOverrides::without_managed_config_for_tests())
+        .codex_home(home.path().to_path_buf())
+        .fallback_cwd(Some(fixture.config.cwd.to_path_buf()))
+        .cloud_config_bundle(
+            CloudConfigBundleFixture::enterprise_config(policy(2))
+                .add_enterprise_requirement("[features]\nuse_xaa = false\n")
+                .into_loader(),
+        )
+        .build()
+        .await?;
+    assert_eq!(
+        fixture
+            .codex
+            .refresh_mcp_config(current_config, refreshed)
+            .await,
+        ConfigRefreshOutcome::Published
+    );
+    let (after, _) = fixture.codex.current_mcp_config_and_runtime_context().await;
+    let old_servers = codex_mcp::configured_mcp_servers(&before);
+    let new_servers = codex_mcp::configured_mcp_servers(&after);
+    let old = old_servers["enterprise"]
+        .ema_registration()
+        .expect("original registration");
+    assert!(old_servers["enterprise"].enabled);
+    assert!(
+        !new_servers["enterprise"].enabled,
+        "managed XAA opt-out must apply"
+    );
+    assert_eq!(old.resource(), Some("resource-1"));
+    assert!(new_servers["enterprise"].ema_registration().is_none());
+    fixture.codex.shutdown_and_wait().await?;
+    assert!(endpoint.received_requests().await.unwrap().is_empty());
+    Ok(())
+}
+
 #[derive(Clone, Copy)]
 enum ActivationScenario {
     FeatureDisabled,
     MissingIdp,
-    PluginSelfOptIn,
     OperatorEnabled,
 }
 
 #[test_case(ActivationScenario::FeatureDisabled; "feature disabled")]
 #[test_case(ActivationScenario::MissingIdp; "missing IdP")]
-#[test_case(ActivationScenario::PluginSelfOptIn; "plugin cannot select enterprise auth")]
 #[test_case(ActivationScenario::OperatorEnabled; "operator registration is eligible for startup")]
 #[tokio::test(flavor = "multi_thread", worker_threads = 2)]
-async fn enterprise_activation_respects_managed_config_and_plugin_ownership(
-    scenario: ActivationScenario,
-) -> Result<()> {
+async fn enterprise_activation_respects_configuration(scenario: ActivationScenario) -> Result<()> {
     skip_if_no_network!(Ok(()));
     let server = MockServer::start().await;
     let responses_server = responses::start_mock_server().await;
@@ -52,19 +109,9 @@ async fn enterprise_activation_respects_managed_config_and_plugin_ownership(
             "[mcp_enterprise_managed_auth.idp]\nissuer = {issuer:?}\nclient_id = \"idp-client\"\n"
         ));
     }
-    if matches!(scenario, ActivationScenario::PluginSelfOptIn) {
-        let plugin_root = super::plugins::write_sample_plugin_manifest_and_config(&home);
-        std::fs::write(
-            plugin_root.join(".mcp.json"),
-            serde_json::to_vec(&json!({"mcpServers": {"enterprise": {
-                "url": resource, "auth": "ema_auth", "oauth": {"client_id": "mcp-client"}
-            }}}))?,
-        )?;
-    } else {
-        managed_config.push_str(&format!(
-            "[mcp_servers.enterprise]\nurl = {resource:?}\nauth = \"ema_auth\"\n[mcp_servers.enterprise.oauth]\nclient_id = \"mcp-client\"\n"
-        ));
-    }
+    managed_config.push_str(&format!(
+        "[mcp_servers.enterprise]\nurl = {resource:?}\nauth = \"ema_auth\"\n[mcp_servers.enterprise.oauth]\nclient_id = \"mcp-client\"\n"
+    ));
     let fixture = test_codex()
         .with_home(home)
         .with_auth(CodexAuth::from_api_key("test-api-key"))
@@ -76,7 +123,6 @@ async fn enterprise_activation_respects_managed_config_and_plugin_ownership(
     let (runtime_config, _) = fixture.codex.current_mcp_config_and_runtime_context().await;
     let expected_enabled = match scenario {
         ActivationScenario::FeatureDisabled | ActivationScenario::MissingIdp => Some(false),
-        ActivationScenario::PluginSelfOptIn => None,
         ActivationScenario::OperatorEnabled => Some(true),
     };
     assert_eq!(

codex-rs/core/tests/suite/mcp_oauth_refresh_tests.rs

@@ -206,16 +206,23 @@ async fn oauth_mode_refresh_replaces_the_live_connection(
     assert_eq!(startup_control.initialize_attempts(), 1);
 
     for (enabled, expected_initializations) in [(true, 2), (true, 2), (false, 3)] {
+        let current_config = fixture.codex.config().await;
         let mut refreshed_config = fixture.config.clone();
         refreshed_config
             .features
             .set_enabled(Feature::McpOAuthRefreshCoordination, enabled)?;
         match refresh_path {
             ConfigRefreshPath::Runtime => {
-                fixture.codex.refresh_runtime_config(refreshed_config).await;
+                let _ = fixture
+                    .codex
+                    .refresh_runtime_config(current_config, refreshed_config)
+                    .await;
             }
             ConfigRefreshPath::Mcp => {
-                fixture.codex.refresh_mcp_config(refreshed_config).await;
+                let _ = fixture
+                    .codex
+                    .refresh_mcp_config(current_config, refreshed_config)
+                    .await;
             }
         }
         // A normal tool call reconciles the refreshed config without forcing a reconnect.

codex-rs/core/tests/suite/mcp_optional_startup_grace.rs

@@ -248,14 +248,21 @@ async fn running_thread_uses_refreshed_optional_mcp_startup_grace(
     );
 
     let refreshed_grace = Duration::from_millis(250);
+    let current_config = fixture.codex.config().await;
     let mut refreshed_config = fixture.config.clone();
     refreshed_config.mcp_optional_startup_grace = refreshed_grace;
     match refresh_path {
         ConfigRefreshPath::Runtime => {
-            fixture.codex.refresh_runtime_config(refreshed_config).await;
+            let _ = fixture
+                .codex
+                .refresh_runtime_config(current_config, refreshed_config)
+                .await;
         }
         ConfigRefreshPath::Mcp => {
-            fixture.codex.refresh_mcp_config(refreshed_config).await;
+            let _ = fixture
+                .codex
+                .refresh_mcp_config(current_config, refreshed_config)
+                .await;
         }
     }
     assert_eq!(

codex-rs/core/tests/suite/mcp_startup_refresh_http_proxy.rs

@@ -160,6 +160,7 @@ async fn local_mcp_startup_and_refresh_use_configured_http_client() -> Result<()
         .await?;
     wait_for_mcp_server(&fixture.codex, SERVER_NAME).await?;
 
+    let current_config = fixture.codex.config().await;
     let mut refreshed_config = fixture.config.clone();
     let mut servers = refreshed_config.mcp_servers.get().clone();
     let server = servers
@@ -177,7 +178,10 @@ async fn local_mcp_startup_and_refresh_use_configured_http_client() -> Result<()
         .mcp_servers
         .set(servers)
         .expect("test MCP servers should accept the refreshed configuration");
-    fixture.codex.refresh_runtime_config(refreshed_config).await;
+    let _ = fixture
+        .codex
+        .refresh_runtime_config(current_config, refreshed_config)
+        .await;
     let result = fixture
         .codex
         .call_mcp_tool(

codex-rs/core/tests/suite/mcp_tool_exposure.rs

@@ -414,8 +414,13 @@ async fn rapid_mcp_refreshes_coalesce_to_the_latest_config() -> Result<()> {
         .await?;
 
     contributor.block_next.store(true, Ordering::SeqCst);
-    test.codex
-        .refresh_runtime_config(config_with_mcp_marker(&test.config, "config-a"))
+    let current_config = test.codex.config().await;
+    let _ = test
+        .codex
+        .refresh_runtime_config(
+            current_config,
+            config_with_mcp_marker(&test.config, "config-a"),
+        )
         .await;
     tokio::time::timeout(Duration::from_secs(5), contributor.entered.acquire())
         .await
@@ -423,11 +428,21 @@ async fn rapid_mcp_refreshes_coalesce_to_the_latest_config() -> Result<()> {
         .expect("entered semaphore should remain open")
         .forget();
 
-    test.codex
-        .refresh_runtime_config(config_with_mcp_marker(&test.config, "config-b"))
+    let current_config = test.codex.config().await;
+    let _ = test
+        .codex
+        .refresh_runtime_config(
+            current_config,
+            config_with_mcp_marker(&test.config, "config-b"),
+        )
         .await;
-    test.codex
-        .refresh_runtime_config(config_with_mcp_marker(&test.config, "config-c"))
+    let current_config = test.codex.config().await;
+    let _ = test
+        .codex
+        .refresh_runtime_config(
+            current_config,
+            config_with_mcp_marker(&test.config, "config-c"),
+        )
         .await;
     contributor.release.add_permits(1);
 
@@ -647,13 +662,17 @@ async fn timeout_refresh_replaces_pending_startup_and_reuses_ready_connection()
         )
         .await?;
 
+    let current_config = test.codex.config().await;
     let mut refresh_config = test.config.clone();
     let mut servers = refresh_config.mcp_servers.get().clone();
     for config in servers.values_mut() {
         config.startup_timeout_sec = None;
     }
     refresh_config.mcp_servers.set(servers)?;
-    test.codex.refresh_mcp_config(refresh_config).await;
+    let _ = test
+        .codex
+        .refresh_mcp_config(current_config, refresh_config)
+        .await;
     // Publish without waiting for the held initialize to finish.
     let error = test
         .codex
@@ -861,6 +880,7 @@ async fn out_of_band_resource_read_reconciles_the_published_mcp_runtime() -> Res
         .expect("thread start should capture the MCP resource client");
     assert!(!resource_client.has_server("refreshed").await);
 
+    let current_config = test.codex.config().await;
     let mut refresh_config = test.config.clone();
     let user_config_path = refresh_config.codex_home.join("config.toml");
     let user_config: toml::Value = toml::from_str(&format!(
@@ -884,7 +904,10 @@ startup_timeout_sec = 0.1
     refresh_config.config_layer_stack = refresh_config
         .config_layer_stack
         .with_user_config(&user_config_path, user_config)?;
-    test.codex.refresh_runtime_config(refresh_config).await;
+    let _ = test
+        .codex
+        .refresh_runtime_config(current_config, refresh_config)
+        .await;
     test.codex.submit(Op::RefreshMcpServers).await?;
 
     let _ = test
@@ -1155,6 +1178,7 @@ async fn deferred_tool_world_state_tracks_initial_unchanged_and_removed_namespac
         "equivalent bindings must preserve MCP handlers and reuse the search index"
     );
 
+    let current_config = test.codex.config().await;
     let mut refresh_config = test.config.clone();
     let user_config_path = refresh_config.codex_home.join("config.toml");
     let user_config = toml::from_str(
@@ -1166,7 +1190,10 @@ enabled = false
     refresh_config.config_layer_stack = refresh_config
         .config_layer_stack
         .with_user_config(&user_config_path, user_config)?;
-    test.codex.refresh_runtime_config(refresh_config).await;
+    let _ = test
+        .codex
+        .refresh_runtime_config(current_config, refresh_config)
+        .await;
     test.codex.submit(Op::RefreshMcpServers).await?;
     test.submit_turn("inspect removed deferred tools").await?;
 

codex-rs/core/tests/suite/request_plugin_install.rs

@@ -1427,7 +1427,11 @@ async fn run_remote_plugin_install_refresh_case(refreshed_tools: RefreshedAppsTo
         "the refreshed installed-plugin cache should filter the cached recommendation"
     );
     drop(requests);
-    test.codex.refresh_runtime_config(test.config.clone()).await;
+    let current_config = test.codex.config().await;
+    let _ = test
+        .codex
+        .refresh_runtime_config(current_config, test.config.clone())
+        .await;
     test.submit_turn("check whether Calendar is still installed")
         .await?;
     let requests = mock.requests();

codex-rs/core/tests/suite/rmcp_client.rs

@@ -4386,6 +4386,7 @@ async fn streamable_http_with_oauth_round_trip_impl() -> anyhow::Result<()> {
         Some("OAuth sign-in is still pending."),
     );
 
+    let current_config = fixture.codex.config().await;
     let mut refreshed_config = fixture.config.clone();
     let mut refreshed_servers = refreshed_config.mcp_servers.get().clone();
     let discovered_server = refreshed_servers
@@ -4398,9 +4399,9 @@ async fn streamable_http_with_oauth_round_trip_impl() -> anyhow::Result<()> {
         .set(refreshed_servers)
         .expect("test MCP servers should accept the discovered OAuth server");
     let discovered_turn = tokio::time::timeout(Duration::from_secs(5), async {
-        fixture
+        let _ = fixture
             .codex
-            .refresh_runtime_config(refreshed_config.clone())
+            .refresh_runtime_config(current_config, refreshed_config.clone())
             .await;
         fixture
             .codex
@@ -4444,7 +4445,11 @@ async fn streamable_http_with_oauth_round_trip_impl() -> anyhow::Result<()> {
         )
         .await?
     );
-    fixture.codex.refresh_runtime_config(refreshed_config).await;
+    let current_config = fixture.codex.config().await;
+    let _ = fixture
+        .codex
+        .refresh_runtime_config(current_config, refreshed_config)
+        .await;
     let logged_out_startup = tokio::time::timeout(
         Duration::from_secs(5),
         wait_for_event(&fixture.codex, |event| {

codex-rs/core/tests/suite/scenarios_guardian_conversation_history_tests.rs

@@ -194,12 +194,18 @@ async fn guardian_conversation_history(scenario: HistoryScenario) -> anyhow::Res
                 "requires approval on the parent",
             ),
         ] {
+            let current_config = test.codex.config().await;
             let mut config = test.config.clone();
             config.config_layer_stack = config.config_layer_stack.with_user_config(
                 &config.codex_home.join("config.toml"),
                 toml::from_str(policy)?,
             )?;
-            test.codex.refresh_runtime_config(config).await;
+            assert_eq!(
+                test.codex
+                    .refresh_runtime_config(current_config, config)
+                    .await,
+                codex_core::ConfigRefreshOutcome::Published
+            );
             let followup =
                 responses::mount_sse_sequence(&server, review_responses(label, &actions[1..]))
                     .await;

codex-rs/tui/src/app/agents_overview_tests.rs

@@ -1694,7 +1694,7 @@ async fn root_switch_preserves_idle_root_with_running_subagent() -> Result<()> {
 
 #[tokio::test]
 async fn overview_selection_applies_user_permissions_only_to_unloaded_threads() -> Result<()> {
-    let mut app = make_test_app().await;
+    let mut app = Box::pin(make_test_app()).await;
     trust_fixture_folders(&mut app);
     std::fs::write(
         app.config.codex_home.join("config.toml"),
@@ -1724,15 +1724,14 @@ async fn overview_selection_applies_user_permissions_only_to_unloaded_threads()
     }
     let mut app_server =
         Box::pin(crate::start_embedded_app_server_for_picker(&server_config)).await?;
-    let loaded = app_server
-        .resume_thread(
-            &crate::local_settings::LocalSettings::from(&server_config),
-            server_config.clone(),
-            thread_ids[0],
-            crate::app_server_session::ResumeModelSettings::RestoreFromThread,
-        )
-        .await?
-        .session;
+    let loaded = Box::pin(app_server.resume_thread(
+        &crate::local_settings::LocalSettings::from(&server_config),
+        server_config.clone(),
+        thread_ids[0],
+        crate::app_server_session::ResumeModelSettings::RestoreFromThread,
+    ))
+    .await?
+    .session;
     app.runtime_approvals_reviewer_override = Some(ApprovalsReviewer::AutoReview);
     app.harness_overrides.sandbox_mode = Some(codex_protocol::config_types::SandboxMode::ReadOnly);
     app.harness_overrides.approval_policy =
@@ -1785,22 +1784,20 @@ async fn overview_selection_applies_user_permissions_only_to_unloaded_threads()
                 AskForApproval::UnlessTrusted,
             ));
         }
-        app.select_agents_overview_thread(&mut tui, &mut app_server, thread_id)
-            .await?;
+        Box::pin(app.select_agents_overview_thread(&mut tui, &mut app_server, thread_id)).await?;
         assert_eq!(
             app.runtime_approvals_reviewer_override,
             (thread_id == thread_ids[2] || thread_id == thread_ids[3])
                 .then_some(ApprovalsReviewer::User)
         );
-        let observed = app_server
-            .resume_thread(
-                &app.local_settings,
-                app.config.clone(),
-                thread_id,
-                crate::app_server_session::ResumeModelSettings::PreserveExistingThread,
-            )
-            .await?
-            .session;
+        let observed = Box::pin(app_server.resume_thread(
+            &app.local_settings,
+            app.config.clone(),
+            thread_id,
+            crate::app_server_session::ResumeModelSettings::PreserveExistingThread,
+        ))
+        .await?
+        .session;
         assert_eq!(
             (
                 app.primary_thread_id,
@@ -1824,8 +1821,7 @@ async fn overview_selection_applies_user_permissions_only_to_unloaded_threads()
     app.loader_overrides.system_requirements_path = Some(requirements.to_path_buf());
     let view = app.agents_overview_view(Vec::new(), /*selected_thread_id*/ None);
     app.chat_widget.show_bottom_pane_view(Box::new(view));
-    app.select_agents_overview_thread(&mut tui, &mut app_server, thread_ids[4])
-        .await?;
+    Box::pin(app.select_agents_overview_thread(&mut tui, &mut app_server, thread_ids[4])).await?;
     assert_eq!(
         (
             app.primary_thread_id,
@@ -2243,7 +2239,7 @@ async fn resume_failure_keeps_command_center_available() {
 
 #[tokio::test]
 async fn resume_picker_round_trip_preserves_each_threads_input() -> Result<()> {
-    let mut app = make_test_app().await;
+    let mut app = Box::pin(make_test_app()).await;
     trust_fixture_folders(&mut app);
     std::fs::write(
         app.config.codex_home.join("config.toml"),
@@ -2280,11 +2276,11 @@ async fn resume_picker_round_trip_preserves_each_threads_input() -> Result<()> {
     for target in targets.iter().chain(&targets) {
         let view = app.agents_overview_view(Vec::new(), /*selected_thread_id*/ None);
         app.chat_widget.show_bottom_pane_view(Box::new(view));
-        app.apply_resume_picker_selection(
+        Box::pin(app.apply_resume_picker_selection(
             &mut tui,
             &mut server,
             SessionSelection::Resume(target.clone()),
-        )
+        ))
         .await?;
         assert_eq!(app.chat_widget.thread_id(), Some(target.thread_id));
         if expected_states.len() < targets.len() {
@@ -2342,7 +2338,7 @@ async fn resume_picker_round_trip_preserves_each_threads_input() -> Result<()> {
 
 #[tokio::test]
 async fn command_center_handles_resume_failure_and_success() -> Result<()> {
-    let mut app = make_test_app().await;
+    let mut app = Box::pin(make_test_app()).await;
     trust_fixture_folders(&mut app);
     std::fs::write(
         app.config.codex_home.join("config.toml"),
@@ -2365,7 +2361,7 @@ async fn command_center_handles_resume_failure_and_success() -> Result<()> {
     let mut tui = crate::tui::test_support::make_test_tui()?;
     let mut server = Box::pin(crate::start_embedded_app_server_for_picker(&app.config)).await?;
     assert!(matches!(
-        app.apply_resume_picker_selection(
+        Box::pin(app.apply_resume_picker_selection(
             &mut tui,
             &mut server,
             SessionSelection::Resume(SessionTarget {
@@ -2374,7 +2370,7 @@ async fn command_center_handles_resume_failure_and_success() -> Result<()> {
                 cwd: None,
                 history_mode: None,
             })
-        )
+        ))
         .await?,
         AppRunControl::Continue
     ));
@@ -2389,7 +2385,7 @@ async fn command_center_handles_resume_failure_and_success() -> Result<()> {
     app.chat_widget
         .handle_key_event(KeyEvent::new(KeyCode::Esc, KeyModifiers::NONE));
     assert!(matches!(
-        app.apply_resume_picker_selection(
+        Box::pin(app.apply_resume_picker_selection(
             &mut tui,
             &mut server,
             SessionSelection::Resume(SessionTarget {
@@ -2398,7 +2394,7 @@ async fn command_center_handles_resume_failure_and_success() -> Result<()> {
                 cwd: None,
                 history_mode: None,
             })
-        )
+        ))
         .await?,
         AppRunControl::Continue
     ));

codex-rs/tui/src/app/tests.rs

@@ -8175,7 +8175,7 @@ async fn remembered_current_cwd_stays_at_launch_across_in_app_resumes() -> Resul
         Arc::new(EnvironmentManager::default_for_tests()),
     ))
     .await?;
-    let (mut app, _app_event_rx, _op_rx) = make_test_app_with_channels().await;
+    let (mut app, _app_event_rx, _op_rx) = Box::pin(make_test_app_with_channels()).await;
     app.config = config;
     app.launch_cwd = launch_cwd.clone();
     app.state_db = state_db;

codex-rs/tui/src/app/tests/session_lifecycle_requests.rs

@@ -1078,7 +1078,7 @@ async fn archive_current_thread_returns_shared_servers_to_agents() -> Result<()>
 
 #[tokio::test]
 async fn local_daemon_registers_approval_gated_mcp_tools_for_both_start_paths() -> Result<()> {
-    let (mut app, events, _ops) = make_test_app_with_channels().await;
+    let (mut app, events, _ops) = Box::pin(make_test_app_with_channels()).await;
     let codex_home = tempdir()?;
     app.config.codex_home = codex_home.path().to_path_buf().abs();
     app.config.sqlite = SqliteConfig::new_for_testing(codex_home.path().abs());
@@ -1096,15 +1096,14 @@ async fn local_daemon_registers_approval_gated_mcp_tools_for_both_start_paths()
         /*failed_thread_name*/ None,
     ))
     .await?;
-    app_server
-        .start_dynamic_tool_mcp(
-            app.config.clone(),
-            app.app_event_tx.clone(),
-            app.dynamic_tool_status_updates.clone(),
-        )
-        .await?;
+    Box::pin(app_server.start_dynamic_tool_mcp(
+        app.config.clone(),
+        app.app_event_tx.clone(),
+        app.dynamic_tool_status_updates.clone(),
+    ))
+    .await?;
 
-    let started = app_server.start_thread(&app.config).await?;
+    let started = Box::pin(app_server.start_thread(&app.config)).await?;
     let thread_id = started.session.thread_id;
     assert!(started.task_tools_available);
     assert!(app_server.task_tools_available(thread_id));
@@ -2532,7 +2531,10 @@ async fn assert_remote_legacy_history_retry(request: LegacyHistoryRequest) -> Re
 
 #[tokio::test]
 async fn remote_legacy_history_resume_retries_generic_method_not_found() -> Result<()> {
-    assert_remote_legacy_history_retry(LegacyHistoryRequest::Resume).await
+    Box::pin(assert_remote_legacy_history_retry(
+        LegacyHistoryRequest::Resume,
+    ))
+    .await
 }
 
 #[tokio::test]
@@ -2542,7 +2544,7 @@ async fn remote_legacy_history_fork_avoids_unsupported_fields() -> Result<()> {
 
 #[tokio::test]
 async fn paginated_fork_survives_post_response_hydration_failure() -> Result<()> {
-    let (app, _codex_home) = make_history_test_app().await?;
+    let (app, _codex_home) = Box::pin(make_history_test_app()).await?;
     let parent_thread_id = create_history_rollout(
         &app.config,
         ThreadHistoryMode::Paginated,
@@ -2558,19 +2560,18 @@ async fn paginated_fork_survives_post_response_hydration_failure() -> Result<()>
     )
     .await?;
 
-    let started = app_server
-        .resume_thread(
-            &app.local_settings,
-            app.config.clone(),
-            parent_thread_id,
-            crate::app_server_session::ResumeModelSettings::RestoreFromThread,
-        )
-        .await?;
+    let started = Box::pin(app_server.resume_thread(
+        &app.local_settings,
+        app.config.clone(),
+        parent_thread_id,
+        crate::app_server_session::ResumeModelSettings::RestoreFromThread,
+    ))
+    .await?;
     assert_eq!(started.session.thread_id, parent_thread_id);
 
-    let forked = app_server
-        .fork_thread(&app.local_settings, app.config.clone(), parent_thread_id)
-        .await?;
+    let forked =
+        Box::pin(app_server.fork_thread(&app.local_settings, app.config.clone(), parent_thread_id))
+            .await?;
 
     assert_ne!(forked.session.thread_id, parent_thread_id);
     assert_eq!(recorded_params(&requests, "thread/fork").len(), 1);

codex-rs/tui/src/lib.rs

@@ -294,7 +294,8 @@ async fn start_embedded_app_server(
     environment_manager: Arc<EnvironmentManager>,
     embedded_network_policy: codex_app_server_client::EmbeddedNetworkPolicy,
 ) -> color_eyre::Result<InProcessAppServerClient> {
-    start_embedded_app_server_with(
+    // Keep embedded startup state off the caller's stack during session transitions.
+    Box::pin(start_embedded_app_server_with(
         arg0_paths,
         config,
         cli_kv_overrides,
@@ -307,7 +308,7 @@ async fn start_embedded_app_server(
         environment_manager,
         embedded_network_policy,
         InProcessAppServerClient::start,
-    )
+    ))
     .await
 }