diff --git a/codex-rs/app-server/README.md b/codex-rs/app-server/README.md index a6bcb1ab33..5331d42b0a 100644 --- a/codex-rs/app-server/README.md +++ b/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`. diff --git a/codex-rs/app-server/src/mcp_refresh.rs b/codex-rs/app-server/src/mcp_refresh.rs index b654ad6a28..ce1f7871ad 100644 --- a/codex-rs/app-server/src/mcp_refresh.rs +++ b/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, 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, 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> = 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(¤t_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(¤t_config) + } + } + } else { + disabled_enterprise_config(¤t_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 { - let thread_config = thread.config().await; config_manager .load_latest_config_with_session_layers( - &thread_config.config_layer_stack, - &thread_config.cwd, + ¤t_config.config_layer_stack, + ¤t_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(¤t_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(¤t_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::>(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>>, } impl CountingThreadConfigLoader { @@ -394,6 +544,15 @@ enabled = false ) -> Result, 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); diff --git a/codex-rs/app-server/src/request_processors/config_processor.rs b/codex-rs/app-server/src/request_processors/config_processor.rs index 4176a55996..d636b125dc 100644 --- a/codex-rs/app-server/src/request_processors/config_processor.rs +++ b/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( - ¤t_config.config_layer_stack, - ¤t_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( + ¤t_config.config_layer_stack, + ¤t_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) -> #[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, diff --git a/codex-rs/app-server/src/request_processors/mcp_processor.rs b/codex-rs/app-server/src/request_processors/mcp_processor.rs index 120e891863..ed9a640c2c 100644 --- a/codex-rs/app-server/src/request_processors/mcp_processor.rs +++ b/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, JSONRPCErrorError> { - self.mcp_server_refresh_response(params) - .await - .map(|response| Some(response.into())) + ) -> impl Future, 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( diff --git a/codex-rs/app-server/tests/suite/v2/plugin_install.rs b/codex-rs/app-server/tests/suite/v2/plugin_install.rs index 99ffcd08c5..3d62550c2f 100644 --- a/codex-rs/app-server/tests/suite/v2/plugin_install.rs +++ b/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::(&mcp_settings)?) + Some(&toml::from_str::(mcp_settings)?) ); Ok(()) } diff --git a/codex-rs/codex-mcp/src/catalog.rs b/codex-rs/codex-mcp/src/catalog.rs index 00a08c51ba..332516533c 100644 --- a/codex-rs/codex-mcp/src/catalog.rs +++ b/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!(®istration.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; } diff --git a/codex-rs/codex-mcp/src/catalog_tests.rs b/codex-rs/codex-mcp/src/catalog_tests.rs index ed791c4963..eaf50e79b8 100644 --- a/codex-rs/codex-mcp/src/catalog_tests.rs +++ b/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); +} diff --git a/codex-rs/codex-mcp/src/mcp/mod_tests.rs b/codex-rs/codex-mcp/src/mcp/mod_tests.rs index c90d39eb68..61f2a7f6f0 100644 --- a/codex-rs/codex-mcp/src/mcp/mod_tests.rs +++ b/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); } } diff --git a/codex-rs/config/src/mcp_ema.rs b/codex-rs/config/src/mcp_ema.rs index bee6d43f5d..327b471d35 100644 --- a/codex-rs/config/src/mcp_ema.rs +++ b/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 { - 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::() - .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(()) } diff --git a/codex-rs/config/src/mcp_ema_tests.rs b/codex-rs/config/src/mcp_ema_tests.rs index 585f62ba37..836bf37197 100644 --- a/codex-rs/config/src/mcp_ema_tests.rs +++ b/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(), ®istration), - (project.clone(), &overlay) - ]), - "{section}: {change}" - ); - } - let higher = registration.replace("tools", "managed-tools"); - assert!( - !valid_ema_layers(vec![ - (system.clone(), ®istration), - (user.clone(), &higher), - (project.clone(), ®istration) - ]), - "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(), ®istration), - (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(), ®istration.replace(required, ""))]), - "missing {required}" - ); - } - } else { - assert!(valid_ema_layers(vec![ - (system.clone(), ®istration), - (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"); diff --git a/codex-rs/config/src/mcp_types.rs b/codex-rs/config/src/mcp_types.rs index 4b627edc77..c85e7c99b8 100644 --- a/codex-rs/config/src/mcp_types.rs +++ b/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, - - /// 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 diff --git a/codex-rs/config/src/state.rs b/codex-rs/config/src/state.rs index ec271fd820..f635eff83d 100644 --- a/codex-rs/config/src/state.rs +++ b/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) -> Self { self.startup_warnings = Some(startup_warnings); self diff --git a/codex-rs/config/src/types.rs b/codex-rs/config/src/types.rs index e44f737c18..23fe035f10 100644 --- a/codex-rs/config/src/types.rs +++ b/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, + /// 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, - 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 +where + D: serde::Deserializer<'de>, +{ + serde::de::IgnoredAny::deserialize(deserializer).map(|_| true) } #[derive(Serialize, Deserialize, Debug, Clone, PartialEq, Eq, Default, JsonSchema)] diff --git a/codex-rs/core-plugins/src/loader.rs b/codex-rs/core-plugins/src/loader.rs index 92500be8cc..eb5103d6a6 100644 --- a/codex-rs/core-plugins/src/loader.rs +++ b/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(); diff --git a/codex-rs/core-plugins/src/loader_tests.rs b/codex-rs/core-plugins/src/loader_tests.rs index 894574d9dc..e997825b0c 100644 --- a/codex-rs/core-plugins/src/loader_tests.rs +++ b/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 = + 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"); diff --git a/codex-rs/core/config.schema.json b/codex-rs/core/config.schema.json index ac6171c01c..1e736a1a66 100644 --- a/codex-rs/core/config.schema.json +++ b/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.", diff --git a/codex-rs/core/src/codex_thread.rs b/codex-rs/core/src/codex_thread.rs index 316af05c08..c54fd5454d 100644 --- a/codex-rs/core/src/codex_thread.rs +++ b/codex-rs/core/src/codex_thread.rs @@ -161,6 +161,18 @@ pub struct CodexThreadSettingsOverrides { pub disabled_plugin_ids: Option>, } +/// 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, + 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, + 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. diff --git a/codex-rs/core/src/compact_tests.rs b/codex-rs/core/src/compact_tests.rs index 217d6b63fd..19d6ed2e5d 100644 --- a/codex-rs/core/src/compact_tests.rs +++ b/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( diff --git a/codex-rs/core/src/config/managed_features.rs b/codex-rs/core/src/config/managed_features.rs index 9fa6c01c1b..80c951c488 100644 --- a/codex-rs/core/src/config/managed_features.rs +++ b/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) } diff --git a/codex-rs/core/src/config/mod.rs b/codex-rs/core/src/config/mod.rs index 60c6f65353..65a8aa3eae 100644 --- a/codex-rs/core/src/config/mod.rs +++ b/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; diff --git a/codex-rs/core/src/config/runtime_refresh.rs b/codex-rs/core/src/config/runtime_refresh.rs new file mode 100644 index 0000000000..0d819cd849 --- /dev/null +++ b/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, +} + +impl Config { + pub(crate) fn resolve_runtime_refresh( + &self, + incoming: &Self, + scope: RuntimeConfigRefresh, + ) -> std::io::Result { + 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"); + } + } +} diff --git a/codex-rs/core/src/config/runtime_refresh_tests.rs b/codex-rs/core/src/config/runtime_refresh_tests.rs new file mode 100644 index 0000000000..add31508e2 --- /dev/null +++ b/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 { + 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::>() + }; + assert_eq!(preserved_layers(&refreshed), preserved_layers(¤t)); + let server = &runtime_servers(&refreshed)["enterprise"]; + assert!(!server.enabled); + assert_eq!( + server.ema_registration(), + runtime_servers(¤t)["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(¤t, 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)); +} diff --git a/codex-rs/core/src/guardian/tests.rs b/codex-rs/core/src/guardian/tests.rs index c58d4a76a0..5e9ec6813f 100644 --- a/codex-rs/core/src/guardian/tests.rs +++ b/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(), diff --git a/codex-rs/core/src/lib.rs b/codex-rs/core/src/lib.rs index 2595ce5bb9..45ff58ab6d 100644 --- a/codex-rs/core/src/lib.rs +++ b/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; diff --git a/codex-rs/core/src/session/config_refresh.rs b/codex-rs/core/src/session/config_refresh.rs new file mode 100644 index 0000000000..410721925b --- /dev/null +++ b/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, + 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, + next_config: Config, + ) -> ConfigRefreshOutcome { + self.refresh_config(expected_config, next_config, RuntimeConfigRefresh::Mcp) + .await + } + + pub(super) async fn refresh_config( + &self, + expected_config: Arc, + 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 + } +} diff --git a/codex-rs/core/src/session/handlers.rs b/codex-rs/core/src/session/handlers.rs index 94c47023ea..5114a80b24 100644 --- a/codex-rs/core/src/session/handlers.rs +++ b/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) { - sess.reload_user_config_layer().await; + Box::pin(sess.reload_user_config_layer()).await; } pub async fn compact(sess: &Arc, sub_id: String) { diff --git a/codex-rs/core/src/session/mod.rs b/codex-rs/core/src/session/mod.rs index e08a0fa5a8..e3493e6eb1 100644 --- a/codex-rs/core/src/session/mod.rs +++ b/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, + 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) { @@ -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::>(); - 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::>(); + 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::(&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::(&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. diff --git a/codex-rs/core/src/session/tests.rs b/codex-rs/core/src/session/tests.rs index 13ebc68245..aadfc55d33 100644 --- a/codex-rs/core/src/session/tests.rs +++ b/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::(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| { @@ -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(¤t_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(¤t_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" + ); +} diff --git a/codex-rs/core/src/stream_events_utils_tests.rs b/codex-rs/core/src/stream_events_utils_tests.rs index 2f5bc57072..aec39664b3 100644 --- a/codex-rs/core/src/stream_events_utils_tests.rs +++ b/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") diff --git a/codex-rs/core/src/tools/executed_tool_calls/request_metadata_tests.rs b/codex-rs/core/src/tools/executed_tool_calls/request_metadata_tests.rs index 032eaab27c..c92841f413 100644 --- a/codex-rs/core/src/tools/executed_tool_calls/request_metadata_tests.rs +++ b/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, 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)]; diff --git a/codex-rs/core/tests/suite/agents_md.rs b/codex-rs/core/tests/suite/agents_md.rs index eec7d13d0b..04f4d023f9 100644 --- a/codex-rs/core/tests/suite/agents_md.rs +++ b/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, diff --git a/codex-rs/core/tests/suite/code_mode.rs b/codex-rs/core/tests/suite/code_mode.rs index 3773f265cc..4094ed06f3 100644 --- a/codex-rs/core/tests/suite/code_mode.rs +++ b/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, diff --git a/codex-rs/core/tests/suite/direct_tool_metadata.rs b/codex-rs/core/tests/suite/direct_tool_metadata.rs index f19cd36d60..65c03e3b02 100644 --- a/codex-rs/core/tests/suite/direct_tool_metadata.rs +++ b/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| { diff --git a/codex-rs/core/tests/suite/hooks_executor.rs b/codex-rs/core/tests/suite/hooks_executor.rs index 567d1c8190..d017742969 100644 --- a/codex-rs/core/tests/suite/hooks_executor.rs +++ b/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 diff --git a/codex-rs/core/tests/suite/mcp_ema_config.rs b/codex-rs/core/tests/suite/mcp_ema_config.rs index 08324fadae..32681ee323 100644 --- a/codex-rs/core/tests/suite/mcp_ema_config.rs +++ b/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!( diff --git a/codex-rs/core/tests/suite/mcp_oauth_refresh_tests.rs b/codex-rs/core/tests/suite/mcp_oauth_refresh_tests.rs index 6ecfde4d06..6f7d1a5618 100644 --- a/codex-rs/core/tests/suite/mcp_oauth_refresh_tests.rs +++ b/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. diff --git a/codex-rs/core/tests/suite/mcp_optional_startup_grace.rs b/codex-rs/core/tests/suite/mcp_optional_startup_grace.rs index d366b75f41..73a2028093 100644 --- a/codex-rs/core/tests/suite/mcp_optional_startup_grace.rs +++ b/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!( diff --git a/codex-rs/core/tests/suite/mcp_startup_refresh_http_proxy.rs b/codex-rs/core/tests/suite/mcp_startup_refresh_http_proxy.rs index 6baaf8a9a4..04dfcb48de 100644 --- a/codex-rs/core/tests/suite/mcp_startup_refresh_http_proxy.rs +++ b/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( diff --git a/codex-rs/core/tests/suite/mcp_tool_exposure.rs b/codex-rs/core/tests/suite/mcp_tool_exposure.rs index b24a4d860a..67440412c0 100644 --- a/codex-rs/core/tests/suite/mcp_tool_exposure.rs +++ b/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?; diff --git a/codex-rs/core/tests/suite/request_plugin_install.rs b/codex-rs/core/tests/suite/request_plugin_install.rs index 33a094a03b..2d8766b4fd 100644 --- a/codex-rs/core/tests/suite/request_plugin_install.rs +++ b/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(); diff --git a/codex-rs/core/tests/suite/rmcp_client.rs b/codex-rs/core/tests/suite/rmcp_client.rs index 54d31beb9f..2833288bb6 100644 --- a/codex-rs/core/tests/suite/rmcp_client.rs +++ b/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| { diff --git a/codex-rs/core/tests/suite/scenarios_guardian_conversation_history_tests.rs b/codex-rs/core/tests/suite/scenarios_guardian_conversation_history_tests.rs index 86f4832819..0b7b10c991 100644 --- a/codex-rs/core/tests/suite/scenarios_guardian_conversation_history_tests.rs +++ b/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; diff --git a/codex-rs/tui/src/app/agents_overview_tests.rs b/codex-rs/tui/src/app/agents_overview_tests.rs index b5f4d88924..c4b6bd3502 100644 --- a/codex-rs/tui/src/app/agents_overview_tests.rs +++ b/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 )); diff --git a/codex-rs/tui/src/app/tests.rs b/codex-rs/tui/src/app/tests.rs index 65271708dc..ab54b10d05 100644 --- a/codex-rs/tui/src/app/tests.rs +++ b/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; diff --git a/codex-rs/tui/src/app/tests/session_lifecycle_requests.rs b/codex-rs/tui/src/app/tests/session_lifecycle_requests.rs index 0fbf795370..bb74f6c17c 100644 --- a/codex-rs/tui/src/app/tests/session_lifecycle_requests.rs +++ b/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); diff --git a/codex-rs/tui/src/lib.rs b/codex-rs/tui/src/lib.rs index d1c672d9ac..b72a33c776 100644 --- a/codex-rs/tui/src/lib.rs +++ b/codex-rs/tui/src/lib.rs @@ -294,7 +294,8 @@ async fn start_embedded_app_server( environment_manager: Arc, embedded_network_policy: codex_app_server_client::EmbeddedNetworkPolicy, ) -> color_eyre::Result { - 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 }