starred/codex-proxy-rs
1
0
Fork 0
mirror of https://github.com/zyycn/codex-proxy-rs.git synced 2026-11-06 16:06:08 +00:00
Code Issues Packages Projects Releases Wiki Activity

fix(gateway-api): translate upstream WebSocket error events into SSE response.failed (#90)

Browse Source 
* fix(gateway-api): translate upstream WebSocket error events into SSE response.failed

WS 上游协议用 `error` 事件报告流中途失败（如 OpenAI 的 server_is_overloaded），
SSE 协议的对应形态是 `response.failed`。SSE 交付边界此前把 `error` 事件按
未知事件原样透传，而 codex 等 SSE 客户端会静默忽略未知事件类型：客户端只看到
无原因的流 EOF（"stream closed before response.completed"），丢失了上游的
完整错误负载（message/code/type），也无法据此分类重试。

现在 SSE 编码器在交付边界把 `error` wire 事件翻译成 `response.failed`：
错误负载（type/code/message/param）原样保留，response id 沿用已交付的
response.created。WS 客户端交付不变（WS 协议原生支持 error 帧）。
原有的会话级失败兜底（finish_with_gateway_error 合成 response.failed）不变。

* fix(gateway-api): preserve upstream response failures

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

---------

Co-authored-by: Zyy <zyycn@users.noreply.github.com>
Co-authored-by: Codex <noreply@openai.com>
This commit is contained in:
2han9wen71an
2026-09-15 10:49:29 +08:00
committed by Zyy
co-authored by Codex Zyy
parent fc07ac0803
commit 1bbae67040
5 changed files with 348 additions and 9 deletions
Download Patch File Download Diff File Expand all files Collapse all files
backend/crates/gateway-api/src/openai/responses/response.rs
+31 -1
View File
@@ -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<String>,
response_snapshot: Option<Value>,
wire_terminal: Option<Value>,
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")
backend/crates/gateway-api/tests/openai/responses/http.rs
+108
View File
@@ -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());
backend/crates/gateway-protocol/src/openai/sse.rs
+46 -7
View File
@@ -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<Value> {
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<usize> {
sse_frame_separator_bytes(bytes).map(|(position, separator_len)| position + separator_len)
backend/crates/gateway-protocol/tests/openai/sse.rs
+54 -1
View File
@@ -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!(
backend/crates/providers/openai/tests/provider/contract.rs
+109
View File
@@ -470,6 +470,7 @@ fn contract_account_scope() -> Arc<FrozenAccountScope> {
"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";