diff --git a/src/providers/codex/auth/manager.rs b/src/providers/codex/auth/manager.rs index 2226e14..d779e2d 100644 --- a/src/providers/codex/auth/manager.rs +++ b/src/providers/codex/auth/manager.rs @@ -27,22 +27,42 @@ impl> CodexAuthManager { } pub fn get_auth(&self) -> Result { - let cached = { - let guard = self.cached.lock().map_err(|e| anyhow::anyhow!("{e}"))?; - guard.clone() - }; - let stored = match cached { - Some(ref auth) => auth.clone(), - None => { - let loaded = self.store.load_auth()?; - match loaded { - Some(auth) => { - let mut guard = self.cached.lock().map_err(|e| anyhow::anyhow!("{e}"))?; + let stored = if self.store.reload_each_request() { + let loaded = self.store.load_auth()?; + match loaded { + Some(auth) => { + let mut guard = self.cached.lock().map_err(|e| anyhow::anyhow!("{e}"))?; + if guard.as_ref() != Some(&auth) { *guard = Some(auth.clone()); - auth } - None => { - anyhow::bail!("Not authenticated. Run: claude-code-proxy codex auth login"); + auth + } + None => { + self.reset_cache(); + anyhow::bail!("Not authenticated. Run: claude-code-proxy codex auth login"); + } + } + } else { + let cached = { + let guard = self.cached.lock().map_err(|e| anyhow::anyhow!("{e}"))?; + guard.clone() + }; + match cached { + Some(ref auth) => auth.clone(), + None => { + let loaded = self.store.load_auth()?; + match loaded { + Some(auth) => { + let mut guard = + self.cached.lock().map_err(|e| anyhow::anyhow!("{e}"))?; + *guard = Some(auth.clone()); + auth + } + None => { + anyhow::bail!( + "Not authenticated. Run: claude-code-proxy codex auth login" + ); + } } } } @@ -197,4 +217,60 @@ mod tests { .contains("Not authenticated") ); } + + #[test] + fn reloading_store_observes_profile_swap_on_next_request() { + let store = CodexTokenStore::new_reloading(InMemoryAuthStore::new()); + store + .save_auth(StoredAuth { + access: "first_access".into(), + refresh: String::new(), + expires: 9999999999999, + account_id: Some("acct_1".into()), + }) + .unwrap(); + let manager = CodexAuthManager::new(store); + + assert_eq!(manager.get_auth().unwrap().access, "first_access"); + manager + .store + .save_auth(StoredAuth { + access: "second_access".into(), + refresh: String::new(), + expires: 9999999999999, + account_id: Some("acct_2".into()), + }) + .unwrap(); + + let swapped = manager.get_auth().unwrap(); + assert_eq!(swapped.access, "second_access"); + assert_eq!(swapped.account_id.as_deref(), Some("acct_2")); + } + + #[test] + fn normal_store_keeps_cached_auth() { + let store = test_store(); + store + .save_auth(StoredAuth { + access: "first_access".into(), + refresh: "first_refresh".into(), + expires: 9999999999999, + account_id: Some("acct_1".into()), + }) + .unwrap(); + let manager = CodexAuthManager::new(store); + + assert_eq!(manager.get_auth().unwrap().access, "first_access"); + manager + .store + .save_auth(StoredAuth { + access: "second_access".into(), + refresh: "second_refresh".into(), + expires: 9999999999999, + account_id: Some("acct_2".into()), + }) + .unwrap(); + + assert_eq!(manager.get_auth().unwrap().access, "first_access"); + } } diff --git a/src/providers/codex/auth/token_store.rs b/src/providers/codex/auth/token_store.rs index e3a725a..e4953c3 100644 --- a/src/providers/codex/auth/token_store.rs +++ b/src/providers/codex/auth/token_store.rs @@ -25,6 +25,7 @@ pub struct StoredAuth { pub struct CodexTokenStore> { store: S, codex_cli_fallback: bool, + reload_each_request: bool, } impl> CodexTokenStore { @@ -32,6 +33,7 @@ impl> CodexTokenStore { Self { store, codex_cli_fallback: false, + reload_each_request: false, } } @@ -39,6 +41,15 @@ impl> CodexTokenStore { Self { store, codex_cli_fallback: true, + reload_each_request: false, + } + } + + pub fn new_reloading(store: S) -> Self { + Self { + store, + codex_cli_fallback: false, + reload_each_request: true, } } @@ -74,6 +85,10 @@ impl> CodexTokenStore { pub fn auth_path(&self) -> String { self.store.path() } + + pub fn reload_each_request(&self) -> bool { + self.reload_each_request + } } pub type DefaultCodexAuthStore = KeychainFileAuthStore; @@ -92,7 +107,7 @@ pub fn file_store() -> CodexTokenStore { if std::env::var_os("CCP_CONFIG_DIR").is_none() { CodexTokenStore::new_with_codex_cli_fallback(store) } else { - CodexTokenStore::new(store) + CodexTokenStore::new_reloading(store) } } diff --git a/src/providers/codex/mod.rs b/src/providers/codex/mod.rs index f775f8b..c0b2e03 100644 --- a/src/providers/codex/mod.rs +++ b/src/providers/codex/mod.rs @@ -40,6 +40,7 @@ use self::translate::reducer::finish_metadata_from_upstream; use self::translate::request::{TranslateOptions, translate_request}; const MAX_RETRYABLE_LIVE_STREAM_RETRIES: u32 = 10; +const CODEX_CONTEXT_WINDOW_TOKENS: u64 = 372_000; use self::translate::stream::translate_stream_bytes_with_traffic; // --------------------------------------------------------------------------- @@ -121,6 +122,7 @@ impl Provider for CodexProvider { ); } }; + let estimated_input_tokens = count_translated_tokens(&translated); // Check continuation let previous_response_id_enabled = config::codex_previous_response_id(); @@ -144,6 +146,7 @@ impl Provider for CodexProvider { ctx, stream_request, continuation, + estimated_input_tokens, ) .await; } @@ -155,7 +158,7 @@ impl Provider for CodexProvider { Ok(r) => r, Err(e) => { clear_continuation(ctx.session_id.as_deref()); - return map_codex_error_to_response(&e); + return map_codex_error_to_response(&e, estimated_input_tokens); } }; @@ -169,10 +172,9 @@ impl Provider for CodexProvider { Ok(b) => b, Err(e) => { clear_continuation(ctx.session_id.as_deref()); - return json_error( - StatusCode::BAD_GATEWAY, - "api_error", - format!("Stream translation error: {e}"), + return map_codex_failure_to_response( + &format!("Stream translation error: {e}"), + estimated_input_tokens, ); } }; @@ -223,10 +225,9 @@ impl Provider for CodexProvider { } Err(e) => { clear_continuation(ctx.session_id.as_deref()); - json_error( - StatusCode::BAD_GATEWAY, - "api_error", - format!("Accumulation error: {e}"), + map_codex_failure_to_response( + &format!("Accumulation error: {e}"), + estimated_input_tokens, ) } } @@ -303,6 +304,7 @@ async fn live_stream_response( ctx: RequestContext, request_body: translate::request::ResponsesRequest, continuation: ContinuationCandidate, + estimated_input_tokens: u64, ) -> Response { let model = model.to_string(); let mut attempt = 0_u32; @@ -314,6 +316,10 @@ async fn live_stream_response( .await { Ok(events) => events, + Err(err) if is_codex_rate_limit_error(&err) => { + clear_continuation(ctx.session_id.as_deref()); + return map_codex_error_to_response(&err, estimated_input_tokens); + } Err(err) if retryable_live_start_codex_error(&err) => { if retry_with_full_context_for_live_error(&err) && drop_live_continuation_for_retry(&mut continuation, &ctx) @@ -323,12 +329,12 @@ async fn live_stream_response( } if attempt >= MAX_RETRYABLE_LIVE_STREAM_RETRIES { clear_continuation(ctx.session_id.as_deref()); - return map_codex_error_to_response(&err); + return map_codex_error_to_response(&err, estimated_input_tokens); } let delay = compute_backoff_delay(attempt, err.retry_after.as_deref()); if delay.exceeds_budget { clear_continuation(ctx.session_id.as_deref()); - return map_codex_error_to_response(&err); + return map_codex_error_to_response(&err, estimated_input_tokens); } attempt += 1; sleep(delay.wait_ms).await; @@ -336,7 +342,7 @@ async fn live_stream_response( } Err(err) => { clear_continuation(ctx.session_id.as_deref()); - return map_codex_error_to_response(&err); + return map_codex_error_to_response(&err, estimated_input_tokens); } }; @@ -346,6 +352,7 @@ async fn live_stream_response( &model, ctx.clone(), request_body.clone(), + estimated_input_tokens, ) .await { @@ -361,12 +368,20 @@ async fn live_stream_response( } if attempt >= MAX_RETRYABLE_LIVE_STREAM_RETRIES { clear_continuation(ctx.session_id.as_deref()); - return json_error(StatusCode::BAD_GATEWAY, "api_error", message); + return map_codex_failure_with_retry_to_response( + &message, + retry_after.as_deref(), + estimated_input_tokens, + ); } let delay = compute_backoff_delay(attempt, retry_after.as_deref()); if delay.exceeds_budget { clear_continuation(ctx.session_id.as_deref()); - return json_error(StatusCode::BAD_GATEWAY, "api_error", message); + return map_codex_failure_with_retry_to_response( + &message, + retry_after.as_deref(), + estimated_input_tokens, + ); } attempt += 1; sleep(delay.wait_ms).await; @@ -381,6 +396,7 @@ async fn live_stream_response_once( model: &str, ctx: RequestContext, request_body: translate::request::ResponsesRequest, + estimated_input_tokens: u64, ) -> LiveStreamStart { let mut translator = LiveStreamTranslator::new(message_id, model.to_string()); let mut upstream_sse_body = Vec::new(); @@ -389,6 +405,13 @@ async fn live_stream_response_once( let payload = match item { Ok(payload) => payload, Err(err) => { + if is_codex_rate_limit_error(&err) { + clear_continuation(ctx.session_id.as_deref()); + return LiveStreamStart::Response(map_codex_error_to_response( + &err, + estimated_input_tokens, + )); + } if retryable_live_start_codex_error(&err) { let full_context = retry_with_full_context_for_live_error(&err); return LiveStreamStart::Retry { @@ -398,7 +421,10 @@ async fn live_stream_response_once( }; } clear_continuation(ctx.session_id.as_deref()); - return LiveStreamStart::Response(map_codex_error_to_response(&err)); + return LiveStreamStart::Response(map_codex_error_to_response( + &err, + estimated_input_tokens, + )); } }; append_upstream_sse_payload(&mut upstream_sse_body, &payload); @@ -406,6 +432,15 @@ async fn live_stream_response_once( { Ok(result) => result, Err(message) => { + if is_live_rate_limit_payload(&payload, &message) { + clear_continuation(ctx.session_id.as_deref()); + let retry_after = retry_after_from_live_payload(&payload); + return LiveStreamStart::Response(map_codex_failure_with_retry_to_response( + &message, + retry_after.as_deref(), + estimated_input_tokens, + )); + } if retryable_live_start_payload(&payload, &message) { return LiveStreamStart::Retry { message, @@ -414,10 +449,9 @@ async fn live_stream_response_once( }; } clear_continuation(ctx.session_id.as_deref()); - return LiveStreamStart::Response(json_error( - StatusCode::BAD_GATEWAY, - "api_error", - message, + return LiveStreamStart::Response(map_codex_failure_to_response( + &message, + estimated_input_tokens, )); } }; @@ -628,6 +662,10 @@ fn retryable_live_start_codex_error(err: &client::CodexError) -> bool { || (err.status == 0 && retryable_live_message(codex_error_message(err))) } +fn is_codex_rate_limit_error(err: &client::CodexError) -> bool { + err.status == 429 || is_rate_limit_message(codex_error_message(err)) +} + fn retry_with_full_context_for_live_error(err: &client::CodexError) -> bool { matches!( err.detail.as_deref(), @@ -713,6 +751,61 @@ fn retryable_live_start_payload(payload: &serde_json::Value, message: &str) -> b || retryable_live_message(message) } +fn is_live_rate_limit_payload(payload: &serde_json::Value, message: &str) -> bool { + if payload.get("type").and_then(|v| v.as_str()) == Some("codex.rate_limits") + && payload + .get("rate_limits") + .and_then(|r| r.get("limit_reached")) + .and_then(|v| v.as_bool()) + == Some(true) + { + return true; + } + + let status = payload + .get("status") + .or_else(|| payload.get("status_code")) + .and_then(|v| v.as_u64()) + .or_else(|| { + payload + .get("error") + .and_then(|e| e.get("status")) + .and_then(|v| v.as_u64()) + }) + .or_else(|| { + payload + .get("response") + .and_then(|r| r.get("error")) + .and_then(|e| e.get("status")) + .and_then(|v| v.as_u64()) + }); + if status == Some(429) { + return true; + } + + let classification = payload + .get("error") + .and_then(live_payload_error_classification) + .or_else(|| { + payload + .get("response") + .and_then(|r| r.get("error")) + .and_then(live_payload_error_classification) + }); + + matches!( + classification, + Some("rate_limit_error" | "rate_limit_exceeded") + ) || is_rate_limit_message(message) +} + +fn live_payload_error_classification(error: &serde_json::Value) -> Option<&str> { + error + .get("code") + .and_then(|v| v.as_str()) + .or_else(|| error.get("type").and_then(|v| v.as_str())) +} + fn retry_after_from_live_payload(payload: &serde_json::Value) -> Option { payload .get("rate_limits") @@ -800,31 +893,79 @@ fn update_continuation_from_upstream( // Error mapping // --------------------------------------------------------------------------- -fn map_codex_error_to_response(err: &client::CodexError) -> Response { +fn map_codex_error_to_response(err: &client::CodexError, estimated_input_tokens: u64) -> Response { + let message = codex_error_message(err); + if is_context_window_overflow(message) { + return map_codex_failure_to_response(message, estimated_input_tokens); + } + match err.status { 401 | 403 => json_error( StatusCode::UNAUTHORIZED, "authentication_error", err.detail.as_deref().unwrap_or("Authentication failed"), ), - 429 => { - let retry_after = err.retry_after.as_deref().unwrap_or("5"); - let resp = json_error( - StatusCode::TOO_MANY_REQUESTS, - "rate_limit_error", - &err.message, - ); - let headers = [(http::header::RETRY_AFTER, retry_after)]; - (headers, resp).into_response() - } - _ => json_error( - StatusCode::BAD_GATEWAY, - "api_error", - codex_error_message(err), + 429 => map_codex_rate_limit_to_response( + &err.message, + err.retry_after.as_deref(), + estimated_input_tokens, + ), + _ if is_rate_limit_message(message) => map_codex_rate_limit_to_response( + message, + err.retry_after.as_deref(), + estimated_input_tokens, ), + _ => json_error(StatusCode::BAD_GATEWAY, "api_error", message), } } +fn map_codex_failure_to_response(message: &str, estimated_input_tokens: u64) -> Response { + map_codex_failure_with_retry_to_response(message, None, estimated_input_tokens) +} + +fn map_codex_failure_with_retry_to_response( + message: &str, + retry_after: Option<&str>, + estimated_input_tokens: u64, +) -> Response { + if is_context_window_overflow(message) { + json_error(StatusCode::PAYLOAD_TOO_LARGE, "request_too_large", message) + } else if is_rate_limit_message(message) { + map_codex_rate_limit_to_response(message, retry_after, estimated_input_tokens) + } else { + json_error(StatusCode::BAD_GATEWAY, "api_error", message) + } +} + +fn map_codex_rate_limit_to_response( + message: &str, + retry_after: Option<&str>, + estimated_input_tokens: u64, +) -> Response { + if estimated_input_tokens >= CODEX_CONTEXT_WINDOW_TOKENS { + return json_error( + StatusCode::PAYLOAD_TOO_LARGE, + "request_too_large", + format!( + "Estimated input ({estimated_input_tokens} tokens) exceeds the Codex context window ({CODEX_CONTEXT_WINDOW_TOKENS} tokens); upstream reported: {message}" + ), + ); + } + + let retry_after = retry_after.unwrap_or("5"); + let response = json_error(StatusCode::TOO_MANY_REQUESTS, "rate_limit_error", message); + let headers = [(http::header::RETRY_AFTER, retry_after)]; + (headers, response).into_response() +} + +fn is_rate_limit_message(message: &str) -> bool { + message.to_ascii_lowercase().contains("rate limit") +} + +fn is_context_window_overflow(message: &str) -> bool { + message.to_ascii_lowercase().contains("context window") +} + fn codex_error_message(err: &client::CodexError) -> &str { err.detail.as_deref().unwrap_or({ if err.status == 0 { @@ -1082,7 +1223,7 @@ mod tests { origin: client::CodexErrorOrigin::WebSocket, }; - let response = map_codex_error_to_response(&err); + let response = map_codex_error_to_response(&err, 100); assert_eq!(response.status(), StatusCode::BAD_GATEWAY); let body = axum::body::to_bytes(response.into_body(), usize::MAX) @@ -1095,6 +1236,52 @@ mod tests { ); } + #[tokio::test] + async fn rate_limit_below_context_window_returns_429_with_retry_after() { + let response = map_codex_failure_with_retry_to_response( + "rate limit reached", + Some("30"), + CODEX_CONTEXT_WINDOW_TOKENS - 1, + ); + + assert_eq!(response.status(), StatusCode::TOO_MANY_REQUESTS); + assert_eq!( + response + .headers() + .get(http::header::RETRY_AFTER) + .and_then(|value| value.to_str().ok()), + Some("30") + ); + let body = axum::body::to_bytes(response.into_body(), usize::MAX) + .await + .unwrap(); + let body: serde_json::Value = serde_json::from_slice(&body).unwrap(); + assert_eq!(body["error"]["type"], "rate_limit_error"); + assert_eq!(body["error"]["message"], "rate limit reached"); + } + + #[tokio::test] + async fn rate_limit_at_context_window_returns_413_for_compaction() { + let response = map_codex_failure_with_retry_to_response( + "rate limit reached", + Some("30"), + CODEX_CONTEXT_WINDOW_TOKENS, + ); + + assert_eq!(response.status(), StatusCode::PAYLOAD_TOO_LARGE); + assert!(response.headers().get(http::header::RETRY_AFTER).is_none()); + let body = axum::body::to_bytes(response.into_body(), usize::MAX) + .await + .unwrap(); + let body: serde_json::Value = serde_json::from_slice(&body).unwrap(); + assert_eq!(body["error"]["type"], "request_too_large"); + assert!( + body["error"]["message"] + .as_str() + .is_some_and(|message| message.contains("372000")) + ); + } + #[test] fn live_start_payload_retry_detection_covers_rate_limit_and_overload() { assert!(retryable_live_start_payload( diff --git a/tests/smoke_cutover.rs b/tests/smoke_cutover.rs index 7bfa167..b586eac 100644 --- a/tests/smoke_cutover.rs +++ b/tests/smoke_cutover.rs @@ -271,6 +271,30 @@ async fn spawn_websocket_error_upstream(message: &'static str) -> String { addr_str } +async fn spawn_websocket_rate_limit_upstream(retry_after_seconds: u64) -> String { + let listener = TcpListener::bind("127.0.0.1:0").await.unwrap(); + let addr = listener.local_addr().unwrap(); + let addr_str = format!("http://{addr}"); + + tokio::spawn(async move { + if let Ok((stream, _)) = listener.accept().await + && let Ok(mut ws) = tokio_tungstenite::accept_async(stream).await + { + let _ = ws.next().await; + let event = json!({ + "type": "codex.rate_limits", + "rate_limits": { + "limit_reached": true, + "primary": {"reset_after_seconds": retry_after_seconds} + } + }); + let _ = ws.send(Message::Text(event.to_string())).await; + } + }); + + addr_str +} + async fn spawn_websocket_sequence_upstream(captured: Arc>>) -> String { let listener = TcpListener::bind("127.0.0.1:0").await.unwrap(); let addr = listener.local_addr().unwrap(); @@ -653,6 +677,42 @@ async fn smoke_codex_http_messages_uses_mock_upstream() { assert_eq!(sent["stream"], true); } +#[allow(clippy::await_holding_lock)] +#[tokio::test] +async fn smoke_codex_http_context_window_error_requests_compaction() { + let _guard = env_lock(); + let config = TempDir::new().unwrap(); + write_auth(config.path(), "codex"); + + let upstream = spawn_http_upstream(|_body: Value| { + "data: {\"type\":\"response.failed\",\"response\":{\"error\":{\"message\":\"input exceeds context window\"}}}\n\n" + .as_bytes() + .to_vec() + }) + .await; + + let _config_env = EnvGuard::set("CCP_CONFIG_DIR", config.path()); + let _base_url_env = EnvGuard::set("CCP_CODEX_BASE_URL", &upstream); + let _transport_env = EnvGuard::set("CCP_CODEX_TRANSPORT", "http"); + let response = call_messages("gpt-5.5").await; + let status = response.status(); + let body = axum::body::to_bytes(response.into_body(), usize::MAX) + .await + .unwrap(); + + assert_eq!(status, StatusCode::PAYLOAD_TOO_LARGE); + let value: Value = serde_json::from_slice(&body).unwrap(); + assert_eq!(value["type"], "error"); + assert_eq!(value["error"]["type"], "request_too_large"); + assert!( + value["error"]["message"] + .as_str() + .is_some_and(|message| message.contains("input exceeds context window")), + "response body: {}", + String::from_utf8_lossy(&body) + ); +} + #[allow(clippy::await_holding_lock)] #[tokio::test] async fn smoke_codex_http_traffic_capture_writes_upstream_artifacts() { @@ -887,7 +947,7 @@ async fn smoke_codex_websocket_stream_returns_delta_before_terminal() { #[allow(clippy::await_holding_lock)] #[tokio::test(flavor = "multi_thread")] -async fn smoke_codex_websocket_stream_returns_json_for_early_error() { +async fn smoke_codex_websocket_context_window_error_requests_compaction() { let _guard = env_lock(); let config = TempDir::new().unwrap(); write_auth(config.path(), "codex"); @@ -913,14 +973,90 @@ async fn smoke_codex_websocket_stream_returns_json_for_early_error() { assert_eq!( status, - StatusCode::BAD_GATEWAY, + StatusCode::PAYLOAD_TOO_LARGE, "response body: {}", String::from_utf8_lossy(&body) ); let value: Value = serde_json::from_slice(&body).unwrap(); + assert_eq!(value["type"], "error"); + assert_eq!(value["error"]["type"], "request_too_large"); assert_eq!(value["error"]["message"], "input exceeds context window"); } +#[allow(clippy::await_holding_lock)] +#[tokio::test(flavor = "multi_thread")] +async fn smoke_codex_websocket_rate_limit_returns_429_with_retry_after() { + let _guard = env_lock(); + let config = TempDir::new().unwrap(); + write_auth(config.path(), "codex"); + clear_codex_websocket_pool_for_tests(); + + let upstream = spawn_websocket_rate_limit_upstream(30).await; + + let _config_env = EnvGuard::set("CCP_CONFIG_DIR", config.path()); + let _base_url_env = EnvGuard::set("CCP_CODEX_BASE_URL", &upstream); + let _transport_env = EnvGuard::set("CCP_CODEX_TRANSPORT", "websocket"); + + let response = call_messages_body(json!({ + "model": "gpt-5.5", + "max_tokens": 64, + "stream": true, + "messages": [{"role":"user","content":"hello"}] + })) + .await; + + assert_eq!(response.status(), StatusCode::TOO_MANY_REQUESTS); + assert_eq!( + response + .headers() + .get(http::header::RETRY_AFTER) + .and_then(|value| value.to_str().ok()), + Some("30") + ); + let body = axum::body::to_bytes(response.into_body(), usize::MAX) + .await + .unwrap(); + let value: Value = serde_json::from_slice(&body).unwrap(); + assert_eq!(value["error"]["type"], "rate_limit_error"); + assert_eq!(value["error"]["message"], "rate limit reached"); +} + +#[allow(clippy::await_holding_lock)] +#[tokio::test(flavor = "multi_thread")] +async fn smoke_codex_websocket_oversized_rate_limit_requests_compaction() { + let _guard = env_lock(); + let config = TempDir::new().unwrap(); + write_auth(config.path(), "codex"); + clear_codex_websocket_pool_for_tests(); + + let upstream = spawn_websocket_rate_limit_upstream(30).await; + + let _config_env = EnvGuard::set("CCP_CONFIG_DIR", config.path()); + let _base_url_env = EnvGuard::set("CCP_CODEX_BASE_URL", &upstream); + let _transport_env = EnvGuard::set("CCP_CODEX_TRANSPORT", "websocket"); + + let response = call_messages_body(json!({ + "model": "gpt-5.5", + "max_tokens": 64, + "stream": true, + "messages": [{"role":"user","content":".".repeat(372_100)}] + })) + .await; + + assert_eq!(response.status(), StatusCode::PAYLOAD_TOO_LARGE); + assert!(response.headers().get(http::header::RETRY_AFTER).is_none()); + let body = axum::body::to_bytes(response.into_body(), usize::MAX) + .await + .unwrap(); + let value: Value = serde_json::from_slice(&body).unwrap(); + assert_eq!(value["error"]["type"], "request_too_large"); + assert!( + value["error"]["message"] + .as_str() + .is_some_and(|message| message.contains("372000")) + ); +} + #[allow(clippy::await_holding_lock)] #[tokio::test(flavor = "multi_thread")] async fn smoke_codex_websocket_stream_uses_previous_response_id() {