diff --git a/codex-rs/app-server/src/application_network.rs b/codex-rs/app-server/src/application_network.rs index 0641a80e0b..bb5d0987d8 100644 --- a/codex-rs/app-server/src/application_network.rs +++ b/codex-rs/app-server/src/application_network.rs @@ -77,18 +77,23 @@ impl ConfigManager { let local_sources = self.load_local_network_policy(revision).await?; let cloud_config = self.current_cloud_config_bundle(); let result = async { - let cloud = cloud_config.get().await.map_err(io::Error::other)?; + let cloud_snapshot = cloud_config.get_snapshot().await; + let cloud = cloud_snapshot.bundle.clone().map_err(io::Error::other)?; let application = local_sources.compose( cloud .as_ref() .map(|bundle| bundle.requirements_toml.clone()) .unwrap_or_default(), )?; - Ok::<_, io::Error>((destination_policy(application.as_ref()), cloud)) + Ok::<_, io::Error>(( + destination_policy(application.as_ref()), + cloud, + cloud_snapshot, + )) } .await; match result { - Ok((policy, cloud)) => { + Ok((policy, cloud, cloud_snapshot)) => { let mut current = self.network_policy_snapshot.write().map_err(|_| { io::Error::other("application network policy snapshot lock poisoned") })?; @@ -99,7 +104,7 @@ impl ConfigManager { let next = ApplicationPolicySnapshot { revision, policy, - cloud: cloud.clone(), + cloud, }; // Concurrent config loads may finish together if their actual inputs are unchanged. let snapshot = match current.as_ref() { @@ -111,7 +116,7 @@ impl ConfigManager { } }; Ok(ApplicationPolicyLoad { - cloud_config: CloudConfigBundleLoader::new(async move { Ok(cloud) }), + cloud_config: CloudConfigBundleLoader::from_snapshot(cloud_snapshot), snapshot, }) } diff --git a/codex-rs/app-server/src/config_manager.rs b/codex-rs/app-server/src/config_manager.rs index 2467d4d4f1..21d1780910 100644 --- a/codex-rs/app-server/src/config_manager.rs +++ b/codex-rs/app-server/src/config_manager.rs @@ -160,6 +160,7 @@ impl ConfigManager { http_client_factory, ); if let Ok(mut guard) = self.cloud_config_bundle.write() { + guard.retire_ema_policy(); *guard = loader; } else { warn!("failed to update cloud config bundle loader"); @@ -168,6 +169,7 @@ impl ConfigManager { pub(crate) fn clear_cloud_config_bundle_loader(&self) { if let Ok(mut guard) = self.cloud_config_bundle.write() { + guard.retire_ema_policy(); *guard = CloudConfigBundleLoader::default(); } else { warn!("failed to clear cloud config bundle loader"); @@ -212,21 +214,19 @@ impl ConfigManager { session_layers: &ConfigLayerStack, cwd: &Path, ) -> std::io::Result { - let refreshed_config = self.load_latest_config(Some(cwd.to_path_buf())).await?; + let codex_home = AbsolutePathBuf::from_absolute_path(&self.codex_home)?; + let layers = self + .load_config_layers_for_cwd(AbsolutePathBuf::from_absolute_path(cwd)?) + .await?; let mut config = Config::rebuild_with_session_layers( session_layers, cwd.to_path_buf(), - &refreshed_config.config_layer_stack, - refreshed_config.codex_home.clone(), - refreshed_config - .zsh_path - .clone() - .map(AbsolutePathBuf::try_from) - .transpose()?, + &layers, + codex_home, + /*default_zsh_path*/ None, ) .await?; - config.application_network_policy = refreshed_config.application_network_policy; - config.application_auth_route_config = refreshed_config.application_auth_route_config; + self.apply_network_policy(&mut config); self.apply_runtime_feature_enablement(&mut config); self.apply_arg0_paths(&mut config); Ok(config) @@ -594,6 +594,10 @@ pub(crate) fn apply_runtime_feature_enablement( config: &mut Config, runtime_feature_enablement: &BTreeMap, ) { + config.runtime_feature_defaults = runtime_feature_enablement + .iter() + .filter_map(|(name, enabled)| feature_for_key(name).map(|feature| (feature, *enabled))) + .collect(); let protected_features = protected_feature_keys(&config.config_layer_stack); for (name, enabled) in runtime_feature_enablement { if protected_features.contains(name) { diff --git a/codex-rs/app-server/src/config_manager_service_tests.rs b/codex-rs/app-server/src/config_manager_service_tests.rs index c3947b0e76..af60acb652 100644 --- a/codex-rs/app-server/src/config_manager_service_tests.rs +++ b/codex-rs/app-server/src/config_manager_service_tests.rs @@ -1441,6 +1441,77 @@ url_prefix_from_env = "VENDOR_ENDPOINT" Ok(()) } +#[tokio::test] +async fn thread_refresh_resolves_preserved_layers_before_materialization() -> Result<()> { + use codex_core::config::ConfigOverrides; + use codex_features::Feature; + use std::collections::HashMap; + use std::sync::Arc; + + let tmp = tempdir()?; + let selected_path = tmp.path().join("work.config.toml"); + std::fs::write(&selected_path, "log_dir = \"old-logs\"\n")?; + let mut loader_overrides = LoaderOverrides::without_managed_config_for_tests(); + loader_overrides.user_config_path = Some(AbsolutePathBuf::from_absolute_path(&selected_path)?); + loader_overrides.user_config_profile = Some("work".parse()?); + let service = ConfigManager::new( + tmp.path().to_path_buf(), + Vec::new(), + loader_overrides, + /*strict_config*/ true, + CloudConfigBundleLoader::default(), + Default::default(), + Arc::new(codex_config::NoopThreadConfigLoader), + ); + let thread = service + .load_for_cwd( + Some(HashMap::from([( + "model_provider".to_string(), + serde_json::json!("openai"), + )])), + ConfigOverrides::default(), + Some(tmp.path().to_path_buf()), + ) + .await?; + service + .extend_runtime_feature_enablement([ + ("secret_auth_storage".to_string(), true), + ("mcp_oauth_refresh_coordination".to_string(), true), + ]) + .expect("set runtime features"); + std::fs::write( + &selected_path, + "model_provider = \"missing-provider\"\nlog_dir = \"new-logs\"\n\ + [features]\nmcp_oauth_refresh_coordination = false\n", + )?; + // A throwaway host configuration is invalid; the actual thread's preserved + // provider is valid. Refresh must materialize only the final composed layers. + service + .load_latest_config(Some(tmp.path().to_path_buf())) + .await + .expect_err("host provider is unavailable"); + let refreshed = service + .load_latest_config_with_session_layers(&thread.config_layer_stack, &thread.cwd) + .await?; + assert_eq!(refreshed.model_provider_id, "openai"); + assert_eq!(refreshed.cwd, thread.cwd); + assert_eq!(refreshed.log_dir, tmp.path().join("new-logs")); + assert!(refreshed.features.enabled(Feature::SecretAuthStorage)); + assert!( + !refreshed + .features + .enabled(Feature::McpOAuthRefreshCoordination) + ); + std::fs::write(&selected_path, "approval_policy = false\n")?; + let error = service + .load_latest_config_with_session_layers(&thread.config_layer_stack, &thread.cwd) + .await + .expect_err("invalid final config is rejected"); + assert_eq!(error.kind(), std::io::ErrorKind::InvalidData); + assert!(error.to_string().contains("work.config.toml"), "{error}"); + Ok(()) +} + #[tokio::test] async fn invalid_user_value_rejected_even_if_overridden_by_managed() { let tmp = tempdir().expect("tempdir"); diff --git a/codex-rs/app-server/src/mcp_refresh.rs b/codex-rs/app-server/src/mcp_refresh.rs index ce1f7871ad..ac965523fb 100644 --- a/codex-rs/app-server/src/mcp_refresh.rs +++ b/codex-rs/app-server/src/mcp_refresh.rs @@ -54,10 +54,7 @@ async fn reload_mcp_config_with_policy( 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 - { + ReloadPolicy::Strict => match config_manager.load_config_layers(/*cwd*/ None).await { Ok(_) => true, Err(err) => { load_error.get_or_insert(err); @@ -147,6 +144,9 @@ async fn reload_mcp_config_with_policy( fn disabled_enterprise_config(current_config: &Config) -> Config { let mut config = current_config.clone(); config.disable_mcp_enterprise_auth(); + config.config_layer_stack = config + .config_layer_stack + .with_cloud_config_binding(/*binding*/ None); config } diff --git a/codex-rs/app-server/src/request_processors.rs b/codex-rs/app-server/src/request_processors.rs index 2f4731ab23..38110ec115 100644 --- a/codex-rs/app-server/src/request_processors.rs +++ b/codex-rs/app-server/src/request_processors.rs @@ -542,6 +542,9 @@ mod bedrock_auth; mod catalog_processor; mod command_exec_processor; mod config_processor; +#[cfg(test)] +#[path = "request_processors/config_reload_tests.rs"] +mod config_reload_tests; mod diagnostics; mod environment_processor; mod feedback_doctor_report; 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 d636b125dc..56a8b4b316 100644 --- a/codex-rs/app-server/src/request_processors/config_processor.rs +++ b/codex-rs/app-server/src/request_processors/config_processor.rs @@ -362,10 +362,7 @@ pub(super) async fn reload_user_config( } 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 - { + if let Err(err) = config_manager.load_config_layers(/*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 { diff --git a/codex-rs/app-server/src/request_processors/config_reload_tests.rs b/codex-rs/app-server/src/request_processors/config_reload_tests.rs new file mode 100644 index 0000000000..35fe5adcc5 --- /dev/null +++ b/codex-rs/app-server/src/request_processors/config_reload_tests.rs @@ -0,0 +1,250 @@ +//! Regressions for reload entry points composing the final thread configuration. + +use super::config_processor::reload_user_config; +use crate::config_manager::ConfigManager; +use crate::mcp_refresh::reload_mcp_config; +use codex_config::types::AuthKeyringBackendKind; +use codex_core::config::ConfigOverrides; +use codex_exec_server::EnvironmentManager; +use codex_login::CodexAuth; +use pretty_assertions::assert_eq; +use serde_json::json; +use std::collections::HashMap; +use std::sync::Arc; + +#[tokio::test] +async fn reloads_accept_a_provider_supplied_by_thread_overrides() -> anyhow::Result<()> { + let home = tempfile::tempdir()?; + let manager = ConfigManager::without_managed_config_for_tests(home.path().to_path_buf()); + let initial = manager + .load_for_cwd( + Some(HashMap::from([ + ("model_provider".to_string(), json!("thread-provider")), + ( + "model_providers.thread-provider".to_string(), + json!({"name": "Thread provider", "base_url": "https://provider.example/v1", "wire_api": "responses"}), + ), + ])), + ConfigOverrides::default(), + Some(home.path().to_path_buf()), + ) + .await?; + let threads = Arc::new( + 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 = threads + .start_thread(codex_core::StartThreadOptions::new(initial)) + .await? + .thread; + + for enabled in [true, false] { + std::fs::write( + home.path().join(codex_config::CONFIG_TOML_FILE), + format!( + "model_provider = 'missing-provider'\n[features]\nsecret_auth_storage = {enabled}\n" + ), + )?; + if enabled { + reload_mcp_config(&threads, &manager).await?; + } else { + reload_user_config(&manager, &threads).await; + } + let config = thread.config().await; + assert_eq!(config.model_provider_id, "thread-provider"); + assert_eq!( + config.auth_keyring_backend_kind(), + if enabled { + AuthKeyringBackendKind::Secrets + } else { + AuthKeyringBackendKind::Direct + } + ); + } + Ok(()) +} + +#[tokio::test] +async fn user_reload_promotes_plugin_and_feature_requirements_without_server_changes() +-> anyhow::Result<()> { + let home = tempfile::tempdir()?; + let requirements_path = home.path().join("requirements.toml"); + std::fs::write(&requirements_path, "")?; + std::fs::write( + home.path().join(codex_config::CONFIG_TOML_FILE), + "[features]\nenable_mcp_apps = true\n", + )?; + let mut overrides = codex_config::LoaderOverrides::without_managed_config_for_tests(); + overrides.ignore_managed_requirements = false; + overrides.system_requirements_path = Some(requirements_path.clone()); + let manager = ConfigManager::new( + home.path().to_path_buf(), + Vec::new(), + overrides, + /*strict_config*/ false, + codex_config::CloudConfigBundleLoader::default(), + codex_arg0::Arg0DispatchPaths::default(), + Arc::new(codex_config::NoopThreadConfigLoader), + ); + let initial = manager + .load_latest_config(Some(home.path().to_path_buf())) + .await?; + let threads = Arc::new( + 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()), + ), + ); + assert!( + initial + .features + .enabled(codex_features::Feature::EnableMcpApps) + ); + let original_servers = initial.mcp_servers.get().clone(); + let thread = threads + .start_thread(codex_core::StartThreadOptions::new(initial)) + .await? + .thread; + std::fs::write( + &requirements_path, + "[features]\nenable_mcp_apps = false\n[plugins.example.mcp_servers]\n", + )?; + tokio::time::timeout( + std::time::Duration::from_secs(10), + reload_user_config(&manager, &threads), + ) + .await + .expect("policy promotion must terminate"); + let config = thread.config().await; + assert_eq!(config.mcp_servers.get(), &original_servers); + assert!( + !config + .features + .enabled(codex_features::Feature::EnableMcpApps) + ); + assert!( + config + .config_layer_stack + .requirements() + .plugins + .as_ref() + .unwrap() + .value + .get("example") + .unwrap() + .mcp_servers + .as_ref() + .unwrap() + .is_empty() + ); + assert!( + config + .config_layer_stack + .requirements() + .feature_requirements + .is_some() + ); + Ok(()) +} + +#[test_case::test_case(codex_config::CloudConfigBundleBindingStatus::Stale)] +#[test_case::test_case(codex_config::CloudConfigBundleBindingStatus::Suspended)] +#[tokio::test] +async fn mcp_refresh_rejects_superseded_app_server_policy( + status: codex_config::CloudConfigBundleBindingStatus, +) -> anyhow::Result<()> { + use codex_config::CloudConfigBundle; + use codex_config::CloudConfigBundleLoader; + use codex_config::CloudConfigBundlePolicy; + use codex_config::CloudConfigBundleSnapshot; + use codex_core::ConfigRefreshOutcome; + + let home = tempfile::tempdir()?; + let policy = CloudConfigBundlePolicy::default(); + let mut snapshot = CloudConfigBundleSnapshot { + bundle: Ok(None), + binding: None, + }; + policy.publish_snapshot(&mut snapshot); + let loader = + CloudConfigBundleLoader::default().with_ema_policy_snapshots(policy.clone(), move || { + let snapshot = snapshot.clone(); + async move { snapshot } + }); + let manager = ConfigManager::new( + home.path().to_path_buf(), + Vec::new(), + codex_config::LoaderOverrides::without_managed_config_for_tests(), + /*strict_config*/ false, + loader, + codex_arg0::Arg0DispatchPaths::default(), + Arc::new(codex_config::NoopThreadConfigLoader), + ); + let initial = manager + .load_latest_config(Some(home.path().to_path_buf())) + .await?; + let threads = 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 = threads + .start_thread(codex_core::StartThreadOptions::new(initial)) + .await? + .thread; + let current = thread.config().await; + let next = manager + .load_latest_config_with_session_layers(¤t.config_layer_stack, ¤t.cwd) + .await?; + let mut replacement = CloudConfigBundle::default(); + replacement + .config_toml + .enterprise_managed + .push(codex_config::CloudConfigFragment { + id: "replacement".into(), + name: "replacement".into(), + contents: "model = 'replacement-model'".into(), + }); + policy.observe_remote_bundle(&replacement); + if status == codex_config::CloudConfigBundleBindingStatus::Stale { + policy.publish_snapshot(&mut CloudConfigBundleSnapshot { + bundle: Ok(Some(replacement)), + binding: None, + }); + } + assert_eq!( + next.config_layer_stack + .cloud_config_binding() + .expect("app-server must retain the policy binding") + .read() + .status, + status, + ); + let outcome = thread.refresh_mcp_config(Arc::clone(¤t), next).await; + let expected = match status { + codex_config::CloudConfigBundleBindingStatus::Stale => ConfigRefreshOutcome::Stale, + codex_config::CloudConfigBundleBindingStatus::Suspended => ConfigRefreshOutcome::Rejected, + codex_config::CloudConfigBundleBindingStatus::Current => unreachable!(), + }; + assert_eq!(outcome, expected); + if status == codex_config::CloudConfigBundleBindingStatus::Stale { + assert!(Arc::ptr_eq(¤t, &thread.config().await)); + } else { + assert!( + thread + .config() + .await + .config_layer_stack + .cloud_config_binding() + .is_none() + ); + } + Ok(()) +} diff --git a/codex-rs/cloud-config/Cargo.toml b/codex-rs/cloud-config/Cargo.toml index 48ab0d870c..2077502017 100644 --- a/codex-rs/cloud-config/Cargo.toml +++ b/codex-rs/cloud-config/Cargo.toml @@ -21,6 +21,7 @@ hmac = "0.12.1" serde = { workspace = true, features = ["derive"] } serde_json = { workspace = true } sha2 = { workspace = true } +tempfile = { workspace = true } thiserror = { workspace = true } tokio = { workspace = true, features = ["fs", "rt", "sync", "time"] } tracing = { workspace = true } @@ -29,7 +30,6 @@ tracing = { workspace = true } codex-agent-identity = { workspace = true } codex-core = { workspace = true } pretty_assertions = { workspace = true } -tempfile = { workspace = true } tokio = { workspace = true, features = ["macros", "rt", "test-util", "time"] } [lib] diff --git a/codex-rs/cloud-config/src/bundle_loader.rs b/codex-rs/cloud-config/src/bundle_loader.rs index eb99944010..5b027d21cd 100644 --- a/codex-rs/cloud-config/src/bundle_loader.rs +++ b/codex-rs/cloud-config/src/bundle_loader.rs @@ -59,9 +59,15 @@ where refresh_task, }); + let snapshot_lifetime = Arc::clone(&lifetime); + let policy = lifetime.service.policy.clone(); let loader = CloudConfigBundleLoader::from_getter(move || { let lifetime = Arc::clone(&lifetime); async move { lifetime.service.get_latest().await } + }) + .with_ema_policy_snapshots(policy, move || { + let lifetime = Arc::clone(&snapshot_lifetime); + async move { lifetime.service.get_latest_snapshot().await } }); (loader, abort_handle) } diff --git a/codex-rs/cloud-config/src/cache.rs b/codex-rs/cloud-config/src/cache.rs index e5d1d4dc84..94748ec052 100644 --- a/codex-rs/cloud-config/src/cache.rs +++ b/codex-rs/cloud-config/src/cache.rs @@ -10,13 +10,17 @@ use chrono::Duration as ChronoDuration; use chrono::Utc; use codex_config::AbsolutePathBuf; use codex_config::CloudConfigBundle; +use codex_config::CloudConfigBundlePolicyRevision; use hmac::Hmac; use hmac::Mac; use serde::Deserialize; use serde::Serialize; use sha2::Sha256; +use std::io::Write; use std::path::Path; +use std::path::PathBuf; use std::time::Duration; +use tempfile::NamedTempFile; use thiserror::Error; use tokio::fs; @@ -125,12 +129,12 @@ impl CloudConfigBundleCache { } } - pub(super) async fn save( + pub(super) async fn prepare( &self, chatgpt_user_id: Option, account_id: Option, bundle: CloudConfigBundle, - ) -> Result<(), CloudConfigBundleCacheError> { + ) -> Result { let now = Utc::now(); let expires_at = now .checked_add_signed( @@ -154,16 +158,45 @@ impl CloudConfigBundleCache { }) .map_err(|_| CloudConfigBundleCacheError)?; - if let Some(parent) = self.path.parent() { - fs::create_dir_all(parent) - .await + let destination = self.path.to_path_buf(); + tokio::task::spawn_blocking(move || { + let parent = destination.parent().ok_or(CloudConfigBundleCacheError)?; + std::fs::create_dir_all(parent).map_err(|_| CloudConfigBundleCacheError)?; + let mut temporary = + NamedTempFile::new_in(parent).map_err(|_| CloudConfigBundleCacheError)?; + temporary + .write_all(&serialized) .map_err(|_| CloudConfigBundleCacheError)?; - } + Ok(StagedCloudConfigBundleCache { + temporary, + destination, + }) + }) + .await + .map_err(|_| CloudConfigBundleCacheError)? + } +} - fs::write(&self.path, serialized) - .await - .map_err(|_| CloudConfigBundleCacheError)?; - Ok(()) +pub(super) struct StagedCloudConfigBundleCache { + temporary: NamedTempFile, + destination: PathBuf, +} + +impl StagedCloudConfigBundleCache { + pub(super) async fn publish_if_current( + self, + revision: CloudConfigBundlePolicyRevision, + ) -> Result<(), CloudConfigBundleCacheError> { + tokio::task::spawn_blocking(move || { + revision.commit_if_current(|| { + self.temporary + .persist(self.destination) + .map(|_| ()) + .map_err(|_| CloudConfigBundleCacheError) + }) + }) + .await + .map_err(|_| CloudConfigBundleCacheError)? } } diff --git a/codex-rs/cloud-config/src/cache_tests.rs b/codex-rs/cloud-config/src/cache_tests.rs index 28899f930d..29d4a57922 100644 --- a/codex-rs/cloud-config/src/cache_tests.rs +++ b/codex-rs/cloud-config/src/cache_tests.rs @@ -67,13 +67,19 @@ async fn save_writes_signed_payload_and_loads_for_matching_identity() { let cache = create_test_cache(codex_home.path()); let bundle = test_bundle(); + let revision = codex_config::CloudConfigBundlePolicy::default() + .observe_remote_bundle(&bundle) + .expect("cache revision"); cache - .save( + .prepare( Some("user-12345".to_string()), Some("account-12345".to_string()), bundle.clone(), ) .await + .expect("stage cache") + .publish_if_current(revision) + .await .expect("save cache"); let cache_file: CloudConfigBundleCacheFile = diff --git a/codex-rs/cloud-config/src/service.rs b/codex-rs/cloud-config/src/service.rs index e27db63b0c..6291091737 100644 --- a/codex-rs/cloud-config/src/service.rs +++ b/codex-rs/cloud-config/src/service.rs @@ -18,6 +18,8 @@ use codex_config::AbsolutePathBuf; use codex_config::CloudConfigBundle; use codex_config::CloudConfigBundleLoadError; use codex_config::CloudConfigBundleLoadErrorCode; +use codex_config::CloudConfigBundlePolicy; +use codex_config::CloudConfigBundleSnapshot; use codex_login::AuthManager; use codex_login::CodexAuth; use codex_login::RefreshTokenError; @@ -82,7 +84,8 @@ pub(crate) struct CloudConfigBundleService { cache_enabled: bool, codex_home: AbsolutePathBuf, timeout: Duration, - latest_bundle: OnceCell, CloudConfigBundleLoadError>>>, + latest_bundle: OnceCell>, + pub(crate) policy: CloudConfigBundlePolicy, } impl CloudConfigBundleService @@ -104,6 +107,7 @@ where codex_home, timeout, latest_bundle: OnceCell::new(), + policy: CloudConfigBundlePolicy::default(), } } @@ -115,12 +119,24 @@ where pub(crate) async fn get_latest( &self, ) -> Result, CloudConfigBundleLoadError> { + self.get_latest_snapshot().await.bundle + } + + pub(crate) async fn get_latest_snapshot(&self) -> CloudConfigBundleSnapshot { + self.latest_snapshot().await.lock().await.clone() + } + + async fn latest_snapshot(&self) -> &Mutex { self.latest_bundle - .get_or_init(|| async { Mutex::new(self.load_startup_bundle_with_timeout().await) }) + .get_or_init(|| async { + let mut snapshot = CloudConfigBundleSnapshot { + bundle: self.load_startup_bundle_with_timeout().await, + binding: None, + }; + self.policy.publish_snapshot(&mut snapshot); + Mutex::new(snapshot) + }) .await - .lock() - .await - .clone() } pub(crate) async fn load_startup_bundle_with_timeout( @@ -326,6 +342,7 @@ where bundle: CloudConfigBundle, ) -> Result, CloudConfigBundleLoadError> { emit_fetch_attempt_metric(trigger, attempt, "success", /*status_code*/ None); + let revision = self.policy.observe_remote_bundle(&bundle); if let Err(err) = validate_bundle(&bundle, &self.codex_home) { emit_fetch_final_metric( trigger, @@ -340,15 +357,19 @@ where let (chatgpt_user_id, account_id) = auth_identity(auth); if self.cache_enabled - && let Err(err) = self - .cache - .save(chatgpt_user_id, account_id, bundle.clone()) - .await + && let Some(revision) = revision { - tracing::warn!( - error = %err, - "Failed to write cloud config bundle cache" - ); + let result = match self + .cache + .prepare(chatgpt_user_id, account_id, bundle.clone()) + .await + { + Ok(staged) => staged.publish_if_current(revision).await, + Err(err) => Err(err), + }; + if let Err(err) = result { + tracing::warn!(error = %err, "Failed to write cloud config bundle cache"); + } } emit_fetch_final_metric( @@ -488,7 +509,7 @@ where let mut refresh_interval = CLOUD_CONFIG_BUNDLE_CACHE_REFRESH_INTERVAL; if let Some(latest_bundle) = self.latest_bundle.get() && matches!( - &*latest_bundle.lock().await, + &latest_bundle.lock().await.bundle, Err(error) if error.code() == CloudConfigBundleLoadErrorCode::Timeout ) { @@ -553,8 +574,15 @@ where return; }; let mut latest = latest.lock().await; - if result.is_ok() || latest.is_err() { - *latest = result; + if result.is_ok() { + *latest = CloudConfigBundleSnapshot { + bundle: result, + binding: None, + }; + self.policy.publish_snapshot(&mut latest); + } else if latest.bundle.is_err() { + latest.bundle = result; + latest.binding = None; } } } diff --git a/codex-rs/cloud-config/src/service_tests.rs b/codex-rs/cloud-config/src/service_tests.rs index e78b60978a..eb3396543b 100644 --- a/codex-rs/cloud-config/src/service_tests.rs +++ b/codex-rs/cloud-config/src/service_tests.rs @@ -11,6 +11,7 @@ use base64::engine::general_purpose::URL_SAFE_NO_PAD; use codex_backend_client::ConfigBundleResponse; use codex_backend_client::DeliveredTomlFragment; use codex_config::AbsolutePathBuf; +use codex_config::CloudConfigBundleBindingStatus; use codex_config::CloudConfigFragment; use codex_config::CloudConfigTomlBundle; use codex_config::CloudRequirementsFragment; @@ -41,6 +42,23 @@ fn create_test_cache(codex_home: &Path) -> CloudConfigBundleCache { CloudConfigBundleCache::new(AbsolutePathBuf::resolve_path_against_base(codex_home, "/")) } +async fn write_test_cache(cache: &CloudConfigBundleCache, bundle: CloudConfigBundle) { + let revision = CloudConfigBundlePolicy::default() + .observe_remote_bundle(&bundle) + .expect("fixture revision"); + cache + .prepare( + Some("user-12345".to_string()), + Some("account-12345".to_string()), + bundle, + ) + .await + .expect("stage test cache") + .publish_if_current(revision) + .await + .expect("write test cache"); +} + async fn auth_manager_with_api_key() -> Arc { let tmp = tempdir().expect("tempdir"); let auth_json = json!({ @@ -220,6 +238,120 @@ fn request_error() -> BundleRequestError { BundleRequestError::Retryable(RetryableFailureKind::Request { status_code: None }) } +#[tokio::test] +async fn superseded_staged_cache_write_cannot_replace_newer_service_bundle() { + let codex_home = tempdir().expect("tempdir"); + let auth_manager = auth_manager_with_plan("business").await; + let auth = auth_manager.auth().await.expect("auth"); + let service = Arc::new(CloudConfigBundleService::new( + auth_manager, + Arc::new(StaticBundleClient::new(test_bundle())), + codex_home.path().to_path_buf(), + Duration::from_secs(10), + )); + let old = test_bundle(); + let new = CloudConfigBundle::default(); + let (staged, staging) = tokio::sync::oneshot::channel(); + let (resume, resumed) = tokio::sync::oneshot::channel(); + let older = Arc::clone(&service); + let pending = tokio::spawn(async move { + let revision = older + .policy + .observe_remote_bundle(&old) + .expect("old revision"); + let output = older + .cache + .prepare( + Some("user-12345".to_string()), + Some("account-12345".to_string()), + old, + ) + .await + .expect("stage older cache"); + staged.send(()).expect("announce staging"); + resumed.await.expect("resume older writer"); + output + .publish_if_current(revision) + .await + .expect("discard older cache"); + }); + staging.await.expect("older cache is staged"); + service + .validate_and_cache_remote_bundle(&auth, "refresh", /*attempt*/ 1, new.clone()) + .await + .expect("publish newer bundle"); + resume.send(()).expect("resume older writer"); + pending.await.expect("older writer completed"); + + assert_eq!( + service + .cache + .load(Some("user-12345"), Some("account-12345")) + .await + .expect("load cache") + .bundle, + new + ); + let files: Vec<_> = std::fs::read_dir(codex_home.path()) + .expect("read cache directory") + .map(|entry| entry.expect("directory entry").path()) + .collect(); + assert_eq!(files, vec![service.cache.path().to_path_buf()]); +} + +#[tokio::test] +async fn retired_policy_discards_staged_cache_and_retains_previous_payload() { + let codex_home = tempdir().expect("tempdir"); + let auth_manager = auth_manager_with_plan("business").await; + let auth = auth_manager.auth().await.expect("auth"); + let service = CloudConfigBundleService::new( + auth_manager, + Arc::new(StaticBundleClient::new(test_bundle())), + codex_home.path().to_path_buf(), + Duration::from_secs(10), + ); + let previous = test_bundle(); + service + .validate_and_cache_remote_bundle(&auth, "refresh", /*attempt*/ 1, previous.clone()) + .await + .expect("publish previous bundle"); + let changed = CloudConfigBundle::default(); + let revision = service + .policy + .observe_remote_bundle(&changed) + .expect("changed revision"); + let output = service + .cache + .prepare( + Some("user-12345".to_string()), + Some("account-12345".to_string()), + changed, + ) + .await + .expect("stage changed cache"); + let loader = codex_config::CloudConfigBundleLoader::new(async { Ok(None) }) + .with_ema_policy_snapshots(service.policy.clone(), || async { + CloudConfigBundleSnapshot { + bundle: Ok(None), + binding: None, + } + }); + loader.retire_ema_policy(); + output + .publish_if_current(revision) + .await + .expect("discard retired cache"); + assert_eq!( + service + .cache + .load(Some("user-12345"), Some("account-12345")) + .await + .expect("load cache") + .bundle, + previous + ); +} + struct StaticBundleClient { bundle: CloudConfigBundle, request_count: AtomicUsize, @@ -558,14 +690,7 @@ async fn invalid_cloud_provider_does_not_replace_cached_bundle() { let codex_home = tempdir().expect("tempdir"); let cache = create_test_cache(codex_home.path()); let previous = test_bundle(); - cache - .save( - Some("user-12345".to_string()), - Some("account-12345".to_string()), - previous.clone(), - ) - .await - .expect("cache valid bundle"); + write_test_cache(&cache, previous.clone()).await; for contents in [ "[model_providers.openai]\nname = 'Reserved'", @@ -606,14 +731,7 @@ async fn invalid_cloud_provider_does_not_replace_cached_bundle() { async fn get_bundle_ignores_invalid_cache_and_refetches() { let codex_home = tempdir().expect("tempdir"); let cache = create_test_cache(codex_home.path()); - cache - .save( - Some("user-12345".to_string()), - Some("account-12345".to_string()), - invalid_config_bundle(), - ) - .await - .expect("write invalid cache"); + write_test_cache(&cache, invalid_config_bundle()).await; let replacement_bundle = test_bundle(); let fetcher = Arc::new(StaticBundleClient::new(replacement_bundle.clone())); let service = CloudConfigBundleService::new( @@ -687,14 +805,7 @@ async fn get_bundle_uses_cache_when_valid() { async fn get_bundle_without_cache_ignores_and_preserves_valid_cache() { let codex_home = tempdir().expect("tempdir"); let cache = create_test_cache(codex_home.path()); - cache - .save( - Some("user-12345".to_string()), - Some("account-12345".to_string()), - CloudConfigBundle::default(), - ) - .await - .expect("write empty cache"); + write_test_cache(&cache, CloudConfigBundle::default()).await; let cached_bytes = std::fs::read(cache.path()).expect("read cache"); let bundle = test_bundle(); let fetcher = Arc::new(StaticBundleClient::new(bundle.clone())); @@ -720,14 +831,7 @@ async fn get_bundle_without_cache_ignores_and_preserves_valid_cache() { #[tokio::test(start_paused = true)] async fn get_bundle_without_cache_fails_closed_on_request_failure() { let codex_home = tempdir().expect("tempdir"); - create_test_cache(codex_home.path()) - .save( - Some("user-12345".to_string()), - Some("account-12345".to_string()), - test_bundle(), - ) - .await - .expect("write valid cache"); + write_test_cache(&create_test_cache(codex_home.path()), test_bundle()).await; let fetcher = Arc::new(SequenceBundleClient::new(vec![ Err(request_error()); CLOUD_CONFIG_BUNDLE_MAX_ATTEMPTS @@ -1245,6 +1349,7 @@ async fn production_loader_recovers_startup_timeouts_in_background() { tokio::task::yield_now().await; } let config = ConfigBuilder::default() + .loader_overrides(codex_config::LoaderOverrides::without_managed_config_for_tests()) .codex_home(codex_home.path().to_path_buf()) .cloud_config_bundle(loader.clone()) .build() @@ -1474,14 +1579,7 @@ async fn dropping_loader_cancels_in_flight_startup_and_refresh() { for starts_from_cache in [false, true] { let codex_home = tempdir().expect("tempdir"); if starts_from_cache { - create_test_cache(codex_home.path()) - .save( - Some("user-12345".to_string()), - Some("account-12345".to_string()), - test_bundle(), - ) - .await - .expect("write initial cache"); + write_test_cache(&create_test_cache(codex_home.path()), test_bundle()).await; } let request_started = Arc::new(tokio::sync::Notify::new()); let request_cancelled = Arc::new(tokio::sync::Notify::new()); @@ -1513,7 +1611,14 @@ async fn dropping_loader_cancels_in_flight_startup_and_refresh() { #[tokio::test(start_paused = true)] async fn refresh_can_clear_preserve_and_restore_the_latest_bundle() { let codex_home = tempdir().expect("tempdir"); - let mut responses = vec![Ok(test_bundle()), Ok(CloudConfigBundle::default())]; + let mut invalid = test_bundle(); + invalid.requirements_toml.enterprise_managed[0].contents = "invalid = [".to_string(); + let mut responses = vec![ + Ok(test_bundle()), + Ok(invalid), + Ok(test_bundle()), + Ok(CloudConfigBundle::default()), + ]; responses.extend(vec![Err(request_error()); CLOUD_CONFIG_BUNDLE_MAX_ATTEMPTS]); responses.push(Ok(test_bundle())); let service = Arc::new(CloudConfigBundleService::new( @@ -1524,8 +1629,42 @@ async fn refresh_can_clear_preserve_and_restore_the_latest_bundle() { )); assert_eq!(service.get_latest().await, Ok(Some(test_bundle()))); + let initial = service + .get_latest_snapshot() + .await + .binding + .expect("initial binding"); + assert!(service.refresh_cache_once().await); + assert_eq!(service.get_latest().await, Ok(Some(test_bundle()))); + assert_eq!( + initial.read().status, + CloudConfigBundleBindingStatus::Suspended + ); + + assert!(service.refresh_cache_once().await); + assert_eq!(service.get_latest().await, Ok(Some(test_bundle()))); + let recovered = service + .get_latest_snapshot() + .await + .binding + .expect("recovered binding"); + assert_eq!(initial.read().status, CloudConfigBundleBindingStatus::Stale); + assert_eq!( + recovered.read().status, + CloudConfigBundleBindingStatus::Current + ); + assert!(service.refresh_cache_once().await); assert_eq!(service.get_latest().await, Ok(None)); + let cleared = service + .get_latest_snapshot() + .await + .binding + .expect("cleared binding"); + assert_eq!( + recovered.read().status, + CloudConfigBundleBindingStatus::Stale + ); let refresh_service = Arc::clone(&service); let refresh = tokio::spawn(async move { refresh_service.refresh_cache_once().await }); @@ -1534,6 +1673,10 @@ async fn refresh_can_clear_preserve_and_restore_the_latest_bundle() { tokio::task::yield_now().await; assert!(refresh.await.expect("failed refresh task")); assert_eq!(service.get_latest().await, Ok(None)); + assert_eq!( + cleared.read().status, + CloudConfigBundleBindingStatus::Current + ); assert!(service.refresh_cache_once().await); assert_eq!(service.get_latest().await, Ok(Some(test_bundle()))); diff --git a/codex-rs/config/src/cloud_config_bundle.rs b/codex-rs/config/src/cloud_config_bundle.rs index e8e7950a1c..3a8898dd4f 100644 --- a/codex-rs/config/src/cloud_config_bundle.rs +++ b/codex-rs/config/src/cloud_config_bundle.rs @@ -20,6 +20,7 @@ use std::fmt; use std::future::Future; use std::sync::Arc; use thiserror::Error; +use tokio::sync::watch; #[derive(Clone, Debug, Default, Deserialize, Eq, PartialEq, Serialize)] pub struct CloudConfigBundle { @@ -180,14 +181,202 @@ impl CloudConfigBundleLoadError { } } +type CloudConfigBundleGetter = dyn Fn() -> BoxFuture<'static, Result, CloudConfigBundleLoadError>> + + Send + + Sync; + +/// Tracks which delivered policy may authorize new enterprise MCP requests. +#[derive(Clone)] +pub struct CloudConfigBundlePolicy(watch::Sender); + +struct CloudConfigBundlePolicyState { + revision: u64, + bundle: Option, + phase: CloudConfigBundlePolicyPhase, +} + +#[derive(Clone, Copy, Eq, PartialEq)] +enum CloudConfigBundlePolicyPhase { + Uninitialized, + Suspended, + Active, + Retired, +} + +impl Default for CloudConfigBundlePolicy { + fn default() -> Self { + Self( + watch::channel(CloudConfigBundlePolicyState { + revision: 0, + bundle: None, + phase: CloudConfigBundlePolicyPhase::Uninitialized, + }) + .0, + ) + } +} + +impl fmt::Debug for CloudConfigBundlePolicy { + fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { + f.debug_struct("CloudConfigBundlePolicy").finish() + } +} + +impl CloudConfigBundlePolicy { + /// Suspends old EMA admission before a changed delivered bundle is validated or cached. + pub fn observe_remote_bundle( + &self, + bundle: &CloudConfigBundle, + ) -> Option { + let bundle = (!bundle.is_empty()).then(|| bundle.clone()); + let mut observed_revision = None; + self.0.send_if_modified(|state| { + if state.phase == CloudConfigBundlePolicyPhase::Retired { + return false; + } + if state.phase != CloudConfigBundlePolicyPhase::Uninitialized && state.bundle == bundle + { + observed_revision = Some(state.revision); + return false; + } + if state.phase != CloudConfigBundlePolicyPhase::Uninitialized { + let Some(revision) = state.revision.checked_add(1) else { + state.phase = CloudConfigBundlePolicyPhase::Retired; + return true; + }; + state.revision = revision; + } + state.bundle = bundle; + state.phase = CloudConfigBundlePolicyPhase::Suspended; + observed_revision = Some(state.revision); + true + }); + observed_revision.map(|revision| CloudConfigBundlePolicyRevision { + revision, + policy: self.clone(), + }) + } + + /// Pairs a service result with its EMA revision before the snapshot becomes visible. + pub fn publish_snapshot(&self, snapshot: &mut CloudConfigBundleSnapshot) { + let Ok(bundle) = &snapshot.bundle else { + return; + }; + self.0.send_if_modified(|state| { + let current = state.phase != CloudConfigBundlePolicyPhase::Retired + && (state.phase == CloudConfigBundlePolicyPhase::Uninitialized + || state.bundle == *bundle); + snapshot.binding = Some(CloudConfigBundleBinding { + revision: current.then_some(state.revision), + policy: self.clone(), + }); + if current && state.phase != CloudConfigBundlePolicyPhase::Active { + state.bundle = bundle.clone(); + state.phase = CloudConfigBundlePolicyPhase::Active; + true + } else { + false + } + }); + } + + fn retire(&self) { + self.0 + .send_modify(|state| state.phase = CloudConfigBundlePolicyPhase::Retired); + } +} + +/// An observed remote policy revision permitted to commit staged persistence. +pub struct CloudConfigBundlePolicyRevision { + revision: u64, + policy: CloudConfigBundlePolicy, +} + +impl CloudConfigBundlePolicyRevision { + /// Executes a synchronous commit only while this observation is still current. + /// Prepare expensive work first; the guard blocks newer observations and retirement. + pub fn commit_if_current(self, commit: impl FnOnce() -> Result<(), E>) -> Result<(), E> { + let state = self.policy.0.borrow(); + if state.revision == self.revision + && matches!( + state.phase, + CloudConfigBundlePolicyPhase::Suspended | CloudConfigBundlePolicyPhase::Active + ) + { + commit() + } else { + Ok(()) + } + } +} + +/// The policy revision used to resolve one set of enterprise MCP inputs. +#[derive(Clone, Debug)] +pub struct CloudConfigBundleBinding { + revision: Option, + policy: CloudConfigBundlePolicy, +} + +#[derive(Clone, Copy, Debug, Eq, PartialEq)] +pub enum CloudConfigBundleBindingStatus { + Current, + Stale, + Suspended, +} + +impl CloudConfigBundleBinding { + /// Keeps the policy observation stable until this guard is dropped. + pub fn read(&self) -> CloudConfigBundleBindingGuard<'_> { + let state = self.policy.0.borrow(); + let status = match (state.phase, self.revision) { + (CloudConfigBundlePolicyPhase::Active, Some(expected)) + if state.revision == expected => + { + CloudConfigBundleBindingStatus::Current + } + (CloudConfigBundlePolicyPhase::Active, Some(_)) => { + CloudConfigBundleBindingStatus::Stale + } + _ => CloudConfigBundleBindingStatus::Suspended, + }; + CloudConfigBundleBindingGuard { + _state: state, + status, + } + } +} + +pub struct CloudConfigBundleBindingGuard<'a> { + _state: watch::Ref<'a, CloudConfigBundlePolicyState>, + pub status: CloudConfigBundleBindingStatus, +} + +pub struct CloudConfigBundlePolicyChanges(watch::Receiver); + +impl CloudConfigBundlePolicyChanges { + pub fn is_active(&self) -> bool { + self.0.borrow().phase == CloudConfigBundlePolicyPhase::Active + } + + pub async fn changed(&mut self) -> Result<(), watch::error::RecvError> { + self.0.changed().await + } +} + +#[derive(Clone)] +pub struct CloudConfigBundleSnapshot { + pub bundle: Result, CloudConfigBundleLoadError>, + pub binding: Option, +} + +type CloudConfigBundleSnapshotGetter = + dyn Fn() -> BoxFuture<'static, CloudConfigBundleSnapshot> + Send + Sync; + #[derive(Clone)] pub struct CloudConfigBundleLoader { - getter: Arc< - dyn Fn() - -> BoxFuture<'static, Result, CloudConfigBundleLoadError>> - + Send - + Sync, - >, + getter: Arc, + snapshot_getter: Option>, + ema_policy: Option, } impl CloudConfigBundleLoader { @@ -211,6 +400,59 @@ impl CloudConfigBundleLoader { { Self { getter: Arc::new(move || getter().boxed()), + snapshot_getter: None, + ema_policy: None, + } + } + + /// Freezes the delivered bundle while retaining its live policy revision binding. + pub fn from_snapshot(snapshot: CloudConfigBundleSnapshot) -> Self { + let bundle = snapshot.bundle.clone(); + let mut loader = Self::from_getter(move || std::future::ready(bundle.clone())); + loader.ema_policy = snapshot + .binding + .as_ref() + .map(|binding| binding.policy.clone()); + loader.snapshot_getter = Some(Arc::new(move || { + std::future::ready(snapshot.clone()).boxed() + })); + loader + } + + pub fn with_ema_policy_snapshots( + mut self, + policy: CloudConfigBundlePolicy, + getter: F, + ) -> Self + where + F: Fn() -> Fut + Send + Sync + 'static, + Fut: Future + Send + 'static, + { + self.ema_policy = Some(policy); + self.snapshot_getter = Some(Arc::new(move || getter().boxed())); + self + } + + pub async fn get_snapshot(&self) -> CloudConfigBundleSnapshot { + match &self.snapshot_getter { + Some(getter) => getter().await, + None => CloudConfigBundleSnapshot { + bundle: self.get().await, + binding: None, + }, + } + } + + pub fn ema_policy_changes(&self) -> Option { + self.ema_policy + .as_ref() + .map(|policy| CloudConfigBundlePolicyChanges(policy.0.subscribe())) + } + + /// Stops new EMA admission from this owner without changing generic loader behavior. + pub fn retire_ema_policy(&self) { + if let Some(policy) = &self.ema_policy { + policy.retire(); } } diff --git a/codex-rs/config/src/lib.rs b/codex-rs/config/src/lib.rs index 34ce7687d7..0893930ba0 100644 --- a/codex-rs/config/src/lib.rs +++ b/codex-rs/config/src/lib.rs @@ -64,10 +64,17 @@ pub use browser_computer_use_requirements::ComputerUseWindowsRequirementsToml; pub use browser_use::BrowserUseConfigToml; pub use browser_use::BrowserUseOriginPolicyConfigToml; pub use cloud_config_bundle::CloudConfigBundle; +pub use cloud_config_bundle::CloudConfigBundleBinding; +pub use cloud_config_bundle::CloudConfigBundleBindingGuard; +pub use cloud_config_bundle::CloudConfigBundleBindingStatus; pub use cloud_config_bundle::CloudConfigBundleLayers; pub use cloud_config_bundle::CloudConfigBundleLoadError; pub use cloud_config_bundle::CloudConfigBundleLoadErrorCode; pub use cloud_config_bundle::CloudConfigBundleLoader; +pub use cloud_config_bundle::CloudConfigBundlePolicy; +pub use cloud_config_bundle::CloudConfigBundlePolicyChanges; +pub use cloud_config_bundle::CloudConfigBundlePolicyRevision; +pub use cloud_config_bundle::CloudConfigBundleSnapshot; pub use cloud_config_bundle::CloudConfigTomlBundle; pub use cloud_config_bundle::CloudRequirementsFragment; pub use cloud_config_bundle::CloudRequirementsTomlBundle; diff --git a/codex-rs/config/src/loader/mod.rs b/codex-rs/config/src/loader/mod.rs index 456436df40..4c2d851041 100644 --- a/codex-rs/config/src/loader/mod.rs +++ b/codex-rs/config/src/loader/mod.rs @@ -205,22 +205,25 @@ pub async fn load_config_layers_state( overrides.ignore_user_and_project_exec_policy_rules; let mut bundle_requirements_layers = Vec::new(); let mut cloud_config_layers = Vec::new(); + let mut cloud_config_binding = None; - if !overrides.ignore_managed_requirements - && let Some(bundle) = cloud_config_bundle.get().await.map_err(io::Error::other)? - { - let cloud_config_base_dir = AbsolutePathBuf::from_absolute_path(codex_home)?; - let bundle_layers = if strict_config { - CloudConfigBundleLayers::from_bundle_strict_config(bundle, &cloud_config_base_dir)? - } else { - CloudConfigBundleLayers::from_bundle(bundle, &cloud_config_base_dir)? - }; - let CloudConfigBundleLayers { - enterprise_managed_config, - enterprise_managed_requirements, - } = bundle_layers; - bundle_requirements_layers = enterprise_managed_requirements; - cloud_config_layers = enterprise_managed_config; + if !overrides.ignore_managed_requirements { + let snapshot = cloud_config_bundle.get_snapshot().await; + cloud_config_binding = snapshot.binding; + if let Some(bundle) = snapshot.bundle.map_err(io::Error::other)? { + let cloud_config_base_dir = AbsolutePathBuf::from_absolute_path(codex_home)?; + let bundle_layers = if strict_config { + CloudConfigBundleLayers::from_bundle_strict_config(bundle, &cloud_config_base_dir)? + } else { + CloudConfigBundleLayers::from_bundle(bundle, &cloud_config_base_dir)? + }; + let CloudConfigBundleLayers { + enterprise_managed_config, + enterprise_managed_requirements, + } = bundle_layers; + bundle_requirements_layers = enterprise_managed_requirements; + cloud_config_layers = enterprise_managed_config; + } } let (config_requirements_toml, loaded_config_layers, requirements_layers) = @@ -491,6 +494,7 @@ pub async fn load_config_layers_state( config_requirements_toml.clone().try_into()?, config_requirements_toml.into_toml(), )? + .with_cloud_config_binding(cloud_config_binding) .with_user_and_project_exec_policy_rules_ignored(ignore_user_and_project_exec_policy_rules); config_layer_stack.is_projectless = is_projectless; startup_warnings.extend(ignored_config_warning( diff --git a/codex-rs/config/src/state.rs b/codex-rs/config/src/state.rs index f635eff83d..ed3d13ccce 100644 --- a/codex-rs/config/src/state.rs +++ b/codex-rs/config/src/state.rs @@ -243,8 +243,10 @@ impl ConfigLayerEntry { } } -#[derive(Debug, Clone, Default, PartialEq)] +#[derive(Debug, Clone, Default)] pub struct ConfigLayerStack { + /// Runtime-only EMA authority; it does not change the configuration's contents. + cloud_config_binding: Option, /// Cached TOML projection derived only from `requirements_toml`. /// Construction validates provider definitions and reports serialization errors, /// so `effective_config()` can replace complete entries without a fallible conversion. @@ -273,6 +275,29 @@ pub struct ConfigLayerStack { pub(crate) is_projectless: bool, } +impl PartialEq for ConfigLayerStack { + fn eq(&self, other: &Self) -> bool { + let Self { + cloud_config_binding: _, + model_provider_requirements, + layers, + requirements, + requirements_toml, + ignore_user_and_project_exec_policy_rules, + startup_warnings, + is_projectless, + } = self; + model_provider_requirements == &other.model_provider_requirements + && layers == &other.layers + && requirements == &other.requirements + && requirements_toml == &other.requirements_toml + && *ignore_user_and_project_exec_policy_rules + == other.ignore_user_and_project_exec_policy_rules + && startup_warnings == &other.startup_warnings + && is_projectless == &other.is_projectless + } +} + impl ConfigLayerStack { pub fn new( layers: Vec, @@ -285,6 +310,7 @@ impl ConfigLayerStack { &requirements_toml, )?); Ok(Self { + cloud_config_binding: None, model_provider_requirements, layers, requirements, @@ -307,6 +333,18 @@ impl ConfigLayerStack { self.ignore_user_and_project_exec_policy_rules } + pub fn with_cloud_config_binding( + mut self, + binding: Option, + ) -> Self { + self.cloud_config_binding = binding; + self + } + + pub fn cloud_config_binding(&self) -> Option<&crate::CloudConfigBundleBinding> { + self.cloud_config_binding.as_ref() + } + /// 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 { @@ -437,6 +475,7 @@ impl ConfigLayerStack { } Ok(Self { layers, + cloud_config_binding: self.cloud_config_binding.clone(), model_provider_requirements: self.model_provider_requirements.clone(), requirements: self.requirements.clone(), requirements_toml: self.requirements_toml.clone(), @@ -473,6 +512,7 @@ impl ConfigLayerStack { } Self { layers, + cloud_config_binding: self.cloud_config_binding.clone(), model_provider_requirements: self.model_provider_requirements.clone(), requirements: self.requirements.clone(), requirements_toml: self.requirements_toml.clone(), diff --git a/codex-rs/core/src/agent/role.rs b/codex-rs/core/src/agent/role.rs index f6633e7590..494ec0d30c 100644 --- a/codex-rs/core/src/agent/role.rs +++ b/codex-rs/core/src/agent/role.rs @@ -253,6 +253,7 @@ mod role_overrides { config.config_layer_stack.requirements().clone(), config.config_layer_stack.requirements_toml().clone(), )? + .with_cloud_config_binding(config.config_layer_stack.cloud_config_binding().cloned()) .with_user_and_project_exec_policy_rules_ignored( config .config_layer_stack diff --git a/codex-rs/core/src/agent/role_tests.rs b/codex-rs/core/src/agent/role_tests.rs index fad4e30459..bcf8892aa6 100644 --- a/codex-rs/core/src/agent/role_tests.rs +++ b/codex-rs/core/src/agent/role_tests.rs @@ -2,6 +2,9 @@ use super::*; use crate::config::ConfigBuilder; use crate::plugins::plugins_manager_for_config; use crate::skills_load_input_from_config; +use codex_config::CloudConfigBundleBindingStatus; +use codex_config::CloudConfigBundlePolicy; +use codex_config::CloudConfigBundleSnapshot; use codex_config::test_support::CloudConfigBundleFixture; use codex_login::test_support::auth_manager_from_optional_auth; use codex_protocol::config_types::ServiceTier; @@ -432,6 +435,15 @@ writable_roots = ["./sandbox-root"] #[tokio::test] async fn apply_role_cannot_expand_parent_authority() { let (home, mut config) = test_config_with_cli_overrides(Vec::new()).await; + let policy = CloudConfigBundlePolicy::default(); + let mut snapshot = CloudConfigBundleSnapshot { + bundle: Ok(None), + binding: None, + }; + policy.publish_snapshot(&mut snapshot); + config.config_layer_stack = config + .config_layer_stack + .with_cloud_config_binding(snapshot.binding); config.notify = Some(vec!["parent-notifier".to_string()]); for feature in [Feature::MemoryTool, Feature::RequestPermissionsTool] { config @@ -480,6 +492,18 @@ command = "attacker-command" .await .expect("custom role should apply"); + policy.observe_remote_bundle( + &CloudConfigBundleFixture::enterprise_config("model = 'managed'").into_bundle(), + ); + assert_eq!( + config + .config_layer_stack + .cloud_config_binding() + .expect("role should preserve managed policy binding") + .read() + .status, + CloudConfigBundleBindingStatus::Suspended + ); assert_eq!( config.developer_instructions.as_deref(), Some("Stay focused") diff --git a/codex-rs/core/src/config/config_loader_tests.rs b/codex-rs/core/src/config/config_loader_tests.rs index d50816a84d..39df96b292 100644 --- a/codex-rs/core/src/config/config_loader_tests.rs +++ b/codex-rs/core/src/config/config_loader_tests.rs @@ -1769,6 +1769,7 @@ async fn load_config_layers_includes_cloud_config_bundle() -> anyhow::Result<()> Some(cwd), &[] as &[(String, TomlValue)], ConfigLoadOptions { + loader_overrides: LoaderOverrides::without_managed_config_for_tests(), cloud_config_bundle, ..Default::default() }, diff --git a/codex-rs/core/src/config/config_tests.rs b/codex-rs/core/src/config/config_tests.rs index b2a2dc45e8..4fb1e74cc9 100644 --- a/codex-rs/core/src/config/config_tests.rs +++ b/codex-rs/core/src/config/config_tests.rs @@ -5532,21 +5532,6 @@ async fn rebuild_with_session_layers_refreshes_requirements() -> std::io::Result requirements_toml, ) .map_err(std::io::Error::other)?; - let refreshed_toml = refreshed_layer_stack - .effective_config() - .try_into() - .map_err(|err| std::io::Error::new(std::io::ErrorKind::InvalidData, err))?; - let refreshed_config = Config::load_config_with_layer_stack( - LOCAL_FS.as_ref(), - refreshed_toml, - ConfigOverrides { - cwd: Some(codex_home.path().to_path_buf()), - ..Default::default() - }, - codex_home.abs(), - refreshed_layer_stack, - ) - .await?; let thread_layer_stack = ConfigLayerStack::new( vec![ ConfigLayerEntry::new( @@ -5614,13 +5599,12 @@ async fn rebuild_with_session_layers_refreshes_requirements() -> std::io::Result thread_layer_stack, ) .await?; - let zsh_path = refreshed_config.zsh_path.clone(); let config = Config::rebuild_with_session_layers( &thread_config.config_layer_stack, thread_config.cwd.to_path_buf(), - &refreshed_config.config_layer_stack, - refreshed_config.codex_home.clone(), - zsh_path.map(AbsolutePathBuf::try_from).transpose()?, + &refreshed_layer_stack, + codex_home.abs(), + /*default_zsh_path*/ None, ) .await?; @@ -5702,17 +5686,6 @@ async fn rebuild_with_session_layers_refreshes_plugin_derived_mcp_config() -> an Default::default(), Default::default(), )?; - let refreshed_config = Config::load_config_with_layer_stack( - LOCAL_FS.as_ref(), - refreshed_layer_stack.effective_config().try_into()?, - ConfigOverrides { - cwd: Some(codex_home.path().to_path_buf()), - ..Default::default() - }, - codex_home.abs(), - refreshed_layer_stack, - ) - .await?; let thread_layer_stack = ConfigLayerStack::new( vec![ConfigLayerEntry::new( ConfigLayerSource::User { @@ -5742,13 +5715,12 @@ async fn rebuild_with_session_layers_refreshes_plugin_derived_mcp_config() -> an thread_layer_stack, ) .await?; - let zsh_path = refreshed_config.zsh_path.clone(); let config = Config::rebuild_with_session_layers( &thread_config.config_layer_stack, thread_config.cwd.to_path_buf(), - &refreshed_config.config_layer_stack, - refreshed_config.codex_home.clone(), - zsh_path.map(AbsolutePathBuf::try_from).transpose()?, + &refreshed_layer_stack, + codex_home.abs(), + /*default_zsh_path*/ None, ) .await?; let plugins_manager = @@ -11577,6 +11549,7 @@ use_xaa = true assert!(config.features.enabled(Feature::ViewImage)); assert!(!config.features.enabled(Feature::ShellTool)); + assert!(config.features.enabled(Feature::UseXaa)); assert!( !config .startup_warnings @@ -11616,6 +11589,7 @@ use_xaa = false assert!(!config.features.enabled(Feature::UnifiedExec)); assert!(config.features.enabled(Feature::ShellTool)); assert!(!config.features.enabled(Feature::UnifiedExecZshFork)); + assert!(!config.features.enabled(Feature::UseXaa)); assert!( !config .startup_warnings diff --git a/codex-rs/core/src/config/mod.rs b/codex-rs/core/src/config/mod.rs index 7f30f7bd2b..57f8140720 100644 --- a/codex-rs/core/src/config/mod.rs +++ b/codex-rs/core/src/config/mod.rs @@ -1118,6 +1118,9 @@ pub struct Config { /// Local rollout preference after checking network restrictions and native availability. pub prefer_mxc: bool, + /// Host feature defaults retained beneath explicit configuration overrides. + pub runtime_feature_defaults: BTreeMap, + /// When `true`, suppress warnings about unstable (under development) features. pub suppress_unstable_features_warning: bool, @@ -1532,30 +1535,7 @@ impl ConfigBuilder { .unwrap_or(&codex_config::NoopThreadConfigLoader), ) .await?; - let merged_toml = config_layer_stack.effective_config(); - - // Note that each layer in ConfigLayerStack should have resolved - // relative paths to absolute paths based on the parent folder of the - // respective config file, so we should be safe to deserialize without - // AbsolutePathBufGuard here. - let config_toml: ConfigToml = match merged_toml.try_into() { - Ok(config_toml) => config_toml, - Err(err) => { - if let Some(config_error) = codex_config::first_layer_config_error::( - &config_layer_stack, - codex_config::CONFIG_TOML_FILE, - ) - .await - { - return Err(codex_config::io_error_from_config_error( - std::io::ErrorKind::InvalidData, - config_error, - Some(err), - )); - } - return Err(std::io::Error::new(std::io::ErrorKind::InvalidData, err)); - } - }; + let config_toml = config_toml_from_layers(&config_layer_stack).await?; Config::load_config_with_layer_stack( LOCAL_FS.as_ref(), config_toml, @@ -1572,6 +1552,28 @@ impl ConfigBuilder { } } +async fn config_toml_from_layers(layers: &ConfigLayerStack) -> std::io::Result { + // The loader resolves paths relative to each layer's file before deserialization. + match layers.effective_config().try_into() { + Ok(config_toml) => Ok(config_toml), + Err(err) => { + if let Some(config_error) = codex_config::first_layer_config_error::( + layers, + codex_config::CONFIG_TOML_FILE, + ) + .await + { + return Err(codex_config::io_error_from_config_error( + std::io::ErrorKind::InvalidData, + config_error, + Some(err), + )); + } + Err(std::io::Error::new(std::io::ErrorKind::InvalidData, err)) + } + } +} + impl Config { pub fn sqlite_config(&self) -> &codex_state::SqliteConfig { &self.sqlite @@ -1914,10 +1916,7 @@ impl Config { ) -> std::io::Result { let config_layer_stack = Self::layer_stack_preserving_session(session_layers, refreshed_layers)?; - let cfg: ConfigToml = config_layer_stack - .effective_config() - .try_into() - .map_err(|err| std::io::Error::new(std::io::ErrorKind::InvalidData, err))?; + let cfg = config_toml_from_layers(&config_layer_stack).await?; Self::load_config_with_layer_stack( LOCAL_FS.as_ref(), cfg, @@ -1954,6 +1953,7 @@ impl Config { refreshed_layers.requirements().clone(), refreshed_layers.requirements_toml().clone(), )? + .with_cloud_config_binding(refreshed_layers.cloud_config_binding().cloned()) .with_user_and_project_exec_policy_rules_ignored( refreshed_layers.ignore_user_and_project_exec_policy_rules(), )) @@ -4471,6 +4471,7 @@ impl Config { current_time_reminder, sleep_tool_mode, features, + runtime_feature_defaults: BTreeMap::new(), suppress_unstable_features_warning: cfg .suppress_unstable_features_warning .unwrap_or(false), diff --git a/codex-rs/core/src/config/runtime_refresh.rs b/codex-rs/core/src/config/runtime_refresh.rs index 0d819cd849..995117dedf 100644 --- a/codex-rs/core/src/config/runtime_refresh.rs +++ b/codex-rs/core/src/config/runtime_refresh.rs @@ -23,6 +23,7 @@ mod tests; pub(crate) enum RuntimeConfigRefresh { User, Mcp, + UserFiles, } #[derive(Default, Deserialize)] @@ -38,7 +39,7 @@ impl Config { scope: RuntimeConfigRefresh, ) -> std::io::Result { let layers = match scope { - RuntimeConfigRefresh::User => self + RuntimeConfigRefresh::User | RuntimeConfigRefresh::UserFiles => self .config_layer_stack .with_user_layer_from(&incoming.config_layer_stack), RuntimeConfigRefresh::Mcp => Self::layer_stack_preserving_session( @@ -59,14 +60,28 @@ impl Config { FeatureConfigSource::default(), FeatureOverrides::default(), ); - // These existing operational controls also accept host runtime overrides. + // These operational controls accept host defaults beneath explicit layer values. // 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)); + if matches!(scope, RuntimeConfigRefresh::UserFiles) { + let explicitly_configured = cfg.features.as_ref().is_some_and(|features| { + features + .entries() + .keys() + .any(|key| codex_features::feature_for_key(key) == Some(feature)) + }); + if !explicitly_configured + && let Some(enabled) = self.runtime_feature_defaults.get(&feature) + { + configured_features.set_enabled(feature, *enabled); + } + } else { + // 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, @@ -95,6 +110,11 @@ impl Config { .map(|(name, server)| (name.clone(), server.clone())), ); let mut config = self.clone(); + if !matches!(scope, RuntimeConfigRefresh::UserFiles) { + config.runtime_feature_defaults = incoming.runtime_feature_defaults.clone(); + config.mcp_optional_startup_grace = incoming.mcp_optional_startup_grace; + config.mcp_oauth_credentials_store_mode = incoming.mcp_oauth_credentials_store_mode; + } config .features .refresh_mcp_features(&features) @@ -138,10 +158,10 @@ impl Config { .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) { + if matches!(scope, RuntimeConfigRefresh::User) { config.active_project = incoming.active_project.clone(); + } + if !matches!(scope, RuntimeConfigRefresh::Mcp) { config.tool_suggest = resolve_tool_suggest_config_from_layer_stack(&layers); } config.config_layer_stack = layers; diff --git a/codex-rs/core/src/config/runtime_refresh_tests.rs b/codex-rs/core/src/config/runtime_refresh_tests.rs index add31508e2..c52c89348c 100644 --- a/codex-rs/core/src/config/runtime_refresh_tests.rs +++ b/codex-rs/core/src/config/runtime_refresh_tests.rs @@ -106,6 +106,7 @@ async fn ordinary_transport_transitions_use_the_resolved_incoming_map() { layered_config(&base, command, "", "", "").await, layered_config(&base, "", url, "", "").await, ), + RuntimeConfigRefresh::UserFiles => unreachable!(), }; let refreshed = current.resolve_runtime_refresh(&incoming, scope).unwrap(); assert_eq!(refreshed.mcp_servers.get(), incoming.mcp_servers.get()); diff --git a/codex-rs/core/src/session/config_refresh.rs b/codex-rs/core/src/session/config_refresh.rs index 410721925b..0fe59df6c6 100644 --- a/codex-rs/core/src/session/config_refresh.rs +++ b/codex-rs/core/src/session/config_refresh.rs @@ -7,6 +7,7 @@ use crate::config::Config; use crate::config::Constrained; use crate::config::RuntimeConfigRefresh; use crate::session::Session; +use codex_config::CloudConfigBundleBindingStatus; use codex_config::McpServerAuth; use codex_features::Feature; use tracing::warn; @@ -51,7 +52,7 @@ impl Session { next_config: Config, scope: RuntimeConfigRefresh, ) -> ConfigRefreshOutcome { - let (mut config, outcome, rejection_kind) = + let (mut config, mut outcome, rejection_kind) = match expected_config.resolve_runtime_refresh(&next_config, scope) { Ok(config) => (config, ConfigRefreshOutcome::Published, None), Err(error) => { @@ -90,18 +91,20 @@ impl Session { (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}"); + if !matches!(scope, RuntimeConfigRefresh::UserFiles) { + 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 mut config = Arc::new(config); let notify_contributors = !matches!(scope, RuntimeConfigRefresh::Mcp) && !self.services.extensions.config_contributors().is_empty(); let (previous_config, new_config) = { @@ -113,12 +116,41 @@ impl Session { ) { return ConfigRefreshOutcome::Stale; } + // User-only refreshes retain the current managed inputs and their + // admission binding. Only MCP refreshes can publish a new policy. + let policy_guard = matches!(scope, RuntimeConfigRefresh::Mcp) + .then(|| next_config.config_layer_stack.cloud_config_binding()) + .flatten() + .map(|binding| binding.read()); + let policy_status = policy_guard + .as_ref() + .map(|guard| guard.status) + .unwrap_or(CloudConfigBundleBindingStatus::Current); + match policy_status { + CloudConfigBundleBindingStatus::Current => {} + CloudConfigBundleBindingStatus::Stale => return ConfigRefreshOutcome::Stale, + CloudConfigBundleBindingStatus::Suspended => { + // Publish ordinary changes under the last valid requirements, + // but revoke enterprise authority until a valid policy is loaded. + let config = Arc::make_mut(&mut config); + config.disable_mcp_enterprise_auth(); + config.config_layer_stack = config + .config_layer_stack + .clone() + .with_cloud_config_binding(/*binding*/ None); + outcome = ConfigRefreshOutcome::Rejected; + } + } if let Some(rejection_kind) = rejection_kind { warn!( ?rejection_kind, "disabling enterprise MCP after rejected configuration refresh" ); } + if matches!(scope, RuntimeConfigRefresh::UserFiles) { + self.services.skills_service.clear_cache(); + self.services.plugins_manager.clear_cache(); + } // 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) { @@ -129,6 +161,7 @@ impl Session { 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); + drop(policy_guard); if matches!(scope, RuntimeConfigRefresh::Mcp) { self.services.mcp_runtime.invalidate_resource_caches(); } diff --git a/codex-rs/core/src/session/config_refresh_tests.rs b/codex-rs/core/src/session/config_refresh_tests.rs new file mode 100644 index 0000000000..72663109c3 --- /dev/null +++ b/codex-rs/core/src/session/config_refresh_tests.rs @@ -0,0 +1,129 @@ +//! File-reload integration regressions; pure authority resolution is tested with Config. + +use super::*; +use pretty_assertions::assert_eq; + +async fn runtime_servers(session: &Session) -> HashMap { + let config = session.get_config().await; + codex_mcp::configured_mcp_servers( + &config + .to_mcp_config(session.services.plugins_manager.as_ref()) + .await, + ) +} + +#[test_case::test_case(true; "host enabled")] +#[test_case::test_case(false; "host disabled")] +#[tokio::test] +async fn file_reload_restores_host_feature_defaults_after_override_removal(host_default: bool) { + let (session, _) = make_session_and_context().await; + let home = session.codex_home().await; + let file = home.join(CONFIG_TOML_FILE); + std::fs::create_dir_all(&home).unwrap(); + std::fs::write(&file, "").unwrap(); + let mut initial = ConfigBuilder::without_managed_config_for_tests() + .codex_home(home.to_path_buf()) + .build() + .await + .unwrap(); + let features = [ + Feature::SecretAuthStorage, + Feature::McpOAuthRefreshCoordination, + ]; + initial.runtime_feature_defaults = features + .into_iter() + .map(|feature| (feature, host_default)) + .collect(); + for feature in features { + initial.features.set_enabled(feature, host_default).unwrap(); + } + session + .state + .lock() + .await + .session_configuration + .original_config_do_not_use = Arc::new(initial); + + for override_value in [None, Some(true), Some(false), None] { + let contents = override_value.map_or_else(String::new, |enabled| { + format!("[features]\nsecret_auth_storage = {enabled}\nmcp_oauth_refresh_coordination = {enabled}\n") + }); + std::fs::write(&file, contents).unwrap(); + session.reload_user_config_layer().await; + let config = session.get_config().await; + assert_eq!( + features.map(|feature| config.features.enabled(feature)), + [override_value.unwrap_or(host_default); 2] + ); + } +} + +#[test_case::test_case("config.toml"; "user config")] +#[test_case::test_case("profiles/work.toml"; "profile config")] +#[tokio::test] +async fn reload_user_config_layer_resolves_paths_without_resetting_runtime_settings( + config_file: &str, +) { + let (session, _) = make_session_and_context().await; + let home = session.codex_home().await; + let file = home.join(config_file); + std::fs::create_dir_all(file.parent().unwrap()).unwrap(); + let ordinary = + "[mcp_servers.ordinary]\nurl = 'https://ordinary.example/mcp'\nenabled = false\n"; + std::fs::write( + &file, + format!("model = 'original-model'\nnotify = ['original-notify']\n{ordinary}"), + ) + .unwrap(); + let mut initial = ConfigBuilder::without_managed_config_for_tests() + .codex_home(home.to_path_buf()) + .loader_overrides(LoaderOverrides { + user_config_path: Some(file.clone()), + user_config_profile: (config_file != CONFIG_TOML_FILE).then(|| "work".parse().unwrap()), + ..LoaderOverrides::without_managed_config_for_tests() + }) + .build() + .await + .unwrap(); + let mut servers = initial.mcp_servers.get().clone(); + let McpServerTransportConfig::StreamableHttp { http_headers, .. } = + &mut servers.get_mut("ordinary").unwrap().transport + else { + unreachable!() + }; + *http_headers = Some(HashMap::from([( + "Authorization".into(), + "Bearer synthetic-host".into(), + )])); + initial.mcp_servers.set(servers).unwrap(); + session + .state + .lock() + .await + .session_configuration + .original_config_do_not_use = Arc::new(initial.clone()); + let original_servers = runtime_servers(&session).await; + std::fs::write( + &file, + format!( + "model = 'new-model'\nnotify = ['new-notify']\nlog_dir = 'relative-logs'\n{ordinary}" + ), + ) + .unwrap(); + + session.reload_user_config_layer().await; + + let config = session.get_config().await; + assert_eq!( + config.config_layer_stack.effective_config()["log_dir"].as_str(), + file.parent() + .unwrap() + .join("relative-logs") + .as_path() + .to_str() + ); + assert_eq!(config.log_dir, initial.log_dir); + assert_eq!(config.model, initial.model); + assert_eq!(config.notify, initial.notify); + assert_eq!(runtime_servers(&session).await, original_servers); +} diff --git a/codex-rs/core/src/session/mod.rs b/codex-rs/core/src/session/mod.rs index e3493e6eb1..fbeea01823 100644 --- a/codex-rs/core/src/session/mod.rs +++ b/codex-rs/core/src/session/mod.rs @@ -21,7 +21,6 @@ use crate::attestation::AttestationProvider; use crate::compact; use crate::compact::CompactedHistoryMetadata; use crate::config::ManagedFeatures; -use crate::config::resolve_tool_suggest_config_from_layer_stack; use crate::context::ContextualUserFragment; use crate::context::DeveloperInstructions; use crate::context::GuardianPolicy; @@ -2047,76 +2046,6 @@ impl Session { state.session_configuration.provider.info().clone() } - 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(); - config.active_project = next_config.active_project.clone(); - config.config_layer_stack = config - .config_layer_stack - .with_user_layer_from(&next_config.config_layer_stack); - config.tool_suggest = - resolve_tool_suggest_config_from_layer_stack(&config.config_layer_stack); - 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; - 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}"); - } - let config = Arc::new(config); - state.session_configuration.original_config_do_not_use = Arc::clone(&config); - self.mark_mcp_runtime_dirty(); - let new_config = notify_config_contributors - .then(|| self.build_effective_session_config(&state.session_configuration)); - (previous_config, new_config, config) - }; - 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) { let disabled_plugin_ids = self.state.lock().await.active_disabled_plugin_ids.clone(); let environments = self.services.turn_environments.snapshot().await; @@ -2164,11 +2093,12 @@ impl Session { pub(crate) async fn reload_user_config_layer(&self) { // Refresh layer-backed runtime state for an existing session, including enabled plugin, - // skill, and hook state. Derived config fields such as feature gates and legacy notify - // settings remain session-static. + // skill, hook, and MCP feature state. Other feature gates and legacy notify settings + // remain session-static. // // Prefer `refresh_runtime_config()` when the host can already provide a materialized // config snapshot. This file-based path exists for legacy local reload flows. + // Retry from the current owner if another publication wins while files are read. loop { let (expected_config, config_toml_paths) = { let state = self.state.lock().await; @@ -2212,6 +2142,20 @@ impl Session { return; } }; + let Some(config_dir) = config_toml_path.parent() else { + warn!("user config path has no parent directory"); + return; + }; + let user_config = match codex_config::loader::resolve_relative_paths_in_config_toml( + user_config, + &config_dir, + ) { + Ok(config) => config, + Err(err) => { + warn!("failed to resolve paths while reloading user config: {err}"); + return; + } + }; reloaded_user_configs.push((config_toml_path, user_config)); } @@ -2229,12 +2173,18 @@ impl Session { }; next_config.config_layer_stack = config_layer_stack; } - 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 + match self + .refresh_config( + expected_config, + next_config, + crate::config::RuntimeConfigRefresh::UserFiles, + ) + .await { - return; + crate::ConfigRefreshOutcome::Stale => continue, + crate::ConfigRefreshOutcome::Published | crate::ConfigRefreshOutcome::Rejected => { + return; + } } } } diff --git a/codex-rs/core/src/session/tests.rs b/codex-rs/core/src/session/tests.rs index aadfc55d33..8221619a21 100644 --- a/codex-rs/core/src/session/tests.rs +++ b/codex-rs/core/src/session/tests.rs @@ -11,6 +11,8 @@ use super::step_settings::StepSettingsUpdate; pub(crate) use super::step_settings::tests::update_selected_settings_for_test; use super::turn_context::TurnEnvironment; use super::*; +#[path = "config_refresh_tests.rs"] +mod config_refresh_tests; use crate::agents_md_manager::AgentsMdManager; use crate::agents_md_manager::SessionInstructions; use crate::compact::InitialContextInjection; @@ -2320,6 +2322,7 @@ async fn refresh_mcp_config_replaces_managed_server_and_plugin_requirements(reje #[test_case::test_case(RuntimeConfigRefresh::User; "user refresh")] #[test_case::test_case(RuntimeConfigRefresh::Mcp; "MCP refresh")] +#[test_case::test_case(RuntimeConfigRefresh::UserFiles; "file refresh")] #[tokio::test] async fn refresh_config_rejects_stale_owner_without_effects(scope: RuntimeConfigRefresh) { let (session, _turn_context) = make_session_and_context().await; @@ -2352,30 +2355,6 @@ async fn refresh_config_rejects_stale_owner_without_effects(scope: RuntimeConfig 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] fn collect_explicit_app_ids_from_skill_items_includes_linked_mentions() { let connectors = vec![make_connector("calendar", "Calendar")]; diff --git a/codex-rs/thread-manager-sample/src/main.rs b/codex-rs/thread-manager-sample/src/main.rs index b137e6bc68..5e1d975ae8 100644 --- a/codex-rs/thread-manager-sample/src/main.rs +++ b/codex-rs/thread-manager-sample/src/main.rs @@ -380,6 +380,8 @@ async fn new_config( sleep_tool_mode: Default::default(), features: Default::default(), prefer_mxc: false, + + runtime_feature_defaults: Default::default(), suppress_unstable_features_warning: false, active_project: ProjectConfig { trust_level: None }, notices: Notice::default(),