读懂 Codex（5.7）：SSE 连接建立、事件解码与流结束

HTTP 200 和连接正常关闭都不能证明一次模型响应成功。SSE 适配器只有在解码到 response.completed 后才结束；在此之前的断流、空闲超时和 incomplete 都是失败。

具体问题与启用条件

本节追踪 HTTP POST 到统一 ResponseEvent 的传输生命周期。每种内容事件怎样进入 Runtime，将在 5.13—5.19 拆开。

条件来源	决定字段或状态	对本机制的影响
传输选择	Provider 不支持 WS 或 Session 已回退	使用 HTTP SSE
请求配置	stream=true、Accept text/event-stream	服务端按事件返回
超时	Provider stream_idle_timeout_ms	限制相邻 SSE 事件等待

协议、类型与状态所有权

状态或协议	所有者	生命周期	关键不变量
HTTP request/response	Responses endpoint client	一次请求	响应 Header 在启动流前解析
SSE byte stream	codex-api parser task	直到 Completed/错误	每次 next 受 idle timeout 限制
1600 容量事件队列	ResponseStream	一次流	生产者通过背压等待消费者
response_error	SSE parser task	一次流	failed 事件保存为最终错误

机制调用链如下：

POST /responses
→ 解析 HTTP 状态与响应 Header
→ spawn parser task + bounded channel
→ eventsource 解帧
→ JSON 反序列化为 ResponsesStreamEvent
→ process_responses_event
→ ResponseEvent
→ Completed 后关闭

SSE 请求、解析任务和 Runtime 消费的时序 — 图 5.7-1：Header 元事件和主体事件经同一个有界队列交付，Completed 才让解析任务正常退出。

机制怎样工作

建立请求时发送 JSON 或 zstd body，并带 Accept: text/event-stream。握手成功后，解析层先从 Header 提取 rate limits、models ETag、实际模型、reasoning included、request id、安全缓冲策略和 turn state，再把这些元事件放进同一事件队列。

主体循环对每个 stream.next() 单独施加 idle timeout。无法解析的单个 JSON 事件会记录并跳过；能识别的 response.failed 被保存成有类型错误。解析器不会在看到 failed 时立刻把通道关掉，而是继续读取，最终连接若结束则返回该错误。只有映射出的 Completed 会被发送后立即正常 return。这样“EOF”永远不是成功别名。

Python 风格伪代码

这段伪代码保留生产实现中会改变结果的状态、分支和异步边界；认证 SDK、遥测字段和 Rust 所有权样板被折叠为明确的领域对象。

async def decode_sse(http_response: HttpResponse, idle: float):
    queue: asyncio.Queue[ResponseEvent | Exception] = asyncio.Queue(maxsize=1600)
    await emit_header_events(http_response.headers, queue)
    pending_error: Exception | None = None

    while True:
        try:
            frame = await asyncio.wait_for(http_response.sse_next(), timeout=idle)
        except TimeoutError:
            await queue.put(StreamError("idle timeout waiting for SSE"))
            return
        if frame is None:
            await queue.put(pending_error or StreamError("closed before completed"))
            return
        try:
            event = map_wire_event(json.loads(frame.data))
        except MalformedEvent:
            continue
        if isinstance(event, Exception):
            pending_error = event
        elif event is not None:
            await queue.put(event)
            if isinstance(event, Completed):
                return

失败、取消与恢复

SSE 流的完成与失败状态机 — 图 5.7-2：连接 EOF 只有在 Completed 之后才可能属于正常退出路径。

故障或边界	已发生的状态	对上层的结果	能否直接重试	恢复动作
HTTP 握手失败	未建立 SSE parser	Transport/Provider 错误	按错误分类	外层重试或终止
单事件 JSON 损坏	流仍可继续	该事件被跳过	不重发请求	等待后续终态
idle timeout	可能已有暂态 Delta/Done Item	Stream 错误	是	外层从最新 History 重试
EOF 无 Completed	连接结束	Stream 错误	是	不可当作成功
response.incomplete	服务端明确未完成	Stream 错误含 reason	按外层规则	重新评估请求

设计取舍与验证

独立 parser task 让网络读取与 Runtime 消费解耦，1600 容量吸收短时突发，却增加任务和通道清理成本。跳过单个未知/坏事件提高前向兼容性；Completed 强终态又防止静默接受半截响应。

可验证契约	证据方式	预期结果
无 Completed 的 SSE 必须重试	先返回半流再返回完整流	至少出现第二次请求
Header 元数据先于主体事件	构造 rate limit/model Header	队列先收到元事件
incomplete/failed 映射为错误	SSE parser 单元测试	不会产生 Completed

Mini Codex 对照

Mini Codex 可用 aiohttp 加自写 SSE frame reader，必须保留 bounded queue、逐事件 idle timeout 和 Completed 标志。可以忽略产品 Header，但不能把 EOF 当成功。

本节边界

SSE 的成功合同已确定。5.8 比较持久 WebSocket 如何在一条连接上仍保持请求串行与响应相关性。