diff --git a/codex-rs/codex-api/src/api_bridge.rs b/codex-rs/codex-api/src/api_bridge.rs index ab1b409eb1..2c4b8ffcc7 100644 --- a/codex-rs/codex-api/src/api_bridge.rs +++ b/codex-rs/codex-api/src/api_bridge.rs @@ -38,6 +38,9 @@ pub fn map_api_error(err: ApiError) -> CodexErr { } ApiError::InvalidRequest { message } => CodexErr::InvalidRequest(message), ApiError::CyberPolicy { message } => CodexErr::CyberPolicy { message }, + ApiError::Transport(ref transport) if is_server_overloaded_transport_error(transport) => { + CodexErr::ServerOverloaded + } ApiError::Transport(transport) => match transport { TransportError::Http { status, @@ -47,19 +50,6 @@ pub fn map_api_error(err: ApiError) -> CodexErr { } => { let body_text = body.unwrap_or_default(); - if status == http::StatusCode::SERVICE_UNAVAILABLE - && let Ok(value) = serde_json::from_str::(&body_text) - && matches!( - value - .get("error") - .and_then(|error| error.get("code")) - .and_then(serde_json::Value::as_str), - Some("server_is_overloaded" | "slow_down") - ) - { - return CodexErr::ServerOverloaded; - } - if status == http::StatusCode::BAD_REQUEST { if let Ok(parsed) = serde_json::from_str::(&body_text) && let Some(error) = parsed.get("error") @@ -141,6 +131,28 @@ pub fn map_api_error(err: ApiError) -> CodexErr { } } +pub(crate) fn is_server_overloaded_transport_error(err: &TransportError) -> bool { + let TransportError::Http { + status, + body: Some(body), + .. + } = err + else { + return false; + }; + *status == http::StatusCode::SERVICE_UNAVAILABLE + && serde_json::from_str::(body) + .ok() + .is_some_and(|value| { + matches!( + value + .pointer("/error/code") + .and_then(serde_json::Value::as_str), + Some("server_is_overloaded" | "slow_down") + ) + }) +} + const ACTIVE_LIMIT_HEADER: &str = "x-codex-active-limit"; const REQUEST_ID_HEADER: &str = "x-request-id"; const OAI_REQUEST_ID_HEADER: &str = "x-oai-request-id"; diff --git a/codex-rs/codex-api/src/api_bridge_tests.rs b/codex-rs/codex-api/src/api_bridge_tests.rs index 5812fc19fe..bcf303bb7f 100644 --- a/codex-rs/codex-api/src/api_bridge_tests.rs +++ b/codex-rs/codex-api/src/api_bridge_tests.rs @@ -26,6 +26,36 @@ fn map_api_error_maps_server_overloaded_from_503_body() { assert!(matches!(err, CodexErr::ServerOverloaded)); } +#[test] +fn server_overload_transport_detection_is_narrow() { + let error = |status, body: &str| TransportError::Http { + status, + url: None, + headers: None, + body: Some(body.to_string()), + }; + for code in ["server_is_overloaded", "slow_down"] { + let body = serde_json::json!({ "error": { "code": code } }).to_string(); + assert!(is_server_overloaded_transport_error(&error( + http::StatusCode::SERVICE_UNAVAILABLE, + &body, + ))); + } + + assert!(!is_server_overloaded_transport_error(&error( + http::StatusCode::SERVICE_UNAVAILABLE, + "unavailable", + ))); + assert!(!is_server_overloaded_transport_error(&error( + http::StatusCode::SERVICE_UNAVAILABLE, + r#"{"error":{"code":"other"}}"#, + ))); + assert!(!is_server_overloaded_transport_error(&error( + http::StatusCode::INTERNAL_SERVER_ERROR, + r#"{"error":{"code":"server_is_overloaded"}}"#, + ))); +} + #[test] fn map_api_error_maps_cloudflare_blocked_response_to_user_message() { let mut headers = HeaderMap::new(); diff --git a/codex-rs/codex-api/src/endpoint/responses.rs b/codex-rs/codex-api/src/endpoint/responses.rs index 804f0027ff..88c878d7dd 100644 --- a/codex-rs/codex-api/src/endpoint/responses.rs +++ b/codex-rs/codex-api/src/endpoint/responses.rs @@ -26,6 +26,7 @@ use tracing::instrument; pub struct ResponsesClient { session: EndpointSession, sse_telemetry: Option>, + defer_server_overloaded_retries: bool, } #[derive(Default)] @@ -43,6 +44,7 @@ impl ResponsesClient { Self { session: EndpointSession::new(transport, provider, auth), sse_telemetry: None, + defer_server_overloaded_retries: false, } } @@ -54,9 +56,16 @@ impl ResponsesClient { Self { session: self.session.with_request_telemetry(request), sse_telemetry: sse, + defer_server_overloaded_retries: self.defer_server_overloaded_retries, } } + /// Return capacity-coded 503s to a caller that applies a slower retry policy. + pub fn with_deferred_server_overload_retries(mut self, defer: bool) -> Self { + self.defer_server_overloaded_retries = defer; + self + } + #[instrument( name = "responses.stream_request", level = "info", @@ -93,8 +102,14 @@ impl ResponsesClient { insert_header(&mut headers, "x-openai-subagent", &subagent); } - self.stream_encoded(body, headers, compression, turn_state) - .await + self.stream_encoded( + body, + headers, + compression, + turn_state, + self.defer_server_overloaded_retries, + ) + .await } fn path() -> &'static str { @@ -121,8 +136,14 @@ impl ResponsesClient { ) -> Result { let body = EncodedJsonBody::encode(&body) .map_err(|e| ApiError::Stream(format!("failed to encode responses request: {e}")))?; - self.stream_encoded(body, extra_headers, compression, turn_state) - .await + self.stream_encoded( + body, + extra_headers, + compression, + turn_state, + self.defer_server_overloaded_retries, + ) + .await } async fn stream_encoded( @@ -131,6 +152,7 @@ impl ResponsesClient { extra_headers: HeaderMap, compression: Compression, turn_state: Option>>, + defer_server_overloaded_retries: bool, ) -> Result { let request_compression = match compression { Compression::None => RequestCompression::None, @@ -144,6 +166,7 @@ impl ResponsesClient { Self::path(), extra_headers, Some(body), + defer_server_overloaded_retries, |req| { req.headers.insert( http::header::ACCEPT, diff --git a/codex-rs/codex-api/src/endpoint/session.rs b/codex-rs/codex-api/src/endpoint/session.rs index 7849225b5c..d6eb017098 100644 --- a/codex-rs/codex-api/src/endpoint/session.rs +++ b/codex-rs/codex-api/src/endpoint/session.rs @@ -1,3 +1,4 @@ +use crate::api_bridge::is_server_overloaded_transport_error; use crate::auth::SharedAuthProvider; use crate::error::ApiError; use crate::provider::Provider; @@ -107,6 +108,7 @@ impl EndpointSession { transport.execute(req).await } }, + |_| true, ) .await?; @@ -125,6 +127,7 @@ impl EndpointSession { path: &str, extra_headers: HeaderMap, body: Option, + defer_server_overloaded_retries: bool, configure: C, ) -> Result where @@ -148,6 +151,7 @@ impl EndpointSession { transport.stream(req).await } }, + |err| !defer_server_overloaded_retries || !is_server_overloaded_transport_error(err), ) .await?; diff --git a/codex-rs/codex-api/src/telemetry.rs b/codex-rs/codex-api/src/telemetry.rs index 91918a65b9..7512e2b3fb 100644 --- a/codex-rs/codex-api/src/telemetry.rs +++ b/codex-rs/codex-api/src/telemetry.rs @@ -5,7 +5,7 @@ use codex_client::Response; use codex_client::RetryPolicy; use codex_client::StreamResponse; use codex_client::TransportError; -use codex_client::run_with_retry; +use codex_client::run_with_retry_if; use http::StatusCode; use std::future::Future; use std::sync::Arc; @@ -70,6 +70,7 @@ pub(crate) async fn run_with_request_telemetry( telemetry: Option>, make_request: impl FnMut() -> Request, send: F, + retry_if: impl Fn(&TransportError) -> bool, ) -> Result where T: WithStatus, @@ -78,21 +79,26 @@ where { // Wraps `run_with_retry` to attach per-attempt request telemetry for both // unary and streaming HTTP calls. - run_with_retry(policy, make_request, move |req, attempt| { - let telemetry = telemetry.clone(); - let send = send.clone(); - async move { - let start = Instant::now(); - let result = send(req).await; - if let Some(t) = telemetry.as_ref() { - let (status, err) = match &result { - Ok(resp) => (Some(resp.status()), None), - Err(err) => (http_status(err), Some(err)), - }; - t.on_request(attempt, status, err, start.elapsed()); + run_with_retry_if( + policy, + make_request, + move |req, attempt| { + let telemetry = telemetry.clone(); + let send = send.clone(); + async move { + let start = Instant::now(); + let result = send(req).await; + if let Some(t) = telemetry.as_ref() { + let (status, err) = match &result { + Ok(resp) => (Some(resp.status()), None), + Err(err) => (http_status(err), Some(err)), + }; + t.on_request(attempt, status, err, start.elapsed()); + } + result } - result - } - }) + }, + retry_if, + ) .await } diff --git a/codex-rs/codex-client/src/lib.rs b/codex-rs/codex-client/src/lib.rs index 4b83ba0dd0..3ad166f140 100644 --- a/codex-rs/codex-client/src/lib.rs +++ b/codex-rs/codex-client/src/lib.rs @@ -6,6 +6,7 @@ pub use crate::retry::RetryOn; pub use crate::retry::RetryPolicy; pub use crate::retry::backoff; pub use crate::retry::run_with_retry; +pub use crate::retry::run_with_retry_if; pub use crate::sse::sse_stream; pub use crate::telemetry::RequestTelemetry; pub use codex_http_client::HttpClient as CodexHttpClient; diff --git a/codex-rs/codex-client/src/retry.rs b/codex-rs/codex-client/src/retry.rs index 9eb408878b..c6ce5d958f 100644 --- a/codex-rs/codex-client/src/retry.rs +++ b/codex-rs/codex-client/src/retry.rs @@ -47,9 +47,22 @@ pub fn backoff(base: Duration, attempt: u64) -> Duration { } pub async fn run_with_retry( + policy: RetryPolicy, + make_req: impl FnMut() -> Request, + op: F, +) -> Result +where + F: Fn(Request, u64) -> Fut, + Fut: Future>, +{ + run_with_retry_if(policy, make_req, op, |_| true).await +} + +pub async fn run_with_retry_if( policy: RetryPolicy, mut make_req: impl FnMut() -> Request, op: F, + retry_if: impl Fn(&TransportError) -> bool, ) -> Result where F: Fn(Request, u64) -> Fut, @@ -60,9 +73,10 @@ where match op(req, attempt).await { Ok(resp) => return Ok(resp), Err(err) - if policy - .retry_on - .should_retry(&err, attempt, policy.max_attempts) => + if retry_if(&err) + && policy + .retry_on + .should_retry(&err, attempt, policy.max_attempts) => { sleep(backoff(policy.base_delay, attempt + 1)).await; } diff --git a/codex-rs/core/src/client.rs b/codex-rs/core/src/client.rs index edcd5dfad1..3aac010f41 100644 --- a/codex-rs/core/src/client.rs +++ b/codex-rs/core/src/client.rs @@ -272,7 +272,6 @@ pub struct ModelClient { pub struct ModelClientSession { client: ModelClient, websocket_session: WebsocketSession, - caller_manages_http_request_retries: bool, /// Turn state for sticky routing. /// /// This is an `OnceLock` that stores the turn state value received from the server @@ -481,7 +480,6 @@ impl ModelClient { ModelClientSession { client: self.clone(), websocket_session: self.take_cached_websocket_session(), - caller_manages_http_request_retries: false, turn_state: Arc::new(OnceLock::new()), } } @@ -1120,10 +1118,6 @@ impl ModelClientSession { Arc::clone(&self.turn_state) } - pub(crate) fn use_caller_managed_http_request_retries(&mut self) { - self.caller_manages_http_request_retries = true; - } - fn reset_websocket_session(&mut self) { self.websocket_session.connection = None; self.websocket_session.last_request = None; @@ -1409,6 +1403,7 @@ impl ModelClientSession { summary: ReasoningSummaryConfig, service_tier: Option, responses_metadata: &CodexResponsesMetadata, + defer_server_overloaded_retries: bool, inference_trace: &InferenceTraceContext, ) -> Result { let auth_manager = self.client.state.provider.auth_manager(); @@ -1459,15 +1454,13 @@ impl ModelClientSession { let inference_trace_attempt = inference_trace.start_attempt(); inference_trace_attempt.add_request_headers(&mut options.extra_headers); inference_trace_attempt.record_started(&request); - let mut responses_api_provider = client_setup.api_provider; - if self.caller_manages_http_request_retries { - // Keep ordinary HTTP open failures in one outer request loop. - responses_api_provider.retry.retry_5xx = false; - responses_api_provider.retry.retry_transport = false; - } - let client = - ApiResponsesClient::new(transport, responses_api_provider, client_setup.api_auth) - .with_telemetry(Some(request_telemetry), Some(sse_telemetry)); + let client = ApiResponsesClient::new( + transport, + client_setup.api_provider, + client_setup.api_auth, + ) + .with_telemetry(Some(request_telemetry), Some(sse_telemetry)) + .with_deferred_server_overload_retries(defer_server_overloaded_retries); let stream_result = client.stream_request(request, options).await; match stream_result { @@ -1792,6 +1785,33 @@ impl ModelClientSession { service_tier: Option, responses_metadata: &CodexResponsesMetadata, inference_trace: &InferenceTraceContext, + ) -> Result { + self.stream_with_deferred_server_overload_retries( + prompt, + model_info, + session_telemetry, + effort, + summary, + service_tier, + responses_metadata, + false, + inference_trace, + ) + .await + } + + #[allow(clippy::too_many_arguments)] + pub(crate) async fn stream_with_deferred_server_overload_retries( + &mut self, + prompt: &Prompt, + model_info: &ModelInfo, + session_telemetry: &SessionTelemetry, + effort: Option, + summary: ReasoningSummaryConfig, + service_tier: Option, + responses_metadata: &CodexResponsesMetadata, + defer_server_overloaded_retries: bool, + inference_trace: &InferenceTraceContext, ) -> Result { let wire_api = self.client.state.provider.info().wire_api; match wire_api { @@ -1828,6 +1848,7 @@ impl ModelClientSession { summary, service_tier, responses_metadata, + defer_server_overloaded_retries, inference_trace, ) .await diff --git a/codex-rs/core/src/compact_remote_v2.rs b/codex-rs/core/src/compact_remote_v2.rs index fadf5934d1..a9b9b6afb6 100644 --- a/codex-rs/core/src/compact_remote_v2.rs +++ b/codex-rs/core/src/compact_remote_v2.rs @@ -17,7 +17,6 @@ use crate::hook_runtime::run_post_compact_hooks; use crate::hook_runtime::run_pre_compact_hooks; use crate::responses_metadata::CodexResponsesMetadata; use crate::responses_metadata::CompactionTurnMetadata; -use crate::responses_retry::ResponsesRetryOutcome; use crate::responses_retry::ResponsesStreamRequest; use crate::responses_retry::handle_retryable_response_stream_error; use crate::session::session::Session; @@ -348,21 +347,17 @@ async fn run_remote_compaction_request_v2( responses_metadata: &CodexResponsesMetadata, cancellation_token: &CancellationToken, ) -> CodexResult { - let max_stream_retries = turn_context + let max_retries = turn_context .provider .info() .stream_max_retries() .min(MAX_REMOTE_COMPACTION_V2_STREAM_RETRIES); - let max_request_retries = turn_context.provider.info().request_max_retries(); - client_session.use_caller_managed_http_request_retries(); - let mut stream_retries = 0; - let mut request_retries = 0; + let mut retries = 0; let mut server_overloaded_retries = 0; loop { - let mut response_stream_opened = false; let result = async { let stream = client_session - .stream( + .stream_with_deferred_server_overload_retries( prompt, &turn_context.model_info, &turn_context.session_telemetry, @@ -370,10 +365,10 @@ async fn run_remote_compaction_request_v2( turn_context.reasoning_summary, turn_context.config.service_tier.clone(), responses_metadata, + true, &InferenceTraceContext::disabled(), ) .await?; - response_stream_opened = true; collect_compaction_output(stream).await } .or_cancel(cancellation_token) @@ -388,17 +383,12 @@ async fn run_remote_compaction_request_v2( if !err.is_retryable() && !should_retry_server_overload { return Err(err); } - let http_request_failed_to_open = - !sess.services.model_client.responses_websocket_enabled() - && !response_stream_opened; - let (retry_counter, max_retries) = if should_retry_server_overload { - (&mut server_overloaded_retries, max_stream_retries) - } else if http_request_failed_to_open { - (&mut request_retries, max_request_retries) + let retry_counter = if should_retry_server_overload { + &mut server_overloaded_retries } else { - (&mut stream_retries, max_stream_retries) + &mut retries }; - let retry_outcome = handle_retryable_response_stream_error( + handle_retryable_response_stream_error( retry_counter, max_retries, err, @@ -409,10 +399,6 @@ async fn run_remote_compaction_request_v2( cancellation_token.child_token(), ) .await?; - if retry_outcome == ResponsesRetryOutcome::TransportFallback { - request_retries = 0; - stream_retries = 0; - } } } } diff --git a/codex-rs/core/src/responses_retry.rs b/codex-rs/core/src/responses_retry.rs index f9b8a05b84..eb55b47f94 100644 --- a/codex-rs/core/src/responses_retry.rs +++ b/codex-rs/core/src/responses_retry.rs @@ -26,14 +26,8 @@ pub(crate) enum ResponsesStreamRequest { RemoteCompactionV2, } -#[derive(Debug, Clone, Copy, PartialEq, Eq)] -pub(crate) enum ResponsesRetryOutcome { - Retrying, - TransportFallback, -} - -/// Handles a retryable stream error and reports whether the retry stays on the -/// current transport or falls back to another one. +/// Handles a retryable stream error and returns `Ok(())` when the caller should +/// retry the request loop. #[allow(clippy::too_many_arguments)] pub(crate) async fn handle_retryable_response_stream_error( retries: &mut u64, @@ -44,7 +38,7 @@ pub(crate) async fn handle_retryable_response_stream_error( turn_context: &TurnContext, request: ResponsesStreamRequest, cancellation_token: CancellationToken, -) -> Result { +) -> Result<(), CodexErr> { let is_server_overloaded = matches!(&err, CodexErr::ServerOverloaded); let max_retries = if is_server_overloaded { SERVER_OVERLOADED_MAX_RETRIES @@ -66,7 +60,7 @@ pub(crate) async fn handle_retryable_response_stream_error( ) .await; *retries = 0; - return Ok(ResponsesRetryOutcome::TransportFallback); + return Ok(()); } if *retries < max_retries { @@ -100,7 +94,7 @@ pub(crate) async fn handle_retryable_response_stream_error( .await; } return tokio::select! { - () = tokio::time::sleep(delay) => Ok(ResponsesRetryOutcome::Retrying), + () = tokio::time::sleep(delay) => Ok(()), () = cancellation_token.cancelled() => Err(CodexErr::TurnAborted), }; } diff --git a/codex-rs/core/src/session/turn.rs b/codex-rs/core/src/session/turn.rs index ba224497c3..d893a1cc12 100644 --- a/codex-rs/core/src/session/turn.rs +++ b/codex-rs/core/src/session/turn.rs @@ -37,7 +37,6 @@ use crate::mentions::collect_tool_mentions_from_messages; use crate::plugins::build_plugin_injections; use crate::responses_metadata::CodexResponsesMetadata; use crate::responses_metadata::CodexResponsesRequestKind; -use crate::responses_retry::ResponsesRetryOutcome; use crate::responses_retry::ResponsesStreamRequest; use crate::responses_retry::handle_retryable_response_stream_error; use crate::session::PreviousTurnSettings; @@ -1153,11 +1152,8 @@ async fn run_sampling_request( Arc::clone(&router), Arc::clone(&turn_diff_tracker), ); - let max_stream_retries = turn_context.provider.info().stream_max_retries(); - let max_request_retries = turn_context.provider.info().request_max_retries(); - client_session.use_caller_managed_http_request_retries(); - let mut stream_retries = 0; - let mut request_retries = 0; + let max_retries = turn_context.provider.info().stream_max_retries(); + let mut retries = 0; let mut server_overloaded_retries = 0; let mut initial_input = Some(input); let mut original_input = None; @@ -1175,14 +1171,12 @@ async fn run_sampling_request( turn_context.as_ref(), base_instructions.clone(), ); - let mut response_stream_opened = false; let err = match try_run_sampling_request( tool_runtime.clone(), Arc::clone(&sess), Arc::clone(&turn_context), Arc::clone(&turn_store), client_session, - &mut response_stream_opened, responses_metadata, Arc::clone(&turn_diff_tracker), &prompt, @@ -1217,16 +1211,12 @@ async fn run_sampling_request( return Err(err); } - let http_request_failed_to_open = - !sess.services.model_client.responses_websocket_enabled() && !response_stream_opened; - let (retry_counter, max_retries) = if should_retry_server_overload { - (&mut server_overloaded_retries, max_stream_retries) - } else if http_request_failed_to_open { - (&mut request_retries, max_request_retries) + let retry_counter = if should_retry_server_overload { + &mut server_overloaded_retries } else { - (&mut stream_retries, max_stream_retries) + &mut retries }; - let retry_outcome = handle_retryable_response_stream_error( + handle_retryable_response_stream_error( retry_counter, max_retries, err, @@ -1237,10 +1227,6 @@ async fn run_sampling_request( cancellation_token.child_token(), ) .await?; - if retry_outcome == ResponsesRetryOutcome::TransportFallback { - request_retries = 0; - stream_retries = 0; - } turn_context.turn_timing_state.record_sampling_retry(); } } @@ -1983,7 +1969,6 @@ async fn try_run_sampling_request( turn_context: Arc, turn_store: Arc, client_session: &mut ModelClientSession, - response_stream_opened: &mut bool, responses_metadata: &CodexResponsesMetadata, turn_diff_tracker: SharedTurnDiffTracker, prompt: &Prompt, @@ -2009,7 +1994,7 @@ async fn try_run_sampling_request( .enabled(Feature::ConcurrentReasoningSummaries) && turn_context.provider.info().is_openai(); let mut stream = client_session - .stream( + .stream_with_deferred_server_overload_retries( prompt, &turn_context.model_info, &turn_context.session_telemetry, @@ -2017,12 +2002,12 @@ async fn try_run_sampling_request( turn_context.reasoning_summary, turn_context.config.service_tier.clone(), responses_metadata, + true, &inference_trace, ) .instrument(trace_span!("stream_request")) .or_cancel(&cancellation_token) .await??; - *response_stream_opened = true; let mut in_flight: FuturesOrdered>> = FuturesOrdered::new(); let mut needs_follow_up = false; diff --git a/codex-rs/core/tests/suite/client.rs b/codex-rs/core/tests/suite/client.rs index 35b74f15de..14cc3e4566 100644 --- a/codex-rs/core/tests/suite/client.rs +++ b/codex-rs/core/tests/suite/client.rs @@ -3614,127 +3614,6 @@ async fn server_overload_retries_use_a_separate_budget_on_the_same_turn() -> any Ok(()) } -#[tokio::test] -async fn http_5xx_retries_respect_request_retry_budget() -> anyhow::Result<()> { - skip_if_no_network!(Ok(())); - let server = MockServer::start().await; - let responses = mount_response_sequence( - &server, - vec![ - ResponseTemplate::new(503).set_body_string("temporary gateway failure"), - ResponseTemplate::new(503).set_body_string("temporary gateway failure"), - ], - ) - .await; - - let test = test_codex() - .with_config(|config| { - config.model_provider.request_max_retries = Some(1); - config.model_provider.stream_max_retries = Some(5); - }) - .build(&server) - .await?; - tokio::time::pause(); - - test.codex - .submit(Op::UserInput { - items: vec![UserInput::Text { - text: "respect request retry budget".into(), - text_elements: Vec::new(), - }], - final_output_json_schema: None, - responsesapi_client_metadata: None, - additional_context: Default::default(), - thread_settings: Default::default(), - }) - .await?; - - let mut saw_terminal_error = false; - loop { - match test.codex.next_event().await?.msg { - EventMsg::StreamError(_) => tokio::time::advance(Duration::from_secs(1)).await, - EventMsg::Error(_) => saw_terminal_error = true, - EventMsg::TurnComplete(_) => break, - _ => {} - } - } - - assert!(saw_terminal_error); - assert_eq!(responses.requests().len(), 2); - Ok(()) -} - -#[tokio::test] -async fn mixed_network_and_5xx_respect_single_request_retry_budget() -> anyhow::Result<()> { - fn connection_reset(_: &wiremock::Request) -> std::io::Error { - std::io::Error::new( - std::io::ErrorKind::ConnectionReset, - "synthetic connection reset", - ) - } - - skip_if_no_network!(Ok(())); - let server = MockServer::start().await; - Mock::given(method("POST")) - .and(path("/v1/responses")) - .respond_with_err(connection_reset) - .up_to_n_times(1) - .expect(1) - .with_priority(1) - .mount(&server) - .await; - Mock::given(method("POST")) - .and(path("/v1/responses")) - .respond_with(ResponseTemplate::new(500).set_body_string("temporary gateway failure")) - .with_priority(2) - .mount(&server) - .await; - - let test = test_codex() - .with_config(|config| { - config.model_provider.request_max_retries = Some(1); - config.model_provider.stream_max_retries = Some(5); - }) - .build(&server) - .await?; - tokio::time::pause(); - - test.codex - .submit(Op::UserInput { - items: vec![UserInput::Text { - text: "respect one shared request retry budget".into(), - text_elements: Vec::new(), - }], - final_output_json_schema: None, - responsesapi_client_metadata: None, - additional_context: Default::default(), - thread_settings: Default::default(), - }) - .await?; - - let mut saw_terminal_error = false; - loop { - match test.codex.next_event().await?.msg { - EventMsg::StreamError(_) => tokio::time::advance(Duration::from_secs(1)).await, - EventMsg::Error(_) => saw_terminal_error = true, - EventMsg::TurnComplete(_) => break, - _ => {} - } - } - - assert!(saw_terminal_error); - let response_requests = server - .received_requests() - .await - .unwrap_or_default() - .into_iter() - .filter(|request| request.method == wiremock::http::Method::POST) - .filter(|request| request.url.path() == "/v1/responses") - .count(); - assert_eq!(response_requests, 2); - Ok(()) -} - /// We try to avoid setting env vars in tests because std::env::set_var() is /// process-wide and unsafe. Though for this test, we want to simulate the /// presence of an environment variable that the provider will read for auth, so diff --git a/codex-rs/core/tests/suite/client_websockets.rs b/codex-rs/core/tests/suite/client_websockets.rs index 7c1e23db34..7ce2cc81c6 100755 --- a/codex-rs/core/tests/suite/client_websockets.rs +++ b/codex-rs/core/tests/suite/client_websockets.rs @@ -1622,42 +1622,6 @@ async fn responses_websocket_connection_limit_error_reconnects_and_completes() { server.shutdown().await; } -#[tokio::test(flavor = "multi_thread", worker_threads = 2)] -async fn responses_websocket_5xx_uses_stream_retry_budget() { - skip_if_no_network!(); - - let websocket_server_error = json!({ - "type": "error", - "status": 500, - "error": { - "type": "server_error", - "message": "temporary websocket failure" - } - }); - let server = start_websocket_server(vec![ - vec![vec![websocket_server_error]], - vec![vec![ev_response_created("resp-1"), ev_completed("resp-1")]], - ]) - .await; - let mut builder = test_codex().with_config(|config| { - config.model_provider.request_max_retries = Some(0); - config.model_provider.stream_max_retries = Some(1); - }); - let test = builder - .build_with_websocket_server(&server) - .await - .expect("build websocket codex"); - - test.submit_turn("hello") - .await - .expect("submission should retry websocket 5xx as a stream failure"); - - let total_websocket_requests: usize = server.connections().iter().map(Vec::len).sum(); - assert_eq!(total_websocket_requests, 2); - - server.shutdown().await; -} - #[tokio::test(flavor = "multi_thread", worker_threads = 2)] async fn responses_websocket_uses_incremental_create_on_prefix() { skip_if_no_network!(); diff --git a/codex-rs/core/tests/suite/compact.rs b/codex-rs/core/tests/suite/compact.rs index 4bf7a9d791..54139b425d 100644 --- a/codex-rs/core/tests/suite/compact.rs +++ b/codex-rs/core/tests/suite/compact.rs @@ -686,7 +686,7 @@ async fn summarize_context_three_requests_and_instructions() { } #[tokio::test] -async fn local_compact_keeps_provider_http_5xx_retries() -> Result<()> { +async fn local_compact_keeps_provider_server_overload_retries() -> Result<()> { skip_if_no_network!(Ok(())); let server = start_mock_server().await; let responses = mount_response_sequence( @@ -696,7 +696,12 @@ async fn local_compact_keeps_provider_http_5xx_retries() -> Result<()> { ev_assistant_message("m1", FIRST_REPLY), ev_completed("r1"), ])), - ResponseTemplate::new(500).set_body_string("temporary compact failure"), + ResponseTemplate::new(503).set_body_json(serde_json::json!({ + "error": { + "code": "server_is_overloaded", + "message": "temporary compact failure" + } + })), sse_response(sse(vec![ ev_assistant_message("m2", SUMMARY_TEXT), ev_completed("r2"),