Add structured telemetry for response retries (#38452)

## What changed

- Emit trace-safe `codex.retry` events immediately before retry delays, including the attempt number, selected delay, retry layer, and operation.
- Cover HTTP requests, sampling streams, remote compaction, and sampling connection recovery. Track connection-recovery attempts separately from the stream retry budget.
- Add integration coverage for retry timing and terminal behavior across HTTP, SSE, WebSocket, and remote-compaction paths, including rate limits, overloads, connection failures, and `Retry-After` inputs.

GitOrigin-RevId: dae38900a0579cf8ba062c3f3d90ee61851c1c99

codex-rs/core/src/responses_retry.rs

@@ -6,6 +6,7 @@ use crate::client::ModelClientSession;
 use crate::session::session::Session;
 use crate::session::turn_context::TurnContext;
 use crate::util::backoff;
+use codex_client::RetryOperation;
 use codex_protocol::error::CodexErr;
 use codex_protocol::error::CodexErrorDetails;
 use codex_protocol::protocol::EventMsg;
@@ -23,6 +24,7 @@ pub(crate) enum ResponsesStreamRequest {
 
 pub(crate) struct ResponsesStreamRetryState {
     retries: u64,
+    connection_retries: u64,
     connection_retry_delay: Duration,
 }
 
@@ -30,6 +32,7 @@ impl Default for ResponsesStreamRetryState {
     fn default() -> Self {
         Self {
             retries: 0,
+            connection_retries: 0,
             connection_retry_delay: INITIAL_CONNECTION_RETRY_DELAY,
         }
     }
@@ -46,6 +49,11 @@ pub(crate) async fn handle_retryable_response_stream_error(
     turn_context: &TurnContext,
     request: ResponsesStreamRequest,
 ) -> Result<(), CodexErr> {
+    let operation = match request {
+        ResponsesStreamRequest::Sampling => RetryOperation::Sampling,
+        ResponsesStreamRequest::RemoteCompactionV2 => RetryOperation::RemoteCompactionV2,
+    };
+
     if matches!(request, ResponsesStreamRequest::Sampling)
         && matches!(err.details(), CodexErrorDetails::ConnectionFailed(_))
         && !turn_context.session_source.is_internal()
@@ -60,6 +68,8 @@ pub(crate) async fn handle_retryable_response_stream_error(
         );
         sess.notify_stream_error(turn_context, "Reconnecting... waiting for network", err)
             .await;
+        retry_state.connection_retries = retry_state.connection_retries.saturating_add(1);
+        codex_client::record_retry!(retry_state.connection_retries, retry_delay, operation);
         tokio::time::sleep(retry_delay).await;
         retry_state.connection_retry_delay = retry_delay
             .saturating_mul(2)
@@ -105,6 +115,7 @@ pub(crate) async fn handle_retryable_response_stream_error(
             )
             .await;
         }
+        codex_client::record_retry!(retry_count, delay, operation);
         tokio::time::sleep(delay).await;
         return Ok(());
     }