mirror of
https://github.com/zyycn/codex-proxy-rs.git
synced 2026-11-06 08:06:19 +00:00
fix(gateway-api): project bare WebSocket errors to response.failed (#99)
codex 的 WS 客户端(codex-api/src/endpoint/responses_websocket.rs)只消费 两类 error 帧:带非 2xx HTTP status 的包装错误,以及 websocket_connection_limit_reached / previous_response_not_found 内置可重 试错误码。OpenAI 上游的中流错误帧(如 server_is_overloaded)没有 status 字段,map_wrapped_websocket_error_event 返回 None,落回 process_responses_event 也没有 error 分支——整帧被静默忽略。 客户端收不到可识别的终态,只能空等到自身 300s stream_idle_timeout 超时 再重连,5 次重连累计 20-30 分钟"正在思考",最后才把失败呈现出来。CPA 等实现之所以秒级报错,正是因为它把错误负载重写成 codex 可消费的终态。 #90 已在 SSE 边界做了同样的投影,但客户端 WS 边界(push_websocket)仍原 样转发;使用 WS transport 的 codex 桌面端不受 #90 保护。本提交在该边界 对不可消费的错误帧投影成 response.failed(消息 JSON 自带 type 字段,经 process_responses_event 的 response.failed 分支映射为可重试错误并立即重 试);带非 2xx status 或特殊错误码的可消费形状原样透传,保持 deliverable_business_failures_are_not_rewritten... 既有契约。
This commit is contained in:
@@ -77,6 +77,21 @@ impl OpenAiResponsesEncoder {
|
||||
return Vec::new();
|
||||
};
|
||||
self.observe_wire(wire);
|
||||
// WS 客户端只对它无法消费的裸 `error` 帧做投影:codex 的 WS 端点
|
||||
// 会静默忽略缺少 status 且不含内置可重试错误码的 `error` 帧,客户
|
||||
// 端只能空等到 idle 超时。投影成 `response.failed`(消息 JSON 自带
|
||||
// type 字段),codex 将其映射为可重试错误并立即重试。可消费形状
|
||||
// (带非 2xx status 或特殊错误码)原样透传,保持业务语义。
|
||||
if wire.event_type() == Some("error")
|
||||
&& !ws_client_consumable_error(wire.data())
|
||||
&& let Some(data) = response_failed_sse_data_from_error_event(
|
||||
self.response_snapshot.as_ref(),
|
||||
self.response_id.as_deref(),
|
||||
wire.data(),
|
||||
)
|
||||
{
|
||||
return vec![data.to_string()];
|
||||
}
|
||||
vec![wire.data().to_string()]
|
||||
}
|
||||
|
||||
@@ -156,3 +171,29 @@ fn openai_wire(event: &ProviderEvent) -> Option<&ProtocolWireEvent> {
|
||||
.wire_event()
|
||||
.filter(|wire| wire.protocol() == OPENAI_PROTOCOL)
|
||||
}
|
||||
|
||||
/// 判断 WS 上游的 `error` 帧是否已经是客户端可直接消费的形状。
|
||||
///
|
||||
/// codex 的 WS 端点只消费两类 `error` 帧:带非 2xx HTTP status 的包装错误,
|
||||
/// 以及连接数上限 / previous_response_not_found 这类内置可重试错误码;其余
|
||||
/// 帧(包括 status 为 2xx 的矛盾形状)会被静默忽略。可消费的帧原样透传,
|
||||
/// 不可消费的才在 WS 边界投影成 `response.failed`。
|
||||
fn ws_client_consumable_error(data: &Value) -> bool {
|
||||
if data
|
||||
.get("error")
|
||||
.and_then(|error| error.get("code"))
|
||||
.and_then(Value::as_str)
|
||||
.is_some_and(|code| {
|
||||
matches!(
|
||||
code,
|
||||
"websocket_connection_limit_reached" | "previous_response_not_found"
|
||||
)
|
||||
})
|
||||
{
|
||||
return true;
|
||||
}
|
||||
data.get("status")
|
||||
.or_else(|| data.get("status_code"))
|
||||
.and_then(Value::as_u64)
|
||||
.is_some_and(|status| !(200..300).contains(&status))
|
||||
}
|
||||
|
||||
@@ -666,6 +666,87 @@ fn transparent_encoder_should_use_identical_json_for_sse_and_websocket() {
|
||||
);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn transparent_encoder_should_translate_error_wire_to_response_failed_for_websocket() {
|
||||
// codex 的 WS 端点只消费带 status 的包装错误帧;上游缺少 status 的裸
|
||||
// `error` 帧会被静默忽略,客户端只能空等到 idle 超时。WS 边界与 SSE
|
||||
// 边界一致投影成 `response.failed`,codex 将其映射为可重试错误并立即
|
||||
// 重试,而不是把失败原因拖到流 EOF。
|
||||
let response_id = "resp_ws_error";
|
||||
let started = openai_wire_event(
|
||||
Vec::new(),
|
||||
"response.created",
|
||||
json!({
|
||||
"type": "response.created",
|
||||
"response": {"id": response_id, "status": "in_progress"}
|
||||
}),
|
||||
);
|
||||
let error = openai_wire_event(
|
||||
Vec::new(),
|
||||
"error",
|
||||
json!({
|
||||
"type": "error",
|
||||
"error": {
|
||||
"type": "service_unavailable_error",
|
||||
"code": "server_is_overloaded",
|
||||
"message": "Our servers are currently overloaded. Please try again later."
|
||||
},
|
||||
"sequence_number": 2
|
||||
}),
|
||||
);
|
||||
|
||||
let mut websocket_encoder = OpenAiResponsesEncoder::new();
|
||||
websocket_encoder.push_websocket(&started);
|
||||
let messages = websocket_encoder.push_websocket(&error);
|
||||
|
||||
assert_eq!(messages.len(), 1);
|
||||
assert!(websocket_encoder.has_wire_failure());
|
||||
let payload: Value = serde_json::from_str(&messages[0]).expect("response.failed is valid JSON");
|
||||
assert_eq!(payload["type"], "response.failed");
|
||||
assert_eq!(payload["response"]["id"], response_id);
|
||||
assert_eq!(payload["response"]["status"], "failed");
|
||||
assert_eq!(payload["response"]["error"]["code"], "server_is_overloaded");
|
||||
|
||||
// 与 SSE 边界投影到同一份 data 负载,两条客户端通道行为一致。
|
||||
let mut sse_encoder = OpenAiResponsesEncoder::new();
|
||||
sse_encoder.push_sse(&started);
|
||||
let frames = sse_encoder.push_sse(&error);
|
||||
let sse_text = String::from_utf8(
|
||||
frames
|
||||
.iter()
|
||||
.flat_map(|frame| frame.as_ref().iter().copied())
|
||||
.collect(),
|
||||
)
|
||||
.expect("SSE is UTF-8");
|
||||
let sse_data = parse_sse_events(&sse_text)
|
||||
.expect("SSE parses")
|
||||
.into_iter()
|
||||
.map(|event| event.data)
|
||||
.collect::<Vec<_>>();
|
||||
assert_eq!(sse_data, messages);
|
||||
|
||||
// 带非 2xx status 的包装错误帧是 codex 可直接消费的形状,必须原样透传。
|
||||
let mut passthrough_encoder = OpenAiResponsesEncoder::new();
|
||||
passthrough_encoder.push_websocket(&started);
|
||||
let wrapped = openai_wire_event(
|
||||
Vec::new(),
|
||||
"error",
|
||||
json!({
|
||||
"type": "error",
|
||||
"status": 429,
|
||||
"headers": {"x-request-id": "req_raw"},
|
||||
"error": {"type": "custom_type", "code": "custom_code", "message": "raw marker"},
|
||||
"future": {"keep": true}
|
||||
}),
|
||||
);
|
||||
let wrapped_messages = passthrough_encoder.push_websocket(&wrapped);
|
||||
assert_eq!(
|
||||
wrapped_messages,
|
||||
vec![wrapped.wire_event().expect("wire").data().to_string()]
|
||||
);
|
||||
assert!(passthrough_encoder.has_wire_failure());
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn transparent_encoder_should_follow_wire_when_canonical_identity_changes() {
|
||||
let mut encoder = OpenAiResponsesEncoder::new();
|
||||
|
||||
Reference in New Issue
Block a user