diff --git a/backend/crates/gateway-api/src/openai/responses/response.rs b/backend/crates/gateway-api/src/openai/responses/response.rs index 67a4cf6ab..6b115c33d 100644 --- a/backend/crates/gateway-api/src/openai/responses/response.rs +++ b/backend/crates/gateway-api/src/openai/responses/response.rs @@ -2,7 +2,9 @@ use bytes::Bytes; use gateway_core::event::{GatewayEvent, ProtocolWireEvent, ProviderEvent}; -use gateway_protocol::openai::sse::encode_sse_event_with_metadata; +use gateway_protocol::openai::sse::{ + encode_sse_event_with_metadata, response_failed_sse_data_from_error_event, +}; use serde_json::Value; use super::error::ResponseEncodeError; @@ -17,6 +19,7 @@ const OPENAI_PROTOCOL: &str = "openai"; #[derive(Debug, Default)] pub struct OpenAiResponsesEncoder { response_id: Option, + response_snapshot: Option, wire_terminal: Option, wire_failure: bool, } @@ -27,6 +30,7 @@ impl OpenAiResponsesEncoder { pub const fn new() -> Self { Self { response_id: None, + response_snapshot: None, wire_terminal: None, wire_failure: false, } @@ -39,6 +43,22 @@ impl OpenAiResponsesEncoder { return Vec::new(); }; self.observe_wire(wire); + // 当前 Codex 不消费 Responses `error` event,会在 EOF 时丢失失败原因。 + // 在客户端 SSE 边界统一投影成它能识别的 `response.failed`。 + if wire.event_type() == Some("error") + && let Some(data) = response_failed_sse_data_from_error_event( + self.response_snapshot.as_ref(), + self.response_id.as_deref(), + wire.data(), + ) + { + return vec![Bytes::from(encode_sse_event_with_metadata( + "response.failed", + &data.to_string(), + wire.sse_id(), + wire.sse_retry(), + ))]; + } if let Some(raw_sse_frame) = wire.raw_sse_frame() { return vec![raw_sse_frame.clone()]; } @@ -110,6 +130,16 @@ impl OpenAiResponsesEncoder { let effective_type = wire .event_type() .or_else(|| wire.data().get("type").and_then(Value::as_str)); + if matches!( + effective_type, + Some("response.created" | "response.in_progress" | "response.queued") + ) && let Some(response) = wire + .data() + .get("response") + .filter(|value| value.is_object()) + { + self.response_snapshot = Some(response.clone()); + } if matches!( effective_type, Some("response.completed" | "response.incomplete") diff --git a/backend/crates/gateway-api/tests/openai/responses/http.rs b/backend/crates/gateway-api/tests/openai/responses/http.rs index 87967e0e6..2499c15d2 100644 --- a/backend/crates/gateway-api/tests/openai/responses/http.rs +++ b/backend/crates/gateway-api/tests/openai/responses/http.rs @@ -34,6 +34,7 @@ use gateway_core::event::{ use gateway_core::operation::{Operation, OperationKind}; use gateway_core::routing::PublicModelId; use gateway_core::upstream::{OpaqueUpstreamValue, UpstreamSendState}; +use gateway_protocol::openai::sse::{encode_sse_event, parse_sse_events}; use serde_json::{Value, json}; use gateway_api::openai::responses::{collect_execution_response, stream_execution_response}; @@ -1572,6 +1573,113 @@ async fn streaming_upstream_wire_failure_should_not_be_rewritten_as_a_gateway_er ); } +#[tokio::test] +async fn streaming_upstream_error_event_should_be_translated_to_response_failed_for_sse() { + let trace = Arc::new(Trace::default()); + let error_data = json!({ + "type": "error", + "error": { + "type": "service_unavailable_error", + "code": "server_is_overloaded", + "message": "Our servers are currently overloaded. Please try again later.", + "param": null, + "future_error_field": {"keep": true} + }, + "sequence_number": 2, + "future_event_field": {"keep": true} + }); + let projected_error_frame = Bytes::from(encode_sse_event("error", &error_data.to_string())); + let error_wire = ProviderEvent::wire( + ProtocolWireEvent::json_with_raw_sse_metadata( + "openai", + Some("error".to_owned()), + error_data, + projected_error_frame, + None, + None, + ) + .expect("valid upstream error wire"), + ); + let started_wire = ProviderEvent::canonical_with_wire( + vec![started()], + ProtocolWireEvent::json( + "openai", + Some("response.created".to_owned()), + json!({ + "type": "response.created", + "response": { + "id": "resp_test", + "model": "public-model", + "status": "in_progress", + "future_response_field": {"keep": true} + } + }), + ) + .expect("valid upstream started wire"), + ); + let session = FakeSession::streaming( + Arc::clone(&trace), + vec![ + NextStep::Event(CoordinatedEvent::single( + started_wire, + CommitRequirement::CommitBeforeDelivery, + )), + NextStep::Event(CoordinatedEvent::single( + error_wire, + CommitRequirement::AlreadyCommitted, + )), + NextStep::Error(EngineError::Provider(ProviderError::new( + ProviderErrorKind::UpstreamCapacityUnavailable, + UpstreamSendState::Sent, + ))), + ], + ); + + let response = stream_execution_response(Box::new(session), None).await; + let body = to_bytes(response.into_body(), usize::MAX) + .await + .expect("read SSE body"); + let body = String::from_utf8(body.to_vec()).expect("SSE is UTF-8"); + let events = parse_sse_events(&body).expect("translated SSE should parse"); + let failed = events + .iter() + .find(|event| event.event.as_deref() == Some("response.failed")) + .expect("translated response.failed event"); + let failed: Value = serde_json::from_str(&failed.data).expect("response.failed JSON"); + + assert_eq!( + ( + failed, + body.matches("event: response.failed").count(), + body.contains("event: error\n"), + body.ends_with("data: [DONE]\n\n"), + ), + ( + json!({ + "type": "response.failed", + "response": { + "id": "resp_test", + "model": "public-model", + "status": "failed", + "error": { + "type": "service_unavailable_error", + "code": "server_is_overloaded", + "message": "Our servers are currently overloaded. Please try again later.", + "param": null, + "future_error_field": {"keep": true} + }, + "future_response_field": {"keep": true} + }, + "sequence_number": 2, + "future_event_field": {"keep": true} + }), + 1, + false, + true, + ) + ); +} + #[tokio::test] async fn atomic_uncommitted_upstream_failure_batch_should_be_forwarded_once() { let trace = Arc::new(Trace::default()); diff --git a/backend/crates/gateway-protocol/src/openai/sse.rs b/backend/crates/gateway-protocol/src/openai/sse.rs index 287f59e40..166a8ab93 100644 --- a/backend/crates/gateway-protocol/src/openai/sse.rs +++ b/backend/crates/gateway-protocol/src/openai/sse.rs @@ -2,7 +2,7 @@ use std::fmt; -use serde_json::json; +use serde_json::{Value, json}; use thiserror::Error; /// 单条 SSE 事件。 @@ -277,12 +277,7 @@ pub fn response_failed_sse_data_with_id( "code": code, "message": message, }); - let response_id = response_id - .filter(|value| !value.trim().is_empty()) - .map_or_else( - || format!("resp_proxy_{}", uuid::Uuid::new_v4().simple()), - ToString::to_string, - ); + let response_id = response_id_or_generated(response_id); json!({ "type": "response.failed", "response": { @@ -294,6 +289,50 @@ pub fn response_failed_sse_data_with_id( }) } +/// 将携带嵌套错误的上游事件投影为 OpenAI Responses `response.failed` 数据。 +/// +/// 该投影供客户端 SSE 兼容转换使用。原始错误对象和事件级元数据保持不透明; +/// 已有 response 快照只更新失败终态字段。 +pub fn response_failed_sse_data_from_error_event( + response: Option<&Value>, + response_id: Option<&str>, + error_event: &Value, +) -> Option { + let mut event = error_event.as_object()?.clone(); + let error = event.remove("error").filter(Value::is_object)?; + let mut response = response + .and_then(Value::as_object) + .cloned() + .unwrap_or_default(); + let has_response_id = response + .get("id") + .and_then(Value::as_str) + .is_some_and(|value| !value.trim().is_empty()); + if !has_response_id { + response.insert( + "id".to_owned(), + Value::String(response_id_or_generated(response_id)), + ); + } + response.insert("status".to_owned(), Value::String("failed".to_owned())); + response.insert("error".to_owned(), error); + event.insert( + "type".to_owned(), + Value::String("response.failed".to_owned()), + ); + event.insert("response".to_owned(), Value::Object(response)); + Some(Value::Object(event)) +} + +fn response_id_or_generated(response_id: Option<&str>) -> String { + response_id + .filter(|value| !value.trim().is_empty()) + .map_or_else( + || format!("resp_proxy_{}", uuid::Uuid::new_v4().simple()), + ToString::to_string, + ) +} + /// 返回下一个完整 SSE 帧结束位置(含分隔符)。 pub fn sse_frame_end(bytes: &[u8]) -> Option { sse_frame_separator_bytes(bytes).map(|(position, separator_len)| position + separator_len) diff --git a/backend/crates/gateway-protocol/tests/openai/sse.rs b/backend/crates/gateway-protocol/tests/openai/sse.rs index 4add156dd..a916bff72 100644 --- a/backend/crates/gateway-protocol/tests/openai/sse.rs +++ b/backend/crates/gateway-protocol/tests/openai/sse.rs @@ -1,8 +1,10 @@ use gateway_protocol::openai::sse::{ DONE_SSE_FRAME, MAX_SSE_EVENT_BUFFER_BYTES, SseError, SseEvent, SseEventDecoder, - encode_sse_event, encode_sse_event_with_metadata, parse_sse_events, response_failed_sse_event, + encode_sse_event, encode_sse_event_with_metadata, parse_sse_events, + response_failed_sse_data_from_error_event, response_failed_sse_event, response_failed_sse_event_with_id, sse_frame_end, sse_frame_is_done, }; +use serde_json::json; fn sse_body_has_done(body: &str) -> bool { body.trim_end_matches(['\r', '\n']) @@ -386,6 +388,57 @@ fn response_failed_event_should_preserve_existing_response_id() { assert_eq!(data["response"]["id"], "resp_existing"); } +#[test] +fn response_failed_projection_should_preserve_response_error_and_event_metadata() { + let response = json!({ + "id": "resp_existing", + "model": "gpt-test", + "status": "in_progress", + "future_response_field": {"keep": true} + }); + let error_event = json!({ + "type": "error", + "error": { + "type": "service_unavailable_error", + "code": "server_is_overloaded", + "message": "overloaded", + "param": null, + "future_error_field": {"keep": true} + }, + "sequence_number": 2, + "future_event_field": {"keep": true} + }); + + let projected = response_failed_sse_data_from_error_event( + Some(&response), + Some("ignored_fallback"), + &error_event, + ) + .expect("wrapped error event should project"); + + assert_eq!( + projected, + json!({ + "type": "response.failed", + "response": { + "id": "resp_existing", + "model": "gpt-test", + "status": "failed", + "error": { + "type": "service_unavailable_error", + "code": "server_is_overloaded", + "message": "overloaded", + "param": null, + "future_error_field": {"keep": true} + }, + "future_response_field": {"keep": true} + }, + "sequence_number": 2, + "future_event_field": {"keep": true} + }) + ); +} + #[test] fn done_frame_detection_should_require_exact_single_data_field() { assert_eq!( diff --git a/backend/crates/providers/openai/tests/provider/contract.rs b/backend/crates/providers/openai/tests/provider/contract.rs index 44abf70c8..f734ccf99 100644 --- a/backend/crates/providers/openai/tests/provider/contract.rs +++ b/backend/crates/providers/openai/tests/provider/contract.rs @@ -470,6 +470,7 @@ fn contract_account_scope() -> Arc { "acct_websocket_turn_state", "acct_ws_quota_a", "acct_ws_quota_b", + "acct_ws_midstream_overload", ] .into_iter() .map(|id| { @@ -2373,6 +2374,114 @@ async fn abrupt_websocket_disconnect_preserves_diagnosis_and_ambiguous_send_stat ); } +#[tokio::test] +async fn websocket_midstream_error_frame_surfaces_upstream_message_after_delivery() { + const ACCOUNT_ID: &str = "acct_ws_midstream_overload"; + const RESPONSE_ID: &str = "resp_midstream_overload"; + const OVERLOAD_MESSAGE: &str = "Our servers are currently overloaded. Please try again later."; + + let store = Arc::new(MemoryAccountStore::default()); + create_account(&store, ACCOUNT_ID).await; + let listener = TcpListener::bind("127.0.0.1:0").await.unwrap(); + let base_url = format!("http://{}", listener.local_addr().unwrap()); + let server = tokio::spawn(async move { + let (stream, _) = listener.accept().await.expect("accept WebSocket"); + let mut websocket = accept_codex_test_websocket(stream).await; + let _request = websocket + .next() + .await + .expect("WebSocket request") + .expect("valid WebSocket request"); + // 真实生产观察:OpenAI 在流已开始后发送带原话的 `error` 帧再断连。 + websocket + .send(Message::Text( + json!({ + "type": "response.created", + "response": {"id": RESPONSE_ID, "model": "gpt-5.4", "status": "in_progress"} + }) + .to_string() + .into(), + )) + .await + .expect("send response.created"); + websocket + .send(Message::Text( + json!({ + "type": "response.in_progress", + "response": {"id": RESPONSE_ID, "model": "gpt-5.4", "status": "in_progress"} + }) + .to_string() + .into(), + )) + .await + .expect("send response.in_progress"); + tokio::time::sleep(Duration::from_millis(1_600)).await; + websocket + .send(Message::Text( + json!({ + "type": "error", + "error": { + "type": "service_unavailable_error", + "code": "server_is_overloaded", + "message": OVERLOAD_MESSAGE, + "param": null + }, + "sequence_number": 2 + }) + .to_string() + .into(), + )) + .await + .expect("send error frame"); + // 不发送任何终止事件,直接断开,模拟上游发完错误帧后的真实行为。 + }); + + let provider = provider_with_base_url(&store, base_url); + let mut stream = provider + .execute( + planned_request("openai", generate_operation()), + context("req_ws_midstream_overload", CancellationToken::new()), + ) + .await + .expect("prepare midstream-failure stream"); + let mut failure = loop { + match stream.next().await { + Some(Ok(_)) => {} + Some(Err(error)) => break error, + None => panic!("midstream error frame must surface a typed failure"), + } + }; + server.abort(); + + assert_eq!( + failure.kind(), + ProviderErrorKind::UpstreamCapacityUnavailable + ); + assert_eq!(failure.send_state(), UpstreamSendState::Sent); + // 交给 Core 的失败必须保留上游原话:客户端只能靠它知道失败原因。 + let visible = failure + .client_visible_upstream_error() + .expect("overload frame must carry a client-visible upstream error"); + assert_eq!(visible.code(), Some("server_is_overloaded")); + assert_eq!(visible.message(), OVERLOAD_MESSAGE); + // 原始错误帧必须随失败一起交给交付边界(SSE 侧据此翻译成 response.failed)。 + let atomic = failure.take_atomic_client_events(); + let error_frame = atomic + .iter() + .find_map(|event| { + let wire = event.wire_event()?; + (wire.event_type() == Some("error")).then(|| wire.data().clone()) + }) + .expect("atomic batch must carry the upstream error frame"); + assert_eq!( + error_frame + .get("error") + .and_then(|error| error.get("message")) + .and_then(Value::as_str), + Some(OVERLOAD_MESSAGE) + ); +} + #[tokio::test] async fn websocket_idle_timeout_diagnosis_survives_ambiguous_send_wrapping() { const ACCOUNT_ID: &str = "acct_abrupt_disconnect";