Codex 的 agent loop 比我们的多了什么？

1三层循环：和第 2 周对一下

第 2 周的 loop 只有一层：for _ in range(max_turns)，每轮调一次模型、执行一批工具。Codex 的一个 turn（用户发一条消息，到 agent 交还控制权）里面套了三层循环：

run_turn                                        core/src/session/turn.rs:163
└─ loop {                 // ① 外层：一个 turn 里的多次采样（step）     :426
     取出用户中途插话 pending_input                                  :430
     run_sampling_request
     └─ loop {            // ② 采样重试                                 :1653
          try_run_sampling_request
          └─ loop { stream.next() }   // ③ 流处理                      :2618
               output_item.done 是工具调用 → 立刻开跑，future 放进 in_flight
               response.completed → 记 token、看 end_turn，退出流循环
          drain_in_flight：按调用顺序把工具结果写进历史                 :3162
          出错 → handle_response_stream_error：分类、退避、降级
        }
     needs_follow_up = 模型要继续 || 有待处理的插话                     :566
     !needs_follow_up → Stop hook → break                             :653
   }

文中路径省略了 codex-rs/ 前缀。完整固定链接见博客正文。

2停止条件：needs_follow_up 由什么决定

Codex 调的是 OpenAI Responses API，没有 Claude 那样的 stop_reason: "tool_use"。它看输出里有没有工具调用 item。三个来源，任何一个成立就再采样一次：

// core/src/stream_events_utils.rs:350-357（流里收到一个完整的工具调用）
let tool_future: InFlightFuture<'static> = Box::pin(
    ctx.tool_runtime
        .clone()
        .handle_tool_call(call, cancellation_token),
);

output.needs_follow_up = true;
output.tool_future = Some(tool_future);

// core/src/session/turn.rs:2994-2996（response.completed 里的 end_turn）
if let Some(false) = end_turn {
    needs_follow_up = true;
}

// core/src/session/turn.rs:566（采样结束后，外层循环再看一眼插话队列）
let needs_follow_up = model_needs_follow_up || has_pending_input;

1. 出现过工具调用：模型要看到工具结果才能继续，所以必须再采样。工具名不认识也走这条路：ToolRouter::build_tool_call 不检查工具名（tools/router.rs:248-301），调用照常派发，注册表里找不到才返回 unsupported call: <工具名>（tools/registry.rs:551-570），作为失败的工具输出按序回填，让模型自己改。build_tool_call 本身返回 RespondToModel 时（目前只有 tool_search 参数解析失败一处，router.rs:274），不派发，直接写一条工具输出回历史并置 needs_follow_up = true（stream_events_utils.rs:400-424）。
2. end_turn == Some(false)：服务端明说"还没完"。response.incomplete 且原因是 interrupted 时，SSE 解析层也会把它转成 end_turn: Some(false)（codex-api/src/sse/responses.rs:450-452）。
3. 有待处理的用户插话：模型这一步说完了，但用户在它说的过程中又发了消息，那就再采样一次，让模型看到新消息（第 6 节）。

三者都不成立，进入 Stop hook。hook 返回 should_block 并附带一段续写提示时，提示会作为新消息写进历史，continue 回到外层循环（turn.rs:672-689）；should_stop 时直接 break。否则检查是否需要 turn 结束时的压缩，然后 break，turn 结束。

3流处理：工具一完整就开跑，结果按顺序回填

模型在一次响应里调 4 个工具，第 1 个调用的参数在第 2 秒就生成完了，整个响应第 5 秒才结束。第 2 周的写法要等到第 5 秒才开始执行第 1 个工具。Codex 在收到 response.output_item.done、发现它是工具调用的那一刻就开跑：

// core/src/session/turn.rs:2589
let mut in_flight: FuturesOrdered<InFlightFuture<'static>> = FuturesOrdered::new();
// ... 流循环里，每收到一个 OutputItemDone：
// core/src/session/turn.rs:2787-2793
if let Some(tool_future) = output_result.tool_future {
    in_flight.push_back(tool_future);
}
// ...
needs_follow_up |= output_result.needs_follow_up;
// ... 流循环结束后（不管成功还是出错）：
// core/src/session/turn.rs:3162-3171
if !in_flight.is_empty() {
    // ...
    drain_in_flight(&mut in_flight, sess.clone(), &step_context).await?;
}

• FuturesOrdered 是一个"按放入顺序吐结果"的 future 队列：里面的 future 可以乱序完成，但取结果时一定先拿第 1 个、再拿第 2 个。所以工具结果写进历史的顺序 = 模型发起调用的顺序，和谁先跑完无关。
• drain_in_flight（turn.rs:2472-2500）就是 while let Some(res) = in_flight.next().await，逐个把结果记进历史。它在流循环之后执行，流中途出错也会执行，所以已经开跑的工具结果不会丢。

断流时怎么办：流断了，已经开跑的工具照样跑完、结果进历史；采样重试时不复用原来的输入，而是从历史重新组装（turn.rs:1656-1662），新请求里已经带着这些工具调用和输出，所以不会把它们再执行一遍。

测开视角：怎么证明"真的提前开跑了"

Codex 有一个专门的集成测试 shell_tools_start_before_response_completed_when_stream_delayed（core/tests/suite/tool_parallelism.rs:304-435）。做法很值得照搬：

1. mock 服务器先只发 4 个 exec_command 调用，response.completed 被一个 oneshot 闸门卡住不发；
2. 每个命令往临时文件里写一个毫秒时间戳；
3. 等文件里攒够 4 个时间戳，才打开闸门发 completed；断言 4 个时间戳都 ≤ completed 的发送时刻。

另一个测试 tool_results_grouped（同文件 :226-302）断言第二次请求里：所有 function_call 都排在所有 function_call_output 之前，并且输出的 call_id 顺序和调用一一对应。断言的是 agent 发给模型的请求体，不是模型说了什么。

4并行工具：一把读写锁

组装 Prompt 时 parallel_tool_calls 固定写成 true（turn.rs:1598），真正发出去的值是 prompt.parallel_tool_calls && !model_info.use_responses_lite（client.rs:1001）：除了走 Responses Lite 的模型，请求里都是 true，模型可以一次发多个调用。但不是所有工具都能同时跑。每次 run_sampling_request 会创建一个 ToolCallRuntime（turn.rs:1638），里面有一把 tokio::sync::RwLock<()>（parallel.rs:50, :63）。锁本身不保护任何数据，只用来排队：

// core/src/tools/parallel.rs:205-209（在 tokio::spawn 出去的任务里）
let guard = if supports_parallel {
    Either::Left(lock.read().await)
} else {
    Either::Right(lock.write().await)
};

• 工具声明 supports_parallel_tool_calls() == true → 拿读锁，多个读锁可以同时持有，这些工具一起跑。
• 否则拿写锁：要等前面的所有工具结束才开始，而且它跑的时候别人都得等。
• 这个方法的默认实现返回 false（tools/src/tool_executor.rs:122-124），不声明就独占，默认是保守的。

两点要注意：

1. "读锁"不等于"只读"。exec_command 跑 shell，完全可以写文件，但它拿的是读锁。读/写只是锁的模式，意思是"能不能和其他调用同时跑"。shell 的安全靠的是审批和沙箱（本周后面的章节），不是这把锁。
2. tokio 的 RwLock 是公平的：等锁的任务按先来后到排队，队头是写者时，后面的读者也拿不到读锁（tokio 文档：read locks will not be given out until the write lock has been released）。所以"读、读、写、读"这个顺序里，第 4 个读者要等第 3 个写者跑完。下面的演示按这个规则计算。

5重试：分类、退避、降级

第 2 周的 loop 里，请求失败就是一个异常。Codex 把失败当常态，分两层重试：

常量出处：DEFAULT_STREAM_MAX_RETRIES = 5、DEFAULT_REQUEST_MAX_RETRIES = 4、DEFAULT_STREAM_IDLE_TIMEOUT_MS = 300_000（model-provider-info/src/lib.rs:63-65）；INITIAL_DELAY_MS = 200、BACKOFF_FACTOR = 2.0（async-utils/src/backoff.rs:7-8）；INITIAL_CONNECTION_RETRY_DELAY = 5s、MAX_CONNECTION_RETRY_DELAY = 60s（responses_retry.rs:23-24）。连接无限重试由 feature flag unbounded_connection_retries 控制，Stable、默认开（features/src/lib.rs:1330-1334）。

WebSocket 握手失败不会变成 ConnectionFailed：map_ws_error 把 IO 错误映射成 TransportError::Network（codex-api/src/endpoint/responses_websocket.rs:566-593），再变成普通的流错误，按采样层计次、用完后降级到 HTTP；握手超过 15 秒（DEFAULT_WEBSOCKET_CONNECT_TIMEOUT_MS，model-provider-info/src/lib.rs:68）算超时，同样计次。降级到 HTTP 之后还连不上，才进入上表第三行的无限重试。

采样层 5 次重试的名义等待是 0.2 + 0.4 + 0.8 + 1.6 + 3.2 = 6.2 秒；加上抖动在 [5.58, 6.82) 秒之间。退避没有封顶，把 stream_max_retries 调大时，第 10 次要等 200ms × 2⁹ ≈ 102 秒。

错误分类：CodexErr::retry_delay（protocol/src/error.rs:389-437）

降级：WebSocket → HTTP

内置的 OpenAI provider 默认走 WebSocket（supports_websockets: true，model-provider-info/src/lib.rs:555）。采样层 5 次重试用完，下一次失败时如果还能降级，就切到 HTTP SSE，重试计数清零（responses_retry.rs:120-139）。降级是粘性的，整个会话只发生一次（client.rs:654-673 用 swap(true) 置位）。切换时如果服务端给过 Retry-After，仍然要等到那个时刻，注释写着 "Changing transport must not bypass the server's retry deadline."

对应的测试

• core/tests/suite/websocket_fallback.rs:85-128：stream_max_retries = 2 时，WebSocket 尝试 1 + 2 次后降级，HTTP 请求 1 次成功。
• core/tests/suite/retry_after.rs:344-400：429 带 Retry-After: 1，断言等了 ≥ 1 秒、总共 2 次请求、turn 正常完成。
• retry_after.rs:502-506：503 server_is_overloaded，request_max_retries = 2。没有 Retry-After 时 3 次请求就结束；带 Retry-After: 0 且 stream_max_retries = 2 时是 3 × 3 = 9 次。两层重试是相乘的。

6用户中途插话（steer）

模型正在跑，用户又发了一句"别改测试文件"。Codex 不开新 turn，而是把这句话塞进当前 turn（core/src/session/turn_input.rs 顶部注释：决定输入是 "starts a turn, steers an active turn, or is rejected"）：

1. steer_input 把消息放进当前 turn 的 pending input 队列（turn_input.rs:734-740）。Review 和 Compact 任务不接受插话（:683-695）。
2. 外层循环每一圈开头取出 pending input，作为用户消息写进历史（turn.rs:430-447）。
3. 采样结束后 needs_follow_up = model_needs_follow_up || has_pending_input：哪怕模型已经说完了，只要有插话，就再采样一次。

所以默认行为是：不取消当前流，模型要到下一次请求才看到插话；但 wait_agent、sleep 这类等待型工具会被新输入提前结束（测试 steer_interrupts_wait_agent_and_is_sent_in_follow_up_request、any_new_input_interrupts_sleep，core/tests/suite/pending_input.rs:593-769），普通工具照常跑完。想更快打断，有一个还在开发中的 instant_interrupt（Under Development、默认关，features/src/lib.rs:1120-1124）：开启后每次采样会监听插话队列，一有用户输入就取消当前流（session/input_queue.rs:253-273）。用户按 Esc 中断则是另一条路：CancellationToken 一路传下去，turn 以 TurnAborted 结束。

7没有 max_turns：取舍和无人值守

在 core/src/session 和 core/src/tasks 里 grep max_turns|max_steps|max_iterations，没有结果。turn 内的步数没有上限。token 预算类的开关有两个，token_budget 和 rollout_budget，都是 Under Development、默认关（features/src/lib.rs:1760-1764, :1772-1776）。本机 codex features list（codex-cli 0.159.2）的输出也是这样：

instant_interrupt                        under development  false
rollout_budget                           under development  false
token_budget                             under development  false
unbounded_connection_retries             stable             true

它靠什么防失控？

• 上下文压缩：token 快满时 turn 中途压缩然后继续。源码注释直接写了这个判断："as long as compaction works well in getting us way below the token limit, we shouldn't worry about being in an infinite loop."（turn.rs:612）
• 人在旁边：交互式编码 agent，用户随时可以 Esc 或者插话纠偏。
• 终止类错误：上下文超窗、额度用完等直接结束 turn；重试有次数上限（连接失败除外）。

8面试要点与动手实验

一句话讲清楚

Codex 的 loop 和我们手写的同构：一个 turn 里反复采样，本次输出有工具调用、end_turn 为 false 或者用户插了话，就再采样一次。生产级多出来的是：流式解析时工具调用一完整就 tokio::spawn 开跑，用一把公平读写锁让声明可并行的工具同时跑、其余独占，流结束后按 FuturesOrdered 的顺序回填；失败按类型分成终止、只听服务端、可重试三类，指数退避加抖动、遵守 Retry-After、重试用完从 WebSocket 降级到 HTTP。它没有 max_turns，因为人在旁边、压缩能让任务继续；无人值守的批量评测里我会保留 max_turns、墙钟超时和 token 预算。

追问准备

1. 提前执行能省多少？省的是"模型还在生成后面调用的参数"这段时间和工具执行的重叠。理论下限是 max(completed 时刻, 每个工具"完整时刻 + 耗时"的最大值)。模型输出越慢、调用越多、前面的工具越慢，省得越多。只有一个调用，或者模型生成很快、工具很慢（重叠窗口很短）时，提前执行几乎不省；工具独占只会让工具之间不能并行，仍能和模型生成重叠（演示 1 默认参数下 4 个都换成 apply_patch：12.0 → 9.0 秒，省 3.0 秒，比全部可并行时的 8.0 → 6.0 还多）。不是"总能省一半"。
2. 为什么结果要按调用顺序回填？让历史可复现：同样的模型输出，不管工具谁先跑完，写进历史的顺序都一样，下一次请求的前缀稳定，也便于测试断言和缓存命中。代价是队头阻塞，但 drain 本来就要等所有工具结束，这里的总耗时不受影响。
3. 断流时正在跑的工具怎么办？照样跑完、结果进历史；重试时从历史重新组装请求，不会重复执行。这一点要测：断流注入后断言工具只执行了一次。
4. 退避为什么要抖动？同一时刻断线的大量客户端如果都按 200、400、800ms 整齐重试，会在同一时刻再次打爆服务端。乘一个 [0.9, 1.1) 的随机因子把重试时间打散。
5. 怎么测这些逻辑？照 Codex 的做法：mock 一个 Responses 服务器，编排 SSE 事件序列，用闸门控制 completed 的发送时机，用 Retry-After 头和状态码注入故障，断言请求次数、等待时长和第二次请求体。

动手实验：把第 2 周的 loop 改成"流式 + 提前执行 + 按序回填"

完整代码在 week03_Codex源码/code/stream_loop_sim.py，纯 Python 标准库、不调 API，python3 stream_loop_sim.py 就能跑。核心是这两段：

async def codex_style(calls):
    """Codex 写法：output_item.done 就 create_task 开跑；流结束后按顺序 await，结果按调用顺序回填。"""
    clock, lock, history = Clock(), FairRWLock(), []
    log = collections.defaultdict(dict)
    in_flight = []                     # 相当于 FuturesOrdered：先进先出
    needs_follow_up = False
    async for ev in fake_stream(calls):
        if ev["type"] == "response.output_item.done":
            item = ev["item"]
            log[item["call_id"]]["done"] = clock.now()
            history.append(item)       # 调用本身先进历史
            in_flight.append(asyncio.create_task(run_tool(item, lock, clock, log)))
            needs_follow_up = True     # 出现过工具调用，就还要再采样一次
        elif ev["type"] == "response.completed":
            if ev["end_turn"] is False:
                needs_follow_up = True
    completed = clock.now()
    for task in in_flight:             # drain_in_flight：按 push 顺序取结果
        out = await task
        history.append(out)
        log[out["call_id"]]["recorded"] = clock.now()
    return completed, clock.now(), log, history, needs_follow_up


def retry_delay(kind, retry_after, attempt, rng):
    """对应 CodexErr::retry_delay：None 表示终止，不再重试。"""
    if kind in TERMINAL:
        return None
    if kind in SERVER_DELAY_ONLY:
        return retry_after
    return retry_after if retry_after is not None else backoff_secs(attempt, rng)

本机运行输出（节选，单位是模拟秒）：

A. 第 2 周写法：等响应结束，串行执行
  调用  工具           收到   开始   结束   回填
  c1    exec_command    2.0    5.0    8.0    8.0
  c2    exec_command    3.0    8.0    9.0    9.0
  c3    exec_command    4.0    9.0   11.0   11.0
  c4    exec_command    5.0   11.0   12.0   12.0
  response.completed 在 5.0 秒，这一步总耗时 12.0 秒

B. Codex 写法：工具一完整就开跑，全部可并行
  调用  工具           收到   开始   结束   回填
  c1    exec_command    2.0    2.0    5.0    5.0
  c2    exec_command    3.0    3.0    4.0    5.0
  c3    exec_command    4.0    4.0    6.0    6.0
  c4    exec_command    5.0    5.0    6.0    6.0
  response.completed 在 5.0 秒，这一步总耗时 6.0 秒
  回填顺序 ['c1', 'c2', 'c3', 'c4']，needs_follow_up = True

C. Codex 写法：c3 换成 apply_patch（写锁，独占）
  调用  工具           收到   开始   结束   回填
  c1    exec_command    2.0    2.0    5.0    5.0
  c2    exec_command    3.0    3.0    4.0    5.0
  c3    apply_patch     4.0    5.0    7.0    7.0
  c4    exec_command    5.0    7.0    8.0    8.0
  response.completed 在 5.0 秒，这一步总耗时 8.0 秒

注意 B 里的 c2：第 4 秒就跑完了，但要等 c1 在第 5 秒回填之后才轮到它。C 里的 c4 是读者，却要等排在它前面的写者 c3 跑完，这就是公平读写锁。这三组数字和演示 1 的读数一致：A 是默认参数下「第 2 周写法」那一格，B、C 分别是「默认」和「c3 换成 apply_patch」两个预设下的 Codex 写法。

下一章：工具系统与 MCP client。RespondToModel 和 Fatal 的边界、apply_patch 为什么用语法约束而不是 JSON 参数、MCP 工具名冲突怎么处理。