fix(observability): preserve upstream failure evidence

Log original OpenAI error payloads and WebSocket close reasons with request, account, attempt, transport, and connection correlation. Project PostgreSQL inet values as bare client addresses in observability queries.

Co-authored-by: Codex <noreply@openai.com>

backend/crates/gateway-store/src/postgres/observability/queries/ops.rs

@@ -19,7 +19,7 @@ const REQUEST_ERROR_SELECT: &str = "select 'model_request'::text as source,
        mr.latency_ms,
        coalesce(mr.error_message, mr.error_kind, 'request failed') as message,
        1::integer as occurrence_count,
-       mr.client_ip::text as client_ip, mr.user_agent,
+       host(mr.client_ip) as client_ip, mr.user_agent,
        mr.reasoning_effort, mr.reasoning_preset, mr.request_kind,
        mr.subagent_kind, mr.compact,
        mr.completed_at as occurred_at,
@@ -40,7 +40,7 @@ const OPS_EVENT_SELECT: &str = "select 'ops_event'::text as source,
        null::integer as client_status_code, oe.status_code as upstream_status_code,
        oe.provider_error_code, mr.client_response_id, oe.upstream_request_id,
        oe.latency_ms, oe.message, oe.occurrence_count,
-       mr.client_ip::text as client_ip, mr.user_agent,
+       host(mr.client_ip) as client_ip, mr.user_agent,
        mr.reasoning_effort, mr.reasoning_preset, mr.request_kind,
        mr.subagent_kind, mr.compact,
        oe.created_at as occurred_at,

backend/crates/gateway-store/src/postgres/observability/queries/usage.rs

@@ -117,7 +117,7 @@ pub(crate) const USAGE_LIST_RECORD_SELECT: &str =
             mr.first_event_ms, mr.first_reasoning_ms, mr.first_text_ms, mr.first_token_ms,
             mr.provider_processing_ms, mr.latency_ms, mr.admission_decision_ms,
             mr.account_selection_wait_ms, mr.capacity_used_slots, mr.capacity_total_slots,
-            mr.client_ip::text as client_ip, mr.user_agent,
+            host(mr.client_ip) as client_ip, mr.user_agent,
             mr.reasoning_effort, mr.reasoning_preset, mr.subagent_kind, mr.compact,
             mr.started_at
      from model_requests mr";
@@ -145,7 +145,7 @@ pub(crate) const USAGE_RECORD_DETAIL_SELECT: &str =
             mr.first_event_ms, mr.first_reasoning_ms, mr.first_text_ms, mr.first_token_ms,
             mr.provider_processing_ms, mr.latency_ms, mr.admission_decision_ms,
             mr.account_selection_wait_ms, mr.capacity_used_slots, mr.capacity_total_slots,
-            mr.client_ip::text as client_ip,
+            host(mr.client_ip) as client_ip,
             mr.user_agent, mr.reasoning_effort, mr.reasoning_preset, mr.request_kind,
             mr.subagent_kind, mr.compact, mr.image_generation_requested,
             mr.image_generation_succeeded, mr.started_at, mr.deadline_at, mr.completed_at

backend/crates/providers/openai/src/provider.rs

@@ -47,6 +47,7 @@ use secrecy::{ExposeSecret, SecretString};
 use serde::{Deserialize, Serialize};
 use serde_json::{Map, Value, json};
 use url::Url;
+use uuid::Uuid;
 
 use crate::credential::{
     CodexAccountFailure, CodexCredentialCatalogError, CodexCredentialCatalogService,
@@ -754,6 +755,12 @@ fn cold_json_response_stream(request: ColdJsonResponse) -> EventStream {
             account_selection,
         )
         .await;
+        if let Err(CodexHandshakeAttemptError::Client(error)) = &response {
+            log_client_upstream_error(
+                UpstreamErrorLogContext::new(&request.context, &active_account, None),
+                error,
+            );
+        }
         let response = match response.map_err(map_handshake_attempt_error) {
             Ok(response) => response,
             Err(mut failure) => {
@@ -880,6 +887,12 @@ fn cold_response_stream(response: ColdResponse) -> EventStream {
             &cancellation,
         )
         .await;
+        if let Err(CodexHandshakeAttemptError::Client(error)) = &response {
+            log_client_upstream_error(
+                UpstreamErrorLogContext::new(&context, &active_account, None),
+                error,
+            );
+        }
         let response = response.map_err(map_handshake_attempt_error);
         let response = match response {
             Ok(response) => response,
@@ -948,6 +961,7 @@ fn cold_response_stream(response: ColdResponse) -> EventStream {
             active_account = current;
         }
         let response_transport = response.transport;
+        let websocket_connection_id = response.websocket_connection_id;
         let mut body = response.body;
         let failure_diagnostics = response.diagnostics.clone();
         let failure_set_cookie_headers = response.set_cookie_headers.clone();
@@ -992,7 +1006,17 @@ fn cold_response_stream(response: ColdResponse) -> EventStream {
                 _ = wait_for_replay_grace(replay_grace_deadline) => Ok(PreCommitPoll::GraceElapsed),
                 chunk = body.next() => match chunk {
                     Some(Ok(chunk)) => Ok(PreCommitPoll::Upstream(Some(chunk))),
-                    Some(Err(error)) => Err(map_stream_error(error)),
+                    Some(Err(error)) => {
+                        log_client_upstream_error(
+                            UpstreamErrorLogContext::new(
+                                &context,
+                                &active_account,
+                                websocket_connection_id,
+                            ),
+                            &error,
+                        );
+                        Err(map_stream_error(error))
+                    }
                     None => Ok(PreCommitPoll::Upstream(None)),
                 },
             };
@@ -1068,6 +1092,15 @@ fn cold_response_stream(response: ColdResponse) -> EventStream {
             let service_tier_changed = observation_state
                 .observe_upstream_service_tier(decoder.response_service_tier());
             let terminal_failure = canonical_failure.map(|(error, semantic_output_seen)| {
+                log_canonical_upstream_error(
+                    UpstreamErrorLogContext::new(
+                        &context,
+                        &active_account,
+                        websocket_connection_id,
+                    ),
+                    response_transport,
+                    &error,
+                );
                 let atomic_upstream_failure = matches!(&error, CodexCanonicalError::Upstream(_));
                 (
                     map_canonical_error(
@@ -1159,6 +1192,15 @@ fn cold_response_stream(response: ColdResponse) -> EventStream {
             }
         };
         let terminal_failure = canonical_failure.map(|(error, semantic_output_seen)| {
+            log_canonical_upstream_error(
+                UpstreamErrorLogContext::new(
+                    &context,
+                    &active_account,
+                    websocket_connection_id,
+                ),
+                response_transport,
+                &error,
+            );
             let atomic_upstream_failure = matches!(&error, CodexCanonicalError::Upstream(_));
             (
                 map_canonical_error(
@@ -2205,6 +2247,141 @@ struct OpenAiFailureContext<'a> {
     allows_account_state_mutation: bool,
 }
 
+#[derive(Clone, Copy)]
+struct UpstreamErrorLogContext<'a> {
+    request_id: &'a str,
+    account_id: &'a str,
+    attempt_index: u32,
+    websocket_connection_id: Option<Uuid>,
+}
+
+impl<'a> UpstreamErrorLogContext<'a> {
+    fn new(
+        context: &'a AttemptContext,
+        account: &'a ProviderAccount,
+        websocket_connection_id: Option<Uuid>,
+    ) -> Self {
+        Self {
+            request_id: context.request_id().as_str(),
+            account_id: account.id().as_str(),
+            attempt_index: context.attempt_index().get(),
+            websocket_connection_id,
+        }
+    }
+
+    fn with_websocket_connection_id(mut self, connection_id: Option<Uuid>) -> Self {
+        self.websocket_connection_id = self.websocket_connection_id.or(connection_id);
+        self
+    }
+}
+
+fn log_client_upstream_error(context: UpstreamErrorLogContext<'_>, error: &CodexClientError) {
+    match error {
+        CodexClientError::Upstream {
+            status,
+            body,
+            transport,
+            ..
+        } => log_raw_upstream_body(
+            context,
+            *transport,
+            "http_error_response",
+            Some(status.as_u16()),
+            None,
+            None,
+            body,
+        ),
+        CodexClientError::WebSocket(CodexWebSocketExchangeError::Upstream(upstream)) => {
+            log_raw_upstream_body(
+                context,
+                CodexBackendTransport::WebSocket,
+                "websocket_opening_response",
+                Some(upstream.status_code),
+                None,
+                None,
+                &upstream.body,
+            );
+        }
+        CodexClientError::WebSocket(error) => {
+            let Some(close) = error.close_before_terminal() else {
+                return;
+            };
+            let context = context.with_websocket_connection_id(close.connection_id());
+            let websocket_connection_id = context
+                .websocket_connection_id
+                .map(|connection_id| connection_id.to_string())
+                .unwrap_or_default();
+            let upstream_error_raw = close.reason().unwrap_or_default();
+            tracing::warn!(
+                request_id = %context.request_id,
+                account_id = %context.account_id,
+                attempt_index = context.attempt_index,
+                websocket_connection_id,
+                upstream_transport = WEBSOCKET_TRANSPORT,
+                upstream_error_kind = "websocket_close",
+                upstream_close_code = close.code().unwrap_or_default(),
+                upstream_close_code_present = close.code().is_some(),
+                upstream_error_raw,
+                upstream_error_raw_bytes = upstream_error_raw.len(),
+                upstream_error_raw_present = close.reason().is_some(),
+                "OpenAI upstream WebSocket closed before terminal event"
+            );
+        }
+        _ => {}
+    }
+}
+
+fn log_canonical_upstream_error(
+    context: UpstreamErrorLogContext<'_>,
+    transport: CodexBackendTransport,
+    error: &CodexCanonicalError,
+) {
+    let CodexCanonicalError::Upstream(failure) = error else {
+        return;
+    };
+    log_raw_upstream_body(
+        context,
+        transport,
+        "responses_error_event",
+        failure.explicit_status_code,
+        failure.upstream_code.as_deref(),
+        failure.upstream_type.as_deref(),
+        failure.raw_body(),
+    );
+}
+
+fn log_raw_upstream_body(
+    context: UpstreamErrorLogContext<'_>,
+    transport: CodexBackendTransport,
+    error_kind: &'static str,
+    status_code: Option<u16>,
+    upstream_code: Option<&str>,
+    upstream_type: Option<&str>,
+    upstream_error_raw: &str,
+) {
+    let websocket_connection_id = context
+        .websocket_connection_id
+        .map(|connection_id| connection_id.to_string())
+        .unwrap_or_default();
+    tracing::warn!(
+        request_id = %context.request_id,
+        account_id = %context.account_id,
+        attempt_index = context.attempt_index,
+        websocket_connection_id,
+        upstream_transport = actual_transport_name(transport),
+        upstream_error_kind = error_kind,
+        upstream_status_code = status_code.unwrap_or_default(),
+        upstream_status_code_present = status_code.is_some(),
+        upstream_error_code = upstream_code.unwrap_or_default(),
+        upstream_error_code_present = upstream_code.is_some(),
+        upstream_error_type = upstream_type.unwrap_or_default(),
+        upstream_error_type_present = upstream_type.is_some(),
+        upstream_error_raw,
+        upstream_error_raw_bytes = upstream_error_raw.len(),
+        "OpenAI upstream returned an error payload"
+    );
+}
+
 async fn apply_failure(
     context: &OpenAiFailureContext<'_>,
     account: &ProviderAccount,

backend/crates/providers/openai/src/transport/canonical.rs

@@ -59,7 +59,7 @@ pub enum CodexCanonicalError {
     Protocol(#[source] ProviderError),
     /// 上游在成功建立流后发送了明确的失败事件。
     #[error("Codex upstream reported a failed response")]
-    Upstream(ResponsesSseFailure),
+    Upstream(Box<ResponsesSseFailure>),
 }
 
 impl fmt::Debug for CodexCanonicalError {
@@ -291,7 +291,11 @@ impl CodexCanonicalDecoder {
         let signals = response_event_signals(event_type, &value);
         self.merge_timing_signals(signals);
         if matches!(event_type, Some("response.failed" | "error")) {
-            let failure = ResponsesSseFailure::from_event(event_type.unwrap_or_default(), &value);
+            let failure = ResponsesSseFailure::from_raw_event(
+                event_type.unwrap_or_default(),
+                &event.data,
+                &value,
+            );
             let mut canonical = Vec::new();
             if !self.started {
                 // 失败首帧可能是唯一携带 response_id 的上游事实。身份投影只做
@@ -307,7 +311,7 @@ impl CodexCanonicalDecoder {
             } else {
                 output.extend(canonical.into_iter().map(ProviderEvent::canonical));
             }
-            return Err(CodexCanonicalError::Upstream(failure));
+            return Err(CodexCanonicalError::Upstream(Box::new(failure)));
         }
         let mut canonical = Vec::new();
         if let Some(event_type) = event_type {

backend/crates/providers/openai/src/transport/client.rs

@@ -22,6 +22,7 @@ use reqwest::{
 };
 use serde_json::{Value, map::Map};
 use thiserror::Error;
+use uuid::Uuid;
 
 use crate::transport::protocol::responses::{CodexResponsesRequest, TransportRequirement};
 
@@ -538,6 +539,8 @@ pub struct CodexBackendStreamingResponse {
     pub body: CodexBackendSseStream,
     /// 实际使用的上游传输。
     pub transport: CodexBackendTransport,
+    /// WebSocket 响应所绑定的连接；HTTP transport 为 `None`。
+    pub websocket_connection_id: Option<Uuid>,
     /// 响应头里的最新 turn state。
     pub turn_state: Option<String>,
     /// 上游透传的 `set-cookie` 列表。

backend/crates/providers/openai/src/transport/client_sse.rs

@@ -151,6 +151,7 @@ impl CodexBackendClient {
         Ok(CodexBackendStreamingResponse {
             body: http_sse_stream(response, Arc::clone(&rate_limit_updates)),
             transport: CodexBackendTransport::HttpSse,
+            websocket_connection_id: None,
             turn_state,
             set_cookie_headers,
             rate_limit_headers,
@@ -387,7 +388,20 @@ impl CodexBackendClient {
                     loop {
                         match await_websocket_delivery_boundary(&mut exchange).await {
                             Ok(DeliveryBoundary::Ready) => break,
-                            Ok(DeliveryBoundary::ConnectionLimitReached { message }) => {
+                            Ok(DeliveryBoundary::ConnectionLimitReached {
+                                message,
+                                raw_frame,
+                            }) => {
+                                let upstream_error_raw = String::from_utf8_lossy(&raw_frame);
+                                tracing::warn!(
+                                    request_id = %context.request_id,
+                                    account_id = context.account_id.unwrap_or_default(),
+                                    websocket_connection_id = %exchange.websocket_connection_id,
+                                    upstream_transport = "websocket",
+                                    upstream_error_kind = "websocket_error_frame",
+                                    upstream_error_raw = %upstream_error_raw,
+                                    "OpenAI upstream returned a recoverable WebSocket error frame"
+                                );
                                 if fresh_retry_used {
                                     // 新连接仍被限流：与官方重试预算耗尽一致，降级同账号 HTTP/SSE。
                                     return self
@@ -513,6 +527,7 @@ impl CodexBackendClient {
                             .map_err(websocket_exchange_error_to_client_error),
                     ),
                     transport: CodexBackendTransport::WebSocket,
+                    websocket_connection_id: Some(exchange.websocket_connection_id),
                     turn_state: exchange.turn_state,
                     set_cookie_headers: exchange.set_cookie_headers,
                     rate_limit_headers: exchange.rate_limit_headers,
@@ -546,6 +561,7 @@ impl CodexBackendClient {
         delivery_wait_started_at: Instant,
         error: CodexWebSocketExchangeError,
     ) -> CodexClientResult<CodexBackendStreamingResponse> {
+        log_raw_websocket_close(context, &error);
         self.http_fallback_before_delivery(
             request,
             context,
@@ -807,6 +823,8 @@ enum DeliveryBoundary {
     ConnectionLimitReached {
         /// 上游错误说明。
         message: String,
+        /// 从 WebSocket 文本事件构造的完整 SSE 错误帧。
+        raw_frame: bytes::Bytes,
     },
 }
 
@@ -819,26 +837,54 @@ async fn await_websocket_delivery_boundary(
             Some(Ok(frame)) if is_websocket_lifecycle_prelude(&frame) => prelude.push(frame),
             Some(Ok(frame)) => {
                 let limit_message = websocket_connection_limit_message(&frame);
+                let raw_frame = limit_message.as_ref().map(|_| frame.clone());
                 prelude.push(frame);
                 let remaining =
                     std::mem::replace(&mut exchange.body, Box::pin(futures::stream::empty()));
                 exchange.body =
                     Box::pin(futures::stream::iter(prelude.into_iter().map(Ok)).chain(remaining));
                 return Ok(match limit_message {
-                    Some(message) => DeliveryBoundary::ConnectionLimitReached { message },
+                    Some(message) => DeliveryBoundary::ConnectionLimitReached {
+                        message,
+                        raw_frame: raw_frame.unwrap_or_default(),
+                    },
                     None => DeliveryBoundary::Ready,
                 });
             }
             Some(Err(error)) => return Err(error),
             None => {
-                return Err(CodexWebSocketExchangeError::closed_before_terminal(
-                    None, None,
+                return Err(CodexWebSocketExchangeError::closed_before_terminal_on(
+                    exchange.websocket_connection_id,
+                    None,
+                    None,
                 ));
             }
         }
     }
 }
 
+fn log_raw_websocket_close(context: CodexRequestContext<'_>, error: &CodexWebSocketExchangeError) {
+    let Some(close) = error.close_before_terminal() else {
+        return;
+    };
+    let websocket_connection_id = close
+        .connection_id()
+        .map(|connection_id| connection_id.to_string())
+        .unwrap_or_default();
+    let upstream_error_raw = close.reason().unwrap_or_default();
+    tracing::warn!(
+        request_id = %context.request_id,
+        account_id = context.account_id.unwrap_or_default(),
+        websocket_connection_id,
+        upstream_transport = "websocket",
+        upstream_error_kind = "websocket_close",
+        upstream_close_code = ?close.code(),
+        upstream_error_raw,
+        upstream_error_raw_present = close.reason().is_some(),
+        "OpenAI upstream WebSocket closed before delivery"
+    );
+}
+
 fn is_websocket_lifecycle_prelude(frame: &[u8]) -> bool {
     frame.starts_with(b"event: response.created\n")
         || frame.starts_with(b"event: response.in_progress\n")

backend/crates/providers/openai/src/transport/protocol/responses.rs

@@ -404,6 +404,8 @@ pub struct ResponsesSseFailure {
     pub explicit_status_code: Option<u16>,
     /// 上游显式重试间隔，或从官方限流消息中解析出的重试间隔。
     pub retry_after_seconds: Option<u64>,
+    /// 上游错误事件的原始 JSON data；只应在明确的失败审计边界读取。
+    raw_body: String,
 }
 
 impl fmt::Debug for ResponsesSseFailure {
@@ -422,6 +424,10 @@ impl fmt::Debug for ResponsesSseFailure {
 
 impl ResponsesSseFailure {
     pub fn from_event(event: &str, value: &Value) -> Self {
+        Self::from_raw_event(event, &value.to_string(), value)
+    }
+
+    pub(crate) fn from_raw_event(event: &str, raw_body: &str, value: &Value) -> Self {
         Self {
             event: event.to_string(),
             message: failure_message(value).unwrap_or_else(|| "Codex upstream SSE failed".into()),
@@ -429,8 +435,15 @@ impl ResponsesSseFailure {
             upstream_type: failure_type(value),
             explicit_status_code: failure_explicit_status_code(value),
             retry_after_seconds: events::retry_after_seconds_from_value(value),
+            raw_body: raw_body.to_owned(),
         }
     }
+
+    /// 返回上游错误事件未经重编码的 JSON data。
+    #[must_use]
+    pub fn raw_body(&self) -> &str {
+        &self.raw_body
+    }
 }
 
 fn failure_message(value: &Value) -> Option<String> {

backend/crates/providers/openai/src/transport/websocket/error.rs

@@ -4,6 +4,7 @@ use std::{fmt, time::Duration};
 
 use gateway_protocol::openai::sse::SseError;
 use thiserror::Error;
+use uuid::Uuid;
 
 use crate::transport::client::CodexClientVisibleUpstreamResponse;
 use crate::transport::diagnostics::CodexUpstreamDiagnostics;
@@ -104,13 +105,29 @@ pub enum CodexWebSocketExchangeError {
 /// close reason 只供原请求的客户端错误响应使用；`Debug` 与 `Display` 均不会输出它。
 #[derive(Clone, PartialEq, Eq)]
 pub struct CodexWebSocketCloseError {
+    connection_id: Option<Uuid>,
     code: Option<u16>,
     reason: Option<String>,
 }
 
 impl CodexWebSocketCloseError {
     pub(crate) fn new(code: Option<u16>, reason: Option<String>) -> Self {
-        Self { code, reason }
+        Self {
+            connection_id: None,
+            code,
+            reason,
+        }
+    }
+
+    pub(crate) fn with_connection_id(mut self, connection_id: Uuid) -> Self {
+        self.connection_id = Some(connection_id);
+        self
+    }
+
+    /// 返回承载该 close frame 的 WebSocket 连接标识。
+    #[must_use]
+    pub const fn connection_id(&self) -> Option<Uuid> {
+        self.connection_id
     }
 
     /// 返回上游 close code；没有 close frame 时为 `None`。
@@ -130,6 +147,7 @@ impl fmt::Debug for CodexWebSocketCloseError {
     fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
         formatter
             .debug_struct("CodexWebSocketCloseError")
+            .field("connection_id", &self.connection_id)
             .field("code", &self.code)
             .field("reason", &self.reason.as_ref().map(|_| "<redacted>"))
             .finish()
@@ -192,8 +210,14 @@ impl fmt::Display for CodexWebSocketUpstreamError {
 }
 
 impl CodexWebSocketExchangeError {
-    pub(crate) fn closed_before_terminal(code: Option<u16>, reason: Option<String>) -> Self {
-        Self::ClosedBeforeTerminal(CodexWebSocketCloseError::new(code, reason))
+    pub(crate) fn closed_before_terminal_on(
+        connection_id: Uuid,
+        code: Option<u16>,
+        reason: Option<String>,
+    ) -> Self {
+        Self::ClosedBeforeTerminal(
+            CodexWebSocketCloseError::new(code, reason).with_connection_id(connection_id),
+        )
     }
 
     pub(crate) fn close_before_terminal(&self) -> Option<&CodexWebSocketCloseError> {

backend/crates/providers/openai/src/transport/websocket/exchange/stream.rs

@@ -208,13 +208,15 @@ async fn forward_websocket_response_stream(state: WebSocketStreamForwardState) {
                 return;
             }
             tungstenite::Message::Close(frame) => {
+                let websocket_connection_id = websocket.connection_id();
                 discard_stream_websocket(
                     websocket,
                     pool_return,
                     StreamWebSocketDiscardReason::UpstreamClosed,
                 )
                 .await;
-                let error = CodexWebSocketExchangeError::closed_before_terminal(
+                let error = CodexWebSocketExchangeError::closed_before_terminal_on(
+                    websocket_connection_id,
                     frame.as_ref().map(|frame| u16::from(frame.code)),
                     frame.map(|frame| frame.reason.to_string()),
                 );
@@ -287,9 +289,11 @@ async fn forward_websocket_response_stream(state: WebSocketStreamForwardState) {
         }
     }
 
+    let websocket_connection_id = websocket.connection_id();
     let upstream_close = websocket
         .exit_reason()
-        .and_then(|reason| reason.upstream_close().cloned());
+        .and_then(|reason| reason.upstream_close().cloned())
+        .map(|close| close.with_connection_id(websocket_connection_id));
     discard_stream_websocket(
         websocket,
         pool_return,
@@ -297,7 +301,13 @@ async fn forward_websocket_response_stream(state: WebSocketStreamForwardState) {
     )
     .await;
     let error = upstream_close.map_or_else(
-        || CodexWebSocketExchangeError::closed_before_terminal(None, None),
+        || {
+            CodexWebSocketExchangeError::closed_before_terminal_on(
+                websocket_connection_id,
+                None,
+                None,
+            )
+        },
         CodexWebSocketExchangeError::ClosedBeforeTerminal,
     );
     let error = if reused_connection {

backend/crates/providers/openai/src/transport/websocket/pump.rs

@@ -556,12 +556,19 @@ fn log_pump_exit(
             "Responses WebSocket pump stopped unexpectedly"
         );
     } else if matches!(reason, PumpExitReason::UpstreamCloseFrame { .. }) {
+        let upstream_close = reason.upstream_close();
+        let upstream_error_raw = upstream_close
+            .and_then(CodexWebSocketCloseError::reason)
+            .unwrap_or_default();
         tracing::info!(
             websocket_connection_id = %connection_id,
             account_id = %account_id,
             conversation_id_hash = %conversation_id_hash,
             pump_exit_reason = reason.as_str(),
             pump_exit_detail = detail,
+            upstream_close_code = ?upstream_close.and_then(CodexWebSocketCloseError::code),
+            upstream_error_raw,
+            upstream_error_raw_present = upstream_close.is_some_and(|close| close.reason().is_some()),
             backpressure_events,
             "Responses WebSocket pump received close frame"
         );

backend/crates/providers/openai/tests/transport/canonical.rs

@@ -854,9 +854,10 @@ fn decoder_should_preserve_changed_upstream_response_ids_as_wire() {
 }
 
 fn assert_failed_event(code: &str, marker: &str) {
-    let body = format!(
-        "event: response.failed\ndata: {{\"response\":{{\"id\":\"resp_failed\",\"status\":\"failed\",\"error\":{{\"code\":\"{code}\",\"message\":\"{marker}\"}}}}}}\n\n"
+    let raw_error = format!(
+        "{{\"response\":{{\"id\":\"resp_failed\",\"status\":\"failed\",\"error\":{{\"code\":\"{code}\",\"message\":\"{marker}\"}}}}}}"
     );
+    let body = format!("event: response.failed\ndata: {raw_error}\n\n");
     let failure = CodexCanonicalDecoder::new("fallback")
         .push(body.as_bytes())
         .expect_err("failed event must become a typed error");
@@ -865,6 +866,7 @@ fn assert_failed_event(code: &str, marker: &str) {
         panic!("response.failed must preserve its typed upstream failure");
     };
     assert_eq!(upstream.upstream_code.as_deref(), Some(code));
+    assert_eq!(upstream.raw_body(), raw_error);
     assert!(!failure.semantic_output_seen());
     assert!(!format!("{failure:?}").contains(marker));
 }

backend/crates/providers/openai/tests/transport/latency.rs

@@ -1117,6 +1117,10 @@ async fn websocket_failure_after_first_delivery_should_not_open_http_fallback()
     let CodexWebSocketExchangeError::ClosedBeforeTerminal(close) = *source else {
         panic!("expected upstream close frame source");
     };
+    assert!(
+        close.connection_id().is_some(),
+        "close failures must retain their WebSocket connection correlation"
+    );
     assert_eq!(close.code(), Some(1009));
     assert_eq!(close.reason(), Some("message too big after output"));
 }

backend/crates/providers/openai/tests/transport/mod.rs

@@ -234,6 +234,7 @@ async fn collect_backend_response(
     let CodexBackendStreamingResponse {
         mut body,
         transport,
+        websocket_connection_id: _,
         mut turn_state,
         set_cookie_headers,
         mut rate_limit_headers,