第 3 周：不调真实模型，怎么测一个 Agent？

/// 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
}

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;

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
    }
}

    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}"
    );

fn sse_incomplete() -> String {
    responses::sse(vec![serde_json::json!({
        "type": "response.output_item.done",
    })])
}
// ...
    let (server, _) = start_streaming_sse_server(vec![
        vec![StreamingSseChunk {
            gate: None,
            body: incomplete_sse,
        }],
        vec![StreamingSseChunk {
            gate: None,
            body: completed_sse,
        }],
    ])
    .await;
// ...
        // 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),
// ...
    let requests = server.requests().await;
    assert_eq!(
        requests.len(),
        2,
        "expected retry after incomplete SSE stream"
    );

            Ok(None) => {
                let error = response_error.unwrap_or(ApiError::Stream(
                    "stream closed before response.completed".into(),
                ));
                let _ = tx_event.send(Err(error)).await;
                return;
            }
            Err(_) => {
                let _ = tx_event
                    .send(Err(ApiError::Stream("idle timeout waiting for SSE".into())))
                    .await;
                return;
            }
        };
// ...
        let event: ResponsesStreamEvent = match serde_json::from_str(&sse.data) {
            Ok(event) => event,
            Err(e) => {
                debug!(
                    // ...
                    "Failed to parse SSE event"
                );
                continue;
            }
        };

const DEFAULT_STREAM_IDLE_TIMEOUT_MS: u64 = 300_000;
const DEFAULT_STREAM_MAX_RETRIES: u64 = 5;
const DEFAULT_REQUEST_MAX_RETRIES: u64 = 4;

/// A 429 with server advice retries in the sampling loop and completes without a terminal error.
#[tokio::test(flavor = "current_thread")]
async fn responses_http_429_uses_retry_after() -> Result<()> {
    // ...
    let response_mock = responses::mount_response_sequence(
        &server,
        vec![
            ResponseTemplate::new(429)
                .insert_header("Retry-After", "1")
                .set_body_json(json!({ "error": { "code": "rate_limit_exceeded" } })),
            responses::sse_response(responses::sse(vec![
                responses::ev_response_created("recovered"),
                responses::ev_completed("recovered"),
            ])),
        ],
    )
    .await;
    // ...
    assert!(wait_for_retry(&mut telemetry, &retry).await >= Duration::from_secs(1));
    // ...
    assert_eq!(response_mock.requests().len(), 2);

    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(),
        )
    );

Scenario: additional context is inserted before the user turn input.

## Window 1
-- request 1 (turn; Request) --
00:message/developer:
    <PERMISSIONS_INSTRUCTIONS>
01:message/developer:
    <automation_info>run one</automation_info>
02:message/user:
    <external_browser_info>tab one</external_browser_info>
03:message/user:
    inspect the active tab

[workspace.lints.clippy]
await_holding_invalid_type = "deny"
await_holding_lock = "deny"
disallowed_methods = "deny"
expect_used = "deny"
# ...
redundant_clone = "deny"
# ...
uninlined_format_args = "deny"
# ...
unwrap_used = "deny"

allow-expect-in-tests = true
allow-unwrap-in-tests = true

  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
      - blob-size-policy
      - cargo-deny
      - codespell
      - repo-checks
      - rust-ci
      - sdk
    runs-on: ubuntu-24.04
    steps:
      # ...
      - name: Require successful dependencies
        env:
          NEEDS: ${{ toJSON(needs) }}
        run: python3 .github/scripts/check_ci_results.py

def ev_tool_use_block(index: int, tool_id: str, name: str, tool_input: dict, pieces: int = 3) -> list[dict]:
    """一个 tool_use 块：参数以 input_json_delta 的 partial_json 分段到达。"""
    raw = json.dumps(tool_input, ensure_ascii=False)
    step = max(1, -(-len(raw) // pieces))
    parts = [raw[i:i + step] for i in range(0, len(raw), step)]
    return (
        [{"type": "content_block_start", "index": index,
          "content_block": {"type": "tool_use", "id": tool_id, "name": name, "input": {}}}]
        + [{"type": "content_block_delta", "index": index, "delta": {"type": "input_json_delta", "partial_json": p}}
           for p in parts]
        + [{"type": "content_block_stop", "index": index}]
    )


def tool_turn(text: str, tool_id: str, name: str, tool_input: dict, msg_id: str = "msg_tool") -> list[dict]:
    """完整的一轮：一段文本 + 一个 tool_use，stop_reason = tool_use。"""
    return [ev_message_start(msg_id), *ev_text_block(0, text), ev_ping(),
            *ev_tool_use_block(1, tool_id, name, tool_input), ev_message_delta("tool_use"), ev_message_stop()]

            def _stream(self, script: SSE) -> None:
                self.send_response(200)
                self.send_header("content-type", "text/event-stream")
                self.send_header("transfer-encoding", "chunked")
                self.end_headers()
                for i, ev in enumerate(script.events):
                    if script.cut_after is not None and i >= script.cut_after:
                        self.wfile.flush()
                        self.connection.shutdown(socket.SHUT_RDWR)  # 不发结束块，直接断开
                        self.close_connection = True
                        return
                    chunk = encode_event(ev)
                    self.wfile.write(f"{len(chunk):X}\r\n".encode() + chunk + b"\r\n")
                    self.wfile.flush()
                    if script.delay_s:
                        time.sleep(script.delay_s)
                self.wfile.write(b"0\r\n\r\n")  # 分块编码的结束块：HTTP 层面"正常说完"
                self.wfile.flush()

class StreamIncomplete(Exception):
    """HTTP 正常结束了，但没收到 message_stop：模型这一轮没说完。"""


# 流层可重试的错误：断开（传输层）、没说完（协议层）、畸形 data（json.JSONDecodeError 是 ValueError）
STREAM_ERRORS = (StreamIncomplete, httpx2.TransportError, anthropic.APIConnectionError, ValueError)


def stream_once(client: anthropic.Anthropic, messages: list) -> anthropic.types.Message:
    saw_stop = False
    with client.messages.stream(model=MODEL, max_tokens=64000, tools=TOOLS, messages=messages) as stream:
        for event in stream:
            if event.type == "message_stop":
                saw_stop = True
        if not saw_stop:
            # SDK 不会因为缺 message_stop 报错，get_final_message() 照样返回半截消息
            raise StreamIncomplete("stream closed before message_stop")
        return stream.get_final_message()


def call_model(client, messages, max_stream_retries=2, backoff_s=0.2,
               sleep: Callable[[float], None] = time.sleep, log: list | None = None):
    for attempt in range(max_stream_retries + 1):
        try:
            return stream_once(client, messages)
        except STREAM_ERRORS as e:
            if log is not None:
                log.append(f"attempt {attempt + 1}: {type(e).__name__}: {e}")
            if attempt == max_stream_retries:
                raise
            sleep(backoff_s * 2 ** attempt)  # 指数退避：0.2s、0.4s……
    raise AssertionError("unreachable")

def test_tool_result_is_sent_back(server, client):
    server.enqueue(
        SSE(tool_turn("先读一下 README。", "toolu_01", "read_file", {"path": "README.md"})),
        SSE(text_turn("这是一个评测找 Bug 的 Agent 的项目。")),
    )

    status, _ = run_agent(client, TASK, impls=IMPLS, **FAST)

    assert status == "done"
    assert len(server.requests) == 2
    second = server.requests[1].body
    # 整个对象一次比较，不逐字段（Codex 评审规范的要求）
    assert second["messages"][1:] == [
        {"role": "assistant", "content": [
            {"type": "text", "text": "先读一下 README。"},
            {"type": "tool_use", "id": "toolu_01", "name": "read_file", "input": {"path": "README.md"}},
        ]},
        {"role": "user", "content": [
            {"type": "tool_result", "tool_use_id": "toolu_01", "content": FILE_TEXT},
        ]},
    ]
    # 整个请求体存成快照：哪天请求形状变了（多了字段、顺序变了），这里会红
    assert_json_snapshot("tool_roundtrip_request_2", second)

def test_429_respects_retry_after(server, client):
    server.enqueue(
        HTTPError(429, "rate_limit_error", "Number of requests has exceeded your rate limit.",
                  headers={"retry-after": "1"}),
        SSE(text_turn("这是一个评测找 Bug 的 Agent 的项目。")),
    )

    status, _ = run_agent(client, TASK, impls=IMPLS, **FAST)

    assert status == "done"
    assert len(server.requests) == 2
    gap = server.requests[1].at - server.requests[0].at
    assert gap >= 1.0, f"只等了 {gap:.3f} 秒"
    assert [r.headers["x-stainless-retry-count"] for r in server.requests] == ["0", "1"]

$ .venv/bin/python -m pytest -v -p no:cacheprovider
collecting ... collected 6 items

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 ===============================

== mutation: 从 STREAM_ERRORS 去掉 httpx2.TransportError
FAILED test_agent_loop.py::test_retries_after_midstream_disconnect - httpx2.R...
1 failed, 5 passed in 4.29s
== mutation: 不检查 message_stop
FAILED test_agent_loop.py::test_missing_message_stop_is_retried - AssertionEr...
1 failed, 5 passed in 4.31s
== mutation: 工具结果截成 10 个字符
FAILED test_agent_loop.py::test_tool_result_is_sent_back - AssertionError: as...
1 failed, 5 passed in 4.41s

cd week03_Codex源码/code/mock_model_server
python3.13 -m venv .venv && .venv/bin/pip install -r requirements.txt   # anthropic==1.11.0, pytest==9.1.1
.venv/bin/python -m pytest -v