feat(soak): 压测入口——预算护栏、并发上限、干跑，以及记分板要的四个产物

--budget-calls 与 --concurrency 都没有默认值：一个能跑飞的压测入口迟早会跑飞。达到调用
上限时停止派发新任务，但已经在跑的让它跑完——半路砍断会制造一批没有结束记录的日志，而
那和崩溃长得一模一样，会污染故障注入那边的判定。

--dry-run 不打模型，只把任务集列出来、把每个 RunRequest 真的装配一遍，确认容器起得来、
语料读得进、脱敏闸过得了。它也刻意不往 runs-dir 写东西，写了的话紧接着的全量会撞上
RunIdentityError。

两个场景同时跑时任务轮流排开而不是拼接：共用一份预算，拼接的话排在前面的场景会把预算
吃光，而报告看起来只是「因预算停在第 N 个任务」——一次只压了一半的跑长得像一次正常的跑。

测试里最有价值的一条是真的把产物喂给记分板：手工搭两个 GovDoc 任务共六次运行（真的走
polyloop.session.run，模型是写死的替身），再 import 记分板判定，验十一条不变量全绿。
两边的 sidecar 约定对不上的话这条会当场红。

实跑过一次干跑：两道 AppWorld 题各装配出 12.4k 字符上下文，GovDoc 两个任务六次运行全部
装配成功，脱敏替换 64 处，跑完零残留容器。

Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>

tools/soak/soak.sh

@@ -0,0 +1,144 @@
+#!/usr/bin/env bash
+#
+# 压测的四步：干跑 → 正常负载 → 故障注入 → 记分板。
+#
+# 这个脚本只做「在当前 shell 里按顺序跑」，自己不建 tmux 会话、也不 attach。**长跑要在 tmux
+# 里跑**（CLAUDE.md §4），所以正确的用法是人先开会话再调它：
+#
+#     tmux new -s polyloop-soak
+#     SOAK_BUDGET_CALLS=4000 SOAK_FAULT_BUDGET_CALLS=600 SOAK_CONCURRENCY=4 \
+#     SOAK_LIMIT=100 SOAK_APPWORLD_DATA_ROOT=/path/to/appworld \
+#         bash tools/soak/soak.sh
+#
+# 会话名约定是 polyloop-soak。跑完不要急着 kill，留着给人复查。
+#
+# **末尾一律不接管道。** `pytest ... | tail` 的退出码来自管道最后一节，于是一次失败的跑会
+# 报成 exit 0。要留日志就设 SOAK_LOG_DIR，那一路用重定向，不用 tee。
+#
+# 全部配置都是环境变量，没有位置参数：
+#
+#   必填
+#     SOAK_BUDGET_CALLS         正常负载的模型调用次数上限
+#     SOAK_FAULT_BUDGET_CALLS   故障注入那一步的模型调用次数上限
+#     SOAK_CONCURRENCY          同时在跑的任务数上限
+#     SOAK_LIMIT                每个场景最多跑几个任务（GovDoc 一个任务是三次运行）
+#     SOAK_APPWORLD_DATA_ROOT   AppWorld 数据根目录（SOAK_SCENARIO 含 appworld 时）
+#
+#   可选
+#     SOAK_OUT           产物根目录，默认 soak-out/<时间戳>
+#     SOAK_SCENARIO      appworld / govdoc / both，默认 both
+#     SOAK_CONTAINERS    AppWorld 容器数，默认 4
+#     SOAK_SPLITS        AppWorld 划分，空格分隔，默认 "train dev"
+#     SOAK_GOVDOC_DB     GovDoc 的审核点 sqlite，不给就用场景里的默认路径
+#     SOAK_GOVDOC_CORPUS GovDoc 的语料目录，同上
+#     SOAK_LOG_DIR       每一步的输出重定向到这里；不给就直接打屏（tmux 里能实时看）
+#     SOAK_SKIP_FAULTS   非空则跳过故障注入那一步
+#     SOAK_CONDA_ENV     conda 环境名，默认 PolyLoop
+
+set -euo pipefail
+
+STEP="启动"
+trap 'echo "!!! 第「${STEP}」步失败（退出码 $?）" >&2' ERR
+
+CONDA_ENV="${SOAK_CONDA_ENV:-PolyLoop}"
+SCENARIO="${SOAK_SCENARIO:-both}"
+CONTAINERS="${SOAK_CONTAINERS:-4}"
+SPLITS="${SOAK_SPLITS:-train dev}"
+LOG_DIR="${SOAK_LOG_DIR:-}"
+OUT="${SOAK_OUT:-soak-out/$(date +%Y%m%d-%H%M%S)}"
+
+BUDGET_CALLS="${SOAK_BUDGET_CALLS:?必须给 SOAK_BUDGET_CALLS：正常负载的模型调用次数上限}"
+FAULT_BUDGET_CALLS="${SOAK_FAULT_BUDGET_CALLS:?必须给 SOAK_FAULT_BUDGET_CALLS：故障注入那一步的上限}"
+CONCURRENCY="${SOAK_CONCURRENCY:?必须给 SOAK_CONCURRENCY：同时在跑的任务数上限}"
+LIMIT="${SOAK_LIMIT:?必须给 SOAK_LIMIT：每个场景最多跑几个任务}"
+
+# conda 和 Python 各缓冲一层，两层都得拆——只加其中一个，长跑命令仍然全程无输出。
+RUN=(env PYTHONUNBUFFERED=1 conda run --live-stream -n "$CONDA_ENV" python)
+
+RUNS_DIR="$OUT/runs"
+FAULT_RUNS_DIR="$OUT/fault-runs"
+
+# 场景相关的参数拼成数组。**用数组不用字符串**：路径里有空格时字符串会在展开时被切开，
+# 而表现是「找不到这个目录」，看起来像数据没准备好。
+SCENARIO_ARGS=(--scenario "$SCENARIO" --limit "$LIMIT" --containers "$CONTAINERS")
+if [[ "$SCENARIO" == "appworld" || "$SCENARIO" == "both" ]]; then
+    APPWORLD_DATA_ROOT="${SOAK_APPWORLD_DATA_ROOT:?SOAK_SCENARIO 含 appworld 时必须给 SOAK_APPWORLD_DATA_ROOT}"
+    SCENARIO_ARGS+=(--appworld-data-root "$APPWORLD_DATA_ROOT")
+    for split in $SPLITS; do
+        SCENARIO_ARGS+=(--split "$split")
+    done
+fi
+# 故障注入那一步的两个 GovDoc 路径没有默认值，而它们的权威在场景模块里。**问 Python 要，
+# 不在这里写第二份**：两处各写一份路径，迟早有一处被改、另一处没改，而表现是「数据不在」。
+GOVDOC_DB="${SOAK_GOVDOC_DB:-$("${RUN[@]}" -c 'from tools.soak.scenarios.govdoc import DEFAULT_DATA_ROOT as R; print(R / "app.sqlite")')}"
+GOVDOC_CORPUS="${SOAK_GOVDOC_CORPUS:-$("${RUN[@]}" -c 'from tools.soak.scenarios.govdoc import DEFAULT_DATA_ROOT as R; print(R / "storage" / "prepared")')}"
+if [[ "$SCENARIO" == "govdoc" || "$SCENARIO" == "both" ]]; then
+    SCENARIO_ARGS+=(--govdoc-db "$GOVDOC_DB" --govdoc-corpus "$GOVDOC_CORPUS")
+fi
+
+# 故障注入的参数是另一套：它自己的 `--split` 只收一个值，AppWorld 数据根目录那一项叫
+# `--data-root`。这里按 `python -m tools.soak.faults --help` 的形状拼。
+FAULT_ARGS=(--govdoc-db "$GOVDOC_DB" --govdoc-corpus "$GOVDOC_CORPUS")
+if [[ "$SCENARIO" == "appworld" || "$SCENARIO" == "both" ]]; then
+    FAULT_ARGS+=(--data-root "$APPWORLD_DATA_ROOT" --split "${SPLITS%% *}")
+fi
+
+mkdir -p "$OUT"
+
+step() {
+    local name="$1"
+    shift
+    STEP="$name"
+    echo ""
+    echo "=== [$name] $* ==="
+    if [[ -n "$LOG_DIR" ]]; then
+        mkdir -p "$LOG_DIR"
+        "$@" >"$LOG_DIR/$name.log" 2>&1
+    else
+        "$@"
+    fi
+}
+
+step 01-干跑 "${RUN[@]}" -m tools.soak.run_soak \
+    "${SCENARIO_ARGS[@]}" \
+    --budget-calls "$BUDGET_CALLS" \
+    --concurrency "$CONCURRENCY" \
+    --runs-dir "$RUNS_DIR" \
+    --report "$OUT/dry-run.md" \
+    --dry-run
+
+step 02-正常负载 "${RUN[@]}" -m tools.soak.run_soak \
+    "${SCENARIO_ARGS[@]}" \
+    --budget-calls "$BUDGET_CALLS" \
+    --concurrency "$CONCURRENCY" \
+    --runs-dir "$RUNS_DIR" \
+    --report "$OUT/normal-load.md"
+
+# 正常负载不该有缺文件，所以这一轮不开 --allow-undetermined：判不了就是要人去看一眼。
+step 03-记分板-正常负载 "${RUN[@]}" -m tools.soak.scoreboard \
+    --runs-dir "$RUNS_DIR" \
+    --report "$OUT/scoreboard-normal.md" \
+    --completing-tool submit_finding
+
+if [[ -n "${SOAK_SKIP_FAULTS:-}" ]]; then
+    echo ""
+    echo "=== [04-故障注入] 按 SOAK_SKIP_FAULTS 跳过 ==="
+else
+    step 04-故障注入 "${RUN[@]}" -m tools.soak.faults \
+        "${FAULT_ARGS[@]}" \
+        --runs-dir "$FAULT_RUNS_DIR" \
+        --budget-calls "$FAULT_BUDGET_CALLS"
+
+    # 崩溃注入那一类天然会缺文件，这一轮才该开 --allow-undetermined。
+    step 05-记分板-故障注入 "${RUN[@]}" -m tools.soak.scoreboard \
+        --runs-dir "$FAULT_RUNS_DIR" \
+        --report "$OUT/scoreboard-faults.md" \
+        --completing-tool submit_finding \
+        --allow-undetermined
+fi
+
+STEP="收尾"
+echo ""
+echo "=== 全部步骤通过 ==="
+echo "产物：$OUT"
+ls -1 "$OUT"

tools/soak/tests/test_run_soak.py

@@ -0,0 +1,599 @@
+"""压测入口驱动的测试。**不打真实模型、不起容器**，模型那一侧全是写死脚本的替身。
+
+分五块：参数校验、预算护栏与并发上限、产物与记分板的对接、错误隔离与取消、GovDoc 的阶段串行。
+
+最有价值的一条是「产物喂给记分板判成全绿」。驱动写四个文件、记分板读四个文件，两边的字段
+约定只写在记分板的模块 docstring 里，没有任何机器约束把它们钉在一起——那条测试就是那个约束：
+少写一个字段、类型写错一个，它当场红。
+
+GovDoc 那几块用手工搭的 `AuditTask`，不读磁盘上的真实语料：那份数据是另一个项目的工作副本，
+换一台机器就没有，而这里要验的是编排（阶段串行、任务并发），跟语料内容无关。
+"""
+
+from __future__ import annotations
+
+import asyncio
+import json
+from typing import TYPE_CHECKING
+
+import pytest
+
+from polyloop.types import ModelReply
+from tools.soak import scoreboard
+from tools.soak.run_soak import (
+    BudgetGuard,
+    RunOutcome,
+    Unit,
+    dispatch,
+    execute_run,
+    interleave,
+    main,
+    run_govdoc_task,
+    select_appworld_tasks,
+)
+from tools.soak.scenarios import govdoc as govdoc_scenario
+
+if TYPE_CHECKING:
+    from collections.abc import Callable, Sequence
+    from pathlib import Path
+
+    from polyloop.ports import ModelCall
+
+# ---------------------------------------------------------------------------
+# 替身
+# ---------------------------------------------------------------------------
+
+
+class _ScriptedModel:
+    """按 run_id 给回复的模型替身。每次调用都让出一次事件循环，好让并发真的发生。"""
+
+    def __init__(self, *, render: Callable[[ModelCall], str]) -> None:
+        self._render = render
+        #: 调用到达的顺序，按 run_id 记。阶段串行与任务并发都从它上面判。
+        self.seen: list[str] = []
+
+    async def call(self, call: ModelCall) -> ModelReply:
+        await asyncio.sleep(0.001)
+        self.seen.append(call.run_id)
+        return ModelReply(
+            call_id=f"call-{len(self.seen)}",
+            content=self._render(call),
+            thinking="",
+        )
+
+    def parameters(self):  # noqa: ANN201 - 替身，签名由 Protocol 定
+        return {"kind": "scripted"}
+
+
+def _guard(*, render: Callable[[ModelCall], str], limit: int = 1000) -> BudgetGuard:
+    return BudgetGuard(inner=_ScriptedModel(render=render), limit=limit)
+
+
+_SUBMIT = json.dumps(
+    {
+        "tool": "submit_finding",
+        "arguments": {
+            "verdict": "存疑",
+            "evidence": "tender.md 第 1 行：示例甲公司",
+            "reasoning": "测试替身给的固定结论",
+        },
+    },
+    ensure_ascii=False,
+)
+
+#: 解析不出来的输出。GovDoc 的预算里连续解析失败上限是 3，所以三步就停——plan 与 execute
+#: 两个阶段没有结束通路（提交工具只在 summarize 那一阶段），不这样它们会一路走到 50 步。
+_GARBAGE = "我先想一想这道题。"
+
+
+def _audit_task(index: int) -> govdoc_scenario.AuditTask:
+    checkpoint = govdoc_scenario.Checkpoint(
+        checkpoint_id=f"cp-{index}",
+        category="资格条件",
+        title="供应商资格要求",
+        description="不得以注册地设置差别待遇。",
+        legal_basis=("政府采购法第五条",),
+        severity="高",
+    )
+    document = govdoc_scenario.CorpusDocument.from_text(
+        logical_name="tender.md",
+        text="第一行：示例甲公司参与投标。\n第二行：投标截止时间为示例日期。",
+    )
+    return govdoc_scenario.AuditTask(index=index, checkpoint=checkpoint, documents=(document,))
+
+
+def _fake_unit(
+    label: str,
+    *,
+    body: Callable[[], object],
+) -> Unit:
+    async def start() -> Sequence[RunOutcome]:
+        result = body()
+        if asyncio.iscoroutine(result):
+            await result
+        return [
+            RunOutcome(
+                run_id=label,
+                scenario="fake",
+                task_id=label,
+                phase=None,
+                stop_reason="task_completed",
+                steps=1,
+                model_calls=1,
+                wall_ms=0,
+            )
+        ]
+
+    return Unit(label=label, scenario="fake", run=start)
+
+
+# ---------------------------------------------------------------------------
+# 一、参数校验
+# ---------------------------------------------------------------------------
+
+
+def test_missing_budget_calls_refuses(tmp_path: Path) -> None:
+    with pytest.raises(SystemExit) as caught:
+        main(["--scenario", "govdoc", "--concurrency", "2", "--runs-dir", str(tmp_path)])
+    assert caught.value.code != 0
+
+
+def test_missing_concurrency_refuses(tmp_path: Path) -> None:
+    with pytest.raises(SystemExit) as caught:
+        main(["--scenario", "govdoc", "--budget-calls", "10", "--runs-dir", str(tmp_path)])
+    assert caught.value.code != 0
+
+
+def test_non_positive_budget_refuses(tmp_path: Path) -> None:
+    with pytest.raises(SystemExit):
+        main(
+            [
+                "--scenario",
+                "govdoc",
+                "--budget-calls",
+                "0",
+                "--concurrency",
+                "1",
+                "--runs-dir",
+                str(tmp_path),
+            ]
+        )
+
+
+def test_appworld_without_data_root_refuses(tmp_path: Path) -> None:
+    with pytest.raises(SystemExit):
+        main(
+            [
+                "--scenario",
+                "appworld",
+                "--budget-calls",
+                "5",
+                "--concurrency",
+                "1",
+                "--runs-dir",
+                str(tmp_path),
+                "--dry-run",
+            ]
+        )
+
+
+# ---------------------------------------------------------------------------
+# 二、预算护栏与并发上限
+# ---------------------------------------------------------------------------
+
+
+async def test_budget_counts_are_per_run_and_total() -> None:
+    guard = _guard(render=lambda call: "ok")
+    calls = [_call("run-a"), _call("run-a"), _call("run-b")]
+    for item in calls:
+        await guard.call(item)
+    assert guard.total == 3
+    assert guard.calls_for("run-a") == 2
+    assert guard.calls_for("run-b") == 1
+    assert guard.exhausted is False
+
+
+async def test_budget_stops_dispatch_and_lets_running_tasks_finish() -> None:
+    guard = _guard(render=lambda call: "ok", limit=2)
+
+    async def slow() -> None:
+        await guard.call(_call("slow"))
+        await asyncio.sleep(0.05)
+
+    async def quick() -> None:
+        await guard.call(_call("quick"))
+
+    units = [
+        _fake_unit("unit-1", body=slow),
+        _fake_unit("unit-2", body=quick),
+        _fake_unit("unit-3", body=quick),
+        _fake_unit("unit-4", body=quick),
+    ]
+    report = await dispatch(units, concurrency=2, guard=guard)
+
+    assert report.dispatched == 2
+    # 「第 N 个任务」从 1 数：第 3 个是第一个没派出去的。
+    assert report.stopped_at == 3
+    # 慢的那个是在预算耗尽时正在跑的，它必须跑完并留下结果，不许被砍断。
+    assert {item.run_id for item in report.outcomes} == {"unit-1", "unit-2"}
+    assert guard.total == 2
+
+
+async def test_concurrency_never_exceeds_the_cap() -> None:
+    live = 0
+    peak = 0
+
+    async def body() -> None:
+        nonlocal live, peak
+        live += 1
+        peak = max(peak, live)
+        await asyncio.sleep(0.005)
+        live -= 1
+
+    units = [_fake_unit(f"unit-{index}", body=body) for index in range(12)]
+    report = await dispatch(units, concurrency=3)
+
+    assert peak == 3
+    assert report.dispatched == 12
+    assert len(report.outcomes) == 12
+
+
+async def test_zero_concurrency_is_rejected() -> None:
+    with pytest.raises(ValueError, match="并发上限"):
+        await dispatch([], concurrency=0)
+
+
+def test_interleave_alternates_between_scenarios() -> None:
+    left = [_fake_unit(f"L{index}", body=lambda: None) for index in range(3)]
+    right = [_fake_unit(f"R{index}", body=lambda: None) for index in range(2)]
+    assert [unit.label for unit in interleave([left, right])] == ["L0", "R0", "L1", "R1", "L2"]
+
+
+def test_select_appworld_tasks_dedupes_across_splits() -> None:
+    class _Pool:
+        def list_task_ids(self, split: str) -> list[str]:
+            return {"train": ["a", "b"], "dev": ["b", "c", "d"]}[split]
+
+    assert select_appworld_tasks(_Pool(), splits=["train", "dev"], limit=None) == [
+        "a",
+        "b",
+        "c",
+        "d",
+    ]
+    assert select_appworld_tasks(_Pool(), splits=["train", "dev"], limit=3) == ["a", "b", "c"]
+
+
+# ---------------------------------------------------------------------------
+# 三、错误隔离与取消
+# ---------------------------------------------------------------------------
+
+
+async def test_one_failing_unit_does_not_stop_the_batch() -> None:
+    def boom() -> None:
+        raise RuntimeError("这一道题炸了")
+
+    units = [
+        _fake_unit("ok-1", body=lambda: None),
+        _fake_unit("boom", body=boom),
+        _fake_unit("ok-2", body=lambda: None),
+    ]
+    report = await dispatch(units, concurrency=1)
+
+    assert {item.run_id for item in report.outcomes} == {"ok-1", "ok-2"}
+    assert len(report.failures) == 1
+    label, message = report.failures[0]
+    assert label == "boom"
+    assert "RuntimeError" in message and "这一道题炸了" in message
+
+
+async def test_cancellation_passes_through_and_cleans_up() -> None:
+    running = asyncio.Event()
+    cleaned = False
+
+    async def body() -> None:
+        nonlocal cleaned
+        running.set()
+        try:
+            await asyncio.sleep(10)
+        finally:
+            cleaned = True
+
+    units = [_fake_unit("slow", body=body)]
+    task = asyncio.create_task(dispatch(units, concurrency=1))
+    await running.wait()
+    task.cancel()
+
+    with pytest.raises(asyncio.CancelledError):
+        await task
+    assert cleaned is True
+
+
+# ---------------------------------------------------------------------------
+# 四、产物：字段齐全，且记分板判得动
+# ---------------------------------------------------------------------------
+
+
+async def _run_summarize(tmp_path: Path, *, index: int = 0) -> None:
+    """跑一次 summarize 阶段：模型第一步就提交结论，运行以 task_completed 结束。"""
+    guard = _guard(render=lambda call: _SUBMIT)
+    await run_govdoc_task(
+        task=_audit_task(index),
+        runs_dir=tmp_path / "runs",
+        workspace_root=tmp_path / "workspaces",
+        guard=guard,
+        phases=("summarize",),
+    )
+
+
+async def test_sidecar_fields_are_complete_and_typed(tmp_path: Path) -> None:
+    await _run_summarize(tmp_path)
+    runs = tmp_path / "runs"
+    run_id = govdoc_scenario.make_run_id(task_index=0, phase="summarize")
+
+    assert (runs / f"{run_id}.jsonl").is_file()
+    assert (runs / f"{run_id}.result.json").is_file()
+    assert (runs / f"{run_id}.events.jsonl").is_file()
+
+    meta = json.loads((runs / f"{run_id}.meta.json").read_text(encoding="utf-8"))
+    assert set(meta) == {
+        "scenario",
+        "task_id",
+        "phase",
+        "wall_ms",
+        "model_calls",
+        "sink_failures",
+        "env_executions",
+        "fault",
+        "success",
+        "resumed_from_step",
+    }
+    assert meta["scenario"] == "govdoc"
+    assert meta["task_id"] == "cp-0"
+    assert meta["phase"] == "summarize"
+    assert isinstance(meta["wall_ms"], int)
+    assert meta["model_calls"] == 1
+    assert meta["sink_failures"] == 0
+    # submit_finding 往审计账里追加一行，那是环境自己记的账。
+    assert meta["env_executions"] == 1
+    assert meta["fault"] is None
+    assert meta["success"] is None
+    assert meta["resumed_from_step"] == 0
+
+    events = [
+        json.loads(line)
+        for line in (runs / f"{run_id}.events.jsonl").read_text(encoding="utf-8").splitlines()
+    ]
+    assert events == [{"kind": "step_finished", "run_id": run_id, "step_idx": 0}]
+
+    result = json.loads((runs / f"{run_id}.result.json").read_text(encoding="utf-8"))
+    assert result["run_id"] == run_id
+    assert result["stop_reason"] == "task_completed"
+    assert result["event_delivery_failures"] == 0
+
+
+async def test_artifacts_pass_the_scoreboard(tmp_path: Path) -> None:
+    """把驱动写出来的产物直接喂给记分板，要它报全绿。
+
+    两边的字段约定没有任何机器约束把它们钉在一起，这条测试就是那个约束。
+    """
+    guard = _guard(render=lambda call: _SUBMIT if "summarize" in call.run_id else _GARBAGE)
+    for index in range(2):
+        await run_govdoc_task(
+            task=_audit_task(index),
+            runs_dir=tmp_path / "runs",
+            workspace_root=tmp_path / "workspaces",
+            guard=guard,
+        )
+
+    board = scoreboard.evaluate(
+        tmp_path / "runs",
+        completing_tools=frozenset({"submit_finding"}),
+    )
+    trouble = [
+        f"{item.name}: {[evidence.describe() for evidence in (*item.breaches, *item.undetermined)]}"
+        for item in (*board.breached, *board.undetermined)
+    ]
+    assert board.verdict is scoreboard.Verdict.PASSED, trouble
+    assert len(board.summaries) == 6
+    assert {item.scenario for item in board.summaries} == {"govdoc"}
+
+
+async def test_sink_failures_match_what_the_library_counted(tmp_path: Path) -> None:
+    """事件文件写不出去时，出口自己数的失败次数与结果里那份必须一致。
+
+    这条不变量是记分板的一条判定，而两边的计数由不同的代码写：出口在自己的 `except` 里加一，
+    库在接住异常之后加一。只要出口漏加或多加，记分板当场报击穿。
+    """
+    runs = tmp_path / "runs"
+    runs.mkdir()
+    run_id = govdoc_scenario.make_run_id(task_index=0, phase="summarize")
+    # 在事件文件该在的位置放一个目录，追加写就必然失败。
+    (runs / f"{run_id}.events.jsonl").mkdir()
+
+    guard = _guard(render=lambda call: _SUBMIT)
+    await run_govdoc_task(
+        task=_audit_task(0),
+        runs_dir=runs,
+        workspace_root=tmp_path / "workspaces",
+        guard=guard,
+        phases=("summarize",),
+    )
+
+    meta = json.loads((runs / f"{run_id}.meta.json").read_text(encoding="utf-8"))
+    result = json.loads((runs / f"{run_id}.result.json").read_text(encoding="utf-8"))
+    assert meta["sink_failures"] == 1
+    assert result["event_delivery_failures"] == meta["sink_failures"]
+
+
+async def test_a_failing_run_still_leaves_meta(tmp_path: Path) -> None:
+    """`run` 抛异常也要留下 `.meta.json`——缺文件在记分板那边只是「无法判定」。
+
+    这里用同一个运行标识跑第二次来触发失败：库对已经有日志的标识直接报 `RunIdentityError`，
+    而那是压测里最可能真的撞上的一种失败（任务集去重漏了一处）。
+    """
+    runs = tmp_path / "runs"
+    await _run_summarize(tmp_path)
+    run_id = govdoc_scenario.make_run_id(task_index=0, phase="summarize")
+    (runs / f"{run_id}.result.json").unlink()
+
+    guard = _guard(render=lambda call: _SUBMIT)
+    task = _audit_task(0)
+    request = govdoc_scenario.build_run_request(
+        task=task,
+        phase="summarize",
+        run_id=run_id,
+        workspace=tmp_path / "ws",
+        model_binding={},
+    )
+    outcome = await execute_run(
+        runs_dir=runs,
+        request=request,
+        model_client=guard,
+        decision_parser=govdoc_scenario.GovDocParser(),
+        synthetic_observations=govdoc_scenario.SYNTHETIC_OBSERVATIONS,
+        scenario="govdoc",
+        task_id=task.checkpoint.checkpoint_id,
+        phase="summarize",
+        count_model_calls=lambda: guard.calls_for(run_id),
+    )
+
+    assert outcome.error is not None
+    assert "RunIdentityError" in outcome.error
+    assert outcome.stop_reason is None
+    meta = json.loads((runs / f"{run_id}.meta.json").read_text(encoding="utf-8"))
+    assert meta["model_calls"] == 0
+    assert meta["success"] is None
+    # 这一次没有结果，所以没有 `.result.json` 可写——记分板会把它记成一条要人看见的观察。
+    assert not (runs / f"{run_id}.result.json").exists()
+
+
+async def test_a_model_error_is_a_stop_reason_not_a_crash(tmp_path: Path) -> None:
+    """模型调用失败不会把异常抛出循环，它是一次以 `llm_error` 结束的正常运行。
+
+    压测的报告里这两件事必须分得开：`error` 是驱动这边出的事，`llm_error` 是库判定的停止原因。
+    """
+
+    class _Exploding:
+        async def call(self, call: ModelCall) -> ModelReply:
+            raise RuntimeError("网关炸了")
+
+        def parameters(self):  # noqa: ANN202 - 替身
+            return {"kind": "exploding"}
+
+    guard = BudgetGuard(inner=_Exploding(), limit=5)
+    task = _audit_task(0)
+    request = govdoc_scenario.build_run_request(
+        task=task,
+        phase="summarize",
+        run_id="govdoc-0-summarize",
+        workspace=tmp_path / "ws",
+        model_binding={},
+    )
+    outcome = await execute_run(
+        runs_dir=tmp_path / "runs",
+        request=request,
+        model_client=guard,
+        decision_parser=govdoc_scenario.GovDocParser(),
+        synthetic_observations=govdoc_scenario.SYNTHETIC_OBSERVATIONS,
+        scenario="govdoc",
+        task_id=task.checkpoint.checkpoint_id,
+        phase="summarize",
+        count_model_calls=lambda: guard.calls_for("govdoc-0-summarize"),
+    )
+
+    assert outcome.error is None
+    assert outcome.stop_reason == "llm_error"
+    meta = json.loads(
+        (tmp_path / "runs" / "govdoc-0-summarize.meta.json").read_text(encoding="utf-8")
+    )
+    assert meta["model_calls"] == 1
+    assert (tmp_path / "runs" / "govdoc-0-summarize.result.json").is_file()
+
+
+# ---------------------------------------------------------------------------
+# 五、GovDoc：阶段串行、任务并发
+# ---------------------------------------------------------------------------
+
+
+async def test_govdoc_phases_run_in_order_within_a_task(tmp_path: Path) -> None:
+    model = _ScriptedModel(render=lambda call: _GARBAGE)
+    guard = BudgetGuard(inner=model, limit=1000)
+    await run_govdoc_task(
+        task=_audit_task(0),
+        runs_dir=tmp_path / "runs",
+        workspace_root=tmp_path / "workspaces",
+        guard=guard,
+    )
+
+    phases = [run_id.rsplit("-", 1)[1] for run_id in model.seen]
+    # 每个阶段的调用连成一段，段与段之间不交错。
+    assert phases == sorted(phases, key=["plan", "execute", "summarize"].index)
+    assert set(phases) == {"plan", "execute", "summarize"}
+
+
+async def test_govdoc_tasks_overlap_while_phases_do_not(tmp_path: Path) -> None:
+    model = _ScriptedModel(render=lambda call: _GARBAGE)
+    guard = BudgetGuard(inner=model, limit=1000)
+    units = [
+        Unit(
+            label=f"govdoc/{index}",
+            scenario="govdoc",
+            run=_govdoc_runner(
+                index=index,
+                runs_dir=tmp_path / "runs",
+                workspace_root=tmp_path / "workspaces",
+                guard=guard,
+            ),
+        )
+        for index in range(2)
+    ]
+    report = await dispatch(units, concurrency=2, guard=guard)
+    assert len(report.outcomes) == 6
+    assert report.failures == ()
+
+    # 阶段串行：同一个任务里，后一阶段的第一次调用晚于前一阶段的最后一次调用。
+    for index in range(2):
+        own = [
+            position
+            for position, run_id in enumerate(model.seen)
+            if run_id.startswith(f"govdoc-{index}-")
+        ]
+        for earlier, later in (("plan", "execute"), ("execute", "summarize")):
+            last_earlier = max(
+                position for position in own if model.seen[position].endswith(earlier)
+            )
+            first_later = min(position for position in own if model.seen[position].endswith(later))
+            assert last_earlier < first_later
+
+    # 任务并发：两个任务的调用在时间上交错。
+    owners = [run_id.split("-")[1] for run_id in model.seen]
+    assert any(left != right for left, right in zip(owners, owners[1:], strict=False))
+
+
+def _govdoc_runner(*, index: int, runs_dir: Path, workspace_root: Path, guard: BudgetGuard):  # noqa: ANN202
+    async def start() -> Sequence[RunOutcome]:
+        return await run_govdoc_task(
+            task=_audit_task(index),
+            runs_dir=runs_dir,
+            workspace_root=workspace_root,
+            guard=guard,
+        )
+
+    return start
+
+
+# ---------------------------------------------------------------------------
+# 小工具
+# ---------------------------------------------------------------------------
+
+
+def _call(run_id: str) -> ModelCall:
+    from polyloop.ports import ModelCall as _ModelCall
+
+    return _ModelCall(
+        messages=(),
+        call_index=0,
+        run_id=run_id,
+        result_id=f"{run_id}-0",
+        binding={},
+    )