From b1c24e276b24387945e4344b94ee898544efc863 Mon Sep 17 00:00:00 2001 From: 2han9wen71an <2han9wen71an@gmail.com> Date: Tue, 15 Sep 2026 16:31:07 +0800 Subject: [PATCH] fix(gateway-api): project bare WebSocket errors to response.failed (#99) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 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... 既有契约。 --- .../src/openai/responses/response.rs | 41 ++++++++++ .../gateway-api/tests/openai/responses/mod.rs | 81 +++++++++++++++++++ 2 files changed, 122 insertions(+) diff --git a/backend/crates/gateway-api/src/openai/responses/response.rs b/backend/crates/gateway-api/src/openai/responses/response.rs index 6b115c33d..97ca3edd0 100644 --- a/backend/crates/gateway-api/src/openai/responses/response.rs +++ b/backend/crates/gateway-api/src/openai/responses/response.rs @@ -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)) +} diff --git a/backend/crates/gateway-api/tests/openai/responses/mod.rs b/backend/crates/gateway-api/tests/openai/responses/mod.rs index 95e9895b0..42f850f5d 100644 --- a/backend/crates/gateway-api/tests/openai/responses/mod.rs +++ b/backend/crates/gateway-api/tests/openai/responses/mod.rs @@ -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::>(); + 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();