diff --git a/codex-rs/codex-api/src/api_bridge.rs b/codex-rs/codex-api/src/api_bridge.rs index c825ab03c5..37271934b4 100644 --- a/codex-rs/codex-api/src/api_bridge.rs +++ b/codex-rs/codex-api/src/api_bridge.rs @@ -9,6 +9,7 @@ use chrono::Utc; use codex_protocol::auth::PlanType; use codex_protocol::error::CodexErr; use codex_protocol::error::CodexErrorDetails; +use codex_protocol::error::ConnectionFailedError; use codex_protocol::error::RetryLimitReachedError; use codex_protocol::error::UnexpectedResponseError; use codex_protocol::error::UsageLimitReachedError; @@ -148,6 +149,9 @@ pub fn map_api_error(err: ApiError) -> CodexErr { request_id: None, }), TransportError::Timeout => CodexErr::RequestTimeout, + TransportError::Connection(source) => { + CodexErr::ConnectionFailed(ConnectionFailedError { source }) + } TransportError::Network(msg) | TransportError::Build(msg) => CodexErr::Stream(msg), }, ApiError::RateLimit(msg) => CodexErr::Stream(msg), diff --git a/codex-rs/codex-client/src/retry.rs b/codex-rs/codex-client/src/retry.rs index 9eb408878b..e95a3afa38 100644 --- a/codex-rs/codex-client/src/retry.rs +++ b/codex-rs/codex-client/src/retry.rs @@ -29,7 +29,9 @@ impl RetryOn { (self.retry_429 && status.as_u16() == 429) || (self.retry_5xx && status.is_server_error()) } - TransportError::Timeout | TransportError::Network(_) => self.retry_transport, + TransportError::Timeout + | TransportError::Connection(_) + | TransportError::Network(_) => self.retry_transport, _ => false, } } diff --git a/codex-rs/core/src/compact_remote_v2.rs b/codex-rs/core/src/compact_remote_v2.rs index 995cd2b700..a0af726c52 100644 --- a/codex-rs/core/src/compact_remote_v2.rs +++ b/codex-rs/core/src/compact_remote_v2.rs @@ -23,6 +23,7 @@ use crate::hook_runtime::run_pre_compact_hooks; use crate::responses_metadata::CodexResponsesMetadata; use crate::responses_metadata::CompactionTurnMetadata; use crate::responses_retry::ResponsesStreamRequest; +use crate::responses_retry::ResponsesStreamRetryState; use crate::responses_retry::handle_retryable_response_stream_error; use crate::session::session::Session; use crate::session::step_context::StepContext; @@ -344,7 +345,7 @@ async fn run_remote_compaction_request_v2( .info() .stream_max_retries() .min(MAX_REMOTE_COMPACTION_V2_STREAM_RETRIES); - let mut retries = 0; + let mut retry_state = ResponsesStreamRetryState::default(); loop { let result = match client_session .stream( @@ -368,7 +369,7 @@ async fn run_remote_compaction_request_v2( Err(err) if !err.is_retryable() => return Err(err), Err(err) => { handle_retryable_response_stream_error( - &mut retries, + &mut retry_state, max_retries, err, client_session, diff --git a/codex-rs/core/src/responses_retry.rs b/codex-rs/core/src/responses_retry.rs index dee517c985..2e822fbba1 100644 --- a/codex-rs/core/src/responses_retry.rs +++ b/codex-rs/core/src/responses_retry.rs @@ -7,20 +7,38 @@ use crate::session::session::Session; use crate::session::turn_context::TurnContext; use crate::util::backoff; use codex_protocol::error::CodexErr; +use codex_protocol::error::CodexErrorDetails; use codex_protocol::protocol::EventMsg; use codex_protocol::protocol::WarningEvent; use tracing::warn; +const INITIAL_CONNECTION_RETRY_DELAY: Duration = Duration::from_secs(5); +const MAX_CONNECTION_RETRY_DELAY: Duration = Duration::from_secs(60); + #[derive(Debug, Clone, Copy)] pub(crate) enum ResponsesStreamRequest { Sampling, RemoteCompactionV2, } +pub(crate) struct ResponsesStreamRetryState { + retries: u64, + connection_retry_delay: Duration, +} + +impl Default for ResponsesStreamRetryState { + fn default() -> Self { + Self { + retries: 0, + connection_retry_delay: INITIAL_CONNECTION_RETRY_DELAY, + } + } +} + /// Handles a retryable stream error and returns `Ok(())` when the caller should /// retry the request loop. pub(crate) async fn handle_retryable_response_stream_error( - retries: &mut u64, + retry_state: &mut ResponsesStreamRetryState, max_retries: u64, err: CodexErr, client_session: &mut ModelClientSession, @@ -28,7 +46,28 @@ pub(crate) async fn handle_retryable_response_stream_error( turn_context: &TurnContext, request: ResponsesStreamRequest, ) -> Result<(), CodexErr> { - if *retries >= max_retries + if matches!(request, ResponsesStreamRequest::Sampling) + && matches!(err.details(), CodexErrorDetails::ConnectionFailed(_)) + && !turn_context.session_source.is_internal() + && !turn_context.provider.info().is_amazon_bedrock() + { + let retry_delay = retry_state.connection_retry_delay; + warn!( + turn_id = %turn_context.sub_id, + error = %err, + ?retry_delay, + "stream connection failed; waiting to retry" + ); + sess.notify_stream_error(turn_context, "Reconnecting... waiting for network", err) + .await; + tokio::time::sleep(retry_delay).await; + retry_state.connection_retry_delay = retry_delay + .saturating_mul(2) + .min(MAX_CONNECTION_RETRY_DELAY); + return Ok(()); + } + + if retry_state.retries >= max_retries && client_session.try_switch_fallback_transport( &turn_context.session_telemetry, &turn_context.model_info, @@ -41,13 +80,13 @@ pub(crate) async fn handle_retryable_response_stream_error( }), ) .await; - *retries = 0; + retry_state.retries = 0; return Ok(()); } - if *retries < max_retries { - *retries += 1; - let retry_count = *retries; + if retry_state.retries < max_retries { + retry_state.retries += 1; + let retry_count = retry_state.retries; let delay = err.retry_delay().unwrap_or_else(|| backoff(retry_count)); log_retry(request, turn_context, &err, retry_count, max_retries, delay); diff --git a/codex-rs/core/src/session/turn.rs b/codex-rs/core/src/session/turn.rs index 96ff9cf0d3..e5157c4953 100644 --- a/codex-rs/core/src/session/turn.rs +++ b/codex-rs/core/src/session/turn.rs @@ -37,6 +37,7 @@ use crate::plugins::build_plugin_injections; use crate::responses_metadata::CodexResponsesMetadata; use crate::responses_metadata::CodexResponsesRequestKind; use crate::responses_retry::ResponsesStreamRequest; +use crate::responses_retry::ResponsesStreamRetryState; use crate::responses_retry::handle_retryable_response_stream_error; use crate::session::PreviousTurnSettings; use crate::session::TurnInput; @@ -1336,7 +1337,7 @@ async fn run_sampling_request( Arc::clone(&turn_diff_tracker), ); let max_retries = turn_context.provider.info().stream_max_retries(); - let mut retries = 0; + let mut retry_state = ResponsesStreamRetryState::default(); let mut initial_input = Some(input); let mut original_input = None; let mut executed_tool_calls_by_output = HashMap::new(); @@ -1402,7 +1403,7 @@ async fn run_sampling_request( } handle_retryable_response_stream_error( - &mut retries, + &mut retry_state, max_retries, err, client_session, diff --git a/codex-rs/core/tests/suite/stream_no_completed.rs b/codex-rs/core/tests/suite/stream_no_completed.rs index 207a2754cf..69f7852f9a 100644 --- a/codex-rs/core/tests/suite/stream_no_completed.rs +++ b/codex-rs/core/tests/suite/stream_no_completed.rs @@ -13,6 +13,9 @@ use core_test_support::streaming_sse::start_streaming_sse_server; use core_test_support::test_codex::TestCodex; use core_test_support::test_codex::test_codex; use core_test_support::wait_for_event; +use pretty_assertions::assert_eq; +use std::net::TcpListener; +use wiremock::MockServer; fn sse_incomplete() -> String { responses::sse(vec![serde_json::json!({ @@ -101,3 +104,75 @@ async fn retries_on_early_close() { server.shutdown().await; } + +#[tokio::test(flavor = "multi_thread", worker_threads = 2)] +async fn connection_failure_pauses_retry_budget_until_provider_is_reachable() -> anyhow::Result<()> +{ + skip_if_no_network!(Ok(())); + + let bootstrap_server = responses::start_mock_server().await; + let unavailable_listener = TcpListener::bind("127.0.0.1:0")?; + let unavailable_address = unavailable_listener.local_addr()?; + drop(unavailable_listener); + + let TestCodex { codex, .. } = test_codex() + .with_config(move |config| { + config.model_provider.base_url = Some(format!("http://{unavailable_address}/v1")); + config.model_provider.request_max_retries = Some(0); + config.model_provider.stream_max_retries = Some(1); + config.model_provider.supports_websockets = false; + }) + .build_with_auto_env(&bootstrap_server) + .await?; + + codex + .submit(Op::UserInput { + items: vec![UserInput::Text { + text: "recover after the network returns".into(), + text_elements: Vec::new(), + }], + final_output_json_schema: None, + responsesapi_client_metadata: None, + additional_context: Default::default(), + thread_settings: Default::default(), + }) + .await?; + + let EventMsg::StreamError(connection_error) = + wait_for_event(&codex, |event| matches!(event, EventMsg::StreamError(_))).await + else { + unreachable!("predicate guarantees a stream error event"); + }; + assert_eq!( + connection_error.message, + "Reconnecting... waiting for network" + ); + + let recovered_server = MockServer::builder() + .listener(TcpListener::bind(unavailable_address)?) + .start() + .await; + let response_mock = responses::mount_sse_sequence( + &recovered_server, + vec![sse_incomplete(), responses::sse_completed("resp_recovered")], + ) + .await; + + let EventMsg::StreamError(stream_error) = + wait_for_event(&codex, |event| matches!(event, EventMsg::StreamError(_))).await + else { + unreachable!("predicate guarantees a stream error event"); + }; + assert_eq!(stream_error.message, "Reconnecting... 1/1"); + + let EventMsg::TurnComplete(completed) = + wait_for_event(&codex, |event| matches!(event, EventMsg::TurnComplete(_))).await + else { + unreachable!("predicate guarantees a turn complete event"); + }; + + assert_eq!(completed.error, None); + assert_eq!(response_mock.requests().len(), 2); + + Ok(()) +} diff --git a/codex-rs/http-client/src/error.rs b/codex-rs/http-client/src/error.rs index 6b49229026..e3a4f8aca6 100644 --- a/codex-rs/http-client/src/error.rs +++ b/codex-rs/http-client/src/error.rs @@ -1,5 +1,6 @@ //! Errors returned by the shared Codex HTTP transport. +use crate::client::HttpError; use http::HeaderMap; use http::StatusCode; use thiserror::Error; @@ -17,6 +18,8 @@ pub enum TransportError { RetryLimit, #[error("timeout")] Timeout, + #[error("connection failed: {0}")] + Connection(#[source] HttpError), #[error("network error: {0}")] Network(String), #[error("request build error: {0}")] diff --git a/codex-rs/http-client/src/transport.rs b/codex-rs/http-client/src/transport.rs index 88c1fe8653..337b1e8675 100644 --- a/codex-rs/http-client/src/transport.rs +++ b/codex-rs/http-client/src/transport.rs @@ -78,7 +78,9 @@ impl ReqwestTransport { } fn map_error(err: reqwest::Error) -> TransportError { - if err.is_timeout() { + if err.is_connect() { + TransportError::Connection(err.without_url()) + } else if err.is_timeout() { TransportError::Timeout } else { TransportError::Network(err.to_string()) diff --git a/codex-rs/http-client/src/transport_tests.rs b/codex-rs/http-client/src/transport_tests.rs index 0562bcf476..a8ab11aba5 100644 --- a/codex-rs/http-client/src/transport_tests.rs +++ b/codex-rs/http-client/src/transport_tests.rs @@ -28,6 +28,28 @@ async fn disabled_request_logging_suppresses_transport_url_and_body() { assert!(!logs.contains("body-secret")); } +#[tokio::test] +async fn connection_failures_are_classified_without_exposing_request_urls() { + let unavailable_server = + std::net::TcpListener::bind(("127.0.0.1", 0)).expect("server port should bind"); + let server_addr = unavailable_server + .local_addr() + .expect("server listener should have an address"); + drop(unavailable_server); + let transport = ReqwestTransport::from_http_client(HttpClient::new(test_reqwest_client())); + let request = Request::new( + Method::POST, + format!("http://{server_addr}/responses?token=url-secret"), + ); + + let error = match transport.stream(request).await { + Err(TransportError::Connection(error)) => error, + Err(error) => panic!("expected a connection failure, got {error}"), + Ok(_) => panic!("an unavailable server should not return a response"), + }; + assert!(!error.to_string().contains("url-secret")); +} + fn test_reqwest_client() -> reqwest::Client { reqwest::Client::builder() .no_proxy() diff --git a/codex-rs/response-debug-context/src/lib.rs b/codex-rs/response-debug-context/src/lib.rs index d9c186e02e..cee5799bab 100644 --- a/codex-rs/response-debug-context/src/lib.rs +++ b/codex-rs/response-debug-context/src/lib.rs @@ -65,6 +65,7 @@ pub fn telemetry_transport_error_message(error: &TransportError) -> String { TransportError::Http { status, .. } => format!("http {}", status.as_u16()), TransportError::RetryLimit => "retry limit reached".to_string(), TransportError::Timeout => "timeout".to_string(), + TransportError::Connection(err) => err.to_string(), TransportError::Network(err) => err.to_string(), TransportError::Build(err) => err.to_string(), }