跳到主要内容

底座引擎:一次回合怎么跑(继承自 Codex)

30 秒导读: 前面两章讲的是 harness(仿真某个客户端的请求塑形与响应回译)。但 harness 不是引擎,它是引擎流水线上的一节。这一章把 harness 之外的 Codex agent 主干讲清楚:任务层怎么启动一个回合、回合循环(run_turn)怎么反复采样、一次采样怎么组装 Prompt 并交给 ModelClient::stream,以及模型返回的流怎么回流成 ResponseItem 喂给下一轮。读完你能指着流程图说出"harness 塞在哪一步"。

本章不重复 harness 内部细节——那是 01(为什么仿真、有哪些 harness、路由矩阵)和 02(一个 harness 到底改了请求/响应的什么)的活。这里只讲引擎主干,以及 harness 挂进主干的那一个挂点。

1. 先建立心智模型:回合 = 采样循环​

这节先给零基础读者一个直觉,不进代码。

一次回合(turn),就是"用户说一句话 → 模型可能来回调用几次工具 → 最后模型给一段收尾话"这整个过程。它不是一次请求,而是一串请求。

引擎把这一串请求组织成一个采样循环(sampling loop)。每转一圈叫一次采样请求(sampling request),循环的规则很简单:

  • 模型这一圈要是要调工具 → 引擎执行工具,把结果塞回历史,再转下一圈让模型看结果。
  • 模型这一圈只说了一段话、没要工具 → 记进历史,回合结束。

这段"要工具就继续、不要就停"的判断,就写在 run_turn 的注释里(core/src/session/turn.rs:130-143,函数 run_turn)。

一句话类比: 把回合想成一场问答接力——模型每次要么"递给你一张工具申请单"(你办完再递回结果),要么"说完了,散会"。引擎就是那个来回跑腿、并在每圈之前把最新的对话历史整理成一份 Prompt 的人。

2. 顶层全景:一次提交,四层下钻​

这节给一张图 + 一张部件表,让你看清"大盘"。

怎么读这张图: 从上到下是调用栈的四层;每层只做自己那件事,把"发一次请求"往下委托。harness 只在最底下第 ④ 层的选路分支里出现——它不改回合循环,只改"这一次请求用什么线路、长什么样"。

用户提交 / 上一回合有残留输入
        │
┌───────▼─────────────────────────────────────────────┐
│ ① 编排层  ThreadManager / Session                    │
│    建线程、存历史、把回合当一个 tokio 任务 spawn 出去 │
└───────┬─────────────────────────────────────────────┘
        │ spawn_task → SessionTask::run
┌───────▼─────────────────────────────────────────────┐
│ ② 任务层  RegularTask::run                           │
│    while 有输入 { run_turn(...) }                     │
└───────┬─────────────────────────────────────────────┘
        │ run_turn
┌───────▼─────────────────────────────────────────────┐
│ ③ 回合循环  run_turn 的 loop(采样循环)             │
│    每圈: 整理历史→ 组 Prompt→ 发一次采样→ 看要不要续 │
│    夹带: 超预算就 auto-compact、drain 待处理输入      │
└───────┬─────────────────────────────────────────────┘
        │ run_sampling_request → client_session.stream
┌───────▼─────────────────────────────────────────────┐
│ ④ 模型客户端  ModelClientSession::stream             │
│    按 StreamTransportRoute 选传输分支 ────► harness   │
│    在此塑形请求 / 回译响应(见 01·02)               │
└───────┬─────────────────────────────────────────────┘
        │ ResponseStream(一串 ResponseEvent)
        ▼
   回流成 ResponseItem → 记进历史 → 回到 ③ 下一圈

部件一句话职责:

部件干什么在哪个文件
ThreadManager建/存线程,把一个回合 tokio::spawn 成任务core/src/thread_manager.rs:182
Session一条线程的运行时状态:历史、待处理输入队列、活动回合、各种 servicecore/src/session/session.rs:29
RegularTask普通回合的任务实现,循环调 run_turncore/src/tasks/regular.rs:20
run_turn回合循环本体(采样循环 + 压缩 + 收尾钩子)core/src/session/turn.rs:144
run_sampling_request一次采样请求:建工具、组 Prompt、重试core/src/session/turn.rs:1155
ModelClientSession::stream按路由把 Prompt 发到某个传输分支,返回事件流core/src/client.rs:2740
ContextManager线程历史的容器,负责"整理成给模型看的样子"core/src/context_manager/history.rs:38

主线走一遍(高层): 用户输入进 Session → ThreadManager spawn 一个 RegularTask → 它调 run_turn → run_turn 每圈把历史 for_prompt 整理好、build_prompt 组装成 Prompt、run_sampling_request 交给 client.stream → stream 选路(可能走 harness 分支)→ 返回 ResponseStream → 引擎逐个 ResponseEvent 消费,组装成 ResponseItem 记进历史 → 判断要不要再转一圈。

3. 任务层:谁启动了一个回合​

这节讲"回合是怎么被拉起来的",对应上图 ①②。

3.1 回合是一个后台任务​

Codex 不在提交调用里同步跑回合,而是把它 tokio::spawn 成一个独立任务。入口是 ThreadManagerState::start_task(被 spawn_task 调用):它先中止上一个任务,再把 SessionTask::run 丢进 tokio::spawn。

// core/src/tasks/mod.rs:401  —— start_task 里真正 spawn 回合任务(节选)
let handle = tokio::spawn(
    async move {
        let task_result = task_for_run
            .run(Arc::clone(&session_ctx), ctx, task_input, /* … */)
            .await;
        // …flush rollout、on_task_finished 统一收尾…
    }
    .instrument(task_span),
);

重点看 task_for_run.run(...):所有回合(普通/压缩/review)都实现同一个 SessionTask trait(core/src/tasks/mod.rs:214),spawn_task 用同一套生命周期把它们跑起来——所以"启动一个回合"和"启动一次压缩"走的是同一条路。

3.2 普通回合:RegularTask 循环调 run_turn​

RegularTask::run 是最常见的那种任务。它做两件事:先发一个 TurnStarted 事件(并尽量复用启动预热的 client session),然后进一个 loop,只要还有待处理输入就反复 run_turn。

// core/src/tasks/regular.rs:73  —— 任务层的回合循环(节选)
loop {
    let last_agent_message = run_turn(
        Arc::clone(&sess), Arc::clone(&ctx), /* … */,
        next_input, prewarmed_client_session.take(),
        cancellation_token.child_token(),
    ).await?;
    if !sess.input_queue.has_pending_input(&sess.active_turn).await {
        return Ok(last_agent_message);  // 没有残留输入了,任务收工
    }
    next_input = Vec::new();  // 有残留(用户中途又说了话)→ 再跑一遍 run_turn
}

注意这里有两层循环:任务层这个 loop(处理"回合结束后又冒出来的用户输入"),和 run_turn 内部的采样循环(处理"模型来回调工具")。别混淆。

3.3 Session:回合能读写的那份状态​

run_turn 的第一个参数永远是 Arc<Session>。Session 是一条线程的运行时状态盒子(core/src/session/session.rs:29),回合要用到的东西几乎都挂在它上面:

字段装什么用途
state: Mutex<SessionState>内含 ContextManager 历史采样前 clone_history()、采样后 record_conversation_items
active_turn当前活动回合判断/中止、挂待处理输入
input_queue待处理输入队列模型跑的时候用户又输入的内容进这里
servicesmodel client、MCP、分析、技能等services.model_client.new_session() 拿每回合的客户端
thread_id线程标识事件、遥测里带上

TurnContext(每回合一份,携带模型、审批策略、环境等)和 StepContext(每次采样一份,快照 MCP 工具/环境/AGENTS.md,core/src/session/step_context.rs:13)是回合内更细的两层上下文——采样循环每圈用 capture_step_context(core/src/session/mod.rs:2894)取一份 StepContext,保证"这一圈的历史、工具、执行"用的是同一份视图。

4. 回合循环:run_turn 主干​

这节把 run_turn(core/src/session/turn.rs:144)拆成"进循环前"和"循环体"两段,对应上图 ③。

4.1 进循环前:预压缩 + 首个 StepContext + 注入​

run_turn 开头做几件一次性准备,顺序如下(每步都是为了让第一次采样"干净"):

  1. 拿本回合的客户端会话 —— prewarmed_client_session.unwrap_or_else(|| sess.services.model_client.new_session())(turn.rs:152)。每回合一个 ModelClientSession,原因见 §6.1。
  2. 预采样压缩 —— run_pre_sampling_compact(turn.rs:836):如果换了模型或已经超预算,先压缩再开工。
  3. 抓首个 StepContext + 记录上下文更新 —— capture_step_context 后 record_context_updates_and_set_reference_context_item(turn.rs:170-175)。
  4. 组装技能/插件注入项 —— build_skills_and_plugins(turn.rs:547):把用户 @ 提到的技能、插件、连接器解析成要注入历史的 ResponseItem。
  5. 跑 session-start 钩子、记录输入,把注入项逐条 record_conversation_items 进历史。

4.2 循环体:一圈采样的骨架​

准备完就进 loop(turn.rs:227)。怎么读下面这张流程图: 从上往下是一圈的顺序,右侧标注是这一圈可能岔出去的分支(压缩 / 收尾 / 报错);命中分支后要么 continue(再转一圈)要么 break(结束回合)。

一圈开始
  │
  ├─ 若允许 → drain 待处理输入,过 hook 记进历史
  │
  ├─ 取本圈 StepContext(首圈复用,之后 capture 新的)
  │
  ├─ clone_history().for_prompt(modalities)  → 得到 sampling_request_input
  │        (把历史整理成"给模型看的样子",见 §7)
  │
  ├─ run_sampling_request(...)  ── 发一次请求,消费事件流 ──►(见 §5·§6)
  │        产出 SamplingRequestResult { needs_follow_up, last_agent_message }
  │
  ├─ 算 token 状态 / 有没有待处理输入
  │        needs_follow_up = 模型要续 || 有待处理输入
  │
  ├─ 若 needs_follow_up 且 (请求了新窗口 || 超token上限)
  │        → run_auto_compact(...) 然后 continue ──────────► 压缩分支
  │
  ├─ 若 !needs_follow_up
  │        → 跑 stop hook / after-agent hook,记 last_agent_message
  │        → break ────────────────────────────────────────► 收尾分支
  │
  └─ 否则 continue(再转一圈)

几个承重点,配真实行号:

  • 组装本圈输入:sess.clone_history().await.for_prompt(&turn_context.model_info.input_modalities)(turn.rs:273-279)。每圈都重新从历史生成,因为上一圈刚把工具结果写进了历史。
  • 续不续的判定:let needs_follow_up = model_needs_follow_up || has_pending_input;(turn.rs:320)。模型说要续、或用户中途又输入了,都算要续。
  • 超预算就压缩:if needs_follow_up && (take_new_context_window_request() || token_limit_reached) { run_auto_compact(...); continue; }(turn.rs:348-372)。压缩是"重置请求",压完 continue 再采样(见 §8)。
  • 收尾:if !needs_follow_up { …stop hooks… break; }(turn.rs:374-452)。模型不再要工具、也没残留输入,回合就结束,返回 last_agent_message。

5. 一次采样请求:build_prompt 与 built_tools​

这节放大上图 ④ 的入口——run_sampling_request(core/src/session/turn.rs:1155),它是"把一圈的输入变成一个 Prompt 并交出去"的地方。

它做三步,然后进一个重试循环:

  1. 建工具路由 —— built_tools(turn.rs:1260):把内建工具、MCP 工具、连接器、动态工具、工具建议全部收拢进一个 ToolRouter。这一步决定"这次请求向模型广告哪些工具"。
  2. 拿基础指令 —— sess.get_base_instructions()。
  3. 组 Prompt —— build_prompt(turn.rs:1122):把输入、工具规格、cwd、输出 schema 打包成 Prompt。
// core/src/session/turn.rs:1122  —— 把一圈的材料打包成 Prompt(节选)
pub(crate) fn build_prompt(
    input: Vec<ResponseItem>, router: &ToolRouter,
    turn_context: &TurnContext, base_instructions: BaseInstructions,
) -> Prompt {
    Prompt {
        input,
        tools: router.model_visible_specs(),               // ← 广告给模型的工具
        parallel_tool_calls: turn_context.model_info.supports_parallel_tool_calls,
        cwd: /* primary environment 的绝对路径 */,
        base_instructions,
        output_schema: turn_context.final_output_json_schema.clone(),
        output_schema_strict: /* guardian 场景下放宽 */,
    }
}

重点看:Prompt 是传输无关的——它不知道自己会被发成 Responses API 还是 Chat Completions 还是某个 harness 的格式。选路和塑形是下一步(§6)的事。这层解耦正是 harness 能"插进来"的前提。

run_sampling_request 外层是重试循环(turn.rs:1186-1249):try_run_sampling_request 失败且可重试时,走 handle_retryable_response_stream_error 退避重试;ContextWindowExceeded / UsageLimitReached 这类不可重试的直接上抛给 run_turn。

6. 选路与消费:client.stream 在哪儿夹进 harness​

这节是本章的核心挂点——Prompt 交给 ModelClientSession::stream 之后发生了什么,以及 harness 到底在哪一行被选中。

6.1 两个客户端:会话级 vs 回合级​

Codex 把模型客户端拆成两层,别搞混:

类型生命周期装什么定义
ModelClient整个 sessionauth、provider 选择、conversation id、传输 fallback 状态core/src/client.rs:293
ModelClientSession单个回合懒开的 Responses WebSocket 连接、x-codex-turn-state 粘性路由 tokencore/src/client.rs:313

每回合用 new_session()(client.rs:525)新建一个 ModelClientSession。不能跨回合复用——否则会把上一回合的粘性路由 token 带进下一回合,违反客户端/服务端契约(client.rs:310-312 的文档注释说得很直白)。回合内的重试、增量续写、压缩则共用同一个 session。

6.2 stream 按 StreamTransportRoute 选分支​

try_run_sampling_request(turn.rs:2024)拿到 Prompt 后,第一件事就是调 client_session.stream(...):

// core/src/session/turn.rs:2054  —— 交给模型客户端,拿回事件流(节选)
let mut stream = client_session
    .stream(prompt, &turn_context.model_info, &turn_context.session_telemetry,
            turn_context.reasoning_effort.clone(), turn_context.reasoning_summary,
            turn_context.config.service_tier.clone(), responses_metadata, &inference_trace)
    .await??;

stream 内部先算路由,再 match 分派(client.rs:2740)。路由由 resolve_stream_transport_route(wire_api, harness)(core/src/harness/routing.rs,stream_transport_route 在 client.rs:1036 调用)决定——输入是 provider 的 wire API + 配置的 harness,输出是一个 StreamTransportRoute 枚举(harness/routing.rs:38)。

这就是 01/02 的路由结果落地的地方。分派表如下:

StreamTransportRoute 分支分派到的方法谁走这条
ResponsesApistream_responses_api(client.rs:1453)OpenAI Responses(默认第一方)
ChatCompletionsCompatstream_chat_completions_compat(client.rs:1580)普通 Chat Completions 兼容后端
ChatHarness(route)stream_chat_harness_api(client.rs:1707)kimi-cli、opencode、mini-swe 等 Chat 系 harness
MessagesHarness(route)stream_messages_harness_api(client.rs:1686)Claude Code / zcode(Anthropic Messages 系)
ClaudeCodeResponses(profile)stream_claude_code_responses_api(client.rs:1952)Claude Code 走 Responses 形态
ClaudeCodeChat(profile)stream_claude_code_chat_api(client.rs:1853)Claude Code 走 Chat 形态

harness 就夹在这一层。 MessagesHarness 分支还会二次分派(client.rs:1695):ClaudeCode → stream_claude_code_api、ZCode → stream_zcode_api。每个 stream_*_harness_api 方法内部才做 02 讲的那套"请求塑形 + 响应回译"——把传输无关的 Prompt 揉成该 harness 期望的线格式,再把返回逐条回译成统一的 ResponseEvent。对 run_turn 而言,不管走哪条分支,拿回来的都是同一种 ResponseStream。

6.3 消费事件流:回流成 ResponseItem​

stream 返回 ResponseStream 后,try_run_sampling_request 进一个 loop(turn.rs:2096),stream.next() 逐个取 ResponseEvent 处理。关键几种:

ResponseEvent引擎怎么处理行号
OutputItemAdded(item)新起一个流式项(assistant 消息/推理/工具调用),开始向客户端发 deltaturn.rs:2236
OutputTextDelta / Reasoning*Delta追加流式增量,发 UI 事件turn.rs:2399+
OutputItemDone(item)一个 ResponseItem 收完 → handle_output_item_done:是工具调用就排进 in-flight 执行,是消息就定稿turn.rs:2139 / 2213
Completed { token_usage, end_turn }本次响应结束,记 token,end_turn == Some(false) 则标记要续,break 出流turn.rs:2371

OutputItemDone 是"回流成 ResponseItem"的关键:模型输出的每个项都是一个 ResponseItem,handle_output_item_done 决定它是要触发工具(产出一个 in-flight future,稍后 drain_in_flight 把工具结果作为新的 ResponseItem 写回历史,turn.rs:1975),还是就是一段收尾消息。流结束后,这些 ResponseItem 已经在历史里,下一圈的 for_prompt 就会带上它们——闭环完成。

7. 数据模型:三个在流水线上流动的类型​

这节把前面反复出现的三个类型点清楚,它们是各层之间的"通用货币"。

7.1 Prompt —— 一次请求的传输无关载荷​

Prompt(core/src/client_common.rs:19)是 build_prompt 的产物、stream 的输入:

// core/src/client_common.rs:19  —— 一次采样请求的载荷(节选)
pub struct Prompt {
    pub input: Vec<ResponseItem>,       // 对话上下文
    pub(crate) tools: Vec<ToolSpec>,    // 广告给模型的工具
    pub(crate) parallel_tool_calls: bool,
    pub(crate) cwd: Option<PathBuf>,
    pub base_instructions: BaseInstructions,
    pub output_schema: Option<Value>,
    pub output_schema_strict: bool,
}

它只描述"要发什么",不描述"怎么发"。同一个 Prompt,stream_responses_api 会序列化成 Responses 请求体,stream_claude_code_api 会揉成 Anthropic Messages 请求体——差异全在 harness/传输方法里,Prompt 本身不变。

7.2 ResponseItem —— 历史的原子​

ResponseItem(来自 codex_protocol::models)是历史的最小单位,也是 harness build_request 的输入原子。一条 assistant 消息、一次推理、一个工具调用、一个工具输出、一个压缩摘要,都是一个 ResponseItem 变体。Prompt.input 是 Vec<ResponseItem>,模型返回的每个 OutputItemDone 也是一个 ResponseItem——所以"输入历史"和"输出" 用的是同一套类型,这让"回流进历史"几乎零转换。

7.3 ResponseStream / ResponseEvent —— 回来的那条流​

stream 返回 ResponseStream(client_common.rs:112),内部是一个 mpsc::Receiver<Result<ResponseEvent>>:各传输方法在后台把 provider 的原始流回译成统一的 ResponseEvent(codex-api/src/common.rs:74)推进 channel,run_turn 侧只认这套统一枚举。ResponseStream 被 drop 时会 cancel 一个 token 通知回译任务停手(client_common.rs:127)——消费者提前不听了,生产端也就不用白跑。

build_prompt ──► Prompt ──► client.stream ──►(harness 塑形/回译)
                                                    │
                                            ResponseStream
                                                    │ 逐个
                                              ResponseEvent
                                                    │ OutputItemDone
                                              ResponseItem ──► 记进历史 ──► 下一圈 for_prompt

8. 上下文与压缩:历史怎么变成"给模型看的样子"​

这节讲历史容器 ContextManager 和它两个关键动作:for_prompt(整理)和 compaction(压缩)。

8.1 ContextManager 与 for_prompt​

ContextManager(core/src/context_manager/history.rs:38)就是那个 items: Vec<ResponseItem>(最老在前)加上 token 统计、参考上下文快照的容器。回合每圈调 clone_history() 拿一份,再 for_prompt 生成给模型的输入:

// core/src/context_manager/history.rs:141  —— 整理成给模型看的样子
pub(crate) fn for_prompt(mut self, input_modalities: &[InputModality]) -> Vec<ResponseItem> {
    self.normalize_history(input_modalities);  // 规整 + 丢掉不适合的项
    self.items
}

normalize_history 会调 core/src/context_manager/normalize.rs 里的几个规整器,保证发给模型的历史是自洽的:

规整器修什么行号
ensure_call_outputs_present有工具调用却缺输出的,补占位,别让配对断裂normalize.rs:17
remove_orphan_outputs有输出却找不到对应调用的,删掉normalize.rs:144
strip_images_when_unsupported模型不支持图片时,把图片从消息/工具输出里剥掉normalize.rs:317

这一步很重要:模型只能看到规整后的历史,而不是磁盘上原样的 rollout。

8.2 auto-compact:历史太长时的"截断重写"​

当历史逼近 token 预算,回合循环会触发自动压缩——用一段摘要替换掉旧历史,腾出窗口。触发点在 run_turn 的 token_limit_reached 判定(turn.rs:348),真正执行的是 run_auto_compact(turn.rs:992)。它按 feature/provider 三选一:

走哪条条件实现
token-budget 版开了 Feature::TokenBudgetcompact_token_budget::run_inline_auto_compact_task
远程压缩provider supports_remote_compaction()(compact.rs:95)compact_remote / compact_remote_v2
本地压缩其余compact::run_inline_auto_compact_task(compact.rs:99)

本地压缩用 SUMMARIZATION_PROMPT(可被 config.compact_prompt 覆盖)让模型自己总结历史,拿摘要重写 ContextManager。

一个要点:压缩要不要重新注入"初始上下文",由 InitialContextInjection(compact.rs:73)决定:

  • 回合中压缩用 BeforeLastUserMessage(world_state):模型被训练成把压缩摘要看作历史最后一项,所以初始上下文要注在"最后一条真实用户消息"之上(turn.rs:356)。
  • 预回合/手动压缩用 DoNotInject:直接替换成摘要并清空参考上下文,下一个正常回合会完整重注初始上下文。

压缩完 run_turn 就 continue(turn.rs:371)——回到采样循环,用压缩后的历史重新开一圈。

9. 边界:这一章刻意不讲的​

诚实划一下界,免得你去 turn.rs 里找不到:

  • harness 内部塑形/回译(每个 stream_*_harness_api 具体改了请求/响应的什么)在 02;为什么仿真、路由矩阵全表在 01。本章只讲到"stream 在这一 match 里选中 harness 分支"为止。
  • 工具怎么执行、沙箱怎么拦、apply_patch/MCP/skills 的细节在 04。本章只讲到"OutputItemDone 产出 in-flight future、drain_in_flight 把结果写回历史"。
  • plan 模式的流式细节(PlanModeStreamState 那一大坨,turn.rs:1406+)和 review/多 agent 任务只点到为止——它们是 run_turn 的分支,不是主干。

10. 代码地图(导航索引)​

用符号名 grep 比行号抗漂移;下表按"从提交到回流"的顺序排。

主题文件路径符号名
把回合 spawn 成任务core/src/tasks/mod.rsThreadManagerState::start_task / spawn_task
任务 traitcore/src/tasks/mod.rsSessionTask
普通回合任务core/src/tasks/regular.rsRegularTask::run
线程编排core/src/thread_manager.rsThreadManager
回合运行时状态core/src/session/session.rsSession
每次采样的快照core/src/session/step_context.rsStepContext
回合循环本体core/src/session/turn.rsrun_turn
一次采样请求core/src/session/turn.rsrun_sampling_request
建工具路由core/src/session/turn.rsbuilt_tools
组 Promptcore/src/session/turn.rsbuild_prompt
消费事件流core/src/session/turn.rstry_run_sampling_request
工具结果写回历史core/src/session/turn.rsdrain_in_flight
请求载荷类型core/src/client_common.rsPrompt
返回流类型core/src/client_common.rsResponseStream
统一事件枚举codex-api/src/common.rsResponseEvent
会话级客户端core/src/client.rsModelClient
回合级客户端core/src/client.rsModelClientSession
选路 + 分派core/src/client.rsModelClientSession::stream
路由解析core/src/harness/routing.rsresolve_stream_transport_route / StreamTransportRoute
历史容器core/src/context_manager/history.rsContextManager / for_prompt
历史规整core/src/context_manager/normalize.rsensure_call_outputs_present / remove_orphan_outputs
自动压缩分派core/src/session/turn.rsrun_auto_compact
本地压缩core/src/compact.rsrun_inline_auto_compact_task / InitialContextInjection