diff --git a/backend/crates/gateway-store/src/postgres/observability/queries/ops.rs b/backend/crates/gateway-store/src/postgres/observability/queries/ops.rs index 99a8e006b..6e0a17c1f 100644 --- a/backend/crates/gateway-store/src/postgres/observability/queries/ops.rs +++ b/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, diff --git a/backend/crates/gateway-store/src/postgres/observability/queries/usage.rs b/backend/crates/gateway-store/src/postgres/observability/queries/usage.rs index 34e6422d6..d91c902ad 100644 --- a/backend/crates/gateway-store/src/postgres/observability/queries/usage.rs +++ b/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 diff --git a/backend/crates/providers/openai/src/provider.rs b/backend/crates/providers/openai/src/provider.rs index 6aad76b9e..c6e086c10 100644 --- a/backend/crates/providers/openai/src/provider.rs +++ b/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, +} + +impl<'a> UpstreamErrorLogContext<'a> { + fn new( + context: &'a AttemptContext, + account: &'a ProviderAccount, + websocket_connection_id: Option, + ) -> 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) -> 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, + 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, diff --git a/backend/crates/providers/openai/src/transport/canonical.rs b/backend/crates/providers/openai/src/transport/canonical.rs index 331ce542d..710a197a6 100644 --- a/backend/crates/providers/openai/src/transport/canonical.rs +++ b/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), } 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 { diff --git a/backend/crates/providers/openai/src/transport/client.rs b/backend/crates/providers/openai/src/transport/client.rs index d787c3acb..abe1b27ad 100644 --- a/backend/crates/providers/openai/src/transport/client.rs +++ b/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, /// 响应头里的最新 turn state。 pub turn_state: Option, /// 上游透传的 `set-cookie` 列表。 diff --git a/backend/crates/providers/openai/src/transport/client_sse.rs b/backend/crates/providers/openai/src/transport/client_sse.rs index f396ed353..764a4a049 100644 --- a/backend/crates/providers/openai/src/transport/client_sse.rs +++ b/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 { + 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") diff --git a/backend/crates/providers/openai/src/transport/protocol/responses.rs b/backend/crates/providers/openai/src/transport/protocol/responses.rs index 30ea5ec0b..a73f3d59d 100644 --- a/backend/crates/providers/openai/src/transport/protocol/responses.rs +++ b/backend/crates/providers/openai/src/transport/protocol/responses.rs @@ -404,6 +404,8 @@ pub struct ResponsesSseFailure { pub explicit_status_code: Option, /// 上游显式重试间隔，或从官方限流消息中解析出的重试间隔。 pub retry_after_seconds: Option, + /// 上游错误事件的原始 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 { diff --git a/backend/crates/providers/openai/src/transport/websocket/error.rs b/backend/crates/providers/openai/src/transport/websocket/error.rs index af5d60776..9f1474da5 100644 --- a/backend/crates/providers/openai/src/transport/websocket/error.rs +++ b/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, code: Option, reason: Option, } impl CodexWebSocketCloseError { pub(crate) fn new(code: Option, reason: Option) -> 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 { + 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(|_| "")) .finish() @@ -192,8 +210,14 @@ impl fmt::Display for CodexWebSocketUpstreamError { } impl CodexWebSocketExchangeError { - pub(crate) fn closed_before_terminal(code: Option, reason: Option) -> Self { - Self::ClosedBeforeTerminal(CodexWebSocketCloseError::new(code, reason)) + pub(crate) fn closed_before_terminal_on( + connection_id: Uuid, + code: Option, + reason: Option, + ) -> Self { + Self::ClosedBeforeTerminal( + CodexWebSocketCloseError::new(code, reason).with_connection_id(connection_id), + ) } pub(crate) fn close_before_terminal(&self) -> Option<&CodexWebSocketCloseError> { diff --git a/backend/crates/providers/openai/src/transport/websocket/exchange/stream.rs b/backend/crates/providers/openai/src/transport/websocket/exchange/stream.rs index 9cb07f76f..1114729b1 100644 --- a/backend/crates/providers/openai/src/transport/websocket/exchange/stream.rs +++ b/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 { diff --git a/backend/crates/providers/openai/src/transport/websocket/pump.rs b/backend/crates/providers/openai/src/transport/websocket/pump.rs index 0e3a27a7c..8464e8a5e 100644 --- a/backend/crates/providers/openai/src/transport/websocket/pump.rs +++ b/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" ); diff --git a/backend/crates/providers/openai/tests/transport/canonical.rs b/backend/crates/providers/openai/tests/transport/canonical.rs index 7500fbc89..340080c88 100644 --- a/backend/crates/providers/openai/tests/transport/canonical.rs +++ b/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)); } diff --git a/backend/crates/providers/openai/tests/transport/latency.rs b/backend/crates/providers/openai/tests/transport/latency.rs index cb9bd8191..e9a8c6de7 100644 --- a/backend/crates/providers/openai/tests/transport/latency.rs +++ b/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")); } diff --git a/backend/crates/providers/openai/tests/transport/mod.rs b/backend/crates/providers/openai/tests/transport/mod.rs index 4b2718869..8cca61c6e 100644 --- a/backend/crates/providers/openai/tests/transport/mod.rs +++ b/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,