读懂 Codex（九）：用 Python 搭建 Mini Codex

前八章一直在拆 Codex：Thread 怎样接收操作，Turn 怎样循环采样，模型为什么只生成工具意图， 审批和沙箱怎样约束副作用，Rollout 又怎样把会话带过进程边界。现在把这些零件重新装起来。

本章不是写一个“模型返回 function call 就执行函数”的演示脚本。那种脚本可以跑通一次工具调用， 却回答不了这些问题：工具执行完后，下一次采样能否看到新文件？工具目录在采样期间变化时，模型 看到的规格和真正执行的 Handler 是否仍然一致？进程死在 Call 与 Result 之间，恢复后怎样生成合法 Prompt？Interrupt 到达时，为什么不会同时收到 Completed 和 Aborted？

配套项目位于 examples/mini-codex。它使用 Python 3.12、asyncio、dataclass、Protocol、 httpx、JSONL、pytest 和 mypy strict；不依赖模型 SDK。默认演示和 27 项测试完全离线，只有同时 设置 OPENAI_API_KEY 与 MINI_CODEX_LIVE_MODEL 时，额外的真实 Responses smoke test 才会运行。 目标不是复刻 Codex 的产品规模，而是让关键架构约束变成可执行断言。

先划定实现和安全边界

当前教学版只支持单进程、单事件循环和一个活动主 Thread。一个 Thread 中每次只能运行一个 Turn， 但一个 Turn 可以包含多次模型采样和多次工具执行。它保留以下能力：

• Op → Session → Event 双向协议；
• 生命周期不同的 TurnContext 与 StepContext；
• 流式 ModelEvent 和完整 ResponseItem；
• 分层 Prompt、History 与动态 World State；
• 同一个快照产生的 ToolSpec 和 ToolRouter；
• Shell、持续进程 Unified Exec 和 Add/Update/Delete Patch Grammar；
• Exec Policy、Approval、可替换执行环境和 Tool Hook；
• 专用 JSONL Writer Task、Interrupt、Resume、Compact、Rollback 和复制式 Fork；
• 真实 Responses/SSE Adapter、录制回放模型、最小 MCP 目录与 Agent Mailbox 控制面。

仍未实现 TUI、App Server、Provider 能力协商、WebSocket、完整 MCP 连接/认证/资源协议、真正的 子 Agent Session 循环、远程执行环境、SQLite 投影、分页 Lineage 或跨平台系统沙箱。这里的 MCP 只覆盖动态 Tool Catalog/Call 适配，AgentPool 只覆盖身份、容量和 Mailbox；不能把它们外推为完整 MCP Client 或多 Agent Runtime。

尤其要明确威胁模型。WorkspacePolicy 能阻止 Apply Patch 使用 ../ 逃出工作区，也会限制 Shell 的 cwd，但它无法阻止 python -c、编译器或其他命令自行访问工作区外的文件和网络。 应用层路径检查只是误操作防线，不是针对恶意代码的 OS 级沙箱。

包结构先表达依赖方向

项目没有把所有逻辑塞进一个 Agent 类，而是按协议、运行时、上下文、模型、工具、策略和持久化 拆分：

src/mini_codex/
├── protocol.py
├── runtime/        # Session、Thread、Turn/Step、取消、Agent Mailbox
├── context/        # Prompt、History、World State
├── models/         # Protocol、Responses/SSE、Accumulator、录制回放
├── tools/          # Plan/Router、Shell、Unified Exec、Patch、Hook、MCP
├── execution/      # 本地进程、持续进程与 Sandbox Adapter
├── policy/         # Workspace、Exec Policy 与 Approval
├── persistence/    # Writer Task、JSONL、Resume/Fork
└── cli.py

Runtime 不直接导入 OpenAI、Anthropic 或其他 SDK。它只依赖一个 ModelClient Protocol。工具也不 直接从全局变量取得当前工作区和审批设置，而是接收一次调用所需的 ToolInvocationContext。 这样测试可以替换边界适配器，同时保留生产代码经过的完整主链。

这张图没有把目录机械画成方框。它强调三种时间尺度：Session 跨多个 Turn 存活，Step 快照只服务 一次模型采样，JSONL 则在进程退出后继续存在。混淆这三个时间尺度，代码仍然可能“能跑”，但 环境刷新、动态工具和恢复都会产生隐蔽错误。

Op 和 Event 是 Runtime 的窄腰

输入协议只定义三个操作：

@dataclass(frozen=True, slots=True)
class UserInputOp:
    text: str

@dataclass(frozen=True, slots=True)
class InterruptOp:
    pass

@dataclass(frozen=True, slots=True)
class ShutdownOp:
    pass

输出协议分为瞬时增量和完整边界：TurnStarted、TextDelta、ItemCompleted、 TurnCompleted、TurnAborted 与 RuntimeErrorEvent。每个 AgentEvent 都带 submission ID， 调用方可以把事件归回某次提交。

Session 内部使用两个容量默认为 32 的 asyncio.Queue。输入队列只有 Session 主循环消费， 输出队列由 CLI 或测试适配器消费。有界队列不是性能装饰：如果客户端处理事件太慢，生产者会在 put() 处等待，内存不会无限增长。代价也很直接——单个消费者必须持续排空事件，否则 Runtime 的推进和关闭都会受到反压。

主循环没有直接 await 完整 Turn，而是把 Turn 创建为活动 Task。这样它仍能继续从输入队列取出 InterruptOp 并设置取消信号。第二个用户输入在已有 Turn 运行时会得到错误事件，第一版不暗中 排队多个 Turn，也不把它解释成 steer。

这里还有一个很小但真实的竞态：调用方可能在 submit(UserInputOp) 后立刻等待 idle。如果只由 主循环在稍后清除 idle，等待者可能读到上一个 Turn 留下的“空闲”。实现因此在用户输入入队前就 清除 idle，只有终态发出后才重新设置。

一个 Turn 为什么仍然需要多个 Step

Turn 是对一次用户请求负责的完整工作单元；Step 是其中一次模型采样及其随后触发的工具处理。 两者的数据快照故意不同：

@dataclass(frozen=True, slots=True)
class TurnContext:
    id: str
    base_instructions: str
    cancellation: CancellationToken

@dataclass(frozen=True, slots=True)
class StepContext:
    id: str
    turn_id: str
    sequence: int
    world_state: WorldState

顶层指令和取消 Token 在当前 Turn 内保持一致。World State 则在每次采样前重新捕获。第一版只扫描 工作区的文件名；如果文件集合发生变化，就追加一条 Developer Message，并在构造模型请求前刷新 Rollout。

这已经足以验证关键行为。假设 Step 1 的 Apply Patch 创建 notes.txt：Step 2 不能继续使用 Turn 开始时的旧文件清单，而必须重新扫描并把 notes.txt 放进新 Prompt。官方 Codex 的 World State 还包含权限、协作模式、项目指令等分区，并维护完整快照和增量 Patch；Mini Codex 只保留刷新时机， 不冒充完整实现。

Turn Loop 的核心可以缩成以下结构：

for sequence in range(1, max_steps_per_turn + 1):
    cancellation.raise_if_cancelled()
    step = await capture_step(turn, sequence)
    tool_plan = tool_planner.plan()
    request = compile_prompt(turn, step, history.model_view(), tool_plan.specs)
    response = await collect_model_stream(request)
    await record_and_flush(response.items)

    if not response.tool_calls:
        await complete_turn()
        return

    results = await tool_plan.router.dispatch(response.tool_calls, invocation_context)
    await record_and_flush(results)

max_steps_per_turn 默认是 12。它防止一个始终要求调用工具的异常模型让 Turn 永远循环。超过限制 不是“正常完成”，而是进入失败路径并发出 TurnAborted。

模型流不能直接当历史保存

ModelClient 只承诺返回一个异步事件迭代器：

class ModelClient(Protocol):
    def stream(
        self,
        request: ModelRequest,
        cancellation: CancellationToken,
    ) -> AsyncIterator[ModelEvent]: ...

ModelEvent 包括文本 Delta、完整 Assistant Message、完整 Tool Call 和 ModelCompleted。 TextDelta 一到达 Session 就可以转发给客户端，但它不会逐块写进 History。ResponseAccumulator 按 item 首次出现的顺序组装完整 ResponseItem，确认收到 ModelCompleted 后才允许 finish()。 如果传输在完成标记前断开，半截响应不会被伪装成一个成功模型结果。

这层分离同时服务三个消费者：UI 要低延迟 Delta，Tool Router 要结构完整的 Call，下一次 Prompt 和 Resume 要稳定的完整 Item。官方 Codex 并没有一个统一同名的 ResponseAccumulator 类型； 相关职责分布在流式解析和 active item 等状态中。Python 版本把它们收拢，是为了让协议边界更容易 测试，不是声称官方 Rust 也使用同一对象。

DemoModelClient 仍是默认离线适配器。第二阶段已经增加 ResponsesModelClient：它把 Message、 Function Call/Output 与 ToolSpec 编译成真实 POST /v1/responses payload；HttpxResponsesTransport 负责认证、HTTP status、typed SSE 和取消。Runtime 没有因此导入 httpx 类型。另有 Recording 与 Replay Adapter 用请求指纹固定离线事件序列。

OpenAI 官方文档要求 Responses 流消费者按 typed SSE event 的 type 分支，文本流至少处理 response.output_text.delta、response.completed 与 error，函数流还会出现 arguments delta/done；Mini 的完整映射和 UTF-8/SSE chunk 测试在 9.6、9.7 展开。

ToolPlan 必须同时冻结“告诉模型什么”和“真正执行什么”

工具 Registry 允许 Handler 更新，并给每一版分配 generation。每次 Step 开始时，Planner 只取一次 快照：

def plan(self) -> ToolPlan:
    snapshot = self._registry.snapshot()
    return ToolPlan(
        generation=snapshot.generation,
        specs=tuple(handler.spec for handler in snapshot.handlers.values()),
        router=ToolRouter(snapshot.handlers),
    )

如果先生成 ToolSpec，等模型返回后再去全局 Registry 查 Handler，就会出现时间穿越：模型按旧参数 定义生成 Call，Runtime 却用新 Handler 执行。Mini Codex 的测试会故意在模型采样期间替换同名 Handler，断言当前 Step 仍由旧 Router 处理，下一 Step 才看到新 generation。

Router 对每个已接收 Call 都产生同 call_id 的 Result。未知工具、审批拒绝、参数错误、执行异常和 取消都作为 is_error=True 的 ToolResult 反馈，而不是让 Call 从历史中悬空。工具失败不等于 Runtime 失败：模型看见失败结果后，仍可能修改参数或选择另一条路径。

这条顺序使第二次模型请求必然包含 Tool Result，也使进程崩溃后的日志能区分“模型尚未请求工具” 和“已经请求，但结果还没写入”。如果为了减少 I/O 把 Call、Result 和 TurnCompleted 全部拖到最后 一次写入，Resume 就无法准确判断故障落点。

Shell、Patch、审批与执行环境

Shell 接受 argv 数组，而不是让 Runtime 先拼成一条 Shell 字符串。它先验证参数和 cwd，再由 Exec Policy 产生 allow/prompt/forbid；只有 prompt 才请求 Approval，最后交给可替换 ExecutionEnvironment。执行环境负责超时、协作式取消、kill/reap 和 stdout/stderr 截断。

Apply Patch 现在解析 *** Begin Patch/*** End Patch，支持多文件 Add、Update hunk 和 Delete。 Parser 先生成 Operation，Planner 再解析全部 workspace 路径、存在性与唯一 context；所有操作规划 成功并经 Approval 后，才 stage 临时文件和 commit。Mini 仍不支持官方 Grammar 的 Move、End of File、宽松 shell 截取和 streaming preview；多个 os.replace 也不是跨文件事务。

Unified Exec 把长进程注册到 ProcessManager。exec_command 首次 yield 返回 session_id，后续 write_stdin 可写输入或轮询增量输出；正常退出、timeout 和 Interrupt 都必须移除 Registry 并清理 reader/watcher Task。9.10 单独给出状态机和 stdin 测试。

审批是 Protocol，不是写死在工具中的 input()。CLI 使用 InteractiveApproval，测试可以注入 AlwaysApprove 或 AlwaysDeny。审批拒绝会形成 ToolResult，例如：

shell command rejected by approval policy

模型因此知道动作没有发生。若只把拒绝显示给 UI 而不回灌模型，下一次采样可能误以为命令已经 执行。相反，批准也不代表安全：它是用户决策记录，真正的文件系统、网络和进程隔离仍需要执行器 或操作系统提供。

Interrupt 必须收束到一个终态

取消 Token 内部使用 asyncio.Event。模型适配器、执行环境和 Tool Router 在自己的等待点检查它； Session 收到 Interrupt 后只修改这个共享取消状态，不从外部直接拼接一个“已中断”回答。

终态由 _run_owned_turn 统一拥有：正常执行完成时发送一个 TurnCompleted；捕获 TurnCancelled 时先追加并刷新 turn_aborted，再发送一个 TurnAborted；其他异常发送错误事件 后也以 Aborted 收束。finally 只处理没有任何路径拿到终态所有权的防御情况。

Tool Router 还有一条局部不变量：同一批 Call 中，取消发生前已执行的 Call 保留真实 Result，尚未 执行的 Call 生成 aborted Result。随后取消继续向上抛出，Turn 进入 Aborted。这样模型可见历史的 Call/Result 结构仍合法，客户端也不会同时收到 Completed。

取消只能保证 Runtime 知道“不再等待”。若子进程、网络请求或外部系统已经产生副作用， TurnAborted 不会自动撤销它们。需要可回滚效果时，工具本身必须提供事务或补偿操作。

JSONL 保存语义，不保存 Python 对象

Rollout 记录 session meta、完整 Item、Turn Started、Completed、Aborted、Compacted 和 RolledBack Checkpoint。它不保存 Queue、Task、文件句柄或模型连接。flush() 把 immutable batch 交给专用 Writer Task；File Sink 逐行 flush/fsync 并返回 confirmed count。若第 k 行失败， PrefixWriteError 只确认已写前缀，pending 保留未确认后缀供下一次重试。

一段最小日志形状如下：

{"type":"session_meta","version":1,"thread_id":"thread_demo"}
{"type":"turn_started","turn_id":"turn_...","submission_id":"submission_..."}
{"type":"item","item":{"id":"item_user","type":"message","role":"user","content":"...","turn_id":"turn_..."}}
{"type":"item","item":{"id":"item_call","type":"tool_call","call_id":"call_...","name":"shell","arguments":{},"turn_id":"turn_..."}}
{"type":"item","item":{"id":"item_result","type":"tool_result","call_id":"call_...","output":"...","is_error":false,"turn_id":"turn_..."}}
{"type":"turn_completed","turn_id":"turn_...","final_message":"..."}

Resume 顺序读取这些记录，遇到 Compacted 或 RolledBack 时用 replacement history 替换此前历史；某个 TurnStarted 到文件结尾仍没有匹配终态，就返回 incomplete_turn_id。单行坏 JSON 会被计数并 跳过，这提高剩余日志的可用性，但不能保证语义无损。

如果崩溃发生在 Tool Call 写入后、Result 写入前，HistoryManager.model_view() 会在模型输入副本 中插入稳定 ID 的 aborted Result；孤儿 Result 则从该副本移除。原始 Rollout 不被改写，恢复性 假设不会伪装成真实工具执行。稳定 ID 由源 Item ID 通过 UUIDv5 派生，多次 Resume 不会每次改变 Prompt 形状。

Compact 接受一段人工摘要和需要保留的最近用户 Turn；Rollback 按 User Message 边界删除最近 N 个 Turn；两者都把 replacement history 写入 checkpoint。Fork 把当前有效历史复制到一个新 Thread 的 JSONL。这三个实现足以验证重放起点和新身份， 但没有官方实现中的自动摘要、Window/WorldState 基线、Ordinal、压缩日志、SQLite 投影或 Reference-backed Lineage。

即使调用了 fsync()，当前实现也不是完整故障恢复系统。它会在追加前为无换行坏尾补 delimiter， 但没有目录 fsync、batch checksum、跨进程 writer lock，更不存在 JSONL 与其他存储之间的事务。 这里验证的是 confirmed prefix、重试与重放算法，不宣称任意断电场景绝对无损。

用测试证明架构约束

项目现有 27 项默认离线测试和 1 项 opt-in 真实模型测试。它们不是按函数做机械覆盖，而是针对会 跨模块失效的不变量：

测试场景	需要同时成立的约束
提交后立即等待 idle	等待的是新 Turn 的终态，不会误读上一次留下的 idle 状态
Function Tool 线格式	type/name/description/strict/parameters 与 Responses 结构一致
Function Tool 内部元数据	output_schema 不会误发进普通工具数组，Deferred 标记按需发送
单次采样	Delta、完整 Item、恰好一个 Completed 的事件顺序
工具后续采样	配对 Result 在第二次请求前进入 History
Registry 热更新	当前 Step 的 Spec 与 Router 同源，下一 Step 才更新
Patch 创建文件	第二次采样重新捕获 World State
审批拒绝	不执行副作用，仍生成模型可见失败 Result
路径穿越	../ 解析后不能逃出工作区
不完整 Turn Resume	报告故障边界，只在 Prompt 副本补稳定 aborted Result
Compact 与 Fork	checkpoint 重放及新 Thread 使用有效历史
Interrupt	取消传播并且只有一个 Aborted 终态
SSE chunk 与 typed event	UTF-8/frame 边界、错误事件和 completed gate
Unified Exec	首次 yield、stdin 续写、timeout kill/reap 与 Registry 清理
Patch 全计划	后置非法操作不会让前置 Add 提前落盘
Exec Policy 与 Hook	longest prefix、参数改写、生命周期与错误回灌
Writer fault injection	部分确认后只重试未确认后缀
Rollback	replacement checkpoint 重放得到正确前缀
MCP Catalog	一个 server 的目录按 generation 整体替换
Agent Mailbox	capacity 拒绝、final 通知和 bounded backpressure
Model record/replay	只有 request fingerprint 一致才回放

在 examples/mini-codex 中运行：

uv sync --extra dev
uv run pytest -q
uv run mypy src
uv run python benchmarks/runtime_baseline.py

当前结果是 27 passed、1 skipped；mypy strict 对 src/mini_codex 下 46 个 Python 文件无报错。 20 次离线单 Turn 本机样本中位数约 14.46 ms、P95 约 19.06 ms；这只是 2026-08-03 当前机器的 观测值，不是跨机器 SLA。 还可以启动离线交互：

uv run mini-codex \
  --workspace ./demo-workspace \
  --rollout ./demo-rollout.jsonl

普通文本由离线模型回答；shell python -c "print('hello')" 会走 Shell Tool；下面的命令会通过 Patch Grammar 创建文件：

patch *** Begin Patch\n*** Add File: a.txt\n+hello\n*** End Patch

CLI 提交普通输入后会立即继续接收命令，因此活动 Turn 中仍可输入 /interrupt；/wait 用来等待 Runtime 回到空闲。Shell 和 Patch 默认逐次询问批准，/compact 摘要、/rollback 1 与 /quit 分别测试压缩、回退和关闭边界。

这次复刻真正保留下来的东西

Mini Codex 的代码量远小于官方 Codex，但它没有把 Agent 简化成一个无限 while True。真正保留 下来的不是 Rust 类型名，而是几条可以迁移到其他 Coding Agent 的契约：

1. 客户端通过 Op/Event 与长期 Runtime 交互，不直接操纵模型循环；
2. Turn 内稳定配置和 Step 级动态环境必须使用不同快照；
3. 流式展示事件、完整模型项和耐久记录不能混为一种数据；
4. 工具描述和执行绑定必须来自同一个 Step 快照；
5. 拒绝、失败和取消也要维持 Call/Result 配对；
6. 恢复依赖可重放的语义记录，不依赖反序列化旧 Session；
7. 路径策略、人工审批和 OS 隔离是三层不同的安全机制。

同时，这个实现也暴露了成本：有界队列需要明确消费者，显式 flush 增加 I/O，动态 Step 快照增加 扫描和编译工作，取消必须贯穿每个等待边界，恢复规则又会把协议不变量带进持久化层。它们不是为了 让架构图更漂亮，而是为失败、并发和时间变化付出的复杂度。

下一章回到 Codex 本身，区分三类设计：Coding Agent 普遍需要的运行时机制，Codex 为产品规模 承担的工程复杂度，以及只在特定执行环境中才值得采用的选择。Mini Codex 已经提供了一把尺子： 一个机制若删掉后，某项测试不变量立刻失效，它就不是单纯的代码组织偏好。

本章细节导航

• 9.1 Python 领域协议：Op、Event 与 ResponseItem
• 9.2 有界 Queue、反压和关闭哨兵
• 9.3 Session 主循环与活动 Turn 所有权
• 9.4 TurnContext、StepContext 和 WorldState 刷新
• 9.5 PromptCompiler 与 HistoryManager
• 9.6 ModelClient Protocol 与真实 Responses Adapter
• 9.7 SSE 事件解析和 ResponseAccumulator
• 9.8 ToolRegistry、ToolPlan 与同 Step Router
• 9.9 Shell Tool 的进程、超时、取消和输出截断
• 9.10 Unified Exec 的持续进程和 stdin 续写
• 9.11 Apply Patch Grammar、Parser 与安全写入
• 9.12 Approval、Exec Policy 与可替换 Sandbox Adapter
• 9.13 Tool Hook、生命周期事件和失败结果回灌
• 9.14 JSONL Writer Task、Flush 和故障注入
• 9.15 Resume、Call/Result 修复与不完整 Turn
• 9.16 Compact、Rollback 和 Fork
• 9.17 MCP Adapter 与动态工具目录
• 9.18 子 Agent、Mailbox 和容量控制
• 9.19 端到端真实模型测试与录制回放测试
• 9.20 故障矩阵、类型检查、并发测试和性能基线

源码与实现导航

• 官方 Op/Event 协议
• 官方 CodexThread 双向接口
• 官方 run_turn 主循环
• 官方 TurnContext
• 官方 StepContext
• 官方 ToolRouter
• 官方 Prompt History 规范化
• 官方 Rollout Recorder
• README.md
• src/mini_codex/runtime/session.py
• tests/

阅读导航

上一章：第八章 · 下一节：9.1