From 553bd29d672fa07546f5c4cb1c6b770c32e6df55 Mon Sep 17 00:00:00 2001 From: viyatb-oai Date: Mon, 13 Apr 2026 12:33:40 -0700 Subject: [PATCH] fix: use stale cloud requirements cache fallback Co-authored-by: Codex noreply@openai.com --- codex-rs/Cargo.lock | 6 + codex-rs/cli/Cargo.toml | 6 + .../cli/tests/cloud_requirements_startup.rs | 336 +++++++++++++++++ codex-rs/cloud-requirements/src/lib.rs | 346 +++++++++++++++--- 4 files changed, 640 insertions(+), 54 deletions(-) create mode 100644 codex-rs/cli/tests/cloud_requirements_startup.rs diff --git a/codex-rs/Cargo.lock b/codex-rs/Cargo.lock index 254c192ef4..ae81d404a6 100644 --- a/codex-rs/Cargo.lock +++ b/codex-rs/Cargo.lock @@ -1667,6 +1667,8 @@ dependencies = [ "anyhow", "assert_cmd", "assert_matches", + "base64 0.22.1", + "chrono", "clap", "clap_complete", "codex-app-server", @@ -1695,13 +1697,17 @@ dependencies = [ "codex-utils-cargo-bin", "codex-utils-cli", "codex-utils-path", + "codex-utils-pty", "codex-windows-sandbox", + "hmac", "libc", "owo-colors", "predicates", "pretty_assertions", "regex-lite", + "serde", "serde_json", + "sha2", "sqlx", "supports-color 3.0.2", "tempfile", diff --git a/codex-rs/cli/Cargo.toml b/codex-rs/cli/Cargo.toml index ab0d86de2c..a7b49ebacb 100644 --- a/codex-rs/cli/Cargo.toml +++ b/codex-rs/cli/Cargo.toml @@ -69,7 +69,13 @@ codex_windows_sandbox = { package = "codex-windows-sandbox", path = "../windows- [dev-dependencies] assert_cmd = { workspace = true } assert_matches = { workspace = true } +base64 = { workspace = true } +chrono = { workspace = true, features = ["serde"] } codex-utils-cargo-bin = { workspace = true } +codex-utils-pty = { workspace = true } +hmac = { workspace = true } predicates = { workspace = true } pretty_assertions = { workspace = true } +serde = { workspace = true, features = ["derive"] } +sha2 = { workspace = true } sqlx = { workspace = true } diff --git a/codex-rs/cli/tests/cloud_requirements_startup.rs b/codex-rs/cli/tests/cloud_requirements_startup.rs new file mode 100644 index 0000000000..be597b4768 --- /dev/null +++ b/codex-rs/cli/tests/cloud_requirements_startup.rs @@ -0,0 +1,336 @@ +use base64::Engine; +use base64::engine::general_purpose::STANDARD as BASE64_STANDARD; +use base64::engine::general_purpose::URL_SAFE_NO_PAD; +use chrono::DateTime; +use chrono::Duration as ChronoDuration; +use chrono::Utc; +use hmac::Hmac; +use hmac::Mac; +use serde::Serialize; +use serde_json::json; +use sha2::Sha256; +use std::collections::HashMap; +use std::io::Read; +use std::io::Write; +use std::net::SocketAddr; +use std::net::TcpListener; +use std::net::TcpStream; +use std::path::Path; +use std::sync::Arc; +use std::sync::atomic::AtomicBool; +use std::sync::atomic::AtomicUsize; +use std::sync::atomic::Ordering; +use std::thread::JoinHandle; +use std::time::Duration; +use tokio::select; +use tokio::time::timeout; + +const CLOUD_REQUIREMENTS_CACHE_FILENAME: &str = "cloud-requirements-cache.json"; +const CLOUD_REQUIREMENTS_CACHE_HMAC_KEY: &[u8] = + b"codex-cloud-requirements-cache-v3-064f8542-75b4-494c-a294-97d3ce597271"; + +type HmacSha256 = Hmac; + +#[tokio::test] +async fn cloud_requirements_cli_error_includes_actionable_context() -> anyhow::Result<()> { + if cfg!(windows) { + return Ok(()); + } + + let tmp = tempfile::tempdir()?; + let codex_home = tmp.path(); + let cwd = std::env::current_dir()?; + let backend = FailingBackend::start()?; + write_chatgpt_business_auth(codex_home)?; + write_startup_config(codex_home, &cwd, backend.url())?; + + let CodexCliOutput { exit_code, output } = run_codex_cli(codex_home, &cwd).await?; + + assert_ne!(0, exit_code, "Codex CLI should exit nonzero."); + assert!( + output.contains("Error loading configuration: failed to load your workspace-managed config after 5 attempts"), + "expected cloud requirements load error in output, got: {output}" + ); + assert!( + output.contains("last backend status: 503"), + "expected backend status in output, got: {output}" + ); + assert!( + output.contains("valid cached copy"), + "expected cache guidance in output, got: {output}" + ); + assert!( + backend.request_count() > 0, + "expected CLI to request cloud requirements from fake backend" + ); + Ok(()) +} + +#[tokio::test] +async fn stale_cloud_requirements_cache_allows_actual_cli_startup_when_remote_fails() +-> anyhow::Result<()> { + if cfg!(windows) { + return Ok(()); + } + + let tmp = tempfile::tempdir()?; + let codex_home = tmp.path(); + let cwd = std::env::current_dir()?; + let backend = FailingBackend::start()?; + std::fs::write( + codex_home.join("rules"), + "rules should be a directory not a file", + )?; + write_chatgpt_business_auth(codex_home)?; + write_stale_cloud_requirements_cache( + codex_home, + r#"allowed_approval_policies = ["on-request"]"#, + )?; + write_startup_config(codex_home, &cwd, backend.url())?; + + let CodexCliOutput { exit_code, output } = run_codex_cli(codex_home, &cwd).await?; + + assert_ne!(0, exit_code, "Codex CLI should exit nonzero."); + assert!( + backend.request_count() > 0, + "expected CLI to retry cloud requirements before using stale cache" + ); + assert!( + output.contains("Error loading rules:"), + "expected startup to proceed beyond config loading, got: {output}" + ); + assert!( + output.contains("failed to read rules files"), + "expected controlled post-config startup error in output, got: {output}" + ); + assert!( + !output.contains("failed to load your workspace-managed config"), + "stale cloud requirements cache should avoid config-load failure, got: {output}" + ); + Ok(()) +} + +struct CodexCliOutput { + exit_code: i32, + output: String, +} + +async fn run_codex_cli( + codex_home: impl AsRef, + cwd: impl AsRef, +) -> anyhow::Result { + let codex_cli = codex_utils_cargo_bin::cargo_bin("codex")?; + let mut env = HashMap::new(); + env.insert( + "CODEX_HOME".to_string(), + codex_home.as_ref().display().to_string(), + ); + + let args = vec!["-c".to_string(), "analytics.enabled=false".to_string()]; + let spawned = codex_utils_pty::spawn_pty_process( + codex_cli.to_string_lossy().as_ref(), + &args, + cwd.as_ref(), + &env, + &None, + codex_utils_pty::TerminalSize::default(), + ) + .await?; + let mut output = Vec::new(); + let codex_utils_pty::SpawnedProcess { + session, + stdout_rx, + stderr_rx, + exit_rx, + } = spawned; + let mut output_rx = codex_utils_pty::combine_output_receivers(stdout_rx, stderr_rx); + let mut exit_rx = exit_rx; + let writer_tx = session.writer_sender(); + let exit_code_result = timeout(Duration::from_secs(10), async { + loop { + select! { + result = output_rx.recv() => match result { + Ok(chunk) => { + if chunk.windows(4).any(|window| window == b"\x1b[6n") { + let _ = writer_tx.send(b"\x1b[1;1R".to_vec()).await; + } + output.extend_from_slice(&chunk); + } + Err(tokio::sync::broadcast::error::RecvError::Closed) => break exit_rx.await, + Err(tokio::sync::broadcast::error::RecvError::Lagged(_)) => {} + }, + result = &mut exit_rx => break result, + } + } + }) + .await; + let exit_code = match exit_code_result { + Ok(Ok(code)) => code, + Ok(Err(err)) => return Err(err.into()), + Err(_) => { + session.terminate(); + anyhow::bail!("timed out waiting for codex CLI to exit"); + } + }; + while let Ok(chunk) = output_rx.try_recv() { + output.extend_from_slice(&chunk); + } + + let output = String::from_utf8_lossy(&output); + Ok(CodexCliOutput { + exit_code, + output: output.to_string(), + }) +} + +#[derive(Serialize)] +struct CloudRequirementsCacheFile { + signed_payload: CloudRequirementsCacheSignedPayload, + signature: String, +} + +#[derive(Clone, Serialize)] +struct CloudRequirementsCacheSignedPayload { + cached_at: DateTime, + expires_at: DateTime, + chatgpt_user_id: Option, + account_id: Option, + contents: Option, +} + +struct FailingBackend { + addr: SocketAddr, + request_count: Arc, + shutdown: Arc, + handle: Option>, +} + +impl FailingBackend { + fn start() -> anyhow::Result { + let listener = TcpListener::bind("127.0.0.1:0")?; + let addr = listener.local_addr()?; + listener.set_nonblocking(true)?; + let request_count = Arc::new(AtomicUsize::new(0)); + let shutdown = Arc::new(AtomicBool::new(false)); + let thread_request_count = request_count.clone(); + let thread_shutdown = shutdown.clone(); + let handle = std::thread::spawn(move || { + while !thread_shutdown.load(Ordering::SeqCst) { + match listener.accept() { + Ok((mut stream, _)) => { + let mut buffer = [0; 4096]; + let _ = stream.read(&mut buffer); + thread_request_count.fetch_add(1, Ordering::SeqCst); + let _ = stream.write_all( + b"HTTP/1.1 503 Service Unavailable\r\nContent-Length: 0\r\nConnection: close\r\n\r\n", + ); + } + Err(err) if err.kind() == std::io::ErrorKind::WouldBlock => { + std::thread::sleep(Duration::from_millis(10)); + } + Err(_) => break, + } + } + }); + + Ok(Self { + addr, + request_count, + shutdown, + handle: Some(handle), + }) + } + + fn url(&self) -> String { + format!("http://{}/backend-api", self.addr) + } + + fn request_count(&self) -> usize { + self.request_count.load(Ordering::SeqCst) + } +} + +impl Drop for FailingBackend { + fn drop(&mut self) { + self.shutdown.store(true, Ordering::SeqCst); + let _ = TcpStream::connect(self.addr); + if let Some(handle) = self.handle.take() { + let _ = handle.join(); + } + } +} + +fn write_startup_config( + codex_home: &Path, + cwd: impl AsRef, + chatgpt_base_url: String, +) -> anyhow::Result<()> { + let cwd = cwd.as_ref(); + let config_contents = format!( + r#" +chatgpt_base_url = "{chatgpt_base_url}" +cli_auth_credentials_store = "file" +model_provider = "ollama" + +[projects] +"{cwd}" = {{ trust_level = "trusted" }} +"#, + cwd = cwd.display() + ); + std::fs::write(codex_home.join("config.toml"), config_contents)?; + Ok(()) +} + +fn write_chatgpt_business_auth(codex_home: &Path) -> anyhow::Result<()> { + let header_b64 = URL_SAFE_NO_PAD.encode(serde_json::to_vec(&json!({ + "alg": "none", + "typ": "JWT", + }))?); + let payload_b64 = URL_SAFE_NO_PAD.encode(serde_json::to_vec(&json!({ + "email": "user@example.com", + "https://api.openai.com/auth": { + "chatgpt_plan_type": "business", + "chatgpt_user_id": "user-12345", + "user_id": "user-12345", + }, + }))?); + let signature_b64 = URL_SAFE_NO_PAD.encode(b"sig"); + let fake_jwt = format!("{header_b64}.{payload_b64}.{signature_b64}"); + let auth_json = json!({ + "OPENAI_API_KEY": null, + "tokens": { + "id_token": fake_jwt, + "access_token": "test-access-token", + "refresh_token": "test-refresh-token", + "account_id": "account-12345", + }, + "last_refresh": "2025-01-01T00:00:00Z", + }); + std::fs::write( + codex_home.join("auth.json"), + serde_json::to_vec_pretty(&auth_json)?, + )?; + Ok(()) +} + +fn write_stale_cloud_requirements_cache(codex_home: &Path, contents: &str) -> anyhow::Result<()> { + let signed_payload = CloudRequirementsCacheSignedPayload { + cached_at: Utc::now() - ChronoDuration::minutes(31), + expires_at: Utc::now() - ChronoDuration::minutes(1), + chatgpt_user_id: Some("user-12345".to_string()), + account_id: Some("account-12345".to_string()), + contents: Some(contents.to_string()), + }; + let payload_bytes = serde_json::to_vec(&signed_payload)?; + let mut mac = HmacSha256::new_from_slice(CLOUD_REQUIREMENTS_CACHE_HMAC_KEY)?; + mac.update(&payload_bytes); + let cache_file = CloudRequirementsCacheFile { + signed_payload, + signature: BASE64_STANDARD.encode(mac.finalize().into_bytes()), + }; + std::fs::write( + codex_home.join(CLOUD_REQUIREMENTS_CACHE_FILENAME), + serde_json::to_vec_pretty(&cache_file)?, + )?; + Ok(()) +} diff --git a/codex-rs/cloud-requirements/src/lib.rs b/codex-rs/cloud-requirements/src/lib.rs index 6ec5960289..d9752dd6f4 100644 --- a/codex-rs/cloud-requirements/src/lib.rs +++ b/codex-rs/cloud-requirements/src/lib.rs @@ -47,6 +47,7 @@ const CLOUD_REQUIREMENTS_MAX_ATTEMPTS: usize = 5; const CLOUD_REQUIREMENTS_CACHE_FILENAME: &str = "cloud-requirements-cache.json"; const CLOUD_REQUIREMENTS_CACHE_REFRESH_INTERVAL: Duration = Duration::from_secs(5 * 60); const CLOUD_REQUIREMENTS_CACHE_TTL: Duration = Duration::from_secs(30 * 60); +const CLOUD_REQUIREMENTS_STALE_CACHE_TTL: Duration = Duration::from_secs(4 * 60 * 60); const CLOUD_REQUIREMENTS_FETCH_ATTEMPT_METRIC: &str = "codex.cloud_requirements.fetch_attempt"; const CLOUD_REQUIREMENTS_FETCH_FINAL_METRIC: &str = "codex.cloud_requirements.fetch_final"; const CLOUD_REQUIREMENTS_LOAD_METRIC: &str = "codex.cloud_requirements.load"; @@ -136,6 +137,51 @@ impl CloudRequirementsCacheSignedPayload { .as_deref() .and_then(|contents| parse_cloud_requirements(contents).ok().flatten()) } + + fn is_fresh(&self, now: DateTime) -> bool { + self.expires_at > now + } + + fn can_use_stale(&self, now: DateTime) -> bool { + if self.is_fresh(now) { + return false; + } + + let Ok(stale_ttl) = ChronoDuration::from_std(CLOUD_REQUIREMENTS_STALE_CACHE_TTL) else { + return false; + }; + self.expires_at + .checked_add_signed(stale_ttl) + .is_some_and(|stale_expires_at| stale_expires_at > now) + } +} + +fn is_cloud_requirements_eligible_auth(auth: &CodexAuth) -> bool { + let Some(plan_type) = auth.account_plan_type() else { + return false; + }; + + auth.is_chatgpt_auth() + && (plan_type.is_business_like() || matches!(plan_type, PlanType::Enterprise)) +} + +fn format_cloud_requirements_load_failed_message( + attempt_count: usize, + status_code: Option, +) -> String { + let status = status_code + .map(|status_code| format!("last backend status: {status_code}")) + .unwrap_or_else(|| "last failure did not include an HTTP status".to_string()); + format!( + "{CLOUD_REQUIREMENTS_LOAD_FAILED_MESSAGE} after {attempt_count} attempts ({status}). Codex could not use a valid cached copy for this account, so startup stopped to avoid running without managed requirements. Please try again." + ) +} + +fn format_cloud_requirements_timeout_message(timeout: Duration) -> String { + format!( + "timed out waiting for workspace-managed config after {}s. Codex could not use a valid cached copy for this account, so startup stopped to avoid running without managed requirements. Please try again.", + timeout.as_secs() + ) } fn sign_cache_payload(payload_bytes: &[u8]) -> Option { let mut mac = HmacSha256::new_from_slice(CLOUD_REQUIREMENTS_CACHE_WRITE_HMAC_KEY).ok()?; @@ -274,26 +320,29 @@ impl CloudRequirementsService { let _timer = codex_otel::start_global_timer("codex.cloud_requirements.fetch.duration_ms", &[]); let started_at = Instant::now(); - let fetch_result = timeout(self.timeout, self.fetch()) - .await - .inspect_err(|_| { - let message = format!( - "Timed out waiting for cloud requirements after {}s", - self.timeout.as_secs() - ); - tracing::error!("{message}"); - emit_load_metric("startup", "error"); - }) - .map_err(|_| { - CloudRequirementsLoadError::new( + let fetch_result = match timeout(self.timeout, self.fetch()).await { + Ok(fetch_result) => fetch_result, + Err(_) => { + let err = CloudRequirementsLoadError::new( CloudRequirementsLoadErrorCode::Timeout, /*status_code*/ None, - format!( - "timed out waiting for cloud requirements after {}s", - self.timeout.as_secs() - ), - ) - })?; + format_cloud_requirements_timeout_message(self.timeout), + ); + if let Some(stale_cache) = self.load_stale_cache_for_current_auth().await { + tracing::warn!( + path = %self.cache_path.display(), + expires_at = %stale_cache.expires_at, + error = %err, + "Using stale cloud requirements cache after remote load timed out" + ); + Ok(stale_cache.requirements()) + } else { + tracing::error!("{err}"); + emit_load_metric("startup", "error"); + return Err(err); + } + } + }; let result = match fetch_result { Ok(result) => result, @@ -328,33 +377,57 @@ impl CloudRequirementsService { let Some(auth) = self.auth_manager.auth().await else { return Ok(None); }; - let Some(plan_type) = auth.account_plan_type() else { - return Ok(None); - }; - if !auth.is_chatgpt_auth() - || !(plan_type.is_business_like() || matches!(plan_type, PlanType::Enterprise)) - { + if !is_cloud_requirements_eligible_auth(&auth) { return Ok(None); } let (chatgpt_user_id, account_id) = auth_identity(&auth); - match self - .load_cache(chatgpt_user_id.as_deref(), account_id.as_deref()) + let cache_payload = match self + .load_cache_payload(chatgpt_user_id.as_deref(), account_id.as_deref()) .await { Ok(signed_payload) => { - tracing::info!( - path = %self.cache_path.display(), - "Using cached cloud requirements" - ); - return Ok(signed_payload.requirements()); + let now = Utc::now(); + if signed_payload.is_fresh(now) { + tracing::info!( + path = %self.cache_path.display(), + "Using cached cloud requirements" + ); + return Ok(signed_payload.requirements()); + } + self.log_cache_load_status(&CacheLoadStatus::CacheExpired); + Some(signed_payload) } Err(cache_load_status) => { self.log_cache_load_status(&cache_load_status); + None + } + }; + let stale_cache = + cache_payload.filter(|signed_payload| signed_payload.can_use_stale(Utc::now())); + + match self.fetch_with_retries(auth, "startup").await { + Ok(requirements) => Ok(requirements), + Err(err) => { + if err.code() == CloudRequirementsLoadErrorCode::RequestFailed + && let Some(stale_cache) = stale_cache + { + tracing::warn!( + path = %self.cache_path.display(), + expires_at = %stale_cache.expires_at, + error = %err, + "Using stale cloud requirements cache after remote load failed" + ); + return Ok(stale_cache.requirements()); + } + tracing::error!( + path = %self.cache_path.display(), + error = %err, + "Failed to load cloud requirements" + ); + Err(err) } } - - self.fetch_with_retries(auth, "startup").await } async fn fetch_with_retries( @@ -517,17 +590,41 @@ impl CloudRequirementsService { CLOUD_REQUIREMENTS_MAX_ATTEMPTS, last_status_code, ); - tracing::error!( - path = %self.cache_path.display(), - "{CLOUD_REQUIREMENTS_LOAD_FAILED_MESSAGE}" - ); Err(CloudRequirementsLoadError::new( CloudRequirementsLoadErrorCode::RequestFailed, last_status_code, - CLOUD_REQUIREMENTS_LOAD_FAILED_MESSAGE, + format_cloud_requirements_load_failed_message( + CLOUD_REQUIREMENTS_MAX_ATTEMPTS, + last_status_code, + ), )) } + async fn load_stale_cache_for_current_auth( + &self, + ) -> Option { + let auth = self.auth_manager.auth().await?; + if !is_cloud_requirements_eligible_auth(&auth) { + return None; + } + + let (chatgpt_user_id, account_id) = auth_identity(&auth); + let cache_payload = match self + .load_cache_payload(chatgpt_user_id.as_deref(), account_id.as_deref()) + .await + { + Ok(cache_payload) => cache_payload, + Err(cache_load_status) => { + self.log_cache_load_status(&cache_load_status); + return None; + } + }; + + cache_payload + .can_use_stale(Utc::now()) + .then_some(cache_payload) + } + async fn refresh_cache_in_background(&self) { loop { sleep(CLOUD_REQUIREMENTS_CACHE_REFRESH_INTERVAL).await; @@ -548,12 +645,7 @@ impl CloudRequirementsService { let Some(auth) = self.auth_manager.auth().await else { return false; }; - let Some(plan_type) = auth.account_plan_type() else { - return false; - }; - if !auth.is_chatgpt_auth() - || !(plan_type.is_business_like() || matches!(plan_type, PlanType::Enterprise)) - { + if !is_cloud_requirements_eligible_auth(&auth) { return false; } @@ -571,7 +663,7 @@ impl CloudRequirementsService { true } - async fn load_cache( + async fn load_cache_payload( &self, chatgpt_user_id: Option<&str>, account_id: Option<&str>, @@ -619,10 +711,6 @@ impl CloudRequirementsService { return Err(CacheLoadStatus::CacheIdentityMismatch); } - if cache_file.signed_payload.expires_at <= Utc::now() { - return Err(CacheLoadStatus::CacheExpired); - } - Ok(cache_file.signed_payload) } @@ -846,6 +934,20 @@ mod tests { Ok(()) } + fn write_signed_cache(codex_home: &Path, signed_payload: CloudRequirementsCacheSignedPayload) { + let payload_bytes = cache_payload_bytes(&signed_payload).expect("payload"); + let signature = sign_cache_payload(&payload_bytes).expect("sign payload"); + let cache_file = CloudRequirementsCacheFile { + signed_payload, + signature, + }; + std::fs::write( + codex_home.join(CLOUD_REQUIREMENTS_CACHE_FILENAME), + serde_json::to_vec_pretty(&cache_file).expect("serialize cache"), + ) + .expect("write cache"); + } + fn auth_manager_with_api_key() -> Arc { let tmp = tempdir().expect("tempdir"); let auth_json = json!({ @@ -1007,6 +1109,19 @@ mod tests { FetchAttemptError::Retryable(RetryableFailureKind::Request { status_code: None }) } + fn request_error_with_status(status_code: u16) -> FetchAttemptError { + FetchAttemptError::Retryable(RetryableFailureKind::Request { + status_code: Some(status_code), + }) + } + + fn approval_policy_requirements(policy: AskForApproval) -> ConfigRequirementsToml { + ConfigRequirementsToml { + allowed_approval_policies: Some(vec![policy]), + ..Default::default() + } + } + struct StaticFetcher { contents: Option, } @@ -1322,7 +1437,11 @@ enabled = false let err = result.expect_err("cloud requirements timeout should fail closed"); assert!( err.to_string() - .contains("timed out waiting for cloud requirements") + .contains("timed out waiting for workspace-managed config") + ); + assert!( + err.to_string() + .contains("Codex could not use a valid cached copy") ); } @@ -1951,6 +2070,125 @@ enabled = false assert_eq!(fetcher.request_count.load(Ordering::SeqCst), 1); } + #[tokio::test(start_paused = true)] + async fn fetch_cloud_requirements_uses_stale_cache_when_remote_fetch_fails() { + let codex_home = tempdir().expect("tempdir"); + write_signed_cache( + codex_home.path(), + CloudRequirementsCacheSignedPayload { + cached_at: Utc::now() - ChronoDuration::minutes(31), + expires_at: Utc::now() - ChronoDuration::minutes(1), + chatgpt_user_id: Some("user-12345".to_string()), + account_id: Some("account-12345".to_string()), + contents: Some(r#"allowed_approval_policies = ["on-request"]"#.to_string()), + }, + ); + + let retryable_error = Err(request_error_with_status(503)); + let fetcher = Arc::new(SequenceFetcher::new(vec![ + retryable_error; + CLOUD_REQUIREMENTS_MAX_ATTEMPTS + ])); + let service = CloudRequirementsService::new( + auth_manager_with_plan("business"), + fetcher.clone(), + codex_home.path().to_path_buf(), + CLOUD_REQUIREMENTS_TIMEOUT, + ); + + let handle = tokio::spawn(async move { service.fetch().await }); + tokio::task::yield_now().await; + tokio::time::advance(Duration::from_secs(5)).await; + tokio::task::yield_now().await; + + assert_eq!( + handle.await.expect("cloud requirements task"), + Ok(Some(approval_policy_requirements( + AskForApproval::OnRequest + ))) + ); + assert_eq!( + fetcher.request_count.load(Ordering::SeqCst), + CLOUD_REQUIREMENTS_MAX_ATTEMPTS + ); + } + + #[tokio::test(start_paused = true)] + async fn fetch_cloud_requirements_rejects_stale_cache_after_grace_period() { + let codex_home = tempdir().expect("tempdir"); + let stale_ttl = + ChronoDuration::from_std(CLOUD_REQUIREMENTS_STALE_CACHE_TTL).expect("stale cache TTL"); + write_signed_cache( + codex_home.path(), + CloudRequirementsCacheSignedPayload { + cached_at: Utc::now() - stale_ttl - ChronoDuration::minutes(31), + expires_at: Utc::now() - stale_ttl - ChronoDuration::minutes(1), + chatgpt_user_id: Some("user-12345".to_string()), + account_id: Some("account-12345".to_string()), + contents: Some(r#"allowed_approval_policies = ["on-request"]"#.to_string()), + }, + ); + + let retryable_error = Err(request_error_with_status(503)); + let fetcher = Arc::new(SequenceFetcher::new(vec![ + retryable_error; + CLOUD_REQUIREMENTS_MAX_ATTEMPTS + ])); + let service = CloudRequirementsService::new( + auth_manager_with_plan("business"), + fetcher.clone(), + codex_home.path().to_path_buf(), + CLOUD_REQUIREMENTS_TIMEOUT, + ); + + let handle = tokio::spawn(async move { service.fetch().await }); + tokio::task::yield_now().await; + tokio::time::advance(Duration::from_secs(5)).await; + tokio::task::yield_now().await; + + let err = handle + .await + .expect("cloud requirements task") + .expect_err("too-stale cache should not be used"); + let err_text = err.to_string(); + assert!(err_text.contains("failed to load your workspace-managed config after 5 attempts")); + assert!(err_text.contains("last backend status: 503")); + assert!(err_text.contains("valid cached copy")); + assert_eq!(err.code(), CloudRequirementsLoadErrorCode::RequestFailed); + } + + #[tokio::test(start_paused = true)] + async fn fetch_cloud_requirements_uses_stale_cache_when_remote_fetch_times_out() { + let codex_home = tempdir().expect("tempdir"); + write_signed_cache( + codex_home.path(), + CloudRequirementsCacheSignedPayload { + cached_at: Utc::now() - ChronoDuration::minutes(31), + expires_at: Utc::now() - ChronoDuration::minutes(1), + chatgpt_user_id: Some("user-12345".to_string()), + account_id: Some("account-12345".to_string()), + contents: Some(r#"allowed_approval_policies = ["on-request"]"#.to_string()), + }, + ); + + let service = CloudRequirementsService::new( + auth_manager_with_plan("business"), + Arc::new(PendingFetcher), + codex_home.path().to_path_buf(), + CLOUD_REQUIREMENTS_TIMEOUT, + ); + + let handle = tokio::spawn(async move { service.fetch_with_timeout().await }); + tokio::time::advance(CLOUD_REQUIREMENTS_TIMEOUT + Duration::from_millis(1)).await; + + assert_eq!( + handle.await.expect("cloud requirements task"), + Ok(Some(approval_policy_requirements( + AskForApproval::OnRequest + ))) + ); + } + #[tokio::test] async fn fetch_cloud_requirements_writes_signed_cache() { let codex_home = tempdir().expect("tempdir"); @@ -2048,10 +2286,10 @@ enabled = false .await .expect("cloud requirements task") .expect_err("cloud requirements retry exhaustion should fail closed"); - assert_eq!( - err.to_string(), - "failed to load your workspace-managed config" - ); + let err_text = err.to_string(); + assert!(err_text.contains("failed to load your workspace-managed config after 5 attempts")); + assert!(err_text.contains("last failure did not include an HTTP status")); + assert!(err_text.contains("valid cached copy")); assert_eq!(err.code(), CloudRequirementsLoadErrorCode::RequestFailed); assert_eq!( fetcher.request_count.load(Ordering::SeqCst),