读懂 Codex（9.7）：SSE 事件解析和 ResponseAccumulator

具体问题与边界

字节分块、SSE frame、typed event、UI Delta 与完整历史 Item 之间分别由谁负责？

SseDecoder 只处理字节到 frame；Transport 只处理 frame JSON；Responses Adapter 只映射事件；ResponseAccumulator 才负责完整 Item 与采样终点。本节以官方 Codex 锁定提交为事实基线，以 Mini Codex 的可执行断言检验 Python 设计；二者不是逐行翻译关系。

官方 OpenAPI 的 /v1/responses 同时声明 JSON 与 text/event-stream 响应；流式示例展示 response.output_item.added、text delta/done、output_item.done 与 response.completed 的次序。

协议、类型与状态所有权

对象	创建/所有者	生命周期与作用域	是否持久化
UTF-8 decoder buffer	SseDecoder	一条 HTTP stream	否
SSE frame buffer	SseDecoder	直到空行分隔符	否
typed wire event	Transport/Adapter	单事件	否
item order/deltas/items	ResponseAccumulator	一次模型采样	否，finish 后的 Items 才持久化

正常路径

HTTP 字节块可能在多字节 UTF-8 字符或 \n\n frame 边界中间切开，因此先用 incremental decoder。
SSE 解析器识别 event/data 字段、忽略 comment，合并多行 data；[DONE] 不是 JSON event。
response.output_text.delta 立即变成 ModelTextDelta，Session 可低延迟发 TextDelta AgentEvent。
response.output_item.done 中的 message 或 function_call 变成完整 ModelAssistantMessage/ModelToolCall；arguments 字符串必须 JSON 解码成 object。
response.completed 变成 ModelCompleted。Accumulator 未见该标志时 finish() 失败；见到后按 item 首次出现顺序产出 ModelResponse。

调用链与等待点

1. HTTP 字节块可能在多字节 UTF-8 字符或 `\n\n` frame 边界中间切开，因此先用 incremental decoder
→ 2. SSE 解析器识别 event/data 字段、忽略 comment，合并多行 data；`[DONE]` 不是 JSON event
→ 3. `response.output_text.delta` 立即变成 ModelTextDelta，Session 可低延迟发 TextDelta AgentEvent
→ 4. `response.output_item.done` 中的 message 或 function_call 变成完整 ModelAssistantMessage/ModelToolCall；arguments 字符串必须 JSON 解码成 object
→ 5. `response.completed` 变成 ModelCompleted

每个 await 都是状态可被取消、外部副作用可能已经发生或其他任务能够推进的边界。正文因此同时写清输入形状、所有者、成功产物和失败残留，而不是只列方法名。

Python 风格伪代码

decoder = IncrementalUTF8AndSSEDecoder()

async for byte_chunk in http_stream:
    for frame in decoder.feed(byte_chunk):
        wire = json.loads(frame.data)
        match wire["type"]:
            case "response.output_text.delta":
                yield ModelTextDelta(wire["item_id"], wire["delta"])
            case "response.output_item.done":
                yield decode_complete_item(wire["item"])
            case "response.completed":
                yield ModelCompleted()
            case "error" | "response.failed" | "response.incomplete":
                raise ResponsesStreamError(wire)

class ResponseAccumulator:
    def accept(event):
        reject_if_already_completed()
        update_delta_or_complete_item(event)
    def finish():
        if not completed: raise ProtocolError("missing completed")
        return assemble_in_first_seen_order()

伪代码表达顺序和责任。真实可运行实现没有把万能调用当作未解释黑箱；对应模块见 Mini 导航。

失败、取消与恢复

SSE 事件解析和 ResponseAccumulator失败路径图 — 图 9.7-2：失败分支按实际状态所有者收口，模型可见失败与 Runtime fatal 分开。

故障点	已发生/残留状态	处理与模型可见结果
UTF-8 字符跨 chunk	直接 decode 每块会产生替换字符或异常	incremental decoder 保留尾字节
SSE JSON 非 object	无法按 type 分派	ResponsesStreamError
函数 arguments 非 object	Router 的参数合同被破坏	在 Adapter 边界拒绝
completed 后还有事件	采样协议出现双终点	Accumulator 明确报错
断流无 completed	可能已有 UI Delta	拒绝规范完成，不写半截 Item

必须保持的不变量

UI 可见 Delta 不等于规范完成；没有 ModelCompleted 的流绝不能伪装成成功 ModelResponse。

设计取舍

官方职责分散在多个流状态与 active item 路径；Mini 用一个 Accumulator 收拢教学边界。它不覆盖 reasoning、citation、image、hosted tool 等全部事件。

“为什么”的表述是从源码状态机、调用顺序和测试反推的工程解释；源码未声明的动机不写成官方承诺。

测试与复现实验

cd examples/mini-codex
uv run pytest -q -k 'test_sse_decoder_handles_utf8_and_frame_chunk_boundaries or test_responses_adapter_rejects_failed_stream or test_single_sample_turn_emits_one_terminal_event'
uv run mypy src

关键断言：

test_sse_decoder_handles_utf8_and_frame_chunk_boundaries
test_responses_adapter_rejects_failed_stream
test_single_sample_turn_emits_one_terminal_event

官方源码导航

codex-rs/core/src/stream_events_utils.rs：流事件组装与 active item 处理
codex-rs/core/src/client.rs：Response 流和取消边界
codex-rs/core/src/stream_events_utils_tests.rs：增量、完整项与事件顺序测试
codex-rs/codex-api/src：typed Responses 事件定义

Mini Codex 对照

src/mini_codex/models/responses.py：SseDecoder 与 typed event 映射
src/mini_codex/models/accumulator.py：完整 Item 累积和完成检查
src/mini_codex/models/events.py：领域 ModelEvent

本节边界

已经证明：UI 可见 Delta 不等于规范完成；没有 ModelCompleted 的流绝不能伪装成成功 ModelResponse。