diff --git a/codex-rs/cloud-config/src/bundle_loader.rs b/codex-rs/cloud-config/src/bundle_loader.rs index bc84f81f78..e3007502c6 100644 --- a/codex-rs/cloud-config/src/bundle_loader.rs +++ b/codex-rs/cloud-config/src/bundle_loader.rs @@ -1,8 +1,7 @@ use crate::backend::BackendBundleClient; +use crate::backend::BundleClient; use crate::service::CLOUD_CONFIG_BUNDLE_TIMEOUT; use crate::service::CloudConfigBundleService; -use codex_config::CloudConfigBundleLoadError; -use codex_config::CloudConfigBundleLoadErrorCode; use codex_config::CloudConfigBundleLoader; use codex_http_client::HttpClientFactory; use codex_login::AuthConfig; @@ -11,13 +10,35 @@ use std::path::PathBuf; use std::sync::Arc; use std::sync::Mutex; use std::sync::OnceLock; +use tokio::task::AbortHandle; use tokio::task::JoinHandle; -fn refresher_task_slot() -> &'static Mutex>> { - static REFRESHER_TASK: OnceLock>>> = OnceLock::new(); +fn refresher_task_slot() -> &'static Mutex> { + static REFRESHER_TASK: OnceLock>> = OnceLock::new(); REFRESHER_TASK.get_or_init(|| Mutex::new(None)) } +pub(crate) fn replace_refresh_task(slot: &Mutex>, next: AbortHandle) { + let mut guard = slot.lock().unwrap_or_else(|err| { + tracing::warn!("cloud config bundle refresher task slot was poisoned"); + err.into_inner() + }); + if let Some(previous) = guard.replace(next) { + previous.abort(); + } +} + +struct CloudConfigBundleLoaderLifetime { + service: Arc>, + refresh_task: JoinHandle<()>, +} + +impl Drop for CloudConfigBundleLoaderLifetime { + fn drop(&mut self) { + self.refresh_task.abort(); + } +} + pub fn cloud_config_bundle_loader( auth_manager: Arc, chatgpt_base_url: String, @@ -33,27 +54,34 @@ pub fn cloud_config_bundle_loader( codex_home, CLOUD_CONFIG_BUNDLE_TIMEOUT, ); - let refresh_service = service.clone(); - let task = tokio::spawn(async move { service.load_startup_bundle_with_timeout().await }); - let refresh_task = - tokio::spawn(async move { refresh_service.refresh_cache_in_background().await }); - let mut refresher_guard = refresher_task_slot().lock().unwrap_or_else(|err| { - tracing::warn!("cloud config bundle refresher task slot was poisoned"); - err.into_inner() + let (loader, refresh_task) = cloud_config_bundle_loader_for_service(service); + replace_refresh_task(refresher_task_slot(), refresh_task); + loader +} + +pub(crate) fn cloud_config_bundle_loader_for_service( + service: CloudConfigBundleService, +) -> (CloudConfigBundleLoader, AbortHandle) +where + C: BundleClient + 'static, +{ + let service = Arc::new(service); + let background_service = Arc::clone(&service); + let refresh_task = tokio::spawn(async move { + let _ = background_service.get_latest().await; + background_service.refresh_cache_in_background().await; }); - if let Some(existing_task) = refresher_guard.replace(refresh_task) { - existing_task.abort(); - } - CloudConfigBundleLoader::new(async move { - task.await.map_err(|err| { - tracing::error!(error = %err, "Cloud config bundle task failed"); - CloudConfigBundleLoadError::new( - CloudConfigBundleLoadErrorCode::Internal, - /*status_code*/ None, - format!("cloud config bundle load failed: {err}"), - ) - })? - }) + let abort_handle = refresh_task.abort_handle(); + let lifetime = Arc::new(CloudConfigBundleLoaderLifetime { + service, + refresh_task, + }); + + let loader = CloudConfigBundleLoader::from_getter(move || { + let lifetime = Arc::clone(&lifetime); + async move { lifetime.service.get_latest().await } + }); + (loader, abort_handle) } pub async fn cloud_config_bundle_loader_for_storage( diff --git a/codex-rs/cloud-config/src/service.rs b/codex-rs/cloud-config/src/service.rs index 1a7104bf52..6bb8d3bc61 100644 --- a/codex-rs/cloud-config/src/service.rs +++ b/codex-rs/cloud-config/src/service.rs @@ -1,8 +1,7 @@ //! Cloud config bundle lifecycle orchestration. //! -//! Startup loads a single shared bundle from cache or backend, and a background -//! refresher keeps the cache warm for future app starts without changing the -//! already-snapshotted runtime config. +//! Startup loads a shared bundle from cache or backend, and background refresh +//! updates both the on-disk cache and the bundle observed by future config loads. use crate::backend::BundleClient; use crate::backend::BundleRequestError; @@ -27,6 +26,8 @@ use std::path::PathBuf; use std::sync::Arc; use std::time::Duration; use std::time::Instant; +use tokio::sync::Mutex; +use tokio::sync::OnceCell; use tokio::time::sleep; use tokio::time::timeout; @@ -77,18 +78,7 @@ pub(crate) struct CloudConfigBundleService { cache: CloudConfigBundleCache, codex_home: AbsolutePathBuf, timeout: Duration, -} - -impl Clone for CloudConfigBundleService { - fn clone(&self) -> Self { - Self { - auth_manager: Arc::clone(&self.auth_manager), - client: Arc::clone(&self.client), - cache: self.cache.clone(), - codex_home: self.codex_home.clone(), - timeout: self.timeout, - } - } + latest_bundle: OnceCell, CloudConfigBundleLoadError>>>, } impl CloudConfigBundleService @@ -108,9 +98,21 @@ where cache: CloudConfigBundleCache::new(codex_home.clone()), codex_home, timeout, + latest_bundle: OnceCell::new(), } } + pub(crate) async fn get_latest( + &self, + ) -> Result, CloudConfigBundleLoadError> { + self.latest_bundle + .get_or_init(|| async { Mutex::new(self.load_startup_bundle_with_timeout().await) }) + .await + .lock() + .await + .clone() + } + pub(crate) async fn load_startup_bundle_with_timeout( &self, ) -> Result, CloudConfigBundleLoadError> { @@ -482,7 +484,12 @@ where .fetch_remote_bundle_and_update_cache_with_retries(auth, "refresh") .await { - Ok(bundle) => emit_load_metric("refresh", "success", bundle.as_ref()), + Ok(bundle) => { + emit_load_metric("refresh", "success", bundle.as_ref()); + if let Some(latest_bundle) = self.latest_bundle.get() { + *latest_bundle.lock().await = Ok(bundle); + } + } Err(err) => { tracing::error!( path = %self.cache.path().display(), @@ -490,6 +497,12 @@ where "Failed to refresh cloud config bundle cache from remote" ); emit_load_metric("refresh", "error", /*bundle*/ None); + if let Some(latest_bundle) = self.latest_bundle.get() { + let mut latest_bundle = latest_bundle.lock().await; + if latest_bundle.is_err() { + *latest_bundle = Err(err); + } + } } } true diff --git a/codex-rs/cloud-config/src/service_tests.rs b/codex-rs/cloud-config/src/service_tests.rs index 1c07ab0bcd..5dccd3ba9a 100644 --- a/codex-rs/cloud-config/src/service_tests.rs +++ b/codex-rs/cloud-config/src/service_tests.rs @@ -16,6 +16,7 @@ use codex_config::CloudConfigTomlBundle; use codex_config::CloudRequirementsFragment; use codex_config::CloudRequirementsTomlBundle; use codex_config::types::AuthCredentialsStoreMode; +use codex_core::config::ConfigBuilder; use codex_login::AuthKeyringBackendKind; use codex_login::auth::AgentIdentityAuth; use codex_login::auth::AgentIdentityAuthRecord; @@ -249,6 +250,25 @@ impl BundleClient for PendingBundleClient { } } +struct NotifyingPendingBundleClient { + request_started: Arc, + request_cancelled: Arc, +} + +impl Drop for NotifyingPendingBundleClient { + fn drop(&mut self) { + self.request_cancelled.notify_one(); + } +} + +impl BundleClient for NotifyingPendingBundleClient { + async fn get_bundle(&self, _auth: &CodexAuth) -> Result { + self.request_started.notify_one(); + pending::<()>().await; + Ok(CloudConfigBundle::default()) + } +} + struct SequenceBundleClient { responses: tokio::sync::Mutex>>, request_count: AtomicUsize, @@ -1030,8 +1050,12 @@ async fn refresh_from_remote_updates_cached_bundle() { CLOUD_CONFIG_BUNDLE_TIMEOUT, ); - assert_eq!(service.load_startup_bundle().await, Ok(Some(test_bundle()))); + assert_eq!(service.get_latest().await, Ok(Some(test_bundle()))); assert!(service.refresh_cache_once().await); + assert_eq!( + service.get_latest().await, + Ok(Some(replacement_bundle.clone())) + ); let cache = create_test_cache(codex_home.path()); let signed_payload = cache @@ -1041,6 +1065,264 @@ async fn refresh_from_remote_updates_cached_bundle() { assert_eq!(signed_payload.bundle, replacement_bundle); } +#[tokio::test(start_paused = true)] +async fn production_loader_refreshes_later_configs_and_preserves_failed_refreshes() { + let codex_home = tempdir().expect("tempdir"); + let initial_bundle = test_bundle(); + let refreshed_bundle = CloudConfigBundle { + config_toml: CloudConfigTomlBundle { + enterprise_managed: vec![CloudConfigFragment { + id: "cfg_refreshed".to_string(), + name: "Refreshed config".to_string(), + contents: "model = \"gpt-5-refreshed\"".to_string(), + }], + }, + requirements_toml: initial_bundle.requirements_toml.clone(), + }; + let mut responses = vec![Ok(initial_bundle.clone()), Ok(refreshed_bundle.clone())]; + responses.extend(vec![Err(request_error()); CLOUD_CONFIG_BUNDLE_MAX_ATTEMPTS]); + let fetcher = Arc::new(SequenceBundleClient::new(responses)); + let service = CloudConfigBundleService::new( + auth_manager_with_plan("business").await, + Arc::clone(&fetcher), + codex_home.path().to_path_buf(), + CLOUD_CONFIG_BUNDLE_TIMEOUT, + ); + let (loader, _) = crate::bundle_loader::cloud_config_bundle_loader_for_service(service); + + let (first, second) = tokio::join!(loader.get(), loader.get()); + assert_eq!(first, Ok(Some(initial_bundle.clone()))); + assert_eq!(second, Ok(Some(initial_bundle))); + assert_eq!(fetcher.request_count.load(Ordering::SeqCst), 1); + tokio::task::yield_now().await; + tokio::time::advance(CLOUD_CONFIG_BUNDLE_CACHE_REFRESH_INTERVAL + Duration::from_millis(1)) + .await; + let refresh_deadline = std::time::Instant::now() + Duration::from_secs(5); + while loader.get().await != Ok(Some(refreshed_bundle.clone())) { + assert!( + std::time::Instant::now() < refresh_deadline, + "the production refresh task should update the latest bundle" + ); + tokio::task::yield_now().await; + } + let refreshed_config = ConfigBuilder::default() + .codex_home(codex_home.path().to_path_buf()) + .cloud_config_bundle(loader.clone()) + .build() + .await + .expect("later session config should load the refreshed bundle"); + assert_eq!(refreshed_config.model.as_deref(), Some("gpt-5-refreshed")); + + tokio::task::yield_now().await; + tokio::time::advance(CLOUD_CONFIG_BUNDLE_CACHE_REFRESH_INTERVAL + Duration::from_millis(1)) + .await; + let failed_refresh_deadline = std::time::Instant::now() + Duration::from_secs(5); + for expected_request_count in 3..=2 + CLOUD_CONFIG_BUNDLE_MAX_ATTEMPTS { + if expected_request_count > 3 { + tokio::task::yield_now().await; + tokio::time::advance(Duration::from_secs(2)).await; + } + while fetcher.request_count.load(Ordering::SeqCst) < expected_request_count { + assert!( + std::time::Instant::now() < failed_refresh_deadline, + "the production refresh task should retry the failed bundle request" + ); + tokio::task::yield_now().await; + } + } + + assert_eq!(loader.get().await, Ok(Some(refreshed_bundle))); +} + +#[tokio::test(start_paused = true)] +async fn refresh_stops_on_replacement_or_after_the_last_loader_clone() { + let codex_home = tempdir().expect("tempdir"); + let fetcher = Arc::new(StaticBundleClient::new(test_bundle())); + let service = CloudConfigBundleService::new( + auth_manager_with_plan("business").await, + Arc::clone(&fetcher), + codex_home.path().to_path_buf(), + CLOUD_CONFIG_BUNDLE_TIMEOUT, + ); + let (loader, abort_handle) = + crate::bundle_loader::cloud_config_bundle_loader_for_service(service); + let task_slot = std::sync::Mutex::new(None); + crate::bundle_loader::replace_refresh_task(&task_slot, abort_handle); + let cloned_loader = loader.clone(); + assert_eq!(loader.get().await, Ok(Some(test_bundle()))); + tokio::task::yield_now().await; + + drop(loader); + tokio::time::advance(CLOUD_CONFIG_BUNDLE_CACHE_REFRESH_INTERVAL + Duration::from_millis(1)) + .await; + let refresh_deadline = std::time::Instant::now() + Duration::from_secs(5); + while fetcher.request_count.load(Ordering::SeqCst) < 2 { + assert!( + std::time::Instant::now() < refresh_deadline, + "the refresh should remain active while another loader clone exists" + ); + tokio::task::yield_now().await; + } + + let replacement_fetcher = Arc::new(StaticBundleClient::new(test_bundle())); + let replacement_service = CloudConfigBundleService::new( + auth_manager_with_plan_and_identity( + "business", + Some("user-replacement"), + Some("account-replacement"), + ) + .await, + Arc::clone(&replacement_fetcher), + codex_home.path().to_path_buf(), + CLOUD_CONFIG_BUNDLE_TIMEOUT, + ); + let (replacement_loader, replacement_handle) = + crate::bundle_loader::cloud_config_bundle_loader_for_service(replacement_service); + let replacement_task = replacement_handle.clone(); + crate::bundle_loader::replace_refresh_task(&task_slot, replacement_handle); + assert_eq!(replacement_loader.get().await, Ok(Some(test_bundle()))); + + tokio::task::yield_now().await; + tokio::time::advance(CLOUD_CONFIG_BUNDLE_CACHE_REFRESH_INTERVAL + Duration::from_millis(1)) + .await; + let refresh_deadline = std::time::Instant::now() + Duration::from_secs(5); + while replacement_fetcher.request_count.load(Ordering::SeqCst) < 2 { + assert!( + std::time::Instant::now() < refresh_deadline, + "the replacement refresher should stay active" + ); + tokio::task::yield_now().await; + } + assert_eq!(fetcher.request_count.load(Ordering::SeqCst), 2); + + drop(cloned_loader); + tokio::task::yield_now().await; + assert!(!replacement_task.is_finished()); + + drop(replacement_loader); + tokio::task::yield_now().await; + assert!(replacement_task.is_finished()); + tokio::time::advance(CLOUD_CONFIG_BUNDLE_CACHE_REFRESH_INTERVAL + Duration::from_secs(1)).await; + tokio::task::yield_now().await; + assert_eq!(replacement_fetcher.request_count.load(Ordering::SeqCst), 2); +} + +#[tokio::test(start_paused = true)] +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"); + } + let request_started = Arc::new(tokio::sync::Notify::new()); + let request_cancelled = Arc::new(tokio::sync::Notify::new()); + let service = CloudConfigBundleService::new( + auth_manager_with_plan("business").await, + Arc::new(NotifyingPendingBundleClient { + request_started: Arc::clone(&request_started), + request_cancelled: Arc::clone(&request_cancelled), + }), + codex_home.path().to_path_buf(), + CLOUD_CONFIG_BUNDLE_TIMEOUT, + ); + let (loader, _) = crate::bundle_loader::cloud_config_bundle_loader_for_service(service); + if starts_from_cache { + assert_eq!(loader.get().await, Ok(Some(test_bundle()))); + tokio::task::yield_now().await; + tokio::time::advance(CLOUD_CONFIG_BUNDLE_CACHE_REFRESH_INTERVAL).await; + } + request_started.notified().await; + + drop(loader); + + tokio::time::timeout(Duration::from_secs(1), request_cancelled.notified()) + .await + .expect("loader drop should cancel the in-flight fetch"); + } +} + +#[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())]; + responses.extend(vec![Err(request_error()); CLOUD_CONFIG_BUNDLE_MAX_ATTEMPTS]); + responses.push(Ok(test_bundle())); + let service = Arc::new(CloudConfigBundleService::new( + auth_manager_with_plan("business").await, + Arc::new(SequenceBundleClient::new(responses)), + codex_home.path().to_path_buf(), + CLOUD_CONFIG_BUNDLE_TIMEOUT, + )); + + assert_eq!(service.get_latest().await, Ok(Some(test_bundle()))); + assert!(service.refresh_cache_once().await); + assert_eq!(service.get_latest().await, Ok(None)); + + let refresh_service = Arc::clone(&service); + let refresh = tokio::spawn(async move { refresh_service.refresh_cache_once().await }); + tokio::task::yield_now().await; + tokio::time::advance(Duration::from_secs(5)).await; + tokio::task::yield_now().await; + assert!(refresh.await.expect("failed refresh task")); + assert_eq!(service.get_latest().await, Ok(None)); + + assert!(service.refresh_cache_once().await); + assert_eq!(service.get_latest().await, Ok(Some(test_bundle()))); +} + +#[tokio::test(start_paused = true)] +async fn refresh_replaces_initial_errors_and_recovers_with_success() { + let codex_home = tempdir().expect("tempdir"); + let initial_error = BundleRequestError::Retryable(RetryableFailureKind::Request { + status_code: Some(500), + }); + let refresh_error = BundleRequestError::Retryable(RetryableFailureKind::Request { + status_code: Some(503), + }); + let mut responses = vec![Err(initial_error); CLOUD_CONFIG_BUNDLE_MAX_ATTEMPTS]; + responses.extend(vec![Err(refresh_error); CLOUD_CONFIG_BUNDLE_MAX_ATTEMPTS]); + responses.push(Ok(test_bundle())); + let service = Arc::new(CloudConfigBundleService::new( + auth_manager_with_plan("business").await, + Arc::new(SequenceBundleClient::new(responses)), + codex_home.path().to_path_buf(), + CLOUD_CONFIG_BUNDLE_TIMEOUT, + )); + + let initial_service = Arc::clone(&service); + let initial = tokio::spawn(async move { initial_service.get_latest().await }); + tokio::task::yield_now().await; + tokio::time::advance(Duration::from_secs(5)).await; + tokio::task::yield_now().await; + let initial_error = initial + .await + .expect("initial task") + .expect_err("initial fetch should fail"); + assert_eq!(initial_error.status_code(), Some(500)); + + let refresh_service = Arc::clone(&service); + let refresh = tokio::spawn(async move { refresh_service.refresh_cache_once().await }); + tokio::task::yield_now().await; + tokio::time::advance(Duration::from_secs(5)).await; + tokio::task::yield_now().await; + assert!(refresh.await.expect("failed refresh task")); + let latest_error = service + .get_latest() + .await + .expect_err("latest failed refresh should replace the initial error"); + assert_eq!(latest_error.status_code(), Some(503)); + + assert!(service.refresh_cache_once().await); + assert_eq!(service.get_latest().await, Ok(Some(test_bundle()))); +} + #[test] fn bundle_response_conversion_preserves_fragment_order() { let response = ConfigBundleResponse { diff --git a/codex-rs/config/src/cloud_config_bundle.rs b/codex-rs/config/src/cloud_config_bundle.rs index cfece32a01..7982d11ea4 100644 --- a/codex-rs/config/src/cloud_config_bundle.rs +++ b/codex-rs/config/src/cloud_config_bundle.rs @@ -14,11 +14,11 @@ use crate::cloud_config_layers_from_fragments; use codex_utils_absolute_path::AbsolutePathBuf; use futures::future::BoxFuture; use futures::future::FutureExt; -use futures::future::Shared; use serde::Deserialize; use serde::Serialize; use std::fmt; use std::future::Future; +use std::sync::Arc; use thiserror::Error; #[derive(Clone, Debug, Default, Deserialize, Eq, PartialEq, Serialize)] @@ -177,7 +177,12 @@ impl CloudConfigBundleLoadError { #[derive(Clone)] pub struct CloudConfigBundleLoader { - fut: Shared, CloudConfigBundleLoadError>>>, + getter: Arc< + dyn Fn() + -> BoxFuture<'static, Result, CloudConfigBundleLoadError>> + + Send + + Sync, + >, } impl CloudConfigBundleLoader { @@ -186,14 +191,27 @@ impl CloudConfigBundleLoader { F: Future, CloudConfigBundleLoadError>> + Send + 'static, + { + let fut = fut.boxed().shared(); + Self::from_getter(move || fut.clone()) + } + + /// Creates a loader that requests the latest bundle on every call. + pub fn from_getter(getter: F) -> Self + where + F: Fn() -> Fut + Send + Sync + 'static, + Fut: Future, CloudConfigBundleLoadError>> + + Send + + 'static, { Self { - fut: fut.boxed().shared(), + getter: Arc::new(move || getter().boxed()), } } + /// Returns the current bundle snapshot. pub async fn get(&self) -> Result, CloudConfigBundleLoadError> { - self.fut.clone().await + (self.getter)().await } } diff --git a/codex-rs/config/src/cloud_config_bundle_tests.rs b/codex-rs/config/src/cloud_config_bundle_tests.rs index 714aca6bc6..e475357cbb 100644 --- a/codex-rs/config/src/cloud_config_bundle_tests.rs +++ b/codex-rs/config/src/cloud_config_bundle_tests.rs @@ -5,6 +5,7 @@ use crate::compose_requirements; use codex_protocol::protocol::AskForApproval; use pretty_assertions::assert_eq; use std::sync::Arc; +use std::sync::RwLock; use std::sync::atomic::AtomicUsize; use std::sync::atomic::Ordering; use tempfile::tempdir; @@ -17,12 +18,51 @@ async fn shared_future_runs_once() { counter_clone.fetch_add(1, Ordering::SeqCst); Ok(Some(CloudConfigBundle::default())) }); + let cloned_loader = loader.clone(); - let (first, second) = tokio::join!(loader.get(), loader.get()); + let (first, second) = tokio::join!(loader.get(), cloned_loader.get()); assert_eq!(first, second); + assert_eq!(loader.get().await, first); assert_eq!(counter.load(Ordering::SeqCst), 1); } +#[tokio::test] +async fn getter_returns_latest_result_across_clones() { + let initial_error = CloudConfigBundleLoadError::new( + CloudConfigBundleLoadErrorCode::RequestFailed, + /*status_code*/ None, + "initial load failed", + ); + let latest = Arc::new(RwLock::new(Err(initial_error.clone()))); + let getter_latest = Arc::clone(&latest); + let loader = CloudConfigBundleLoader::from_getter(move || { + let latest = Arc::clone(&getter_latest); + async move { latest.read().expect("bundle state lock").clone() } + }); + let cloned_loader = loader.clone(); + + assert_eq!(loader.get().await, Err(initial_error)); + + let bundle = CloudConfigBundle { + config_toml: CloudConfigTomlBundle { + enterprise_managed: vec![CloudConfigFragment { + id: "managed".to_string(), + name: "Managed".to_string(), + contents: "model = \"managed\"".to_string(), + }], + }, + ..Default::default() + }; + *latest.write().expect("bundle state lock") = Ok(Some(bundle.clone())); + + assert_eq!(cloned_loader.get().await, Ok(Some(bundle))); + assert_eq!(CloudConfigBundleLoader::default().get().await, Ok(None)); + + *latest.write().expect("bundle state lock") = Ok(None); + + assert_eq!(loader.get().await, Ok(None)); +} + #[test] fn bundle_layers_preserve_enterprise_managed_bucket_order() { let tempdir = tempdir().expect("tempdir"); diff --git a/codex-rs/core/tests/suite/cloud_config.rs b/codex-rs/core/tests/suite/cloud_config.rs new file mode 100644 index 0000000000..94968a1144 --- /dev/null +++ b/codex-rs/core/tests/suite/cloud_config.rs @@ -0,0 +1,71 @@ +use anyhow::Result; +use codex_config::CloudConfigBundleLoader; +use codex_config::test_support::CloudConfigBundleFixture; +use codex_protocol::protocol::AskForApproval; +use core_test_support::responses::start_mock_server; +use core_test_support::test_codex::test_codex; +use pretty_assertions::assert_eq; +use std::sync::Arc; +use std::sync::RwLock; +use tempfile::TempDir; + +#[tokio::test] +async fn refreshed_cloud_bundle_updates_later_sessions() -> Result<()> { + let server = start_mock_server().await; + let home = Arc::new(TempDir::new()?); + let initial_bundle = CloudConfigBundleFixture::enterprise_requirement( + r#"allowed_approval_policies = ["never"]"#, + ) + .add_enterprise_config(r#"developer_instructions = "initial managed instructions""#) + .into_bundle(); + let latest = Arc::new(RwLock::new(initial_bundle)); + let getter_latest = Arc::clone(&latest); + let loader = CloudConfigBundleLoader::from_getter(move || { + let latest = Arc::clone(&getter_latest); + async move { Ok(Some(latest.read().expect("bundle state lock").clone())) } + }); + + let mut initial_builder = test_codex() + .with_home(Arc::clone(&home)) + .with_cloud_config_bundle(loader.clone()); + let initial = initial_builder.build_with_auto_env(&server).await?; + assert_eq!( + initial.session_configured.approval_policy, + AskForApproval::Never + ); + assert_eq!( + initial + .codex + .config() + .await + .developer_instructions + .as_deref(), + Some("initial managed instructions") + ); + + *latest.write().expect("bundle state lock") = CloudConfigBundleFixture::enterprise_requirement( + r#"allowed_approval_policies = ["on-request"]"#, + ) + .add_enterprise_config(r#"developer_instructions = "refreshed managed instructions""#) + .into_bundle(); + + let mut refreshed_builder = test_codex() + .with_home(home) + .with_cloud_config_bundle(loader); + let refreshed = refreshed_builder.build_with_auto_env(&server).await?; + assert_eq!( + refreshed.session_configured.approval_policy, + AskForApproval::OnRequest + ); + assert_eq!( + refreshed + .codex + .config() + .await + .developer_instructions + .as_deref(), + Some("refreshed managed instructions") + ); + + Ok(()) +} diff --git a/codex-rs/core/tests/suite/mod.rs b/codex-rs/core/tests/suite/mod.rs index 0cb63eec73..4bf5295eaa 100644 --- a/codex-rs/core/tests/suite/mod.rs +++ b/codex-rs/core/tests/suite/mod.rs @@ -48,6 +48,7 @@ mod catalog_permission_messages; mod cli_stream; mod client; mod client_websockets; +mod cloud_config; mod code_mode; mod code_mode_elicitation; mod codex_delegate;