Split cloud config bundle service modules (#25668)

## Summary - Splits the monolithic `codex-cloud-config` implementation into focused modules. - Keeps behavior unchanged from the preceding config bundle runtime switch. ## Details This is the reviewability follow-up after the lineage-preserving migration PRs. The split separates backend transport, loader construction, cache handling, metrics, validation, service orchestration, and focused tests into named files. Verification: `just fmt`; `just test -p codex-cloud-config`.
2026-07-01 00:31:56 +08:00 · 2026-06-02 14:30:12 -07:00
parent f6d64bd6ab
commit e7039f9844
11 changed files with 2336 additions and 2219 deletions
@@ -2347,7 +2347,6 @@ dependencies = [
 name = "codex-cloud-config"
 version = "0.0.0"
 dependencies = [
- "async-trait",
 "base64 0.22.1",
 "chrono",
 "codex-backend-client",
@@ -8,7 +8,6 @@ license.workspace = true
 workspace = true

 [dependencies]
-async-trait = { workspace = true }
 base64 = { workspace = true }
 chrono = { workspace = true, features = ["serde"] }
 codex-backend-client = { workspace = true }
@@ -0,0 +1,136 @@
+use codex_backend_client::Client as BackendClient;
+use codex_backend_client::ConfigBundleResponse;
+use codex_backend_client::DeliveredTomlFragment;
+use codex_config::CloudConfigBundle;
+use codex_config::CloudConfigFragment;
+use codex_config::CloudConfigTomlBundle;
+use codex_config::CloudRequirementsFragment;
+use codex_config::CloudRequirementsTomlBundle;
+use codex_login::CodexAuth;
+use std::future::Future;
+
+#[derive(Clone, Copy, Debug, Eq, PartialEq)]
+pub(crate) enum RetryableFailureKind {
+    BackendClientInit,
+    Request { status_code: Option<u16> },
+}
+
+impl RetryableFailureKind {
+    pub(crate) fn status_code(self) -> Option<u16> {
+        match self {
+            Self::BackendClientInit => None,
+            Self::Request { status_code } => status_code,
+        }
+    }
+}
+
+#[derive(Clone, Debug, Eq, PartialEq)]
+pub(crate) enum BundleRequestError {
+    Retryable(RetryableFailureKind),
+    Unauthorized {
+        status_code: Option<u16>,
+        message: String,
+    },
+}
+
+/// Retrieves one cloud config bundle from the backend.
+///
+/// Implementations should return the backend-selected bundle exactly as delivered and leave
+/// validation, caching, and config/requirements parsing decisions to the service layer.
+pub(crate) trait BundleClient: Send + Sync {
+    fn get_bundle(
+        &self,
+        auth: &CodexAuth,
+    ) -> impl Future<Output = Result<CloudConfigBundle, BundleRequestError>> + Send;
+}
+
+pub(crate) struct BackendBundleClient {
+    base_url: String,
+}
+
+impl BackendBundleClient {
+    pub(crate) fn new(base_url: String) -> Self {
+        Self { base_url }
+    }
+}
+
+impl BundleClient for BackendBundleClient {
+    async fn get_bundle(&self, auth: &CodexAuth) -> Result<CloudConfigBundle, BundleRequestError> {
+        let client = BackendClient::from_auth(self.base_url.clone(), auth)
+            .inspect_err(|err| {
+                tracing::warn!(
+                    error = %err,
+                    "Failed to construct backend client for cloud config bundle"
+                );
+            })
+            .map_err(|_| BundleRequestError::Retryable(RetryableFailureKind::BackendClientInit))?;
+
+        let response = client
+            .get_config_bundle()
+            .await
+            .inspect_err(|err| {
+                tracing::warn!(error = %err, "Failed to fetch cloud config bundle");
+            })
+            .map_err(|err| {
+                let status_code = err.status().map(|status| status.as_u16());
+                if err.is_unauthorized() {
+                    BundleRequestError::Unauthorized {
+                        status_code,
+                        message: err.to_string(),
+                    }
+                } else {
+                    BundleRequestError::Retryable(RetryableFailureKind::Request { status_code })
+                }
+            })?;
+
+        Ok(bundle_from_response(response))
+    }
+}
+
+pub(crate) fn bundle_from_response(response: ConfigBundleResponse) -> CloudConfigBundle {
+    let config_toml = response
+        .config_toml
+        .flatten()
+        .map(|config_toml| *config_toml)
+        .and_then(|config_toml| config_toml.enterprise_managed.flatten())
+        .unwrap_or_default()
+        .into_iter()
+        .map(config_fragment_from_delivered)
+        .collect();
+    let requirements_toml = response
+        .requirements_toml
+        .flatten()
+        .map(|requirements_toml| *requirements_toml)
+        .and_then(|requirements_toml| requirements_toml.enterprise_managed.flatten())
+        .unwrap_or_default()
+        .into_iter()
+        .map(requirements_fragment_from_delivered)
+        .collect();
+
+    CloudConfigBundle {
+        config_toml: CloudConfigTomlBundle {
+            enterprise_managed: config_toml,
+        },
+        requirements_toml: CloudRequirementsTomlBundle {
+            enterprise_managed: requirements_toml,
+        },
+    }
+}
+
+fn config_fragment_from_delivered(fragment: DeliveredTomlFragment) -> CloudConfigFragment {
+    CloudConfigFragment {
+        id: fragment.id,
+        name: fragment.name,
+        contents: fragment.contents,
+    }
+}
+
+fn requirements_fragment_from_delivered(
+    fragment: DeliveredTomlFragment,
+) -> CloudRequirementsFragment {
+    CloudRequirementsFragment {
+        id: fragment.id,
+        name: fragment.name,
+        contents: fragment.contents,
+    }
+}
@@ -0,0 +1,68 @@
+use crate::backend::BackendBundleClient;
+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_config::types::AuthCredentialsStoreMode;
+use codex_login::AuthManager;
+use std::path::PathBuf;
+use std::sync::Arc;
+use std::sync::Mutex;
+use std::sync::OnceLock;
+use tokio::task::JoinHandle;
+
+fn refresher_task_slot() -> &'static Mutex<Option<JoinHandle<()>>> {
+    static REFRESHER_TASK: OnceLock<Mutex<Option<JoinHandle<()>>>> = OnceLock::new();
+    REFRESHER_TASK.get_or_init(|| Mutex::new(None))
+}
+
+pub fn cloud_config_bundle_loader(
+    auth_manager: Arc<AuthManager>,
+    chatgpt_base_url: String,
+    codex_home: PathBuf,
+) -> CloudConfigBundleLoader {
+    let service = CloudConfigBundleService::new(
+        auth_manager,
+        Arc::new(BackendBundleClient::new(chatgpt_base_url)),
+        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()
+    });
+    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}"),
+            )
+        })?
+    })
+}
+
+pub async fn cloud_config_bundle_loader_for_storage(
+    codex_home: PathBuf,
+    enable_codex_api_key_env: bool,
+    credentials_store_mode: AuthCredentialsStoreMode,
+    chatgpt_base_url: String,
+) -> CloudConfigBundleLoader {
+    let auth_manager = AuthManager::shared(
+        codex_home.clone(),
+        enable_codex_api_key_env,
+        credentials_store_mode,
+        Some(chatgpt_base_url.clone()),
+    )
+    .await;
+    cloud_config_bundle_loader(auth_manager, chatgpt_base_url, codex_home)
+}
@@ -0,0 +1,253 @@
+//! Signed on-disk cache for cloud config bundles.
+//!
+//! The cache is scoped to the authenticated ChatGPT user and account, has a
+//! short TTL, and is HMAC-signed so malformed or edited files fail closed.
+
+use base64::Engine;
+use base64::engine::general_purpose::STANDARD as BASE64_STANDARD;
+use chrono::DateTime;
+use chrono::Duration as ChronoDuration;
+use chrono::Utc;
+use codex_config::AbsolutePathBuf;
+use codex_config::CloudConfigBundle;
+use hmac::Hmac;
+use hmac::Mac;
+use serde::Deserialize;
+use serde::Serialize;
+use sha2::Sha256;
+use std::path::Path;
+use std::time::Duration;
+use thiserror::Error;
+use tokio::fs;
+
+const CLOUD_CONFIG_BUNDLE_CACHE_VERSION: u32 = 1;
+pub(super) const CLOUD_CONFIG_BUNDLE_CACHE_FILENAME: &str = "cloud-config-bundle-cache.json";
+const CLOUD_CONFIG_BUNDLE_CACHE_TTL: Duration = Duration::from_secs(30 * 60);
+const CLOUD_CONFIG_BUNDLE_CACHE_WRITE_HMAC_KEY: &[u8] =
+    b"codex-cloud-config-bundle-cache-v1-6160ae70-bcfd-4ca8-a99b-40f73b3b072e";
+const CLOUD_CONFIG_BUNDLE_CACHE_READ_HMAC_KEYS: &[&[u8]] =
+    &[CLOUD_CONFIG_BUNDLE_CACHE_WRITE_HMAC_KEY];
+
+type HmacSha256 = Hmac<Sha256>;
+
+#[derive(Clone)]
+pub(super) struct CloudConfigBundleCache {
+    path: AbsolutePathBuf,
+}
+
+impl CloudConfigBundleCache {
+    pub(super) fn new(codex_home: AbsolutePathBuf) -> Self {
+        Self {
+            path: codex_home.join(CLOUD_CONFIG_BUNDLE_CACHE_FILENAME),
+        }
+    }
+
+    pub(super) fn path(&self) -> &Path {
+        &self.path
+    }
+
+    pub(super) async fn load(
+        &self,
+        chatgpt_user_id: Option<&str>,
+        account_id: Option<&str>,
+    ) -> Result<CloudConfigBundleCacheSignedPayload, CacheLoadStatus> {
+        let (Some(chatgpt_user_id), Some(account_id)) = (chatgpt_user_id, account_id) else {
+            return Err(CacheLoadStatus::AuthIdentityIncomplete);
+        };
+
+        let bytes = match fs::read(&self.path).await {
+            Ok(bytes) => bytes,
+            Err(err) => {
+                if err.kind() != std::io::ErrorKind::NotFound {
+                    return Err(CacheLoadStatus::CacheReadFailed(err.to_string()));
+                }
+                return Err(CacheLoadStatus::CacheFileNotFound);
+            }
+        };
+
+        let cache_file: CloudConfigBundleCacheFile = match serde_json::from_slice(&bytes) {
+            Ok(cache_file) => cache_file,
+            Err(err) => {
+                return Err(CacheLoadStatus::CacheParseFailed(err.to_string()));
+            }
+        };
+        let payload_bytes = match cache_payload_bytes(&cache_file.signed_payload) {
+            Some(payload_bytes) => payload_bytes,
+            None => {
+                return Err(CacheLoadStatus::CacheParseFailed(
+                    "failed to serialize cache payload".to_string(),
+                ));
+            }
+        };
+        if !verify_cache_signature(&payload_bytes, &cache_file.signature) {
+            return Err(CacheLoadStatus::CacheSignatureInvalid);
+        }
+        if cache_file.signed_payload.version != CLOUD_CONFIG_BUNDLE_CACHE_VERSION {
+            return Err(CacheLoadStatus::CacheVersionUnsupported(
+                cache_file.signed_payload.version,
+            ));
+        }
+
+        let (Some(cached_chatgpt_user_id), Some(cached_account_id)) = (
+            cache_file.signed_payload.chatgpt_user_id.as_deref(),
+            cache_file.signed_payload.account_id.as_deref(),
+        ) else {
+            return Err(CacheLoadStatus::CacheIdentityIncomplete);
+        };
+
+        if cached_chatgpt_user_id != chatgpt_user_id || cached_account_id != account_id {
+            return Err(CacheLoadStatus::CacheIdentityMismatch);
+        }
+
+        if cache_file.signed_payload.expires_at <= Utc::now() {
+            return Err(CacheLoadStatus::CacheExpired);
+        }
+
+        Ok(cache_file.signed_payload)
+    }
+
+    pub(super) fn log_load_status(&self, status: &CacheLoadStatus) {
+        if matches!(status, CacheLoadStatus::CacheFileNotFound) {
+            return;
+        }
+
+        let warn = matches!(
+            status,
+            CacheLoadStatus::CacheReadFailed(_)
+                | CacheLoadStatus::CacheParseFailed(_)
+                | CacheLoadStatus::CacheSignatureInvalid
+        );
+
+        if warn {
+            tracing::warn!(path = %self.path.display(), "{status}");
+        } else {
+            tracing::info!(path = %self.path.display(), "{status}");
+        }
+    }
+
+    pub(super) async fn save(
+        &self,
+        chatgpt_user_id: Option<String>,
+        account_id: Option<String>,
+        bundle: CloudConfigBundle,
+    ) -> Result<(), CloudConfigBundleCacheError> {
+        let now = Utc::now();
+        let expires_at = now
+            .checked_add_signed(
+                ChronoDuration::from_std(CLOUD_CONFIG_BUNDLE_CACHE_TTL)
+                    .map_err(|_| CloudConfigBundleCacheError)?,
+            )
+            .ok_or(CloudConfigBundleCacheError)?;
+        let signed_payload = CloudConfigBundleCacheSignedPayload {
+            version: CLOUD_CONFIG_BUNDLE_CACHE_VERSION,
+            cached_at: now,
+            expires_at,
+            chatgpt_user_id,
+            account_id,
+            bundle,
+        };
+        let payload_bytes =
+            cache_payload_bytes(&signed_payload).ok_or(CloudConfigBundleCacheError)?;
+        let serialized = serde_json::to_vec_pretty(&CloudConfigBundleCacheFile {
+            signature: sign_cache_payload(&payload_bytes).ok_or(CloudConfigBundleCacheError)?,
+            signed_payload,
+        })
+        .map_err(|_| CloudConfigBundleCacheError)?;
+
+        if let Some(parent) = self.path.parent() {
+            fs::create_dir_all(parent)
+                .await
+                .map_err(|_| CloudConfigBundleCacheError)?;
+        }
+
+        fs::write(&self.path, serialized)
+            .await
+            .map_err(|_| CloudConfigBundleCacheError)?;
+        Ok(())
+    }
+}
+
+#[derive(Clone, Debug, Eq, Error, PartialEq)]
+pub(super) enum CacheLoadStatus {
+    #[error("Skipping cloud config bundle cache read because auth identity is incomplete.")]
+    AuthIdentityIncomplete,
+    #[error("Cloud config bundle cache file not found.")]
+    CacheFileNotFound,
+    #[error("Failed to read cloud config bundle cache: {0}.")]
+    CacheReadFailed(String),
+    #[error("Failed to parse cloud config bundle cache: {0}.")]
+    CacheParseFailed(String),
+    #[error("Cloud config bundle cache failed signature verification.")]
+    CacheSignatureInvalid,
+    #[error("Ignoring cloud config bundle cache because cached identity is incomplete.")]
+    CacheIdentityIncomplete,
+    #[error("Ignoring cloud config bundle cache for different auth identity.")]
+    CacheIdentityMismatch,
+    #[error("Ignoring cloud config bundle cache with unsupported version {0}.")]
+    CacheVersionUnsupported(u32),
+    #[error("Cloud config bundle cache expired.")]
+    CacheExpired,
+    #[error("Ignoring cloud config bundle cache because the cached bundle is invalid.")]
+    CacheInvalidBundle,
+}
+
+#[derive(Debug, Error)]
+#[error("failed to write cloud config bundle cache")]
+pub(super) struct CloudConfigBundleCacheError;
+
+#[derive(Clone, Debug, Deserialize, Eq, PartialEq, Serialize)]
+pub(super) struct CloudConfigBundleCacheFile {
+    pub(super) signed_payload: CloudConfigBundleCacheSignedPayload,
+    pub(super) signature: String,
+}
+
+#[derive(Clone, Debug, Deserialize, Eq, PartialEq, Serialize)]
+pub(super) struct CloudConfigBundleCacheSignedPayload {
+    pub(super) version: u32,
+    pub(super) cached_at: DateTime<Utc>,
+    pub(super) expires_at: DateTime<Utc>,
+    pub(super) chatgpt_user_id: Option<String>,
+    pub(super) account_id: Option<String>,
+    pub(super) bundle: CloudConfigBundle,
+}
+
+pub(super) fn cache_payload_bytes(
+    payload: &CloudConfigBundleCacheSignedPayload,
+) -> Option<Vec<u8>> {
+    serde_json::to_vec(&payload).ok()
+}
+
+pub(super) fn sign_cache_payload(payload_bytes: &[u8]) -> Option<String> {
+    let mut mac = HmacSha256::new_from_slice(CLOUD_CONFIG_BUNDLE_CACHE_WRITE_HMAC_KEY).ok()?;
+    mac.update(payload_bytes);
+    let signature = mac.finalize().into_bytes();
+    Some(BASE64_STANDARD.encode(signature))
+}
+
+pub(super) fn verify_cache_signature(payload_bytes: &[u8], signature: &str) -> bool {
+    let signature_bytes = match BASE64_STANDARD.decode(signature) {
+        Ok(signature_bytes) => signature_bytes,
+        Err(_) => return false,
+    };
+
+    CLOUD_CONFIG_BUNDLE_CACHE_READ_HMAC_KEYS
+        .iter()
+        .any(|key| verify_cache_signature_with_key(payload_bytes, &signature_bytes, key))
+}
+
+fn verify_cache_signature_with_key(
+    payload_bytes: &[u8],
+    signature_bytes: &[u8],
+    key: &[u8],
+) -> bool {
+    let mut mac = match HmacSha256::new_from_slice(key) {
+        Ok(mac) => mac,
+        Err(_) => return false,
+    };
+    mac.update(payload_bytes);
+    mac.verify_slice(signature_bytes).is_ok()
+}
+
+#[cfg(test)]
+#[path = "cache_tests.rs"]
+mod tests;
@@ -0,0 +1,206 @@
+use super::*;
+use codex_config::AbsolutePathBuf;
+use codex_config::CloudConfigFragment;
+use codex_config::CloudConfigTomlBundle;
+use codex_config::CloudRequirementsFragment;
+use codex_config::CloudRequirementsTomlBundle;
+use pretty_assertions::assert_eq;
+use std::path::Path;
+use tempfile::tempdir;
+
+fn test_bundle() -> CloudConfigBundle {
+    CloudConfigBundle {
+        config_toml: CloudConfigTomlBundle {
+            enterprise_managed: vec![CloudConfigFragment {
+                id: "cfg_1".to_string(),
+                name: "Base config".to_string(),
+                contents: "model = \"gpt-5\"".to_string(),
+            }],
+        },
+        requirements_toml: CloudRequirementsTomlBundle {
+            enterprise_managed: vec![CloudRequirementsFragment {
+                id: "req_1".to_string(),
+                name: "Base requirements".to_string(),
+                contents: "allowed_approval_policies = [\"never\"]".to_string(),
+            }],
+        },
+    }
+}
+
+fn signed_cache_file(
+    signed_payload: CloudConfigBundleCacheSignedPayload,
+) -> CloudConfigBundleCacheFile {
+    let payload_bytes = cache_payload_bytes(&signed_payload).expect("payload bytes");
+    CloudConfigBundleCacheFile {
+        signature: sign_cache_payload(&payload_bytes).expect("signature"),
+        signed_payload,
+    }
+}
+
+fn valid_signed_payload() -> CloudConfigBundleCacheSignedPayload {
+    let cached_at = Utc::now();
+    CloudConfigBundleCacheSignedPayload {
+        version: CLOUD_CONFIG_BUNDLE_CACHE_VERSION,
+        cached_at,
+        expires_at: cached_at + ChronoDuration::minutes(30),
+        chatgpt_user_id: Some("user-12345".to_string()),
+        account_id: Some("account-12345".to_string()),
+        bundle: test_bundle(),
+    }
+}
+
+fn write_cache_file(cache: &CloudConfigBundleCache, cache_file: &CloudConfigBundleCacheFile) {
+    std::fs::write(
+        cache.path(),
+        serde_json::to_vec_pretty(cache_file).expect("serialize cache"),
+    )
+    .expect("write cache");
+}
+
+fn create_test_cache(codex_home: &Path) -> CloudConfigBundleCache {
+    CloudConfigBundleCache::new(AbsolutePathBuf::resolve_path_against_base(codex_home, "/"))
+}
+
+#[tokio::test]
+async fn save_writes_signed_payload_and_loads_for_matching_identity() {
+    let codex_home = tempdir().expect("tempdir");
+    let cache = create_test_cache(codex_home.path());
+    let bundle = test_bundle();
+
+    cache
+        .save(
+            Some("user-12345".to_string()),
+            Some("account-12345".to_string()),
+            bundle.clone(),
+        )
+        .await
+        .expect("save cache");
+
+    let cache_file: CloudConfigBundleCacheFile =
+        serde_json::from_slice(&std::fs::read(cache.path()).expect("read cache"))
+            .expect("parse cache");
+    assert!(
+        cache_file.signed_payload.expires_at
+            <= cache_file.signed_payload.cached_at + ChronoDuration::minutes(30)
+    );
+    assert!(cache_file.signed_payload.expires_at > cache_file.signed_payload.cached_at);
+    assert_eq!(
+        cache_file,
+        signed_cache_file(CloudConfigBundleCacheSignedPayload {
+            version: CLOUD_CONFIG_BUNDLE_CACHE_VERSION,
+            cached_at: cache_file.signed_payload.cached_at,
+            expires_at: cache_file.signed_payload.expires_at,
+            chatgpt_user_id: Some("user-12345".to_string()),
+            account_id: Some("account-12345".to_string()),
+            bundle,
+        })
+    );
+
+    assert_eq!(
+        cache.load(Some("user-12345"), Some("account-12345")).await,
+        Ok(cache_file.signed_payload)
+    );
+}
+
+#[tokio::test]
+async fn load_rejects_missing_request_identity_before_reading_cache_file() {
+    let codex_home = tempdir().expect("tempdir");
+    let cache = create_test_cache(codex_home.path());
+
+    assert_eq!(
+        cache
+            .load(/*chatgpt_user_id*/ None, Some("account-12345"))
+            .await,
+        Err(CacheLoadStatus::AuthIdentityIncomplete)
+    );
+    assert_eq!(
+        cache.load(Some("user-12345"), /*account_id*/ None).await,
+        Err(CacheLoadStatus::AuthIdentityIncomplete)
+    );
+}
+
+#[tokio::test]
+async fn load_reports_missing_and_malformed_cache_files() {
+    let codex_home = tempdir().expect("tempdir");
+    let cache = create_test_cache(codex_home.path());
+
+    assert_eq!(
+        cache.load(Some("user-12345"), Some("account-12345")).await,
+        Err(CacheLoadStatus::CacheFileNotFound)
+    );
+
+    std::fs::write(cache.path(), "{").expect("write malformed cache");
+    assert!(matches!(
+        cache.load(Some("user-12345"), Some("account-12345")).await,
+        Err(CacheLoadStatus::CacheParseFailed(_))
+    ));
+}
+
+#[tokio::test]
+async fn load_rejects_tampered_payload() {
+    let codex_home = tempdir().expect("tempdir");
+    let cache = create_test_cache(codex_home.path());
+    let mut cache_file = signed_cache_file(valid_signed_payload());
+    cache_file
+        .signed_payload
+        .bundle
+        .requirements_toml
+        .enterprise_managed[0]
+        .contents = "allowed_approval_policies = [\"on-request\"]".to_string();
+    write_cache_file(&cache, &cache_file);
+
+    assert_eq!(
+        cache.load(Some("user-12345"), Some("account-12345")).await,
+        Err(CacheLoadStatus::CacheSignatureInvalid)
+    );
+}
+
+#[tokio::test]
+async fn load_rejects_cache_for_incomplete_or_different_identity() {
+    let codex_home = tempdir().expect("tempdir");
+    let cache = create_test_cache(codex_home.path());
+    let cache_file = signed_cache_file(valid_signed_payload());
+    write_cache_file(&cache, &cache_file);
+
+    assert_eq!(
+        cache.load(Some("user-99999"), Some("account-12345")).await,
+        Err(CacheLoadStatus::CacheIdentityMismatch)
+    );
+
+    let mut signed_payload = valid_signed_payload();
+    signed_payload.chatgpt_user_id = None;
+    write_cache_file(&cache, &signed_cache_file(signed_payload));
+
+    assert_eq!(
+        cache.load(Some("user-12345"), Some("account-12345")).await,
+        Err(CacheLoadStatus::CacheIdentityIncomplete)
+    );
+}
+
+#[tokio::test]
+async fn load_rejects_expired_cache() {
+    let codex_home = tempdir().expect("tempdir");
+    let cache = create_test_cache(codex_home.path());
+    let mut signed_payload = valid_signed_payload();
+    signed_payload.expires_at = Utc::now() - ChronoDuration::seconds(1);
+    write_cache_file(&cache, &signed_cache_file(signed_payload));
+
+    assert_eq!(
+        cache.load(Some("user-12345"), Some("account-12345")).await,
+        Err(CacheLoadStatus::CacheExpired)
+    );
+}
+
+#[tokio::test]
+async fn load_rejects_unsupported_cache_version() {
+    let codex_home = tempdir().expect("tempdir");
+    let cache = create_test_cache(codex_home.path());
+    let mut signed_payload = valid_signed_payload();
+    signed_payload.version = 2;
+    write_cache_file(&cache, &signed_cache_file(signed_payload));
+
+    assert_eq!(
+        cache.load(Some("user-12345"), Some("account-12345")).await,
+        Err(CacheLoadStatus::CacheVersionUnsupported(2))
+    );
+}
@@ -0,0 +1,95 @@
+use codex_config::CloudConfigBundle;
+
+const CLOUD_CONFIG_BUNDLE_FETCH_ATTEMPT_METRIC: &str = "codex.cloud_config_bundle.fetch_attempt";
+const CLOUD_CONFIG_BUNDLE_FETCH_FINAL_METRIC: &str = "codex.cloud_config_bundle.fetch_final";
+const CLOUD_CONFIG_BUNDLE_LOAD_METRIC: &str = "codex.cloud_config_bundle.load";
+
+pub(crate) fn emit_fetch_attempt_metric(
+    trigger: &str,
+    attempt: usize,
+    outcome: &str,
+    status_code: Option<u16>,
+) {
+    let attempt_tag = attempt.to_string();
+    let status_code_tag = status_code_tag(status_code);
+    emit_metric(
+        CLOUD_CONFIG_BUNDLE_FETCH_ATTEMPT_METRIC,
+        vec![
+            ("trigger", trigger.to_string()),
+            ("attempt", attempt_tag),
+            ("outcome", outcome.to_string()),
+            ("status_code", status_code_tag),
+        ],
+    );
+}
+
+pub(crate) fn emit_fetch_final_metric(
+    trigger: &str,
+    outcome: &str,
+    reason: &str,
+    attempt_count: usize,
+    status_code: Option<u16>,
+    bundle: Option<&CloudConfigBundle>,
+) {
+    let attempt_count_tag = attempt_count.to_string();
+    let status_code_tag = status_code_tag(status_code);
+    emit_metric(
+        CLOUD_CONFIG_BUNDLE_FETCH_FINAL_METRIC,
+        vec![
+            ("trigger", trigger.to_string()),
+            ("outcome", outcome.to_string()),
+            ("reason", reason.to_string()),
+            ("attempt_count", attempt_count_tag),
+            ("status_code", status_code_tag),
+            ("bundle_shape", bundle_shape_tag(bundle)),
+        ],
+    );
+}
+
+pub(crate) fn emit_load_metric(trigger: &str, outcome: &str, bundle: Option<&CloudConfigBundle>) {
+    emit_metric(
+        CLOUD_CONFIG_BUNDLE_LOAD_METRIC,
+        vec![
+            ("trigger", trigger.to_string()),
+            ("outcome", outcome.to_string()),
+            ("bundle_shape", bundle_shape_tag(bundle)),
+        ],
+    );
+}
+
+pub(crate) fn bundle_shape_tag(bundle: Option<&CloudConfigBundle>) -> String {
+    let Some(bundle) = bundle else {
+        return "none".to_string();
+    };
+
+    let mut sources = Vec::new();
+    if !bundle.config_toml.enterprise_managed.is_empty() {
+        sources.push("enterprise_config");
+    }
+    if !bundle.requirements_toml.enterprise_managed.is_empty() {
+        sources.push("enterprise_requirements");
+    }
+
+    if sources.is_empty() {
+        "empty".to_string()
+    } else {
+        sources.sort_unstable();
+        sources.join(",")
+    }
+}
+
+fn status_code_tag(status_code: Option<u16>) -> String {
+    status_code
+        .map(|status_code| status_code.to_string())
+        .unwrap_or_else(|| "none".to_string())
+}
+
+fn emit_metric(metric_name: &str, tags: Vec<(&str, String)>) {
+    if let Some(metrics) = codex_otel::global() {
+        let tag_refs = tags
+            .iter()
+            .map(|(key, value)| (*key, value.as_str()))
+            .collect::<Vec<_>>();
+        let _ = metrics.counter(metric_name, /*inc*/ 1, &tag_refs);
+    }
+}
@@ -0,0 +1,500 @@
+//! 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.
+
+use crate::backend::BundleClient;
+use crate::backend::BundleRequestError;
+use crate::backend::RetryableFailureKind;
+use crate::cache::CacheLoadStatus;
+use crate::cache::CloudConfigBundleCache;
+use crate::metrics::emit_fetch_attempt_metric;
+use crate::metrics::emit_fetch_final_metric;
+use crate::metrics::emit_load_metric;
+use crate::validation::validate_bundle;
+use codex_config::AbsolutePathBuf;
+use codex_config::CloudConfigBundle;
+use codex_config::CloudConfigBundleLoadError;
+use codex_config::CloudConfigBundleLoadErrorCode;
+use codex_core::util::backoff;
+use codex_login::AuthManager;
+use codex_login::CodexAuth;
+use codex_login::RefreshTokenError;
+use codex_login::UnauthorizedRecovery;
+use codex_protocol::account::PlanType;
+use std::path::PathBuf;
+use std::sync::Arc;
+use std::time::Duration;
+use std::time::Instant;
+use tokio::time::sleep;
+use tokio::time::timeout;
+
+pub(crate) const CLOUD_CONFIG_BUNDLE_TIMEOUT: Duration = Duration::from_secs(15);
+const CLOUD_CONFIG_BUNDLE_MAX_ATTEMPTS: usize = 5;
+const CLOUD_CONFIG_BUNDLE_CACHE_REFRESH_INTERVAL: Duration = Duration::from_secs(5 * 60);
+const CLOUD_CONFIG_BUNDLE_LOAD_FAILED_MESSAGE: &str =
+    "Failed to load cloud config bundle (workspace-managed policies).";
+const CLOUD_CONFIG_BUNDLE_AUTH_RECOVERY_FAILED_MESSAGE: &str = concat!(
+    "Your authentication session could not be refreshed automatically. ",
+    "Please log out and sign in again."
+);
+
+fn auth_identity(auth: &CodexAuth) -> (Option<String>, Option<String>) {
+    (auth.get_chatgpt_user_id(), auth.get_account_id())
+}
+
+fn cloud_config_eligible_auth(auth: &CodexAuth) -> bool {
+    let Some(plan_type) = auth.account_plan_type() else {
+        return false;
+    };
+    auth.uses_codex_backend()
+        && (plan_type.is_business_like() || matches!(plan_type, PlanType::Enterprise))
+}
+
+fn optional_bundle(bundle: CloudConfigBundle) -> Option<CloudConfigBundle> {
+    if bundle.is_empty() {
+        None
+    } else {
+        Some(bundle)
+    }
+}
+
+enum CachedBundleLookup {
+    Hit(Option<CloudConfigBundle>),
+    Miss,
+}
+
+enum UnauthorizedRecoveryAction {
+    RetrySameAttempt,
+    RetryNextAttempt,
+}
+
+pub(crate) struct CloudConfigBundleService<C> {
+    auth_manager: Arc<AuthManager>,
+    client: Arc<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,
+        }
+    }
+}
+
+impl<C> CloudConfigBundleService<C>
+where
+    C: BundleClient + 'static,
+{
+    pub(crate) fn new(
+        auth_manager: Arc<AuthManager>,
+        client: Arc<C>,
+        codex_home: PathBuf,
+        timeout: Duration,
+    ) -> Self {
+        let codex_home = AbsolutePathBuf::resolve_path_against_base(codex_home, "/");
+        Self {
+            auth_manager,
+            client,
+            cache: CloudConfigBundleCache::new(codex_home.clone()),
+            codex_home,
+            timeout,
+        }
+    }
+
+    pub(crate) async fn load_startup_bundle_with_timeout(
+        &self,
+    ) -> Result<Option<CloudConfigBundle>, CloudConfigBundleLoadError> {
+        let _timer =
+            codex_otel::start_global_timer("codex.cloud_config_bundle.fetch.duration_ms", &[]);
+        let started_at = Instant::now();
+        let load_result = timeout(self.timeout, self.load_startup_bundle())
+            .await
+            .inspect_err(|_| {
+                let message = format!(
+                    "Timed out waiting for cloud config bundle after {}s",
+                    self.timeout.as_secs()
+                );
+                tracing::error!("{message}");
+                emit_load_metric("startup", "error", /*bundle*/ None);
+            })
+            .map_err(|_| {
+                CloudConfigBundleLoadError::new(
+                    CloudConfigBundleLoadErrorCode::Timeout,
+                    /*status_code*/ None,
+                    format!(
+                        "timed out waiting for cloud config bundle after {}s",
+                        self.timeout.as_secs()
+                    ),
+                )
+            })?;
+
+        let result = match load_result {
+            Ok(result) => result,
+            Err(err) => {
+                emit_load_metric("startup", "error", /*bundle*/ None);
+                return Err(err);
+            }
+        };
+
+        match result.as_ref() {
+            Some(bundle) => {
+                tracing::info!(
+                    elapsed_ms = started_at.elapsed().as_millis(),
+                    config_fragments = bundle.config_toml.enterprise_managed.len(),
+                    requirements_fragments = bundle.requirements_toml.enterprise_managed.len(),
+                    "Cloud config bundle load completed"
+                );
+                emit_load_metric("startup", "success", Some(bundle));
+            }
+            None => {
+                tracing::info!(
+                    elapsed_ms = started_at.elapsed().as_millis(),
+                    "Cloud config bundle load completed (none)"
+                );
+                emit_load_metric("startup", "success", /*bundle*/ None);
+            }
+        }
+
+        Ok(result)
+    }
+
+    async fn load_startup_bundle(
+        &self,
+    ) -> Result<Option<CloudConfigBundle>, CloudConfigBundleLoadError> {
+        let Some(auth) = self.auth_manager.auth().await else {
+            return Ok(None);
+        };
+        if !cloud_config_eligible_auth(&auth) {
+            return Ok(None);
+        }
+
+        // Startup prefers a valid, identity-matched cache entry. The backend is
+        // only consulted on cache miss or invalid cache contents.
+        let (chatgpt_user_id, account_id) = auth_identity(&auth);
+        match self
+            .load_valid_cached_bundle(chatgpt_user_id.as_deref(), account_id.as_deref())
+            .await
+        {
+            CachedBundleLookup::Hit(bundle) => return Ok(bundle),
+            CachedBundleLookup::Miss => {}
+        }
+
+        self.fetch_remote_bundle_and_update_cache_with_retries(auth, "startup")
+            .await
+    }
+
+    async fn load_valid_cached_bundle(
+        &self,
+        chatgpt_user_id: Option<&str>,
+        account_id: Option<&str>,
+    ) -> CachedBundleLookup {
+        match self.cache.load(chatgpt_user_id, account_id).await {
+            Ok(signed_payload) => {
+                if let Err(err) = validate_bundle(&signed_payload.bundle, &self.codex_home) {
+                    tracing::warn!(
+                        path = %self.cache.path().display(),
+                        error = %err,
+                        "Ignoring invalid cached cloud config bundle"
+                    );
+                    self.cache
+                        .log_load_status(&CacheLoadStatus::CacheInvalidBundle);
+                    CachedBundleLookup::Miss
+                } else {
+                    tracing::info!(
+                        path = %self.cache.path().display(),
+                        "Using cached cloud config bundle"
+                    );
+                    CachedBundleLookup::Hit(optional_bundle(signed_payload.bundle))
+                }
+            }
+            Err(cache_load_status) => {
+                self.cache.log_load_status(&cache_load_status);
+                CachedBundleLookup::Miss
+            }
+        }
+    }
+
+    async fn fetch_remote_bundle_and_update_cache_with_retries(
+        &self,
+        mut auth: CodexAuth,
+        trigger: &'static str,
+    ) -> Result<Option<CloudConfigBundle>, CloudConfigBundleLoadError> {
+        let mut attempt = 1;
+        let mut last_status_code: Option<u16> = None;
+        let mut auth_recovery = self.auth_manager.unauthorized_recovery();
+
+        while attempt <= CLOUD_CONFIG_BUNDLE_MAX_ATTEMPTS {
+            match self.client.get_bundle(&auth).await {
+                Ok(bundle) => {
+                    return self
+                        .validate_and_cache_remote_bundle(&auth, trigger, attempt, bundle)
+                        .await;
+                }
+                Err(BundleRequestError::Retryable(status)) => {
+                    last_status_code = status.status_code();
+                    if self
+                        .retry_after_request_failure(trigger, attempt, status)
+                        .await
+                    {
+                        attempt += 1;
+                        continue;
+                    }
+                }
+                Err(BundleRequestError::Unauthorized {
+                    status_code,
+                    message,
+                }) => {
+                    last_status_code = status_code;
+                    match self
+                        .handle_unauthorized(
+                            &mut auth,
+                            &mut auth_recovery,
+                            trigger,
+                            attempt,
+                            status_code,
+                            &message,
+                        )
+                        .await?
+                    {
+                        UnauthorizedRecoveryAction::RetrySameAttempt => continue,
+                        UnauthorizedRecoveryAction::RetryNextAttempt => {
+                            attempt += 1;
+                            continue;
+                        }
+                    }
+                }
+            }
+
+            break;
+        }
+
+        emit_fetch_final_metric(
+            trigger,
+            "error",
+            "request_retry_exhausted",
+            CLOUD_CONFIG_BUNDLE_MAX_ATTEMPTS,
+            last_status_code,
+            /*bundle*/ None,
+        );
+        tracing::error!(
+            path = %self.cache.path().display(),
+            "{CLOUD_CONFIG_BUNDLE_LOAD_FAILED_MESSAGE}"
+        );
+        Err(CloudConfigBundleLoadError::new(
+            CloudConfigBundleLoadErrorCode::RequestFailed,
+            last_status_code,
+            CLOUD_CONFIG_BUNDLE_LOAD_FAILED_MESSAGE,
+        ))
+    }
+
+    async fn validate_and_cache_remote_bundle(
+        &self,
+        auth: &CodexAuth,
+        trigger: &'static str,
+        attempt: usize,
+        bundle: CloudConfigBundle,
+    ) -> Result<Option<CloudConfigBundle>, CloudConfigBundleLoadError> {
+        emit_fetch_attempt_metric(trigger, attempt, "success", /*status_code*/ None);
+        if let Err(err) = validate_bundle(&bundle, &self.codex_home) {
+            emit_fetch_final_metric(
+                trigger,
+                "error",
+                "invalid_bundle",
+                attempt,
+                /*status_code*/ None,
+                /*bundle*/ None,
+            );
+            return Err(err);
+        }
+
+        let (chatgpt_user_id, account_id) = auth_identity(auth);
+        if let Err(err) = self
+            .cache
+            .save(chatgpt_user_id, account_id, bundle.clone())
+            .await
+        {
+            tracing::warn!(
+                error = %err,
+                "Failed to write cloud config bundle cache"
+            );
+        }
+
+        emit_fetch_final_metric(
+            trigger,
+            "success",
+            "none",
+            attempt,
+            /*status_code*/ None,
+            Some(&bundle),
+        );
+        Ok(optional_bundle(bundle))
+    }
+
+    async fn retry_after_request_failure(
+        &self,
+        trigger: &'static str,
+        attempt: usize,
+        status: RetryableFailureKind,
+    ) -> bool {
+        let status_code = status.status_code();
+        emit_fetch_attempt_metric(trigger, attempt, "error", status_code);
+        if attempt < CLOUD_CONFIG_BUNDLE_MAX_ATTEMPTS {
+            tracing::warn!(
+                status = ?status,
+                attempt,
+                max_attempts = CLOUD_CONFIG_BUNDLE_MAX_ATTEMPTS,
+                "Failed to fetch cloud config bundle; retrying"
+            );
+            sleep(backoff(attempt as u64)).await;
+            true
+        } else {
+            false
+        }
+    }
+
+    async fn handle_unauthorized(
+        &self,
+        auth: &mut CodexAuth,
+        auth_recovery: &mut UnauthorizedRecovery,
+        trigger: &'static str,
+        attempt: usize,
+        status_code: Option<u16>,
+        message: &str,
+    ) -> Result<UnauthorizedRecoveryAction, CloudConfigBundleLoadError> {
+        emit_fetch_attempt_metric(trigger, attempt, "unauthorized", status_code);
+        if auth_recovery.has_next() {
+            tracing::warn!(
+                attempt,
+                max_attempts = CLOUD_CONFIG_BUNDLE_MAX_ATTEMPTS,
+                "Cloud config bundle request was unauthorized; attempting auth recovery"
+            );
+            match auth_recovery.next().await {
+                Ok(_) => {
+                    let Some(refreshed_auth) = self.auth_manager.auth().await else {
+                        tracing::error!(
+                            "Auth recovery succeeded but no auth is available for cloud config bundle"
+                        );
+                        emit_fetch_final_metric(
+                            trigger,
+                            "error",
+                            "auth_recovery_missing_auth",
+                            attempt,
+                            status_code,
+                            /*bundle*/ None,
+                        );
+                        return Err(CloudConfigBundleLoadError::new(
+                            CloudConfigBundleLoadErrorCode::Auth,
+                            status_code,
+                            CLOUD_CONFIG_BUNDLE_AUTH_RECOVERY_FAILED_MESSAGE,
+                        ));
+                    };
+                    *auth = refreshed_auth;
+                    return Ok(UnauthorizedRecoveryAction::RetrySameAttempt);
+                }
+                Err(RefreshTokenError::Permanent(failed)) => {
+                    tracing::warn!(
+                        error = %failed,
+                        "Failed to recover from unauthorized cloud config bundle request"
+                    );
+                    emit_fetch_final_metric(
+                        trigger,
+                        "error",
+                        "auth_recovery_unrecoverable",
+                        attempt,
+                        status_code,
+                        /*bundle*/ None,
+                    );
+                    return Err(CloudConfigBundleLoadError::new(
+                        CloudConfigBundleLoadErrorCode::Auth,
+                        status_code,
+                        failed.message,
+                    ));
+                }
+                Err(RefreshTokenError::Transient(recovery_err)) => {
+                    if attempt < CLOUD_CONFIG_BUNDLE_MAX_ATTEMPTS {
+                        tracing::warn!(
+                            error = %recovery_err,
+                            attempt,
+                            max_attempts = CLOUD_CONFIG_BUNDLE_MAX_ATTEMPTS,
+                            "Failed to recover from unauthorized cloud config bundle request; retrying"
+                        );
+                        sleep(backoff(attempt as u64)).await;
+                    }
+                    return Ok(UnauthorizedRecoveryAction::RetryNextAttempt);
+                }
+            }
+        }
+
+        tracing::warn!(
+            error = %message,
+            "Cloud config bundle request was unauthorized and no auth recovery is available"
+        );
+        emit_fetch_final_metric(
+            trigger,
+            "error",
+            "auth_recovery_unavailable",
+            attempt,
+            status_code,
+            /*bundle*/ None,
+        );
+        Err(CloudConfigBundleLoadError::new(
+            CloudConfigBundleLoadErrorCode::Auth,
+            status_code,
+            CLOUD_CONFIG_BUNDLE_AUTH_RECOVERY_FAILED_MESSAGE,
+        ))
+    }
+
+    pub(crate) async fn refresh_cache_in_background(&self) {
+        loop {
+            sleep(CLOUD_CONFIG_BUNDLE_CACHE_REFRESH_INTERVAL).await;
+            match timeout(self.timeout, self.refresh_cache_once()).await {
+                Ok(true) => {}
+                Ok(false) => break,
+                Err(_) => {
+                    tracing::error!(
+                        "Timed out refreshing cloud config bundle cache from remote; keeping existing cache"
+                    );
+                    emit_load_metric("refresh", "error", /*bundle*/ None);
+                }
+            }
+        }
+    }
+
+    async fn refresh_cache_once(&self) -> bool {
+        let Some(auth) = self.auth_manager.auth().await else {
+            return false;
+        };
+        if !cloud_config_eligible_auth(&auth) {
+            return false;
+        }
+
+        match self
+            .fetch_remote_bundle_and_update_cache_with_retries(auth, "refresh")
+            .await
+        {
+            Ok(bundle) => emit_load_metric("refresh", "success", bundle.as_ref()),
+            Err(err) => {
+                tracing::error!(
+                    path = %self.cache.path().display(),
+                    error = %err,
+                    "Failed to refresh cloud config bundle cache from remote"
+                );
+                emit_load_metric("refresh", "error", /*bundle*/ None);
+            }
+        }
+        true
+    }
+}
+
+#[cfg(test)]
+#[path = "service_tests.rs"]
+mod tests;
@@ -0,0 +1,34 @@
+use codex_config::AbsolutePathBuf;
+use codex_config::CloudConfigBundle;
+use codex_config::CloudConfigBundleLayers;
+use codex_config::CloudConfigBundleLoadError;
+use codex_config::CloudConfigBundleLoadErrorCode;
+use codex_config::compose_requirements;
+
+pub(crate) fn validate_bundle(
+    bundle: &CloudConfigBundle,
+    base_dir: &AbsolutePathBuf,
+) -> Result<(), CloudConfigBundleLoadError> {
+    let bundle_layers =
+        CloudConfigBundleLayers::from_bundle(bundle.clone(), base_dir).map_err(|err| {
+            CloudConfigBundleLoadError::new(
+                CloudConfigBundleLoadErrorCode::InvalidBundle,
+                /*status_code*/ None,
+                format!("invalid cloud config bundle: {err}"),
+            )
+        })?;
+    let CloudConfigBundleLayers {
+        enterprise_managed_config: _,
+        enterprise_managed_requirements,
+    } = bundle_layers;
+
+    compose_requirements(enterprise_managed_requirements).map_err(|err| {
+        CloudConfigBundleLoadError::new(
+            CloudConfigBundleLoadErrorCode::InvalidBundle,
+            /*status_code*/ None,
+            format!("invalid cloud config bundle: {err}"),
+        )
+    })?;
+
+    Ok(())
+}