diff --git a/codex-rs/codex-api/src/api_bridge_tests.rs b/codex-rs/codex-api/src/api_bridge_tests.rs index 56c4838ef6..9c4cd1be08 100644 --- a/codex-rs/codex-api/src/api_bridge_tests.rs +++ b/codex-rs/codex-api/src/api_bridge_tests.rs @@ -74,6 +74,56 @@ async fn map_api_error_preserves_retry_delay() { } } +/// Mapping a new error category preserves elapsed advice without making hard failures retryable. +#[tokio::test(start_paused = true)] +async fn http_retry_deadline_survives_mapping_and_respects_terminal_errors() { + use std::time::Duration; + + let advice = RetryAfter::from_delay(Duration::from_secs(10)).expect("retry advice"); + tokio::time::advance(Duration::from_secs(4)).await; + for (status, code, retryable) in [ + (503, "server_is_overloaded", true), + (429, "rate_limit_exceeded", true), + (429, "insufficient_quota", false), + (429, "usage_limit_reached", false), + (400, "cyber_policy", false), + ] { + let error = map_api_error(ApiError::Transport(TransportError::Http { + status: http::StatusCode::from_u16(status).unwrap(), + url: None, + headers: None, + body: Some(serde_json::json!({"error": {"type": code, "code": code, "message": "sensitive upstream detail"}}).to_string()), + retry_after: Some(advice), + })); + assert_eq!( + (error.retry_after(), error.server_retry_delay()), + (Some(advice), Some(Duration::from_secs(6))), + "{code}", + ); + assert_eq!( + error.retry_delay(/*retry_count*/ 1), + retryable.then_some(Duration::from_secs(6)), + "{code}", + ); + if status == 429 && retryable { + assert_eq!( + ( + error.to_codex_protocol_error(), + error.http_status_code_value(), + error.to_string() + ), + ( + CodexErrorInfo::ResponseTooManyFailedAttempts { + http_status_code: Some(429), + }, + Some(429), + "exceeded retry limit, last status: 429 Too Many Requests".to_string(), + ), + ); + } + } +} + #[test] fn map_api_error_distinguishes_capacity_from_slow_down() { for (code, expected, retryable) in [ diff --git a/codex-rs/codex-api/src/endpoint/responses_websocket.rs b/codex-rs/codex-api/src/endpoint/responses_websocket.rs index 7f3ebd7aac..b4463500b9 100644 --- a/codex-rs/codex-api/src/endpoint/responses_websocket.rs +++ b/codex-rs/codex-api/src/endpoint/responses_websocket.rs @@ -14,6 +14,7 @@ use crate::sse::process_responses_event; use crate::telemetry::WebsocketTelemetry; use codex_client::TransportError; use codex_http_client::HttpClientFactory; +use codex_http_client::RetryAfter; use codex_websocket_client::WebSocketConnection; use codex_websocket_client::WebSocketConnector; use futures::FutureExt; @@ -570,6 +571,7 @@ fn map_ws_error(err: WsError, url: &Url) -> ApiError { WsError::Http(response) => { let status = response.status(); let headers = response.headers().clone(); + let retry_after = RetryAfter::from_headers(&headers); let body = response .body() .as_ref() @@ -579,7 +581,7 @@ fn map_ws_error(err: WsError, url: &Url) -> ApiError { url: Some(url.to_string()), headers: Some(headers), body, - retry_after: None, + retry_after, }) } WsError::ConnectionClosed | WsError::AlreadyClosed => { diff --git a/codex-rs/core/src/responses_retry.rs b/codex-rs/core/src/responses_retry.rs index c8edc8b9d4..048a68bfc9 100644 --- a/codex-rs/core/src/responses_retry.rs +++ b/codex-rs/core/src/responses_retry.rs @@ -1,5 +1,6 @@ //! Shared retry and transport fallback decisions for Responses requests. //! Content-filter guidance is recorded for sampling requests before retry decisions. +//! Server advice controls timing without extending configured retry limits. use std::time::Duration; @@ -116,13 +117,16 @@ pub(crate) async fn handle_response_stream_error( return Ok(()); } - // TODO(anp): Respect server retry advice before issuing the fallback HTTP request. if retry_state.retries >= max_retries && client_session.try_switch_fallback_transport( &turn_context.session_telemetry, turn_context.model_info(), ) { + // Changing transport must not bypass the server's retry deadline. + if let Some(retry_after) = retry_after { + tokio::time::sleep_until(retry_after.deadline()).await; + } sess.send_event( turn_context, EventMsg::Warning(WarningEvent { diff --git a/codex-rs/core/tests/suite/compact.rs b/codex-rs/core/tests/suite/compact.rs index 7455ff786f..2dced3650d 100644 --- a/codex-rs/core/tests/suite/compact.rs +++ b/codex-rs/core/tests/suite/compact.rs @@ -23,6 +23,7 @@ use codex_protocol::openai_models::ModelInfo; use codex_protocol::openai_models::ModelServiceTier; use codex_protocol::openai_models::ModelsResponse; use codex_protocol::protocol::AskForApproval; +use codex_protocol::protocol::CodexErrorInfo; use codex_protocol::protocol::EventMsg; use codex_protocol::protocol::HookEventName; use codex_protocol::protocol::HookRunStatus; @@ -3074,6 +3075,193 @@ async fn pre_sampling_compact_falls_back_after_previous_model_stream_retries_are ); } +/// Both compaction models keep the configured retry budget when the server supplies advice, +/// including the selected model's final attempt. +#[test_case::test_case(2; "selected_model_recovers")] +#[test_case::test_case(3; "selected_model_exhausts_retries")] +#[tokio::test(flavor = "multi_thread", worker_threads = 2)] +async fn pre_sampling_compact_advised_errors_fall_back_then_retry_on_selected_model( + selected_model_failures: usize, +) { + skip_if_no_network!(); + + let server = MockServer::start().await; + let previous_model = "gpt-5.4"; + let selected_model = "gpt-5.2"; + let _models_mock = mount_models_once( + &server, + ModelsResponse { + models: vec![ + model_info_with_context_window(previous_model, /*context_window*/ 273_000), + model_info_with_context_window(selected_model, /*context_window*/ 125_000), + ], + }, + ) + .await; + let overload = wiremock::ResponseTemplate::new(/*s*/ 503) + .insert_header("Retry-After", "0") + .set_body_json(json!({ "error": { "code": "server_is_overloaded" } })); + let mut responses = vec![ + sse_response(sse(vec![ + ev_assistant_message("m1", "before switch"), + ev_completed_with_tokens("r1", /*total_tokens*/ 120_000), + ])), + overload.clone(), // Old model: initial attempt and two retries. + overload.clone(), + overload.clone(), + ]; + responses.extend(vec![overload; selected_model_failures]); + if selected_model_failures < 3 { + responses.extend([ + sse_response(remote_v2_compaction_response()), + sse_response(sse(vec![ + ev_assistant_message("m3", "after switch"), + ev_completed_with_tokens("r3", /*total_tokens*/ 100), + ])), + ]); + } + let request_log = mount_response_sequence(&server, responses).await; + let mut model_provider = openai_model_provider(&server); + model_provider.request_max_retries = Some(0); + model_provider.stream_max_retries = Some(2); + let test = test_codex() + .with_auth(CodexAuth::create_dummy_chatgpt_auth_for_testing()) + .with_model(previous_model) + .with_config(move |config| { + config.model_provider = model_provider; + set_test_compact_prompt(config); + }) + .build_with_auto_env(&server) + .await + .expect("build codex"); + test.codex + .start_or_steer_turn(disabled_permission_user_turn( + "before switch", + test.cwd.path().to_path_buf(), + previous_model.to_string(), + )) + .await + .expect("submit first turn"); + wait_for_event(&test.codex, |event| { + matches!(event, EventMsg::TurnComplete(_)) + }) + .await; + test.codex + .start_or_steer_turn(disabled_permission_user_turn( + "after switch", + test.cwd.path().to_path_buf(), + selected_model.to_string(), + )) + .await + .expect("submit selected-model turn"); + if selected_model_failures < 3 { + assert_compaction_uses_turn_lifecycle_id(&test.codex).await; + } else { + let mut errors = Vec::new(); + loop { + match wait_for_event(&test.codex, |_| true).await { + EventMsg::Error(error) => errors.push(error.codex_error_info), + EventMsg::TurnComplete(event) => { + assert_eq!( + event.error.and_then(|error| error.codex_error_info), + Some(CodexErrorInfo::ServerOverloaded) + ); + break; + } + _ => {} + } + } + assert_eq!(errors, vec![Some(CodexErrorInfo::ServerOverloaded)]); + } + let actual_models = request_log + .requests() + .iter() + .map(|request| { + request.body_json()["model"] + .as_str() + .expect("request model") + .to_string() + }) + .collect::>(); + let mut expected_models = vec![ + previous_model, // First turn. + previous_model, // Old-model compaction attempt and retries. + previous_model, + previous_model, + selected_model, // Selected-model compaction attempt and retries. + selected_model, + selected_model, + ]; + if selected_model_failures < 3 { + expected_models.push(selected_model); // Sampling after compaction succeeds. + } + assert_eq!(actual_models, expected_models); +} + +/// Manual compaction keeps the configured retry limit even when the server supplies advice. +#[test_case::test_case(0; "no_retries")] +#[test_case::test_case(2; "two_retries")] +#[tokio::test(flavor = "multi_thread", worker_threads = 2)] +async fn manual_remote_compact_advice_respects_retry_limit(max_retries: u64) { + skip_if_no_network!(); + + let server = start_mock_server().await; + let overload = wiremock::ResponseTemplate::new(/*s*/ 503) + .insert_header("Retry-After", "0") + .set_body_json(json!({ "error": { "code": "server_is_overloaded" } })); + let request_log = + mount_response_sequence(&server, vec![overload; (max_retries + 1) as usize]).await; + let mut model_provider = openai_model_provider(&server); + model_provider.request_max_retries = Some(0); + model_provider.stream_max_retries = Some(max_retries); + let test = test_codex() + .with_auth(CodexAuth::create_dummy_chatgpt_auth_for_testing()) + .with_model("gpt-5.4") + .with_config(move |config| { + config.model_provider = model_provider; + set_test_compact_prompt(config); + }) + .build_with_auto_env(&server) + .await + .expect("build codex"); + + test.codex.submit(Op::Compact).await.expect("run /compact"); + let mut errors = Vec::new(); + let mut reconnect_messages = Vec::new(); + loop { + match wait_for_event(&test.codex, |_| true).await { + EventMsg::Error(error) => errors.push(error.codex_error_info), + EventMsg::StreamError(error) => reconnect_messages.push(error.message), + EventMsg::TurnComplete(event) => { + assert_eq!( + event.error.and_then(|error| error.codex_error_info), + Some(CodexErrorInfo::ServerOverloaded) + ); + break; + } + _ => {} + } + } + assert_eq!(errors, vec![Some(CodexErrorInfo::ServerOverloaded)]); + assert_eq!( + reconnect_messages, + (1..=max_retries) + .map(|retry| format!("Reconnecting... {retry}/{max_retries}")) + .collect::>() + ); + let models = request_log + .requests() + .iter() + .map(|request| { + request.body_json()["model"] + .as_str() + .expect("request model") + .to_string() + }) + .collect::>(); + assert_eq!(models, vec!["gpt-5.4"; (max_retries + 1) as usize]); +} + #[tokio::test(flavor = "multi_thread", worker_threads = 2)] async fn pre_sampling_compact_keeps_unknown_previous_model_for_api_key_auth_and_custom_provider() { skip_if_no_network!(); diff --git a/codex-rs/core/tests/suite/retry_after.rs b/codex-rs/core/tests/suite/retry_after.rs index 4bb5da9a8d..f69caf5208 100644 --- a/codex-rs/core/tests/suite/retry_after.rs +++ b/codex-rs/core/tests/suite/retry_after.rs @@ -51,11 +51,11 @@ const SECOND_RETRY_MIN_DELAY: Duration = Duration::from_millis(360); const SECOND_RETRY_MAX_DELAY: Duration = Duration::from_millis(440); #[derive(Debug, PartialEq, Eq)] -struct RetryTelemetryEvent { +pub(super) struct RetryTelemetryEvent { attempt: u64, - delay: Duration, - layer: String, - operation: String, + pub(super) delay: Duration, + pub(super) layer: String, + pub(super) operation: String, } #[derive(Default)] @@ -193,7 +193,7 @@ where } } -struct RetryTelemetryCapture { +pub(super) struct RetryTelemetryCapture { events: mpsc::UnboundedReceiver, resumptions: mpsc::UnboundedReceiver, _subscriber: DefaultGuard, @@ -201,7 +201,7 @@ struct RetryTelemetryCapture { } impl RetryTelemetryCapture { - fn install() -> Self { + pub(super) fn install() -> Self { let (sender, events) = mpsc::unbounded_channel(); let (resumptions_sender, resumptions) = mpsc::unbounded_channel(); // Avoid caching no interest when another test first reaches a shared callsite. @@ -225,7 +225,7 @@ impl RetryTelemetryCapture { } } - async fn next_retry(&mut self) -> RetryTelemetryEvent { + pub(super) async fn next_retry(&mut self) -> RetryTelemetryEvent { tokio::time::timeout(Duration::from_secs(10), self.events.recv()) .await .expect("timed out waiting for retry telemetry") @@ -275,7 +275,7 @@ async fn wait_for_turn_completion(test: &TestCodex) { assert_eq!(completed.error, None, "turn should complete successfully"); } -/// HTTP overloads use the upstream header before another HTTP request. +/// HTTP overloads use server advice within the request and stream retry budgets. #[tokio::test(flavor = "current_thread")] async fn responses_http_uses_retry_after() -> Result<()> { skip_if_no_network!(Ok(())); @@ -285,6 +285,9 @@ async fn responses_http_uses_retry_after() -> Result<()> { let response_mock = responses::mount_response_sequence( &server, vec![ + ResponseTemplate::new(503) + .insert_header("Retry-After", "1") + .set_body_json(json!({ "error": { "code": "server_is_overloaded" } })), ResponseTemplate::new(503) .insert_header("Retry-After", "1") .set_body_json(json!({ "error": { "code": "server_is_overloaded" } })), @@ -298,7 +301,7 @@ async fn responses_http_uses_retry_after() -> Result<()> { let test = test_codex() .with_config(|config| { config.model_provider.request_max_retries = Some(1); - config.model_provider.stream_max_retries = Some(0); + config.model_provider.stream_max_retries = Some(1); }) .build_with_auto_env(&server) .await?; @@ -316,8 +319,79 @@ async fn responses_http_uses_retry_after() -> Result<()> { } ); assert!(wait_for_retry(&mut telemetry, &retry).await >= Duration::from_secs(1)); + let retry = telemetry.next_retry().await; + assert!(retry.delay <= Duration::from_secs(1)); + assert_eq!( + retry, + RetryTelemetryEvent { + attempt: 1, + delay: retry.delay, + layer: "stream".into(), + operation: "sampling".into(), + } + ); + assert!(wait_for_retry(&mut telemetry, &retry).await >= Duration::from_secs(1)); wait_for_turn_completion(&test).await; + assert_eq!(response_mock.requests().len(), 3); + assert_eq!( + telemetry.events.try_recv(), + Err(mpsc::error::TryRecvError::Empty) + ); + Ok(()) +} + +/// A 429 with server advice retries in the sampling loop and completes without a terminal error. +#[tokio::test(flavor = "current_thread")] +async fn responses_http_429_uses_retry_after() -> Result<()> { + skip_if_no_network!(Ok(())); + + let mut telemetry = RetryTelemetryCapture::install(); + let server = responses::start_mock_server().await; + let response_mock = responses::mount_response_sequence( + &server, + vec![ + ResponseTemplate::new(429) + .insert_header("Retry-After", "1") + .set_body_json(json!({ "error": { "code": "rate_limit_exceeded" } })), + responses::sse_response(responses::sse(vec![ + responses::ev_response_created("recovered"), + responses::ev_completed("recovered"), + ])), + ], + ) + .await; + let test = test_codex() + .with_config(|config| { + config.model_provider.request_max_retries = Some(0); + config.model_provider.stream_max_retries = Some(1); + }) + .build_with_auto_env(&server) + .await?; + + submit_user_input(&test, "retry the rate-limited request").await?; + let retry = telemetry.next_retry().await; + assert!(retry.delay <= Duration::from_secs(1)); + assert_eq!( + retry, + RetryTelemetryEvent { + attempt: 1, + delay: retry.delay, + layer: "stream".into(), + operation: "sampling".into(), + } + ); + assert!(wait_for_retry(&mut telemetry, &retry).await >= Duration::from_secs(1)); + loop { + match wait_for_event(&test.codex, |_| true).await { + EventMsg::Error(error) => panic!("unexpected terminal error: {error:?}"), + EventMsg::TurnComplete(completed) => { + assert_eq!(completed.error, None); + break; + } + _ => {} + } + } assert_eq!(response_mock.requests().len(), 2); assert_eq!( telemetry.events.try_recv(), @@ -425,31 +499,40 @@ async fn exhausted_http_retries_preserve_deadline_through_error_mapping() { assert_eq!(outer_error.server_retry_delay(), Some(Duration::ZERO)); } -/// Headerless HTTP overloads currently exhaust request retries before emitting one terminal error. +/// Retry advice changes delays but never extends either retry budget or its displayed count. +#[test_case::test_case(None, 2, 3; "without_advice")] +#[test_case::test_case(Some("0"), 0, 3; "advice_without_stream_retries")] +#[test_case::test_case(Some("0"), 2, 9; "advice_with_stream_retries")] #[tokio::test(flavor = "current_thread")] -async fn responses_http_overload_without_retry_after_exhausts_request_retries() -> Result<()> { +async fn responses_http_overload_respects_retry_limits( + retry_after: Option<&str>, + stream_max_retries: u64, + expected_requests: usize, +) -> Result<()> { skip_if_no_network!(Ok(())); let server = responses::start_mock_server().await; + let mut overload = ResponseTemplate::new(503) + .set_body_json(json!({ "error": { "code": "server_is_overloaded" } })); + if let Some(retry_after) = retry_after { + overload = overload.insert_header("Retry-After", retry_after); + } Mock::given(method("POST")) .and(path("/v1/responses")) - .respond_with( - ResponseTemplate::new(503) - .set_body_json(json!({ "error": { "code": "server_is_overloaded" } })), - ) + .respond_with(overload) .mount(&server) .await; let test = test_codex() - .with_config(|config| { + .with_config(move |config| { config.model_provider.request_max_retries = Some(2); - config.model_provider.stream_max_retries = Some(2); + config.model_provider.stream_max_retries = Some(stream_max_retries); }) .build_with_auto_env(&server) .await?; submit_user_input(&test, "reject the disabled model").await?; let mut error_events = 0; - let mut stream_error_events = 0; + let mut reconnect_messages = Vec::new(); tokio::time::timeout(Duration::from_secs(10), async { loop { match wait_for_event(&test.codex, |_| true).await { @@ -464,7 +547,7 @@ async fn responses_http_overload_without_retry_after_exhausts_request_retries() "Selected model is at capacity. Please try a different model." ); } - EventMsg::StreamError(_) => stream_error_events += 1, + EventMsg::StreamError(error) => reconnect_messages.push(error.message), EventMsg::TurnComplete(event) => { assert_eq!( event.error.and_then(|error| error.codex_error_info), @@ -480,7 +563,14 @@ async fn responses_http_overload_without_retry_after_exhausts_request_retries() .expect("overload retries should finish the turn within 10 seconds"); assert_eq!(error_events, 1); - assert_eq!(stream_error_events, 0); + let expected_messages = if retry_after.is_some() { + (1..=stream_max_retries) + .map(|retry| format!("Reconnecting... {retry}/{stream_max_retries}")) + .collect::>() + } else { + Vec::new() + }; + assert_eq!(reconnect_messages, expected_messages); let request_count = server .received_requests() .await @@ -489,8 +579,8 @@ async fn responses_http_overload_without_retry_after_exhausts_request_retries() .filter(|request| request.url.path() == "/v1/responses") .count(); assert_eq!( - request_count, 3, - "headerless overload should exhaust the configured request retries" + request_count, expected_requests, + "overload should exhaust the configured retry budgets" ); Ok(()) @@ -1709,6 +1799,79 @@ async fn websocket_connection_limit_retries_with_local_backoff() -> Result<()> { Ok(()) } +/// Advised WebSocket failures reach HTTP after the existing retry budget. HTTP +/// recovery uses its own retry budget and honors the preceding advice deadlines. +#[tokio::test(flavor = "current_thread")] +async fn websocket_upgrade_rejection_uses_retry_after() -> Result<()> { + skip_if_no_network!(Ok(())); + + let server = responses::start_mock_server().await; + let prewarm_rejection = Mock::given(method("GET")) + .and(path("/v1/responses")) + .respond_with(ResponseTemplate::new(/*s*/ 503)) + .up_to_n_times(/*n*/ 1) + .expect(/*r*/ 1) + .mount_as_scoped(&server) + .await; + Mock::given(method("GET")) + .and(path("/v1/responses")) + .respond_with( + ResponseTemplate::new(/*s*/ 503) + .insert_header("Retry-After", "1") + .set_body_json(json!({ "error": { "code": "server_is_overloaded" } })), + ) + .mount(&server) + .await; + let http_overload = ResponseTemplate::new(/*s*/ 503) + .insert_header("Retry-After", "0") + .set_body_json(json!({ "error": { "code": "server_is_overloaded" } })); + let response_mock = responses::mount_response_sequence( + &server, + vec![ + http_overload, + ResponseTemplate::new(/*s*/ 200) + .insert_header("content-type", "text/event-stream") + .set_body_string(responses::sse(vec![ + responses::ev_response_created("recovered"), + responses::ev_completed("recovered"), + ])), + ], + ) + .await; + let test = test_codex() + .with_config(|config| { + config.model_catalog = Some(bundled_models_response().expect("bundled models")); + config.model_provider.supports_websockets = true; + config.model_provider.request_max_retries = Some(0); + config.model_provider.stream_max_retries = Some(1); + }) + .build_with_auto_env(&server) + .await?; + + // Prewarm is best effort; start the advised failures in the sampling retry loop. + tokio::time::timeout( + Duration::from_secs(10), + prewarm_rejection.wait_until_satisfied(), + ) + .await?; + drop(prewarm_rejection); + let start = Instant::now(); + submit_user_input(&test, "retry the rejected websocket upgrade").await?; + wait_for_turn_completion(&test).await; + + // Both WebSocket advice deadlines apply, including the one before transport fallback. + assert!(start.elapsed() >= Duration::from_secs(2)); + let requests = server.received_requests().await.unwrap_or_default(); + let methods = requests + .iter() + .filter(|request| request.url.path() == "/v1/responses") + .map(|request| request.method.as_str()) + .collect::>(); + assert_eq!(methods, vec!["GET", "GET", "GET", "POST", "POST"]); + assert_eq!(response_mock.requests().len(), 2); + Ok(()) +} + // TODO(anp) respect Retry-After /// Nested websocket retry headers are currently ignored, leaving rate-limit errors terminal. #[tokio::test(flavor = "current_thread")] diff --git a/codex-rs/protocol/src/error.rs b/codex-rs/protocol/src/error.rs index d0dad844bd..19134d6f3e 100644 --- a/codex-rs/protocol/src/error.rs +++ b/codex-rs/protocol/src/error.rs @@ -403,18 +403,19 @@ impl CodexErr { | CodexErrorDetails::UnsupportedOperation(_) | CodexErrorDetails::Sandbox(_) | CodexErrorDetails::LandlockSandboxExecutableNotProvided - | CodexErrorDetails::RetryLimit(_) | CodexErrorDetails::ContextWindowExceeded | CodexErrorDetails::ThreadNotFound(_) | CodexErrorDetails::AgentLimitReached { .. } | CodexErrorDetails::Spawn | CodexErrorDetails::SessionConfiguredNotFirstEvent | CodexErrorDetails::UsageLimitReached(_) - | CodexErrorDetails::ServerOverloaded | CodexErrorDetails::FlexUnavailable | CodexErrorDetails::CyberPolicy { .. } | CodexErrorDetails::BioPolicy { .. } | CodexErrorDetails::MisalignmentPolicyViolation { .. } => None, + CodexErrorDetails::ServerOverloaded | CodexErrorDetails::RetryLimit(_) => { + self.server_retry_delay() + } CodexErrorDetails::Stream(..) | CodexErrorDetails::ContentFilter | CodexErrorDetails::RateLimitExceeded(_)