青雲的博客
拆开 Codex 第六部:Agent runtime 怎样被承载与验证 第 34 章

普通 Responses 与 Realtime 为什么是两套会话

普通 Responses 把一次 Turn 的推理请求复用在会话级客户端上;Realtime 则维护独立的音频、文本、handoff、transport 和关闭状态。本章沿源码与本地回环测试划清两套状态机相遇的边界。

源码版本
rust-v0.144.6
验证日期
Commit
5d1fbf26c43abc65a203928b2e31561cb039e06d

第 33 章已经把多 Agent 的 mailbox、turn、terminal-turn completion notification 和 live subtree 拆开了。这里再处理一个很容易被名称带偏的问题:代码里同时出现了 Responses WebSocket、Realtime WebSocket、Realtime V2 和 WebRTC,但它们并不是同一个“实时会话”。

普通 Responses 的问题是:Turn 怎样复用未关闭的 inference 连接,以及后续请求何时能引用上一个 response、只发送增量输入;这种复用既可能发生在同一 Turn 的多次请求间,也可能通过 ModelClient 缓存跨 Turn 发生。Realtime 的问题是:一个持续存在的会话怎样接收音频/文本、怎样排队 response.create、怎样把转写或 handoff 发回 Core,以及结束时谁负责收尾。前者是推理请求的 transport 优化,后者是一套独立的会话状态机。

flowchart LR
  accTitle: Responses 与 Realtime 的两条生命周期
  accDescr: 普通 Turn 从 ModelClient 创建 ModelClientSession,借用缓存的 Responses WebSocket 并按请求条件发送增量 payload,失败后切 HTTP;Realtime 通过 Op 进入独立 manager,建立队列和输入任务,再把事件 fanout 回 Core。
  TURN[普通 Turn] --> MC[ModelClient]
  MC --> MCS[每 Turn 一个 ModelClientSession]
  MCS --> RQ[Responses request]
  RQ --> WS[Responses WebSocket]
  RQ --> HTTP[HTTPS fallback]
  OP[RealtimeConversation Op] --> RM[RealtimeConversationManager]
  RM --> CS[ConversationState]
  CS --> Q[音频 / 文本 / handoff queues]
  Q --> RT[Realtime WebSocket 或 WebRTC sideband]
  RT --> FANOUT[事件 fanout]
  FANOUT --> CORE[Core EventMsg / handoff]

普通 Responses:连接可复用,sticky token 不能跨 Turn

ModelClientModelClientSession 的 owner 不同

ModelClient 是 Session 级对象,持有 provider、认证、thread id 和 transport fallback 状态;它不代表一条正在生成的 response。真正发流的 ModelClientSession 每个 Turn 新建:它会从 ModelClient 借出缓存的 WebsocketSession,用其中的 request/response 基线判断增量请求,再在 Turn 结束时归还。只有 x-codex-turn-stateOnceLock 严格属于这个 Turn,不能带到下一 Turn。这使得“连接和增量基线可以缓存”和“Turn token 不能跨 Turn”同时成立。

本节源码依据(3 处)

连接复用和增量 payload 复用还有第二层区别。未关闭的连接本身可以继续发送完整 request;要引用前一个 response 并只发送 delta,responses_request_properties_match 还要对 model、instructions、tools、reasoning、prompt cache key、text 等字段逐项比较,inputclient_metadata 则另行处理。当前 input 只有在这个属性检查通过、且确实扩展了已有 request/response 基线时,才会和 previous_response_id 一起压成增量 payload。代码故意让字段解构保持穷尽,新加一个 request 字段必须明确回答是否允许增量复用。

本节源码依据(2 处)

fallback 是普通 Responses 的恢复分支

对 retryable stream error,responses_retry 会在达到 retry budget 后调用 try_switch_fallback_transport。切换成功会发一条用户可见的 warning,把 retry 计数清零,然后让同一 Turn 继续。另一个更早的入口是连接握手收到 HTTP 426:stream_responses_websocket 直接返回 FallbackToHttp,由当前请求切换 transport,不必先耗尽 stream retry。两条路径最终都会让 force_http_fallback 关闭后续 WebSocket,并清空缓存 session。这个状态存在 ModelClient 上,所以同一个 Session 后续 Turn 也会走 HTTP,但它仍然不产生 Realtime ConversationState

本节源码依据(3 处)

因此普通 Responses 的时间线可以写成:

Session: ModelClient
  Turn 1: new ModelClientSession -> lazy WS -> request A -> incremental request B
  Turn 1 error: retry budget exhausted -> HTTP fallback -> cache cleared
  Turn 2: new ModelClientSession -> HTTP remains selected for this Session

这里的“session”是 Codex Core 的 ModelClient 生命周期,不是 Realtime provider 返回的 realtime_session_id。把这两个 id 或两种 fallback 混为一谈,会让后面的 resume、handoff 和 transcript 解释全部失真。

Realtime:另一个 manager、另一组队列

ConversationState 是自己的状态机

RealtimeConversationManager 只持有 Mutex<Option<ConversationState>>。start 先把旧 state 从锁里取出,等待它停止,再创建新 state;shutdown 也走同一条 stop_conversation_state 路径。ConversationState 里有 audio_tx、text_tx、handoff state、输入 task、可选 fanout task、realtime_active 和 CancellationToken。这里没有复用 ModelClientSession 的 last request,也没有把 Responses 的 previous_response_id 当成 Realtime 的状态。

本节源码依据(3 处)

启动时会一次性建立五条有界通道:音频输入 256、文本输入 64、handoff 输出 64、事件输出 256,以及只容纳一条 session-end transcript tail 的通道。它们是背压和关闭传播的组成部分,不是一个可无限增长的 Conversation history。输入 task 消费前几类 channel,fanout task 再把解析出的 RealtimeEvent 和可选 tail 投影回 Core。

本节源码依据(2 处)

WebSocket 与 WebRTC 是 Realtime 内部的两条连接路径

没有 Session Description Protocol(SDP)时,manager 用 RealtimeWebsocketClient.connect 建立直接 WebSocket,并把 writer、events 和各个输入 channel 交给 realtime input task。有 SDP 时,ModelClient 先创建 Realtime call,拿到 call_id、sideband headers 和 SDP answer,再启动 WebRTC sideband input task。客户端的媒体连接和服务器的控制连接因此分开;不能把 SDP answer 的存在解释成“已经复用了 Responses WebSocket”。

本节源码依据(2 处)

prepare_realtime_start 还会根据 transport 和 configured version 选择 parser。WebRTC 被限制为 V1 conversational mode;WebSocket 可以配置 V1 或 V2。build_realtime_session_config 再把 model、voice、output modality、backend prompt 和 startup context 收成一次 Realtime session config。这里的 version 是 Realtime protocol version,不是普通 Responses API 的 request schema 版本。

本节源码依据(3 处)

输入、输出和 handoff 不共享同一条历史

audio_in 取出当前 state 的 sender,用 try_send 写入音频队列。队列满时记录 warning 并丢掉这一帧;队列关闭或没有 state 才返回 conversation is not running。text_in 会在 V2 中按 role 给用户文本加 [USER] 前缀,V1 则保持原文,再等待 send;append_speech 走单独的 handoff output。三者都要求 Realtime 正在运行,但可靠性和 owner 不同。

本节源码依据(2 处)

Codex 回答回到 Realtime 时,也不是直接写普通 history。handoff_out 根据 active_handoff、session kind、client_managed_handoffs 和 codex_responses_as_items 选择 StandaloneHandoff、HandoffUpdate、HandoffAppend 或 ConversationItem;V2 完成时还会发 CompletedHandoff 或 acknowledgement。这个分支是“把 Core 的结果交给当前 Realtime transport”,不是把两个模型的上下文 owner 合并。

本节源码依据(2 处)

Realtime response events 经过 fanout 后才回到 Session。HandoffRequested 会提取 input transcript 或 active transcript,包成 realtime_delegation XML,再调用 route_realtime_text_input;包括 HandoffRequested 在内的所有事件仍会作为 RealtimeConversationRealtime 发出。会话结束时可选地 flush transcript tail,随后 finish_if_active 并发送 RealtimeConversationClosed。

本节源码依据(2 处)

协议边界:同一个 Thread,不同的 Op 和 listener

Core 的 Op 明确列出 RealtimeConversationStart、Audio、Text、Speech、Close 和 ListVoices。它们和 UserInput 是不同的提交操作;Realtime 事件也有 Started、Realtime、Closed、Sdp 等独立 EventMsg。普通 Responses stream 的增量事件不会自动变成这些 Realtime 事件。

本节源码依据(3 处)

app-server v2 把这条边界公开成 thread-scoped experimental API。Start 参数有 outputModality、transport、version、realtimeSessionId、startup context 和 handoff flags;append audio/text/speech 与 stop 各自有独立 request/response。turn processor 先加载 Thread、挂上 conversation listener,再检查 Feature::RealtimeConversation,最后把请求映射成 Core Op。feature 未开启时,协议层直接返回 invalid request,而不是偷偷降级成普通 Turn。

本节源码依据(3 处)

Realtime handoff 回到 Core 时,Session 的 route_realtime_text_input 会构造一个新的 UserInput Op。这个调用点正是两套系统的相遇边界:Realtime 的 transcript 可以成为普通 Agent Turn 的输入,但普通 Responses 的 request/session cache 不会因此变成 Realtime state。

本节源码依据(1 处)

startup context 只是一次注入,不是第四种 history

Realtime 启动上下文由当前 thread history、最近 thread metadata 和受限的 workspace scan 组成;每个 section 有自己的 token budget,最后再附上明确说明来源和排除项的 Notes。没有任何 section 时直接不注入。它不读取 repo memory instructions、AGENTS 文件、project-doc prompt blends 或 memory summaries,因此不能把这段 context 写成普通 Responses prompt 的永久副本。

本节源码依据(1 处)

实验:两条本地 transport round-trip 都通过

在固定 commit 导出的 disposable archive / detached worktree 中执行:

: "${ARCHIVE_CODEX_RS:?先执行第六部导读的 archive 准备脚本}"
cd "$ARCHIVE_CODEX_RS"
if [[ -n "${CODEX_SANDBOX_NETWORK_DISABLED+x}" ]]; then
  printf '%s\n' '拒绝运行:网络禁用标记会让这两项测试提前返回' >&2
  exit 1
fi
just test --locked -p codex-core -E 'test(responses_websocket_v2_incremental_requests_are_reused_across_turns) | test(conversation_start_audio_text_close_round_trip)' --no-capture

实际结果为 2/2。just test 使用 nextest,下面只保留与该 runner 一致的摘要,不把 Cargo libtest 的 running N tests 文本混进来:

Starting 2 tests across 2 binaries
PASS responses_websocket_v2_incremental_requests_are_reused_across_turns
PASS conversation_start_audio_text_close_round_trip
Summary: 2 tests run: 2 passed, 2967 skipped

这是两项本地测试,不是一项 fake-WebSocket 测试覆盖两套状态机。第一项启动 fake Responses WebSocket,连续建立三个 Turn 的 session,验证只用一个 handshake、三个 response.create、第二/第三请求分别携带 previous_response_id 和增量 input。第二项另起 loopback Realtime WebSocket,检查 session.updated、audio delta、conversation item、audio/text input、授权 header、session id 与 close event。

测试代码还用 skip_if_no_network!;默认沙箱执行会提前跳过,强行在受限沙箱 bind loopback 会得到 PermissionDenied。命令先要求 CODEX_SANDBOX_NETWORK_DISABLED 不存在,以拒绝这个已知的 early-return marker;它本身不会授予网络权限。实验还必须运行在允许 loopback bind 的环境中,才能真正执行两条本地回环流量。

本节源码依据(2 处)

这个结果证明的是固定 Rust 代码与本地 fake/loopback transport 的协议和状态转换。它没有证明真实 OpenAI service、公开认证、账号配额、WebRTC media path、麦克风/扬声器、TUI 投影或随机生产流量。

Realtime 停在 handoff 边界

本章留下两条可复用的阅读规则。第一,看到 WebSocket 先问 owner:它是 ModelClientSession 的 inference transport,还是 RealtimeConversationManager 的长期输入任务。第二,看到“文本回到 Agent”先找 Op:只有 route_realtime_text_input 明确构造 UserInput,才发生跨状态机 handoff。

Realtime 的责任停在会话关闭或明确的 handoff。第 35 章:Hooks 为什么不是一条统一回调链不沿用这里的 queue、transport 或 session state;它重新从 originating call site 出发,再检查 scope、matcher、payload parser 与阻断语义。只有这样,同名的 continue 字段才不会被错当成同一种控制权。