雨 雨天小六

读懂 Codex（2.8）：EventMsg、Raw Response Item 与界面事件的层次

· 更新于 2026-08-02 · 专栏：读懂 Codex

#Codex#Agent Runtime#软件架构#Events#App Server

Codex 一次模型响应会同时出现在历史、Core Event Stream 和 UI 通知中。它们看起来都像“消息”，但解决 的问题并不相同：模型要看到协议完整的输入输出，Runtime 要表达实时状态，UI 需要稳定且可恢复的展示 Item，App Server 还必须维持版本化 JSON-RPC 合同。

把这些层压成一个 JSON 结构，最直接的后果是 Provider 字段变化穿透 UI，或者为了 UI 简化而丢掉模型 下一步必须看到的 Tool Call/Output 关系。

第一层：ResponseItem 是模型历史单元

ResponseItem 定义在 Protocol models 中，表示 Responses API/模型上下文里的事实，例如：

  • user/developer/assistant message；
  • reasoning；
  • function/tool call 与 output；
  • MCP、web search、image generation 等 provider item；
  • agent-to-agent message；
  • compacted summary 或其他控制 item。

它的第一受众是模型和历史重放，不是 UI。某些字段对用户不适合展示，却必须保留以便下一轮关联 Call ID 或继续 Response。

第二层：EventMsg 是 Core 的实时运行协议

Event 包含关联 id 和 EventMsg payload。EventMsg 的范围远大于 ResponseItem：

事件族示例
Session/Turn 生命周期SessionConfigured、TurnStarted、TurnComplete、TurnAborted、ShutdownComplete
文本与推理流AgentMessage、AgentMessageContentDelta、ReasoningContentDelta、SectionBreak
工具生命周期ItemStarted/Completed、ExecCommandBegin/Delta/End、McpToolCallBegin/End
人机等待ExecApprovalRequest、RequestPermissions、RequestUserInput、ElicitationRequest
配置与历史ThreadSettingsApplied、ContextCompacted、ThreadRolledBack、TokenCount
基础设施EnvironmentConnected、McpStartupUpdate、StreamError、HookStarted/Completed
RawRawResponseItem、RawResponseCompleted

EventMsg 表达“Runtime 正在发生什么”，其中很多事件从来不会写进模型历史。例如环境断开、流重试警告 和 stdout delta 对模型历史不是同一种事实。

第三层：TurnItem/ThreadItem 是展示投影

Core 会把适合展示与恢复的内容转成 Item 生命周期。一个 command execution 可以经历 ItemStarted、多个 output delta，最后 ItemCompleted；最终 Item 聚合 command、cwd、status、exit code、duration 等稳定字段。

App Server 的 ThreadItem 面向公共 API，和 Core 内部 TurnItem/ResponseItem 并非一一同型。比如旧的 PatchApplyBegin/End 仍可在 Core fan-out 与 rollout 中保留，V2 客户端却只接收 canonical FileChange Item， 避免公开两套等价 UI 模型。

第四层：App Server Notification 是版本化协议

App Server 的 bespoke event handling 按 EventMsg 类型维护 ThreadState/TurnSummary、决定是否发通知， 并把它转换成 item/started、item/completed、turn/completed 等 V2 方法。它还负责：

  • 计算完成时间与 Turn status；
  • 在 TurnAborted 时取消 pending server requests；
  • 聚合最后一条 Agent message；
  • 隐藏 deprecated 或内部事件；
  • 根据协商能力决定是否转发 raw events。
ResponseItem、Core EventMsg、TurnItem ThreadItem 与 App Server JSON-RPC Notification 四层事件模型
图 2.8-1：四层之间是选择性投影与聚合，不是把同一对象换四个名字。

record_conversation_items 同时维护三份一致性

当一个 ResponseItem 成为对话事实，Session 执行固定顺序：

  1. prepare_conversation_items_for_history 补齐缺失的 Turn ID/ResponseItem ID 等历史字段；
  2. 在 Session state 中记录，按模型 truncation policy 维护上下文；
  3. 持久化为 Rollout ResponseItem；
  4. 逐项发送 RawResponseItem Event。
async def record_conversation_items(session, turn, items):
    prepared = stamp_and_normalize(turn, items)
    async with session.state_lock:
        session.history.record(prepared, turn.model.truncation_policy)
    await session.rollout.persist(prepared)
    for item in prepared:
        await session.events.send(
            Event(id=turn.id, msg=RawResponseItem(item))
        )

顺序的重要性在于：客户端看到 raw item 时，它已经进入当前内存历史，并已提交到 rollout 写入路径。 这不必然代表底层文件已经 fsync；需要强 durability 的路径仍调用 flush。

RawResponseItem 不等于“原始 SSE 字节”

这里的 Raw 指未投影成 UI ThreadItem 的 Protocol ResponseItem，不是 Provider 网络响应的原始字节流。 它已经反序列化为受类型约束的结构，并可能补过本地 ID/Turn metadata。

RawResponseCompleted 则在一个 Provider response 完成时报告 response ID 和 token usage。一个 Turn 可能因 工具循环包含多次 response，因此不能用一次 RawResponseCompleted 替代 TurnComplete。

模型 ResponseItem 进入历史和 Rollout，同时产生 Raw Event，并选择性投影成 Item 生命周期与 App Server 通知
图 2.8-2：同一 ResponseItem 一路用于模型续跑，一路用于可选 raw 观察，另一路被聚合成稳定 UI Item。

Raw 公共通知是显式协商能力

App Server 定义 rawResponseItem/completed 与 rawResponse/completed，但 Thread lifecycle 只在客户端 启用 experimental raw events 时转发对应 Core Event。默认客户端不应依赖它们存在。

这层开关保护两件事：

  • Raw schema 更接近模型协议，变化频率可能高于稳定 UI Item；
  • Raw 内容可能包含 UI 不需要处理的内部上下文。

需要调试、记录或构建自定义模型历史视图的客户端可以 opt in；普通 UI 使用稳定 Notification。

Delta、Started 和 Completed 各有用途

以命令执行为例：

  • ItemStarted：创建可定位的 UI 行，状态 running；
  • ExecCommandOutputDelta：低延迟追加 stdout/stderr；
  • ItemCompleted：给出最终聚合状态；
  • RawResponseItem：记录模型发出的 Tool Call 或收到的 Tool Output；
  • TurnComplete：整个 Agent Turn 不再有后续动作。

如果客户端只保存 Delta，重连后无法恢复最终聚合对象；只等 Completed 又会失去实时体验。App Server 的 持久 Thread history 和实时 Notification 分别覆盖恢复与流式显示。

Event ID、Turn ID、Item ID 与 Call ID

这些 ID 常常同时出现：

ID关联对象典型用途
Event/Submission IDCore 命令或 Turn将 TurnStarted/Complete 关联到提交
Thread ID整个会话路由 JSON-RPC 与持久记录
Item IDUI/历史 ItemStarted、Delta、Completed 聚合
Tool Call ID模型 Tool Call/Output把 output 返回给正确的 call
Request ID审批/elicitation 等等待点完成特定人机请求

有些路径会复用 Submission ID 作为公开 Turn ID，但这不构成“所有 ID 都相等”的协议。桥接层必须按字段 语义保存，不能只留一个通用 id。

TurnItemsView 控制历史装载成本

App Server V2 的 Turn 带 items_view：

Viewitems 内容适用场景
NotLoaded故意为空只列 Thread/Turn 元数据
Summary仅展示摘要列表和快速恢复
Full持久 App Server history 可得的全部 ThreadItem完整详情页

Full 不等于 RawResponseItem 全量透传。它只承诺 App Server 持久历史中的全部公共 ThreadItem。该边界使 客户端能控制 IO 和序列化成本，也避免为了列表页装载所有命令输出。

App Server 为什么需要 bespoke handling

简单地 serialize(EventMsg) 会产生多项错误：

  • Core v1 wire format 仍有 task_started/task_complete 兼容名，V2 需要稳定方法；
  • TurnComplete 需要结合此前累计的 error/last message 计算公开状态；
  • TurnAborted 需要同时回应 pending interrupt RPC；
  • PatchApply 旧事件应被 FileChange Item 吸收；
  • Raw event 需要能力开关；
  • 某些内部 lifecycle 只更新状态，不应直接广播。
async def project_core_event(event, client_caps, thread_state):
    match event.msg:
        case RawResponseItem(item) if client_caps.experimental_raw:
            emit("rawResponseItem/completed", convert(item))
        case RawResponseItem(_):
            pass
        case ItemStarted(item):
            thread_state.begin(item)
            emit("item/started", public_item(item))
        case TurnComplete(done):
            summary = thread_state.finish_turn(done.turn_id)
            emit("turn/completed", public_turn(summary))
        case PatchApplyBegin() | PatchApplyEnd():
            pass  # V2 用 FileChange item
        case other:
            handle_versioned_projection(other)

失败与重放边界

情况层级行为
Stream 临时断开并重试EventMsg::StreamError；Turn 可继续
Tool output 已写历史但 UI 丢连接Rollout/Thread history 可用于恢复
客户端未启用 raw不发送 raw JSON-RPC，稳定 Item 仍可用
TurnAbortedCore 发 abort；App Server 取消该 Turn 的 pending requests
Item 只有 Started 无 Completed重放/断线恢复需根据 Turn 终态修正显示
NotLoaded 请求items 为空是合同，不代表 Turn 真无 Item

测试重点

async def test_record_stamps_before_raw_event(session, turn):
    item = response_item_without_turn_id()
    await session.record_conversation_items(turn, [item])
    raw = await next_event("raw_response_item")
    assert raw.item.turn_id == turn.id
    assert await session.history.contains(raw.item)


async def test_raw_notification_requires_capability(app_server):
    await app_server.connect(experimental_raw=False)
    await run_turn_with_tool(app_server)
    assert not app_server.received("rawResponseItem/completed")
    assert app_server.received("item/completed")


def test_items_view_not_loaded_is_not_empty_history(turn):
    projected = load_turn(turn.id, items_view="notLoaded")
    assert projected.items == []
    assert projected.items_view == "notLoaded"

事件系统的稳定性来自分层：ResponseItem 保证模型历史完整，EventMsg 保证 Runtime 可观察，ThreadItem 保证 UI 可恢复，App Server Notification 保证跨版本客户端合同。后续追踪一次普通 Turn 时，需要明确 每一步产出的是哪一层对象。

评论

← 返回文章列表