From f9718188915eca7322deb4cbb6dcaf9575a99fa7 Mon Sep 17 00:00:00 2001 From: zyycn Date: Mon, 15 Jun 2026 13:30:34 +0800 Subject: [PATCH] codex: add sse parser golden Co-authored-by: Codex --- docs/OPENAI_CHAIN_PARITY_REAUDIT.md | 7 +++-- tests/codex_gateway/usage_events.rs | 30 ++++++++++++++++++- .../responses/golden/non_sse_body_event.json | 8 +++++ .../golden/sse_pretty_json_events.json | 8 +++++ .../responses/golden/sse_standard_events.json | 8 +++++ 5 files changed, 57 insertions(+), 4 deletions(-) create mode 100644 tests/fixtures/responses/golden/non_sse_body_event.json create mode 100644 tests/fixtures/responses/golden/sse_pretty_json_events.json create mode 100644 tests/fixtures/responses/golden/sse_standard_events.json diff --git a/docs/OPENAI_CHAIN_PARITY_REAUDIT.md b/docs/OPENAI_CHAIN_PARITY_REAUDIT.md index be22e7cef..9a1b87b84 100644 --- a/docs/OPENAI_CHAIN_PARITY_REAUDIT.md +++ b/docs/OPENAI_CHAIN_PARITY_REAUDIT.md @@ -378,6 +378,7 @@ Rust 证据： - 本轮新增 `tests/fixtures/responses/http_sse/previous_response_not_found.sse`、`tests/fixtures/responses/http_sse/unanswered_function_call.sse` 与 `tests/codex_serving/upstream_fallback.rs::v1_responses_should_strip_history_after_http_sse_previous_response_not_found`、`v1_responses_should_strip_history_after_http_sse_unanswered_function_call`，覆盖 HTTP SSE non-streaming collect 阶段收到 stale-history 类 `error` 事件后会走同账号 strip-and-retry。 - 本轮新增 `tests/fixtures/responses/websocket/previous_response_not_found.json`、`tests/fixtures/responses/websocket/unanswered_function_call.json`，并用 `tests/codex_serving/responses_websocket.rs` 覆盖 non-streaming 与 streaming `previous_response_not_found`、non-streaming unanswered function call 的同账号 strip-and-retry。 - `tests/codex_gateway/usage_events.rs` 覆盖 multiline data、非标准 JSON 续行、`[DONE]` 忽略和非 SSE JSON body 转 `non_sse_response`。 +- 本轮新增 `tests/fixtures/responses/golden/sse_standard_events.json`、`tests/fixtures/responses/golden/sse_pretty_json_events.json`、`tests/fixtures/responses/golden/non_sse_body_event.json`，并在 `tests/codex_gateway/usage_events.rs` 中对标准 SSE、漂亮打印 JSON 续行、非 SSE body 合成事件做 exact golden 断言。 - `tests/codex_gateway/websocket.rs:376-760` 覆盖 WS 内部 rate-limit event 不转发、首个 rotatable error、mid-stream close error 和 fallback。 - 本轮新增 `tests/fixtures/responses/golden/websocket_completed.sse`，并在 `tests/codex_gateway/websocket.rs::ordinary_response_should_use_websocket_transport_by_default` 中对 WebSocket `response.completed` JSON frame 转下游 SSE chunk 做 exact golden 断言，锁定 `event:`、`data:` 和 SSE 终止分隔符格式。 - 本轮修正后，HTTP SSE streaming 在上游正常 EOF 但缺少 `response.completed`/`response.failed`/`error` 时，会给客户端追加原版格式的 `response.failed`，错误码为 `stream_disconnected`；`tests/codex_serving/upstream_errors.rs::v1_responses_stream_should_synthesize_response_failed_when_http_sse_closes_before_terminal` 覆盖该行为。 @@ -390,7 +391,7 @@ Rust 证据： 已对齐： - 标准 SSE 事件解析、WS JSON frame 到 SSE chunk 的基本转换、terminal event 判断、`codex.rate_limits` 内部事件过滤、heartbeat 15 秒节奏、usage 提取、function call ids 提取、空响应重试等主路径已经具备。 - WebSocket `response.completed` frame 到 SSE chunk 已有 exact golden 覆盖，确认 raw WS JSON 按原版对下游表现为 `event: response.completed` + 单个 `data:` JSON frame。 -- SSE parser 已对齐原版的三个容错点：漂亮打印 JSON 续行、`[DONE]` 忽略、非 SSE body 合成 `error/non_sse_response`。 +- SSE parser 已对齐原版的标准事件解析和三个容错点：漂亮打印 JSON 续行、`[DONE]` 忽略、非 SSE body 合成 `error/non_sse_response`；标准 SSE、漂亮打印 JSON 续行、非 SSE body 均已有 exact golden 覆盖。 - streaming 路径对客户端保持 SSE 格式，HTTP SSE 与 WebSocket 两种上游传输都能以 SSE 返回给下游。 - `response.failed`/`error` 在 streaming 路径会被透传并写入审计；non-streaming 路径也能识别为上游失败，并已对 429/402/403 进入账号状态机和 fallback。 - request-level `previous_response_not_found` 与 unanswered function call 已按原版优先于账号 fallback 处理：忘记 stale affinity，清除 `previous_response_id`/`turn_state`，同账号重试一次。 @@ -403,10 +404,10 @@ Rust 证据： - Responses pooled path 的非流式最终错误外壳已按原版 `PASSTHROUGH_FORMAT` 对齐：普通上游错误为 `codex_api_error`，429 为 `rate_limit_exceeded`，无账号为 `no_available_accounts`；compact 路径复用同一 Responses 外壳。 未完全对齐： -- response/SSE/WS 转换链路仍缺完整 golden fixture 矩阵。主恢复集合已覆盖 429/402/403、401 token invalid、model unsupported、`previous_response_not_found`/unanswered function call、premature close、WS completed frame 和最终错误外壳；本轮已补 HTTP SSE premature close 与 WS completed exact golden，但还需要更系统的 golden tests 防止格式细节回退。 +- response/SSE/WS 转换链路仍缺完整 golden fixture 矩阵。主恢复集合已覆盖 429/402/403、401 token invalid、model unsupported、`previous_response_not_found`/unanswered function call、premature close、WS completed frame 和最终错误外壳；本轮已补 HTTP SSE premature close、WS completed、标准 SSE、漂亮打印 JSON 续行与非 SSE body exact golden，但还需要更系统的 golden tests 防止格式细节回退。 缺口/后续动作： -- 增加 golden tests：标准 SSE、漂亮打印 JSON 续行、非 SSE body、`response.failed` auth recovery、reasoning replay。 +- 增加 golden tests：`response.failed` auth recovery、reasoning replay。 ## 8. fallback、retry、错误分类 diff --git a/tests/codex_gateway/usage_events.rs b/tests/codex_gateway/usage_events.rs index c3701e6e4..a10435693 100644 --- a/tests/codex_gateway/usage_events.rs +++ b/tests/codex_gateway/usage_events.rs @@ -1,10 +1,17 @@ use serde_json::json; use codex_proxy_rs::codex::gateway::transport::{ - sse::{encode_sse_event, parse_sse_events, MAX_SSE_EVENT_BUFFER_BYTES}, + sse::{encode_sse_event, parse_sse_events, SseEvent, MAX_SSE_EVENT_BUFFER_BYTES}, usage_events::{extract_sse_usage, extract_usage, TokenUsage}, }; +const SSE_STANDARD_PARSE_GOLDEN: &str = + include_str!("../fixtures/responses/golden/sse_standard_events.json"); +const SSE_PRETTY_JSON_PARSE_GOLDEN: &str = + include_str!("../fixtures/responses/golden/sse_pretty_json_events.json"); +const NON_SSE_BODY_PARSE_GOLDEN: &str = + include_str!("../fixtures/responses/golden/non_sse_body_event.json"); + #[test] fn parse_sse_events_combines_multiline_data_and_metadata() { let input = concat!( @@ -27,6 +34,7 @@ fn parse_sse_events_combines_multiline_data_and_metadata() { ); assert_eq!(events[0].data, "{\"delta\":\"hel\"}\n{\"delta\":\"lo\"}"); assert_eq!(events[0].retry, Some(5000)); + assert_eq!(sse_events_json(&events), SSE_STANDARD_PARSE_GOLDEN); } #[test] @@ -48,6 +56,7 @@ fn parse_sse_events_should_keep_non_prefixed_json_continuation_lines() { events[0].data, "{\n \"error\": {\n \"message\": \"bad upstream\"\n }\n}" ); + assert_eq!(sse_events_json(&events), SSE_PRETTY_JSON_PARSE_GOLDEN); } #[test] @@ -66,6 +75,7 @@ fn parse_sse_events_should_convert_non_sse_json_body_to_error_event() { let data: serde_json::Value = serde_json::from_str(&events[0].data).unwrap(); assert_eq!(data["error"]["code"], "non_sse_response"); assert_eq!(data["error"]["message"], "not an SSE stream"); + assert_eq!(sse_events_json(&events), NON_SSE_BODY_PARSE_GOLDEN); } #[test] @@ -215,3 +225,21 @@ fn extract_sse_usage_should_read_completed_image_generation_tool_usage() { assert_eq!(serialized["imageInputTokens"], 31); assert_eq!(serialized["imageOutputTokens"], 9); } + +fn sse_events_json(events: &[SseEvent]) -> String { + let value = serde_json::Value::Array( + events + .iter() + .map(|event| { + json!({ + "event": event.event, + "data": event.data, + "id": event.id, + "retry": event.retry, + }) + }) + .collect(), + ); + + format!("{}\n", serde_json::to_string_pretty(&value).unwrap()) +} diff --git a/tests/fixtures/responses/golden/non_sse_body_event.json b/tests/fixtures/responses/golden/non_sse_body_event.json new file mode 100644 index 000000000..6bdcb9baa --- /dev/null +++ b/tests/fixtures/responses/golden/non_sse_body_event.json @@ -0,0 +1,8 @@ +[ + { + "data": "{\"error\":{\"code\":\"non_sse_response\",\"message\":\"not an SSE stream\",\"type\":\"error\"}}", + "event": "error", + "id": null, + "retry": null + } +] diff --git a/tests/fixtures/responses/golden/sse_pretty_json_events.json b/tests/fixtures/responses/golden/sse_pretty_json_events.json new file mode 100644 index 000000000..30ea436a9 --- /dev/null +++ b/tests/fixtures/responses/golden/sse_pretty_json_events.json @@ -0,0 +1,8 @@ +[ + { + "data": "{\n \"error\": {\n \"message\": \"bad upstream\"\n }\n}", + "event": "error", + "id": null, + "retry": null + } +] diff --git a/tests/fixtures/responses/golden/sse_standard_events.json b/tests/fixtures/responses/golden/sse_standard_events.json new file mode 100644 index 000000000..a8317e348 --- /dev/null +++ b/tests/fixtures/responses/golden/sse_standard_events.json @@ -0,0 +1,8 @@ +[ + { + "data": "{\"delta\":\"hel\"}\n{\"delta\":\"lo\"}", + "event": "response.output_text.delta", + "id": "evt_1", + "retry": 5000 + } +]