Keep response streams alive through connection failures (#37485)

## What changed

- Classify HTTP connection failures separately from other network errors without exposing request URLs.
- For sampling requests, retry connection failures with exponential delays from 5 to 60 seconds and show a `Reconnecting... waiting for network` stream error.
- Preserve the normal stream retry budget while waiting for the provider to become reachable. Keep the existing bounded retry behavior for other retryable errors.

## Testing

- Verify connection errors are classified without leaking URL contents.
- Verify a turn recovers after its provider becomes reachable and still applies the configured retry limit to a subsequent incomplete stream.

GitOrigin-RevId: 646553290c865a1332abd30c4a64ed9266bbfc6f

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),

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,
         }
     }

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,

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);
 

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,

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(())
+}

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}")]

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())

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()

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(),
     }