不调真实模型，怎么测一个 Agent？

1为什么断言请求，而不是断言回答

用真实模型测 agent，有三个绕不开的问题：输出不确定（同一个输入两次回答不一样），花钱，以及故障没法按需复现，你没办法让线上 API"在第 5 个事件断开"。

换个角度：agent loop 是你的代码，模型是它的外部依赖。测你的代码，就把依赖换成假的，固定住它的行为，再看你的代码对每一种行为反应对不对。agent 的"反应"体现在它发出的下一次请求里：工具结果有没有按 id 回填、重试时有没有把半截消息塞进历史、限流后隔了多久才重发。这些都是确定的，可以逐字断言。

2Codex 的假模型服务器：三件工具

测试支持代码在 codex-rs/core/tests/common/responses.rs，底层是 Rust 的 HTTP mock 库 wiremock 0.6（codex-rs/Cargo.toml:555）。读懂三件就够了：

// core/tests/common/responses.rs:752-766
/// Build an SSE stream body from a list of JSON events.
pub fn sse(events: Vec<Value>) -> String {
    use std::fmt::Write as _;
    let mut out = String::new();
    for ev in events {
        let kind = ev.get("type").and_then(|v| v.as_str()).unwrap();
        writeln!(&mut out, "event: {kind}").unwrap();
        if !ev.as_object().map(|o| o.len() == 1).unwrap_or(false) {
            write!(&mut out, "data: {ev}\n\n").unwrap();
        } else {
            out.push('\n');
        }
    }
    out
}

① sse(events)：把一串 JSON 事件拼成 SSE 文本。每个事件取 type 字段写成 event: 行，再把整个 JSON 写成 data: 行，空行分隔。注意 if：事件只有 type 一个字段时不写 data 行。故障注入测试正是利用了这一点造出"半个事件"。

// core/tests/common/responses.rs:1469-1506（节选）
pub async fn mount_sse_sequence(server: &MockServer, bodies: Vec<String>) -> ResponseMock {
    // ...
        fn respond(&self, _: &wiremock::Request) -> ResponseTemplate {
            let call_num = self.num_calls.fetch_add(1, Ordering::SeqCst);
            let missing_response_message = format!("no response for {call_num}");
            let body = self
                .responses
                .get(call_num)
                .expect(&missing_response_message);
            ResponseTemplate::new(200)
                .insert_header("content-type", "text/event-stream")
                .set_body_string(body.clone())
        }
    // ...
    mock.respond_with(responder)
        .up_to_n_times(num_calls as u64)
        .expect(num_calls as u64)
        .mount(server)
        .await;

② mount_sse_sequence：按调用顺序发剧本。第 n 次 POST 拿第 n 段 SSE（fetch_add 是原子计数器，并发安全）。.up_to_n_times(N) 让 mock 最多匹配 N 次：第 N+1 个请求不再匹配，wiremock 回默认的 404，走不到 respond，所以 .expect("no response for n") 实际碰不到（函数注释说会 panic，和实际行为不符）。.expect(N) 保证请求少了一定红；多了红不红，取决于 agent 怎么处理那个 404。我们的 Python 版把多出来的请求记进 unexpected，fixture 收尾时断言它为空，才做到多一个也一定红。

// core/tests/common/responses.rs:738-750
impl Match for ResponseMock {
    fn matches(&self, request: &wiremock::Request) -> bool {
        self.requests
            .lock()
            .unwrap()
            .push(ResponsesRequest(request.clone()));

        // Enforce invariant checks on every request body captured by the mock.
        // Panic on orphan tool outputs or calls to catch regressions early.
        validate_request_body_invariants(request);
        true
    }
}

③ ResponseMock：录下每一个请求，顺便查不变量。它挂在 wiremock 的匹配器上，每来一个请求就存一份，测试结束后用 function_call_output_text(call_id) 之类的方法取出"agent 回给模型的工具输出"。validate_request_body_invariants（:1558 起）检查每个请求里工具调用和工具输出是否成对，孤儿输出直接 panic。也就是说，任何一个用了这套 mock 的测试，都顺带测了配对规则。

另有 start_mock_server()（:1203-1212）启动时先挂一个空的 /models 响应，保证测试完全离线。

3一个完整的测试：工具输出有没有原样回到模型

// core/tests/suite/tools.rs:704-726, 745-748（节选）
    let responses = vec![
        sse(vec![
            ev_response_created("resp-1"),
            ev_function_call(call_id, "exec_command", &serde_json::to_string(&args)?),
            ev_completed("resp-1"),
        ]),
        sse(vec![
            ev_assistant_message("msg-1", "done"),
            ev_completed("resp-2"),
        ]),
    ];
    let mock = mount_sse_sequence(&server, responses).await;

    fixture
        .submit_turn_with_permission_profile(
            "run a command that should be denied by the read-only sandbox",
            PermissionProfile::read_only(),
        )
        .await?;

    let output_text = mock
        .function_call_output_text(call_id)
        .context("shell output present")?;
    // ...
    assert!(
        body.contains(sentinel),
        "expected sentinel output from command to reach the model: {body}"
    );

1. 剧本两段：第一段让"模型"调 exec_command，命令往只读沙箱外写文件并往 stderr 打一个哨兵字符串；第二段让模型说 "done"。
2. 用只读权限提交一轮。agent 真的去执行命令，沙箱真的拒绝。
3. 从请求记录里取 agent 回给模型的工具输出，断言：带有权限拒绝的字样、哨兵字符串到了模型那里、提到了被拒的路径、不是兜底文案、退出码非 0（:736-763）。

和第 2 周的 loop 对照：这就是"tool_result 有没有按 id 回填、内容对不对"的测试，只是 Codex 断言的是 OpenAI Responses API 的 function_call_output，我们换成 Claude 的 tool_result。

4故障注入：三个测试文件各测什么

// core/tests/suite/stream_no_completed.rs:20-24
fn sse_incomplete() -> String {
    responses::sse(vec![serde_json::json!({
        "type": "response.output_item.done",
    })])
}
// :65-68
        // exercise retry path: first attempt yields incomplete stream, so allow 1 retry
        request_max_retries: Some(0),
        stream_max_retries: Some(1),
        stream_idle_timeout_ms: Some(2000),
// :95-100
    let requests = server.requests().await;
    assert_eq!(
        requests.len(),
        2,
        "expected retry after incomplete SSE stream"
    );

被测的生产代码在 codex-api/src/sse/responses.rs：流读到头还没见到 completed，就报 "stream closed before response.completed"（:558-563）；某个事件的 data 解析不了，记一条 debug 日志，跳过这个事件继续读（:575-586）；两次事件之间超过空闲超时（默认 300 秒）也报错。重试分两层：request_max_retries 默认 4，管 HTTP 层的 5xx、超时、网络错误，不含 429（retry_429: false，lib.rs:447-453）；stream_max_retries 默认 5，管流中途的错误和带 Retry-After 的 429（model-provider-info/src/lib.rs:62-64）。5xx 两层都会重试，次数相乘。测试里把一层设 0、另一层设 1，就能精确控制走哪条路径。

5快照、lint、CI 门禁、评审规范

insta 快照

Codex 用 insta 1.46（全仓 1,488 个 .snap，其中 TUI 1,373 个，core 84 个）。core 里的快照多半拍的是发给模型的请求长什么样：

// core/tests/suite/additional_context.rs:89-95
    insta::assert_snapshot!(
        "additional_context_simple_input",
        context_snapshot::format_labeled_requests_snapshot(
            "additional context is inserted before the user turn input.",
            &[("Request", &request)],
            &ContextSnapshotOptions::default().rewrite_known_segments(),
        )
    );

先把请求格式化成人能读的文本（rewrite_known_segments 把权限说明之类的大段固定文本换成 <PERMISSIONS_INSTRUCTIONS> 占位），再和 snapshots/ 下的文件逐字比较。上下文拼接顺序一变，快照就红，评审时看 diff 就知道请求变了什么。

clippy：禁止 unwrap / expect

codex-rs/Cargo.toml:561-597 的 [workspace.lints.clippy] 把 36 条 lint 设成 deny，包括 unwrap_used、expect_used、await_holding_lock、redundant_clone、uninlined_format_args。codex-rs/clippy.toml 前两行又放开了测试：allow-expect-in-tests = true、allow-unwrap-in-tests = true。

为什么：Rust 的 unwrap() 相当于 Python 里"假设一定成功，失败就让进程崩"。在长时间运行的 agent 里，一个没想到的 None 会让整个会话崩掉，而正确做法是把错误往上返回，由 loop 决定是重试、回给模型还是终止。测试里崩了就是测试失败，正是想要的，所以放开。

CI 门禁

# .github/workflows/blocking-ci.yml:48-72（节选）
  required:
    name: CI required
    # Without `always()`, GitHub skips this job after a failed dependency and a
    # required check can appear successful instead of reporting the failure.
    if: ${{ always() }}
    needs:
      - bazel
      # ...
      - sdk
    # ...
      - name: Require successful dependencies
        env:
          NEEDS: ${{ toJSON(needs) }}
        run: python3 .github/scripts/check_ci_results.py

仓库的设计是让 PR 合并只依赖一个汇总检查 "CI required"（yml 开头注释说 main 分支的规则集应当要求它；GitHub 上实际配了哪些必需检查，从仓库文件里看不到，未核实）。它依赖 7 个子工作流（Bazel 测试与 clippy、cargo-deny、codespell、repo-checks、rust-ci 等），用 if: always() 保证依赖失败时它照样运行，再由脚本检查：只有 success 算过，skipped、cancelled 也算失败。全平台 cargo clippy -D warnings 和分片 nextest 放在合并后的 postmerge-ci.yml 里跑。nextest 配置（codex-rs/.config/nextest.toml）：超过 30 秒算慢、2 个周期后终止，失败自动重试 1 次。

PR 评审规范

.github/codex/labels/codex-rust-review.md 是给 Codex 自己审 PR 用的 prompt。和测试相关的几条：PR 描述要说清动机；一个 PR 只做一件事，重构和功能分开；测试里对整个对象做一次 assert_eq!，不要逐字段比较，因为单元测试也是"可执行的文档"；不用 unsafe，不用 std::env::set_var（曾多次导致竞态）。这份文件也有过时的地方：它建议把共享逻辑放进 codex-rs/common，而这个 crate 在当前 commit 里已经不存在。

6迁移到我们自己的 agent：Python 版

本地代码在 week03_Codex源码/code/mock_model_server/：假服务器只用标准库（http.server），被测的是第 2 周的 loop 改成流式之后的版本，用官方 SDK anthropic 1.11.0 指向假服务器。6 个测试在本机真实跑通：

test_agent_loop.py::test_tool_result_is_sent_back PASSED                 [ 16%]
test_agent_loop.py::test_retries_after_midstream_disconnect PASSED       [ 33%]
test_agent_loop.py::test_missing_message_stop_is_retried PASSED          [ 50%]
test_agent_loop.py::test_429_respects_retry_after PASSED                 [ 66%]
test_agent_loop.py::test_429_without_http_retries_fails_fast PASSED      [ 83%]
test_agent_loop.py::test_malformed_json_gives_up_after_bounded_retries PASSED [100%]

============================== 6 passed in 4.37s ===============================

跑测试的过程中发现了 SDK（1.11.0）的四个行为，正是这些测试要防的坑：

再做三次"变异"验证测试真能抓 bug：把断开异常的捕获去掉、把 message_stop 检查去掉、把工具结果截成 10 个字符，每次都恰好有 1 个测试变红（详见博客正文）。

7面试要点与代码

一句话讲清楚

测 agent 不调真实模型：起一个假的模型服务器，按剧本返回 SSE 事件流，把断开、没说完、429、畸形数据都写进剧本；断言的是 agent 发给模型的下一个请求（工具结果按 id 回填、重试时请求体不变、限流后等够了 Retry-After），以及请求次数。Codex 的集成测试就是这么写的，再加上 deny unwrap 的 lint、请求快照和一个 always() 的汇总门禁。

追问准备

1. mock 和真实 API 行为不一致怎么办？事件格式照官方文档的流式示例写，并用官方 SDK 当客户端去解析：格式错了 SDK 会直接报错。另外保留少量真实 API 的冒烟测试，定期跑，不进 PR 门禁。
2. 流断了，重发时要不要带上已经收到的半截内容？官方流式文档的 Error recovery 一节建议保存半截内容续写：Claude 4.5 及更早放进 assistant 消息开头，4.6 及之后改成在 user 消息里带上半截回复、让模型接着写。我们选择整轮重发、历史里不留半截消息，理由有二：同一节说 tool_use / thinking 块不能部分恢复，而断点常落在 tool_use 里；整轮重发时请求体和第一次逐字相同，测试一条等式就能断言。代价是已生成的部分要重新生成。
3. 两层重试为什么要分开？流层（读到一半出问题）SDK 不管，只能 agent 自己重试；HTTP 状态错误（429 / 5xx）我们交给 SDK，它认 retry-after。两层次数分开配，测试才能只打开其中一条路径。这个分工是我们自己的设计，不是"和 Codex 一样"：Codex 的请求层不重试 429（交给采样循环），5xx 两层都重试、次数相乘。我们这边 429 / 5xx 不相乘，但拿到响应前的超时会被 SDK 和流层各重试一遍（实测 3 × 3 = 9 个连接）。
4. 怎么证明这些测试有用？做变异：故意改坏被测代码（去掉某个检查），看对应的测试会不会红。三处变异各让 1 个测试变红，说明每个测试都守着一条具体的路径。

常见错误说法

最小可运行代码

完整代码在本地 week03_Codex源码/code/mock_model_server/（mock_model_server.py、agent_loop.py、test_agent_loop.py、conftest.py），运行：python3.13 -m venv .venv && .venv/bin/pip install -r requirements.txt && .venv/bin/python -m pytest -v。博客正文贴了关键部分。

本周下一步：第 5 周"长任务可靠性"会直接复用这个假服务器做故障注入。