Apply refreshed cloud config bundles to later sessions (#37908)

## Why

Background refreshes previously warmed only the on-disk cache, so new sessions in the same process continued using the startup snapshot.

## What changed

- Make `CloudConfigBundleLoader` retrieve the latest shared bundle on each configuration load.
- Update the in-memory bundle after successful refreshes while preserving the last successful bundle when a refresh fails.
- Stop refresh work when its loader is replaced or its final clone is dropped.

## Testing

- Cover refreshed requirements and managed configuration in later sessions.
- Cover concurrent initial loads, refresh failures and recovery, bundle clearing, and refresh-task cancellation.

GitOrigin-RevId: ae3d79b575bad71a1eed370cee6dc66dc022ffa8

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<Option<JoinHandle<()>>> {
-    static REFRESHER_TASK: OnceLock<Mutex<Option<JoinHandle<()>>>> = OnceLock::new();
+fn refresher_task_slot() -> &'static Mutex<Option<AbortHandle>> {
+    static REFRESHER_TASK: OnceLock<Mutex<Option<AbortHandle>>> = OnceLock::new();
     REFRESHER_TASK.get_or_init(|| Mutex::new(None))
 }
 
+pub(crate) fn replace_refresh_task(slot: &Mutex<Option<AbortHandle>>, 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<C> {
+    service: Arc<CloudConfigBundleService<C>>,
+    refresh_task: JoinHandle<()>,
+}
+
+impl<C> Drop for CloudConfigBundleLoaderLifetime<C> {
+    fn drop(&mut self) {
+        self.refresh_task.abort();
+    }
+}
+
 pub fn cloud_config_bundle_loader(
     auth_manager: Arc<AuthManager>,
     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<C>(
+    service: CloudConfigBundleService<C>,
+) -> (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(

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<C> {
     cache: CloudConfigBundleCache,
     codex_home: AbsolutePathBuf,
     timeout: Duration,
-}
-
-impl<C> Clone for CloudConfigBundleService<C> {
-    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<Mutex<Result<Option<CloudConfigBundle>, CloudConfigBundleLoadError>>>,
 }
 
 impl<C> CloudConfigBundleService<C>
@@ -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<Option<CloudConfigBundle>, 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<Option<CloudConfigBundle>, 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

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<tokio::sync::Notify>,
+    request_cancelled: Arc<tokio::sync::Notify>,
+}
+
+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<CloudConfigBundle, BundleRequestError> {
+        self.request_started.notify_one();
+        pending::<()>().await;
+        Ok(CloudConfigBundle::default())
+    }
+}
+
 struct SequenceBundleClient {
     responses: tokio::sync::Mutex<VecDeque<Result<CloudConfigBundle, BundleRequestError>>>,
     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 {

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<BoxFuture<'static, Result<Option<CloudConfigBundle>, CloudConfigBundleLoadError>>>,
+    getter: Arc<
+        dyn Fn()
+                -> BoxFuture<'static, Result<Option<CloudConfigBundle>, CloudConfigBundleLoadError>>
+            + Send
+            + Sync,
+    >,
 }
 
 impl CloudConfigBundleLoader {
@@ -186,14 +191,27 @@ impl CloudConfigBundleLoader {
         F: Future<Output = Result<Option<CloudConfigBundle>, 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<F, Fut>(getter: F) -> Self
+    where
+        F: Fn() -> Fut + Send + Sync + 'static,
+        Fut: Future<Output = Result<Option<CloudConfigBundle>, 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<Option<CloudConfigBundle>, CloudConfigBundleLoadError> {
-        self.fut.clone().await
+        (self.getter)().await
     }
 }
 

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");

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(())
+}

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;