feat(soak): 补上 context_overflow 与 env_error 两类，七类变九类

一次 193 个运行的全量跑完之后，停止原因的十个取值里有两个一次都没出现过。没出现不等于
它们是对的，只等于没验过——这正是「全绿要先怀疑负载」该指向的地方。

context_overflow 的判据不能照字面写成「最后一步的提示词超过上限」：规模判定在调模型之前
做，命中时不产生步记录，所以落盘的每条步记录必定不超上限，那样断言等于断言契约的反面。
改成判「再走一步会有多大」，公式拿全量里 865 对相邻步验过，0 处不符。

env_error 是真把容器 docker kill 掉，不是用测试替身。它自己起一个池、用另一个端口——
共用那个 size=1 的池的话，排在它后面的每一类都会跑在一个不存在的环境上。

实跑：两类都通过，击穿 0、无法判定 0。

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

tools/soak/faults.py

@@ -1,9 +1,20 @@
-"""压测的故障注入：在崩溃、取消、撞预算、解析连击四条路径上验库有没有守住承诺。
+"""压测的故障注入：在崩溃、取消、撞预算、解析连击、提示词超限、环境故障这几条路径上验库有
+没有守住承诺。
 
-一百个任务顺利跑完什么都证明不了——顺利那条路上库只要不崩就算过。能证明东西的是这四条：
+一百个任务顺利跑完什么都证明不了——顺利那条路上库只要不崩就算过。能证明东西的是这几条：
-进程被杀在半路、调用方中途取消、目标根本完不成、模型的输出一句都解析不了。库在这些地方给
+进程被杀在半路、调用方中途取消、目标根本完不成、模型的输出一句都解析不了、提示词撑到装不下、
-下游的承诺（崩溃前的轨迹逐字节不变、声明绝不重放的动作不会执行两次、取消原样穿透且资源
+环境跑到一半不能接着服务了。库在这些地方给下游的承诺（崩溃前的轨迹逐字节不变、声明绝不重放
-归还、撞上限时以正确的停止原因干净收尾、解析失败不碰环境）只有在这里才检验得到。
+的动作不会执行两次、取消原样穿透且资源归还、撞上限时以正确的停止原因干净收尾、解析失败不碰
+环境、提示词超限时终止而不是静默截断历史、环境坏掉那一步的观察换成合成观察）只有在这里才
+检验得到。
+
+**九类里最后添的两类是在补一次全量跑留下的空白。** 193 个运行跑完之后，`polyloop.types.
+StopReason` 的十个取值里有两个一次都没出现过：`context_overflow` 与 `env_error`。没出现不等于
+它们是对的，只等于没验过——那两条路上的承诺到那一刻为止一次都没有被检验，而它们错了的形态
+恰好都不会当场炸：静默截断历史在数据里看起来跟「模型不行」一模一样，环境故障那一步的观察
+没被替换掉则表现成模型收到一段来路不明的文本。所以这两类各自把那个停止原因造出来一次：
+GovDoc 那一路把 `max_prompt_chars` 压到刚好等于装配出来的初始提示词，AppWorld 那一路在跑完
+一步之后真的把容器 `docker kill` 掉。
 
 **崩溃续跑用 GovDoc 场景，不用 AppWorld。** GovDoc 的「环境」是一个工作区目录加一份审计
 日志（`tools/soak/scenarios/govdoc.py` 里 `GovDocTools._append_audit` 写的那份），它天然活过
@@ -39,6 +50,7 @@ from __future__ import annotations
 
 import argparse
 import asyncio
+import contextlib
 import json
 import os
 import sys
@@ -48,6 +60,8 @@ from dataclasses import dataclass, replace
 from enum import StrEnum
 from pathlib import Path
 
+import httpx
+
 from polyloop import session
 from polyloop.adapters import GatewayModelClient
 from polyloop.ports import Event, InvalidDecision, ParsedReply, RunLog, RunStore
@@ -55,6 +69,8 @@ from polyloop.serialization import encode
 from polyloop.session import AgentDefinition, ParameterDriftError, RunRequest
 from polyloop.stores import RECORD_KEY, JsonlRunStore
 from polyloop.types import (
+    ActionOutcome,
+    ActionStatus,
     Budget,
     Intent,
     ModelCallResult,
@@ -64,8 +80,9 @@ from polyloop.types import (
     RunResult,
     RunStarted,
     StepCompleted,
+    TextBlock,
 )
-from tools.soak.appworld import AppWorldPool
+from tools.soak.appworld import CONTAINER_NAME_PREFIX, DEFAULT_PORT_BASE, AppWorldPool
 from tools.soak.scenarios import appworld as appworld_scenario
 from tools.soak.scenarios import govdoc as govdoc_scenario
 from tools.soak.scenarios.appworld import AppWorldParser
@@ -89,10 +106,12 @@ FAULT_NAMES: tuple[str, ...] = (
     "step_budget",
     "action_budget",
     "parse_failures",
+    "context_overflow",
+    "env_error",
 )
 
-#: 用 GovDoc 场景做的那两类。其余用 AppWorld。
+#: 用 GovDoc 场景做的那三类。其余用 AppWorld。
-GOVDOC_FAULTS = frozenset({"crash_resume_a", "crash_resume_b"})
+GOVDOC_FAULTS = frozenset({"crash_resume_a", "crash_resume_b", "context_overflow"})
 APPWORLD_FAULTS = frozenset(FAULT_NAMES) - GOVDOC_FAULTS
 
 
@@ -273,6 +292,19 @@ def _step_payloads(read: LogRead) -> tuple[Mapping[str, object], ...]:
     return tuple(steps)
 
 
+def parameter_snapshot_of(read: LogRead) -> Mapping[str, object] | None:
+    """运行开始那条记录带的参数快照。没有那条记录就是 None。
+
+    判据要从这里取这次运行实际用的参数，而不是从本文件的常量取：常量是我们**打算**用的值，
+    快照是库**真的**收到的值，两者不一致时该被判据看见的正是后者。
+    """
+    started = tagged(read, "run_started")
+    if not started:
+        return None
+    snapshot = started[0].get("parameter_snapshot")
+    return snapshot if isinstance(snapshot, Mapping) else None
+
+
 # ---------------------------------------------------------------------------
 # 三、判据：崩溃续跑
 # ---------------------------------------------------------------------------
@@ -482,9 +514,17 @@ def check_resume_made_progress(*, crashed_steps: int, final_steps: int) -> Crite
 
 
 # ---------------------------------------------------------------------------
-# 四、判据：停止原因、步数、取消、环境
+# 四、判据：停止原因、步数、取消、环境、提示词规模
 # ---------------------------------------------------------------------------
 
+#: 提示词上限在参数快照里的键名。快照由 `RunRequest.parameter_snapshot()` 拼出来，值是十进制
+#: 字符串。判据从快照读它、不硬编码：这一路的上限是按上下文现算出来的，写死一份的话判据比的
+#: 就是另一个数字，而两边不一致时它照样给得出「通过」。
+MAX_PROMPT_CHARS_SNAPSHOT_KEY = "request.max_prompt_chars"
+
+#: 观察模板在参数快照里的键名。算「再走一步的提示词会有多大」要用它。
+OBSERVATION_TEMPLATE_SNAPSHOT_KEY = "request.observation_template"
+
 
 def check_stop_reason(read: LogRead, expected: str) -> Criterion:
     """结束记录在场，且它带的停止原因是期望的那个。
@@ -502,6 +542,19 @@ def check_stop_reason(read: LogRead, expected: str) -> Criterion:
     return passed(f"stop_reason_is_{expected}", f"日志的 run_finished 里停止原因是 {actual}")
 
 
+def check_run_finished_present(read: LogRead) -> Criterion:
+    """日志里有结束记录。
+
+    它由库在把结果交给调用方**之前**写下，所以它在不在与停止原因取值对不对是两件事：没有它
+    的话，恢复会把这次运行读成一次可以续跑的运行，而它其实已经结束了。分成两条判据是因为两者
+    的成因不同——一条查的是那次写有没有发生，另一条查的是写下去的取值对不对。
+    """
+    finished = tagged(read, "run_finished")
+    if not finished:
+        return breached("run_finished_recorded", "日志里没有 run_finished 记录")
+    return passed("run_finished_recorded", f"日志里有 {len(finished)} 条 run_finished 记录")
+
+
 def check_step_count(read: LogRead, *, expected: int, name: str) -> Criterion:
     actual = len(_step_payloads(read))
     if actual != expected:
@@ -545,6 +598,212 @@ def check_no_env_error_step(read: LogRead) -> Criterion:
     return passed("no_env_error_step", f"{len(entries)} 步里没有 env_error")
 
 
+def check_at_least_one_step(read: LogRead) -> Criterion:
+    """至少完整走过一步。
+
+    撞提示词上限那一类要靠它兜底：一步都没走的话，撞上的是「上下文本身就比上限大」，而那个
+    长度是装配决定的、不是循环一步步撑出来的，这一类什么都没验到。报「无法判定」而不是
+    「击穿」——那种情况下库做的事仍然是对的，错的是这一路的上限选得太小。
+    """
+    steps = _step_payloads(read)
+    if not steps:
+        return undetermined(
+            "at_least_one_step",
+            "日志里一条步记录都没有：提示词在第一次装配时就超了上限，循环一步都没走过",
+        )
+    return passed("at_least_one_step", f"日志里 {len(steps)} 条步记录")
+
+
+def check_prompt_chars_monotonic(read: LogRead) -> Criterion:
+    """`prompt_chars` 在整个运行里单调不减：历史一次都没有被截断过。
+
+    静默截断正是这一档要防住的坏法。它把提示词的前缀改掉，于是后面每一步都重新全价计费，而
+    在数据里看起来跟「模型不行」一模一样。历史只增不减时这一列只会往上走，截断会在它上面留下
+    一次下降——那是这件事在轨迹里唯一看得见的痕迹。
+    """
+    steps = _step_payloads(read)
+    values = [step.get("prompt_chars") for step in steps]
+    if not values:
+        return undetermined("prompt_chars_never_shrinks", "日志里一条步记录都没有，单调性无从判起")
+    bad = [value for value in values if not isinstance(value, int) or isinstance(value, bool)]
+    if bad:
+        return breached(
+            "prompt_chars_never_shrinks", f"有 {len(bad)} 条步记录的 prompt_chars 不是整数"
+        )
+    drops = [
+        f"第 {index} 步从 {previous} 掉到 {current}"
+        for index, (previous, current) in enumerate(zip(values, values[1:], strict=False), start=1)
+        if current < previous  # type: ignore[operator]
+    ]
+    if drops:
+        return breached(
+            "prompt_chars_never_shrinks",
+            f"提示词字符数出现下降（按步记录次序计）：{'；'.join(drops)}，历史被截断过",
+        )
+    return passed(
+        "prompt_chars_never_shrinks",
+        f"{len(values)} 步的提示词字符数从 {values[0]} 一路不减到 {values[-1]}",
+    )
+
+
+def check_prompt_reached_max_prompt_chars(read: LogRead) -> Criterion:
+    """这次运行真的撑到了提示词上限那条线上。
+
+    **判的不是「最后一条步记录的 `prompt_chars` 超过了上限」。** 那条在一个正确的库上永远不
+    成立，断言它等于断言契约的反面：规模判定在调模型之前做，命中时不产生步记录也不写任何意图
+    （`polyloop._stopping.prompt_size_admission`），所以落进日志的每一条步记录都是通过了那一档
+    的，它们的 `prompt_chars` 必定不超过上限。真正超限的那一次装配没有在任何地方留下记录。
+
+    能从日志里判的是那一次装配会有多大。最后一步之后历史里多出两条消息：它的 `raw_output`，
+    以及套上观察模板的 `observation`（`polyloop._assembly.history_messages`）。两者的字符数加上
+    最后一步的 `prompt_chars`，就是下一次装配出来的规模。这个数不超过上限的话，说明库在还装得
+    下的时候就报了 `context_overflow`。
+
+    另一头也要判：最后一步自己的 `prompt_chars` 超过上限，说明有一次超限的装配被放行去调了
+    模型，那正是这一档该拦住的东西。
+
+    上限与观察模板都从参数快照里读，不从本文件的常量读——这一路的上限是按上下文现算的。
+    """
+    name = "prompt_reached_max_prompt_chars"
+    snapshot = parameter_snapshot_of(read)
+    if snapshot is None:
+        return undetermined(name, "日志里没有 run_started 记录，读不到这次运行的提示词上限")
+    limit_text = snapshot.get(MAX_PROMPT_CHARS_SNAPSHOT_KEY)
+    template = snapshot.get(OBSERVATION_TEMPLATE_SNAPSHOT_KEY)
+    if not isinstance(limit_text, str) or not isinstance(template, str):
+        return undetermined(
+            name,
+            f"参数快照里缺 {MAX_PROMPT_CHARS_SNAPSHOT_KEY} 或 "
+            f"{OBSERVATION_TEMPLATE_SNAPSHOT_KEY}，算不出该不该撞上限",
+        )
+    try:
+        limit = int(limit_text)
+    except ValueError:
+        return undetermined(name, f"参数快照里的提示词上限不是一个整数：{limit_text!r}")
+    steps = _step_payloads(read)
+    if not steps:
+        return undetermined(name, "日志里一条步记录都没有，撞没撞上上限无从判起")
+    last = steps[-1]
+    chars = last.get("prompt_chars")
+    raw_output = last.get("raw_output")
+    observation = last.get("observation")
+    if (
+        not isinstance(chars, int)
+        or isinstance(chars, bool)
+        or not isinstance(raw_output, str)
+        or not isinstance(observation, str)
+    ):
+        return undetermined(name, "最后一条步记录缺 prompt_chars、raw_output 或 observation")
+    try:
+        rendered = template.format(observation=observation)
+    except (KeyError, IndexError, ValueError) as exc:
+        return undetermined(name, f"观察模板渲染不了（{exc}），算不出下一步的提示词规模")
+    if chars > limit:
+        return breached(
+            name,
+            f"最后一步自己的提示词就有 {chars} 字符，已经超过上限 {limit}："
+            "有一次超限的装配被放行去调了模型",
+        )
+    projected = chars + len(raw_output) + len(rendered)
+    if projected <= limit:
+        return breached(
+            name,
+            f"最后一步的提示词 {chars} 字符，加上它的输出与套过模板的观察一共 {projected} 字符，"
+            f"仍不超过上限 {limit}：还装得下就报了 context_overflow",
+        )
+    return passed(
+        name,
+        f"最后一步的提示词 {chars} 字符没超上限 {limit}，再走一步会是 {projected} 字符，超了",
+    )
+
+
+def check_last_step_action_status(read: LogRead, *, expected: str) -> Criterion:
+    """最后一条步记录的动作状态是期望的那个。"""
+    name = f"last_step_action_status_is_{expected}"
+    steps = _step_payloads(read)
+    if not steps:
+        return undetermined(name, "日志里一条步记录都没有，动作状态无从判起")
+    actual = steps[-1].get("action_status")
+    if actual != expected:
+        return breached(name, f"最后一步的动作状态是 {actual!r}，期望 {expected!r}")
+    return passed(name, f"最后一步的动作状态是 {expected}")
+
+
+def check_last_observation_is_synthetic(read: LogRead, *, expected: str, name: str) -> Criterion:
+    """最后一步的观察被换成了库自己那段合成文本，而且逐字就是装配时给的那一段。
+
+    环境故障那一档下，执行接缝返回的观察不进历史，库改填 `SyntheticObservations.env_failed`
+    （`polyloop.session` 的 `_project_observation`）。这么定是因为观察属于「模型看得见的东西」，
+    那种东西必须能进参数快照；执行接缝每次现造一段的话，同一份配置跑出来的两次运行在模型看来
+    其实不同，而没有任何地方会报错。
+
+    期望值由调用处从装配时用的那份取，这里不抄字面量。
+
+    **证据只报长度不报正文。** 替换要是没发生，那一列里躺着的是环境（或与环境通信那一层）给的
+    文本，它可以是任何东西，而这份报告是要贴给人看的。
+    """
+    steps = _step_payloads(read)
+    if not steps:
+        return undetermined(name, "日志里一条步记录都没有，观察无从判起")
+    last = steps[-1]
+    flag = last.get("observation_is_synthetic")
+    if flag is not True:
+        return breached(name, f"最后一步的 observation_is_synthetic 是 {flag!r}，期望 True")
+    actual = last.get("observation")
+    if not isinstance(actual, str):
+        return breached(name, f"最后一步的 observation 不是字符串，是 {type(actual).__name__}")
+    if actual != expected:
+        return breached(
+            name,
+            f"最后一步的观察不是装配时给的那段合成观察（实际 {len(actual)} 字符，"
+            f"期望那段 {len(expected)} 字符）",
+        )
+    return passed(name, f"最后一步的观察逐字是装配时给的那段合成观察（{len(expected)} 字符）")
+
+
+def check_steps_before_last_all_executed(read: LogRead) -> Criterion:
+    """环境坏掉之前的那些步没有被牵连：它们的动作状态都是「已执行」。
+
+    这一条守的是「环境故障只影响它发生的那一步」。前面的步被改写或被补上一个别的状态，说明
+    库把一次局部故障扩散到了已经落地的轨迹上，而那些步的观察是模型接下来要看的东西。
+    """
+    name = "steps_before_env_error_executed"
+    steps = _step_payloads(read)
+    if len(steps) < 2:
+        return undetermined(
+            name,
+            f"日志里只有 {len(steps)} 条步记录，环境故障那一步之前一步都没有，这一条真空成立",
+        )
+    bad = [
+        index for index, step in enumerate(steps[:-1]) if step.get("action_status") != "executed"
+    ]
+    if bad:
+        return breached(
+            name, f"第 {bad} 步（按步记录次序计）的动作状态不是 executed，它们在环境坏掉之前"
+        )
+    return passed(name, f"环境坏掉之前的 {len(steps) - 1} 步动作状态全是 executed")
+
+
+def check_env_broken_after_a_full_step(*, broken: bool, executions_before_break: int) -> Criterion:
+    """动手弄坏环境之前，至少已经完整执行过一次动作。
+
+    第一次执行之前就把容器打掉的话，验的是「环境起不来」而不是「跑到一半环境不能接着服务
+    了」——前者落在会话初始化上，根本走不到动作执行接缝，而这一类要压的正是那个接缝。
+    """
+    name = "env_broken_after_a_full_step"
+    if not broken:
+        return undetermined(
+            name, "这次运行结束时环境一次都没被弄坏过：动作执行没走到该动手的那一次"
+        )
+    if executions_before_break < 1:
+        return undetermined(
+            name,
+            f"弄坏环境之前只完整执行过 {executions_before_break} 次动作，"
+            "压到的是初始化而不是执行接缝",
+        )
+    return passed(name, f"完整执行过 {executions_before_break} 次动作之后才把环境弄坏")
+
+
 def check_all_steps_parse_failed(read: LogRead) -> Criterion:
     """这几步全部解析失败，且一个动作都没被分发。
 
@@ -1018,6 +1277,152 @@ class TriggeringExecutor:
         return await self._inner.execute(action)  # type: ignore[attr-defined]
 
 
+def container_name_for_port(port: int) -> str:
+    """池里跑在这个宿主端口上的容器叫什么。
+
+    形状是「前缀 + 端口号」，前缀取 `tools/soak/appworld.py` 公开的那个常量，端口取池的 `ports`
+    属性。**不去调容器池那个同名的私有方法**：从外面 import 一个下划线开头的名字等于把它的实现
+    细节钉死。代价是这里拼错了没人会当场发现，所以拼错的后果被做成显式失败——`docker kill` 会
+    以「没有这个容器」的退出码告诉我们，那一路直接抛错，不会静默地跑成一次「环境没坏」的运行。
+    """
+    return f"{CONTAINER_NAME_PREFIX}-{port}"
+
+
+def should_break_env(
+    *, executions_done: int, break_after_executions: int, already_broken: bool
+) -> bool:
+    """现在这一次动作执行之前，该不该动手把环境弄坏。纯函数。
+
+    `executions_done` 数的是**已经成功走完**的动作次数，所以 `break_after_executions` 取 1 就是
+    「第一步完整走完之后、第二步的执行之前动手」。取 0 会把容器打在第一次执行之前，那时压到的
+    是会话初始化而不是执行接缝。
+
+    弄坏过一次之后不再动手：容器已经没了，再发一次 `docker kill` 只会拿到一个「没有这个容器」
+    的错误，而那个错误会被当成「弄坏失败」报出来。
+    """
+    if already_broken:
+        return False
+    return executions_done >= break_after_executions
+
+
+async def _docker(*args: str) -> tuple[int, str, str]:
+    """跑一条 docker 命令，返回 (退出码, stdout, stderr)。
+
+    `tools/soak/appworld.py` 里有一个形状相同的私有函数，这里不 import 它——理由与
+    `container_name_for_port` 那条相同。
+    """
+    process = await asyncio.create_subprocess_exec(
+        "docker", *args, stdout=asyncio.subprocess.PIPE, stderr=asyncio.subprocess.PIPE
+    )
+    stdout, stderr = await process.communicate()
+    return (
+        process.returncode or 0,
+        stdout.decode(errors="replace"),
+        stderr.decode(errors="replace"),
+    )
+
+
+async def kill_and_remove_container(name: str) -> str:
+    """把这个容器打死并删掉，返回一句给 note 用的说明。
+
+    **两步分开发。** `docker rm -f` 自己也会先发 SIGKILL，一条命令就够，但分成两条之后
+    「环境从这一刻起服务不了了」与「容器有没有被留在盘上」在报告里各是一句话：第二步失败只是
+    留下一个已经死掉的容器要人手动删，故障本身照样注入成功。
+
+    第一步失败则直接抛错：那说明容器名拼错了或者容器根本不在，环境没被弄坏，这一路接下来跑出
+    来的东西什么都不是。抛出去由 `guarded()` 记成「无法判定」。
+    """
+    code, stdout, stderr = await _docker("kill", name)
+    if code != 0:
+        raise FaultInjectionError(
+            f"docker kill {name} 失败（退出码 {code}）：{(stderr or stdout).strip()}；"
+            "环境没有被弄坏，这一类没有意义"
+        )
+    remove_code, remove_out, remove_err = await _docker("rm", "-f", name)
+    if remove_code != 0:
+        return (
+            f"容器 {name} 已被 docker kill，但随后的 docker rm -f 失败"
+            f"（退出码 {remove_code}）：{(remove_err or remove_out).strip()}，要人手动删"
+        )
+    return f"容器 {name} 已被 docker kill 并删除"
+
+
+class EnvBreakingExecutor:
+    """转发动作执行，并在跑完指定次数之后真的把容器打死。
+
+    **弄坏的是真环境，不是一个测试替身。** 容器没了之后，后续的 `execute` 连不上那个端口，
+    这一路要验的就是库拿到 `ENV_ERROR` 之后做了什么。
+
+    **它还要把连不上翻译成 `ENV_ERROR`，因为场景那侧不翻译。**
+    `tools/soak/scenarios/appworld.py` 的 `AppWorldExecutor` 只接 `AppWorldError`，而
+    `AppWorldSession.execute` 在容器没了时抛的是 `httpx.ConnectError`——`_post_json` 直接
+    `await client.post(...)`，没有把传输层的异常包起来。实测过（`docker kill` 之后 execute 抛
+    `httpx.ConnectError: All connection attempts failed`）。不翻译的话这次运行会整个抛出去，
+    根本产不出 `env_error` 这个停止原因。翻译这件事本来就归执行接缝：判据是「环境还能不能接着
+    服务」，由适配器判、库不判（`research-wiki/design/0007-seam-behaviour.md` 决策一）。
+
+    内层自己给出 `ENV_ERROR` 时原样透传，不覆盖——那样场景那边哪天补上了转换，这里不用跟着改。
+    """
+
+    __slots__ = (
+        "_break_after",
+        "_break_env",
+        "_inner",
+        "broken",
+        "break_note",
+        "executions",
+        "executions_before_break",
+    )
+
+    def __init__(
+        self,
+        *,
+        inner: object,
+        break_after_executions: int,
+        break_env: Callable[[], Awaitable[str]],
+    ) -> None:
+        """Args:
+        inner: 真正打环境的那个执行器。
+        break_after_executions: 完整执行过几次动作之后动手。见 `should_break_env`。
+        break_env: 怎么弄坏，返回一句说明。**做成参数是为了能测**——默认那条路要起真容器。
+        """
+        self._inner = inner
+        self._break_after = break_after_executions
+        self._break_env = break_env
+        self.executions = 0
+        self.broken = False
+        self.executions_before_break = 0
+        self.break_note = ""
+
+    def parameters(self) -> Mapping[str, str]:
+        return dict(self._inner.parameters()) | {"env_break_probe": "docker_kill"}  # type: ignore[attr-defined]
+
+    async def execute(self, action: object) -> ActionOutcome:
+        if should_break_env(
+            executions_done=self.executions,
+            break_after_executions=self._break_after,
+            already_broken=self.broken,
+        ):
+            self.executions_before_break = self.executions
+            self.break_note = await self._break_env()
+            self.broken = True
+        try:
+            outcome = await self._inner.execute(action)  # type: ignore[attr-defined]
+        except httpx.HTTPError as exc:
+            # 环境不能接着服务了。观察填异常的类名与文本：它是与环境通信那一层给的，不是库
+            # 合成的占位，所以 `observation_is_synthetic` 为假——库随后会把这一列换成自己的
+            # 那段合成观察，而这一类要验的正是那次替换。
+            return ActionOutcome(
+                status=ActionStatus.ENV_ERROR,
+                observation=f"{type(exc).__name__}: {exc}",
+                observation_is_synthetic=False,
+                env_reported_completion=False,
+                observation_truncated_chars=0,
+            )
+        self.executions += 1
+        return outcome  # type: ignore[no-any-return]
+
+
 @dataclass(slots=True)
 class CallGuard:
     """调用数护栏：花掉多少次真实模型调用，还剩多少。
@@ -1152,6 +1557,65 @@ CRASH_AFTER_STEPS = 1
 
 APPWORLD_MODEL_BINDING: Mapping[str, str] = {"scenario": "appworld"}
 
+#: 撞提示词上限那一路，除提示词上限外的三个数。
+#:
+#: 步数上限取 3 是**兜底**：正常情况下第 1 步走完、第 2 步装配时就撞上提示词那一档，走不到这
+#: 里。它存在是为了万一那一档没命中时这次跑仍然会停下来，而不是一路烧到 GovDoc 场景自己那份
+#: 50 步的预算。取 3 不取 2 是留一步余量给「第一次模型输出解析不了」这种意外。
+CONTEXT_OVERFLOW_BUDGET_STEPS = 3
+
+#: 提示词上限在初始提示词之上留的余量，字符。
+#:
+#: **取值的两头都是实测出来的。** 下界是 0：`initial_prompt_chars` 算出来的数与库自己装配出来
+#: 的第一步 `prompt_chars` 逐字符相同（拿 GovDoc execute 阶段实测，两边都是 2828），所以留 0
+#: 也够第一步过关。上界是「一步能把提示词撑大多少」：48 次真实 GovDoc execute 运行里，从第 0
+#: 步到第 1 步的增长最少 157 字符、中位数 989。余量必须小于那个最小值，否则第 2 步的装配可能
+#: 仍然装得下，这一类就撞不上上限。64 落在两者中间，两边都留着一倍以上的距离。
+CONTEXT_OVERFLOW_SLACK_CHARS = 64
+
+
+def initial_prompt_chars(request: RunRequest) -> int:
+    """一步都还没走时，这次运行装配出来的提示词有多少字符。
+
+    照 `polyloop._assembly.assemble` 在零步下的形态数：run 级片段、注入槽、目标级片段，历史
+    是空的。**自己数而不是 import 那个模块**：它是下划线开头的内部模块，从压测工具里 import
+    等于把它的内部结构钉死。代价是这里与库的度量口径可能漂移，所以取值不是拍脑袋来的——实测
+    过它与库写进第一条步记录的 `prompt_chars` 相同，而且余量的取法（`CONTEXT_OVERFLOW_SLACK_
+    CHARS`）本来就容得下几十个字符的偏差。
+
+    认不得的内容块类型直接抛错，不当成 0：当成 0 会让上限算小，于是第一步就撞上限、一步都走
+    不成，而报出来只是一句「无法判定」。
+    """
+    total = 0
+    messages = (*request.context.run_level, *request.context.goal_level)
+    for message in messages:
+        for block in message.content:
+            if not isinstance(block, TextBlock):
+                raise FaultInjectionError(
+                    f"上下文里有认不得的内容块类型 {type(block).__name__}，数不出提示词规模"
+                )
+            total += len(block.text)
+    for entries in request.injections.values():
+        for entry in entries:
+            total += len(entry.content)
+    return total
+
+
+def build_context_overflow_budget(request: RunRequest) -> Budget:
+    """把提示词上限压到「刚好装得下第一步、装不下第二步」。
+
+    **上限按这次运行的上下文现算，不写死一个数字。** 上下文的大小取决于挑中的是哪个审核点、
+    语料清单有多长，写死的话它今天够用、换一份数据就要么第一步就撞上限（什么都没验到），要么
+    宽到几步都撞不上（白烧模型调用）。现算出来的这个数在 GovDoc execute 阶段是三千字符上下。
+    """
+    return Budget(
+        max_steps=CONTEXT_OVERFLOW_BUDGET_STEPS,
+        max_actions=CONTEXT_OVERFLOW_BUDGET_STEPS,
+        max_consecutive_parse_failures=CONTEXT_OVERFLOW_BUDGET_STEPS,
+        max_prompt_chars=initial_prompt_chars(request) + CONTEXT_OVERFLOW_SLACK_CHARS,
+    )
+
+
 #: 崩溃那一刻的日志快照落盘时用的后缀。
 CRASH_SNAPSHOT_SUFFIX = ".crash-snapshot"
 
@@ -1421,8 +1885,84 @@ async def _resume_and_judge(
     return FaultReport(fault=fault, criteria=tuple(criteria), notes=tuple(notes))
 
 
+async def run_context_overflow_fault(
+    *,
+    runs_dir: Path,
+    workspace_root: Path,
+    govdoc_db: Path,
+    govdoc_corpus: Path,
+    model_client: object,
+    guard: CallGuard,
+) -> FaultReport:
+    """撞提示词上限：把上限压到刚好等于初始提示词，验库终止而不是静默截断历史。
+
+    **用 GovDoc 不用 AppWorld**：它的观察是文档片段，一步就能把提示词撑过去，而且上下文本身
+    只有三千字符上下，上限压到那个量级之后第一步仍然过得去。
+
+    工作区里照崩溃续跑那一路预置同一份计划，理由也相同：execute 阶段的提示词让模型先读 plan.md
+    再动笔，没有它第一步会浪费在一次读不到文件的工具调用上。这一路只走一步，但那一步的观察越
+    像真的越好——它正是把提示词撑过上限的那一段。
+    """
+    fault = "context_overflow"
+    run_id = f"fault-{fault}"
+    log_path = runs_dir / f"{run_id}.jsonl"
+    task = load_govdoc_task(govdoc_db=govdoc_db, govdoc_corpus=govdoc_corpus)
+    workspace = workspace_root / run_id
+    workspace.mkdir(parents=True, exist_ok=True)
+    (workspace / SEEDED_PLAN_NAME).write_text(SEEDED_PLAN_TEXT, encoding="utf-8")
+
+    # 绑定沿用崩溃续跑那一份：它记的是「哪个场景、哪个阶段」，而这一路两样都相同。
+    base = govdoc_scenario.build_run_request(
+        task=task,
+        phase=CRASH_PHASE,
+        run_id=run_id,
+        workspace=workspace,
+        model_binding=CRASH_MODEL_BINDING,
+    )
+    budget = build_context_overflow_budget(base)
+    request = replace(base, budget=budget)
+    store = JsonlRunStore(directory=runs_dir)
+    sink = JsonlEventSink(runs_dir / f"{run_id}.events.jsonl")
+    definition = build_crash_definition(model_client=model_client, store=store, sink=sink)
+
+    started = time.monotonic()
+    result = await session.run(definition, request)
+    wall_ms = int((time.monotonic() - started) * 1000)
+
+    read = read_log(log_path)
+    guard.charge(count_model_calls(read))
+    criteria = (
+        check_log_readable(read),
+        check_run_finished_present(read),
+        check_stop_reason(read, "context_overflow"),
+        check_at_least_one_step(read),
+        check_prompt_chars_monotonic(read),
+        check_prompt_reached_max_prompt_chars(read),
+    )
+    notes = (
+        "提示词上限按这次运行的上下文现算："
+        f"{budget.max_prompt_chars - CONTEXT_OVERFLOW_SLACK_CHARS} 字符的初始提示词加 "
+        f"{CONTEXT_OVERFLOW_SLACK_CHARS} 字符余量，一共 {budget.max_prompt_chars}",
+    )
+    write_sidecars(
+        runs_dir=runs_dir,
+        run_id=run_id,
+        scenario="govdoc",
+        fault=fault,
+        result=result,
+        wall_ms=wall_ms,
+        model_calls=count_model_calls(read),
+        sink_failures=sink.failures,
+        env_executions=len(read_audit_lines(workspace)),
+        task_id=task.checkpoint.checkpoint_id,
+        phase=CRASH_PHASE,
+        resumed_from_step=None,
+    )
+    return FaultReport(fault=fault, criteria=criteria, notes=notes)
+
+
 # ---------------------------------------------------------------------------
-# 九、AppWorld：取消、预算、解析连击
+# 九、AppWorld：取消、预算、解析连击、环境故障
 # ---------------------------------------------------------------------------
 
 #: 撞步数上限那一路的预算。**动作上限必须比步数上限宽**，否则先撞上的是动作那一维，停止
@@ -1441,6 +1981,24 @@ PARSE_FAILURE_BUDGET = Budget(
     max_steps=10, max_actions=10, max_consecutive_parse_failures=3, max_prompt_chars=400_000
 )
 
+#: 环境故障那一路的预算。四步是给「走一步 → 打死容器 → 第二步撞环境故障」留的余量：正常只用
+#: 两步，多出来的两步是留给一次解析失手的。
+ENV_ERROR_BUDGET = Budget(
+    max_steps=4, max_actions=4, max_consecutive_parse_failures=3, max_prompt_chars=400_000
+)
+
+#: 完整执行过几次动作之后才把容器打死。见 `should_break_env`：取 0 压到的是初始化不是执行接缝。
+ENV_ERROR_BREAK_AFTER_EXECUTIONS = 1
+
+#: 环境故障那一路自己那个容器池的起始端口。
+#:
+#: **它必须自己起一个池，不能用其余几类共用的那个。** 那个池只有一个容器（容器租约那条判据
+#: 要求如此），而这一类会把容器打死；共用的话，排在它后面的每一类都会跑在一个已经没了的环境
+#: 上，而报出来是一串「连不上」，看起来像 docker 挂了。
+#:
+#: 端口取共用池那一个的下一个：共用池大小是 1，只占 `DEFAULT_PORT_BASE` 本身。
+ENV_ERROR_PORT_BASE = DEFAULT_PORT_BASE + 1
+
 #: 取消之后再借一个容器的等待上限。池满时 `lease()` 会一直阻塞，所以超时就是击穿。
 LEASE_TIMEOUT_S = 120.0
 
@@ -1693,6 +2251,103 @@ async def run_parse_failure_fault(
     return FaultReport(fault=fault, criteria=criteria, notes=())
 
 
+async def run_env_error_fault(
+    *,
+    data_root: Path,
+    split: str,
+    runs_dir: Path,
+    model_client: object,
+    guard: CallGuard,
+) -> FaultReport:
+    """环境故障：正常跑一步，把容器打死，验库换成合成观察并以 `env_error` 收尾。
+
+    **容器是真的被杀掉的，不是一个返回 `ENV_ERROR` 的替身。** 替身验不到「环境真的不在了
+    之后，在飞的连接、会话关闭、容器池收尾这一串还能不能干净地收场」，而那串正是这一类跑完
+    要留下的东西。
+
+    收尾这件事实测过一遍：容器被 `docker kill` 之后，`pool.session()` 退出时的 `/close` 会失败
+    ——那由 `AppWorldPool._close_quietly` 记一次账、不抛（连续三次才抛），所以这一类不会把整轮
+    压测带走；随后 `pool.stop()` 里的 `docker rm -f` 撞上一个已经删掉的容器，退出码非零而它本来
+    就不看退出码。两处都不需要改 `tools/soak/appworld.py`。
+    """
+    fault = "env_error"
+    run_id = f"fault-{fault}"
+    log_path = runs_dir / f"{run_id}.jsonl"
+    store = JsonlRunStore(directory=runs_dir)
+    sink = JsonlEventSink(runs_dir / f"{run_id}.events.jsonl")
+    synthetic = appworld_scenario.build_synthetic_observations()
+    notes: list[str] = []
+    result: RunResult | None = None
+    env_executions = 0
+    started = time.monotonic()
+
+    async with AppWorldPool(data_root=data_root, size=1, port_base=ENV_ERROR_PORT_BASE) as pool:
+        task_ids = pool.list_task_ids(split)
+        if not task_ids:
+            raise FaultInjectionError(f"{split} 划分里一道题都没有，环境故障那一类跑不了")
+        task_id = task_ids[0]
+        app_descriptions = await appworld_scenario.load_app_descriptions(pool, task_id=task_id)
+        container = container_name_for_port(pool.ports[0])
+        async with pool.session(task_id) as handle:
+            request = replace(
+                appworld_scenario.build_run_request(
+                    run_id=run_id,
+                    session=handle,
+                    app_descriptions=app_descriptions,
+                    model_binding=APPWORLD_MODEL_BINDING,
+                ),
+                budget=ENV_ERROR_BUDGET,
+            )
+            breaker = EnvBreakingExecutor(
+                inner=request.action_executor,
+                break_after_executions=ENV_ERROR_BREAK_AFTER_EXECUTIONS,
+                break_env=lambda: kill_and_remove_container(container),
+            )
+            request = replace(request, action_executor=breaker)  # type: ignore[arg-type]
+            definition = _appworld_definition(
+                model_client=model_client, store=store, sink=sink, parser=AppWorldParser()
+            )
+            result = await session.run(definition, request)
+            env_executions = handle.n_executions
+
+    wall_ms = int((time.monotonic() - started) * 1000)
+    if breaker.break_note:
+        notes.append(breaker.break_note)
+    read = read_log(log_path)
+    guard.charge(count_model_calls(read))
+
+    criteria = (
+        check_log_readable(read),
+        check_run_finished_present(read),
+        check_stop_reason(read, "env_error"),
+        check_last_step_action_status(read, expected="env_error"),
+        check_last_observation_is_synthetic(
+            read, expected=synthetic.env_failed, name="env_failed_observation_substituted"
+        ),
+        check_steps_before_last_all_executed(read),
+        check_env_broken_after_a_full_step(
+            broken=breaker.broken, executions_before_break=breaker.executions_before_break
+        ),
+    )
+    write_sidecars(
+        runs_dir=runs_dir,
+        run_id=run_id,
+        scenario="appworld",
+        fault=fault,
+        result=result,
+        wall_ms=wall_ms,
+        model_calls=count_model_calls(read),
+        sink_failures=sink.failures,
+        # 环境侧自己数的执行次数。**打死容器之后那一次不计入**：`AppWorldSession` 在请求返回
+        # 之后才加一，而那一次请求根本没到过环境。
+        env_executions=env_executions,
+        task_id=task_id,
+        phase=None,
+        resumed_from_step=None,
+    )
+    return FaultReport(fault=fault, criteria=criteria, notes=tuple(notes))
+
+
 # ---------------------------------------------------------------------------
 # 十、子进程入口
 # ---------------------------------------------------------------------------
@@ -1746,7 +2401,9 @@ async def run_child(args: argparse.Namespace) -> int:
 def build_parser() -> argparse.ArgumentParser:
     parser = argparse.ArgumentParser(
         prog="python -m tools.soak.faults",
-        description="PolyLoop 压测的故障注入：崩溃续跑、取消、撞预算、解析失败连击",
+        description=(
+            "PolyLoop 压测的故障注入：崩溃续跑、取消、撞预算、解析失败连击、撞提示词上限、环境故障"
+        ),
     )
     parser.add_argument("--runs-dir", required=True, help="产物目录：日志与三种 sidecar 都写在这里")
     parser.add_argument(
@@ -1835,14 +2492,32 @@ async def run_all(args: argparse.Namespace) -> int:
 
         govdoc_selected = [name for name in selected if name in GOVDOC_FAULTS]
         if govdoc_selected:
-            govdoc_db = _require(args, "govdoc-db", "崩溃续跑要读审核点")
+            govdoc_db = _require(args, "govdoc-db", "GovDoc 那几类要读审核点")
-            govdoc_corpus = _require(args, "govdoc-corpus", "崩溃续跑要读语料")
+            govdoc_corpus = _require(args, "govdoc-corpus", "GovDoc 那几类要读语料")
             for name in govdoc_selected:
-                timing = KillTiming.AFTER_STEP if name == "crash_resume_a" else KillTiming.AT_INTENT
+                coroutine: Awaitable[FaultReport]
+                if name == "context_overflow":
+                    if not guard.affordable(CONTEXT_OVERFLOW_BUDGET_STEPS):
                         reports.append(
-                    await guarded(
+                            _unaffordable(
-                        name,
+                                name, guard=guard, worst_case=CONTEXT_OVERFLOW_BUDGET_STEPS
-                        run_crash_fault(
+                            )
+                        )
+                        continue
+                    coroutine = run_context_overflow_fault(
+                        runs_dir=runs_dir,
+                        workspace_root=runs_dir / "workspaces",
+                        govdoc_db=govdoc_db,
+                        govdoc_corpus=govdoc_corpus,
+                        model_client=model_client,
+                        guard=guard,
+                    )
+                else:
+                    # 崩溃续跑那两类自己按轮次查护栏（每重试一轮都要再问一次），不在这里查。
+                    timing = (
+                        KillTiming.AFTER_STEP if name == "crash_resume_a" else KillTiming.AT_INTENT
+                    )
+                    coroutine = run_crash_fault(
                         fault=name,
                         timing=timing,
                         runs_dir=runs_dir,
@@ -1853,9 +2528,8 @@ async def run_all(args: argparse.Namespace) -> int:
                         guard=guard,
                         attempts=args.attempts,
                         self_kill_grace_s=args.self_kill_grace,
-                        ),
-                    )
                     )
+                reports.append(await guarded(name, coroutine))
 
         appworld_selected = [name for name in selected if name in APPWORLD_FAULTS]
         if appworld_selected:
@@ -1881,6 +2555,30 @@ async def run_all(args: argparse.Namespace) -> int:
     return 1 if breaches else 0
 
 
+#: 每一类 AppWorld 故障最坏会花掉几次真实模型调用。护栏按这个数在每类开跑前问一次。
+_APPWORLD_WORST_CASE_CALLS: Mapping[str, int] = {
+    "cancel_model": 3,
+    "cancel_env": 3,
+    "step_budget": STEP_BUDGET_OVERRIDE.max_steps,
+    "action_budget": ACTION_BUDGET_OVERRIDE.max_steps,
+    "parse_failures": PARSE_FAILURE_BUDGET.max_consecutive_parse_failures,
+    "env_error": ENV_ERROR_BUDGET.max_steps,
+}
+
+
+def _unaffordable(fault: str, *, guard: CallGuard, worst_case: int) -> FaultReport:
+    """护栏不够跑这一类了。**报「无法判定」不报「通过」**：它一次都没跑过。"""
+    return FaultReport(
+        fault=fault,
+        criteria=(
+            undetermined(
+                "call_budget_available",
+                f"调用数护栏只剩 {guard.remaining} 次，最坏要 {worst_case} 次，这一类没跑",
+            ),
+        ),
+    )
+
+
 async def _run_appworld_faults(
     *,
     names: Sequence[str],
@@ -1890,41 +2588,45 @@ async def _run_appworld_faults(
     model_client: object,
     guard: CallGuard,
 ) -> list[FaultReport]:
-    """AppWorld 那几类共用一个池。
+    """AppWorld 那几类共用一个池，环境故障那一类除外。
 
     **池大小固定为 1**：容器租约那条判据要的是「取消之后空闲名额回到满」，而池里有第二个
     容器的话，借得到只说明还剩别的名额，什么都证明不了。
+
+    **环境故障那一类自己起一个池**（见 `ENV_ERROR_PORT_BASE`）：它会把容器打死，共用的话排在
+    它后面的每一类都会跑在一个已经没了的环境上。所以共用池只在真有别的类要跑时才起——只选了
+    环境故障那一类时，起它纯属白等一次容器启动。
     """
     reports: list[FaultReport] = []
-    async with AppWorldPool(data_root=data_root, size=1) as pool:
+    shared_names = [name for name in names if name != "env_error"]
+    async with contextlib.AsyncExitStack() as stack:
+        pool: AppWorldPool | None = None
+        task_id = ""
+        app_descriptions = ""
+        if shared_names:
+            pool = await stack.enter_async_context(AppWorldPool(data_root=data_root, size=1))
             task_ids = pool.list_task_ids(split)
             if not task_ids:
                 raise FaultInjectionError(f"{split} 划分里一道题都没有，AppWorld 那几类跑不了")
             task_id = task_ids[0]
             app_descriptions = await appworld_scenario.load_app_descriptions(pool, task_id=task_id)
         for name in names:
-            worst_case = {
+            worst_case = _APPWORLD_WORST_CASE_CALLS[name]
-                "cancel_model": 3,
-                "cancel_env": 3,
-                "step_budget": STEP_BUDGET_OVERRIDE.max_steps,
-                "action_budget": ACTION_BUDGET_OVERRIDE.max_steps,
-                "parse_failures": PARSE_FAILURE_BUDGET.max_consecutive_parse_failures,
-            }[name]
             if not guard.affordable(worst_case):
-                reports.append(
+                reports.append(_unaffordable(name, guard=guard, worst_case=worst_case))
-                    FaultReport(
-                        fault=name,
-                        criteria=(
-                            undetermined(
-                                "call_budget_available",
-                                f"调用数护栏只剩 {guard.remaining} 次，最坏要 {worst_case} 次，这一类没跑",
-                            ),
-                        ),
-                    )
-                )
                 continue
             coroutine: Awaitable[FaultReport]
-            if name in {"cancel_model", "cancel_env"}:
+            if name == "env_error":
+                coroutine = run_env_error_fault(
+                    data_root=data_root,
+                    split=split,
+                    runs_dir=runs_dir,
+                    model_client=model_client,
+                    guard=guard,
+                )
+            elif pool is None:  # pragma: no cover - 共用池只在有别的类要跑时才建，走不到这里
+                raise FaultInjectionError(f"{name} 要用共用容器池，而它没有被建起来")
+            elif name in {"cancel_model", "cancel_env"}:
                 coroutine = run_cancel_fault(
                     fault=name,
                     pool=pool,
@@ -1982,15 +2684,22 @@ if __name__ == "__main__":
 
 __all__ = [
     "APPWORLD_FAULTS",
+    "CONTEXT_OVERFLOW_BUDGET_STEPS",
+    "CONTEXT_OVERFLOW_SLACK_CHARS",
     "CRASH_BUDGET",
     "CRASH_EXIT_CODE",
+    "ENV_ERROR_BREAK_AFTER_EXECUTIONS",
+    "ENV_ERROR_PORT_BASE",
     "FAULT_NAMES",
     "GOVDOC_FAULTS",
+    "MAX_PROMPT_CHARS_SNAPSHOT_KEY",
+    "OBSERVATION_TEMPLATE_SNAPSHOT_KEY",
     "SELF_KILL_GRACE_S",
     "AlwaysInvalidParser",
     "CallGuard",
     "Criterion",
     "CriterionStatus",
+    "EnvBreakingExecutor",
     "FaultInjectionError",
     "FaultReport",
     "JsonlEventSink",
@@ -1998,26 +2707,40 @@ __all__ = [
     "KillTiming",
     "LogRead",
     "SelfKillingStore",
+    "build_context_overflow_budget",
     "check_all_steps_parse_failed",
+    "check_at_least_one_step",
     "check_audit_unchanged",
     "check_cancelled_raised",
     "check_crash_prefix_preserved",
+    "check_env_broken_after_a_full_step",
     "check_env_untouched",
     "check_executed_action_count",
     "check_intents_settled",
+    "check_last_observation_is_synthetic",
+    "check_last_step_action_status",
     "check_lease_returned",
     "check_log_readable",
     "check_never_action_not_replayed",
     "check_no_env_error_step",
+    "check_prompt_chars_monotonic",
+    "check_prompt_reached_max_prompt_chars",
     "check_resume_made_progress",
+    "check_run_finished_present",
     "check_step_count",
     "check_step_indices_dense",
+    "check_steps_before_last_all_executed",
     "check_stop_reason",
+    "container_name_for_port",
     "count_model_calls",
+    "initial_prompt_chars",
+    "kill_and_remove_container",
     "main",
+    "parameter_snapshot_of",
     "parse_audit_line",
     "parse_terminated",
     "read_log",
+    "should_break_env",
     "should_kill",
     "stop_reason_of",
     "tagged",

tools/soak/tests/test_faults.py

@@ -21,57 +21,81 @@ import json
 import sys
 from pathlib import Path
 
+import httpx
 import pytest
 
 from polyloop.ports import EventKind, InvalidDecision, RunLog
+from polyloop.session import RunRequest
+from polyloop.tools import ToolRegistry
 from polyloop.types import (
     ActionOutcome,
     ActionStatus,
+    Budget,
+    Context,
+    Injection,
     Intent,
     IntentKind,
+    Message,
     ModelCallResult,
     ModelReply,
     ReplayPolicy,
+    Role,
     RunFinished,
     RunResult,
     RunStarted,
     StepCompleted,
     StepRecord,
     StopReason,
+    TextBlock,
 )
 from tools.soak.faults import (
+    CONTEXT_OVERFLOW_SLACK_CHARS,
     CRASH_EXIT_CODE,
+    FAULT_NAMES,
     AlwaysInvalidParser,
     CallGuard,
     Criterion,
     CriterionStatus,
+    EnvBreakingExecutor,
     FaultReport,
     JsonlEventSink,
     KillTiming,
     LogRead,
     SelfKillingStore,
+    build_context_overflow_budget,
     build_parser,
     check_all_steps_parse_failed,
+    check_at_least_one_step,
     check_audit_unchanged,
     check_cancelled_raised,
     check_crash_prefix_preserved,
+    check_env_broken_after_a_full_step,
     check_env_untouched,
     check_executed_action_count,
     check_intents_settled,
+    check_last_observation_is_synthetic,
+    check_last_step_action_status,
     check_lease_returned,
     check_log_readable,
     check_never_action_not_replayed,
     check_no_env_error_step,
+    check_prompt_chars_monotonic,
+    check_prompt_reached_max_prompt_chars,
     check_resume_made_progress,
+    check_run_finished_present,
     check_step_count,
     check_step_indices_dense,
+    check_steps_before_last_all_executed,
     check_stop_reason,
+    container_name_for_port,
     count_model_calls,
     guarded,
+    initial_prompt_chars,
     main,
     parse_audit_line,
     parse_terminated,
     read_log,
+    should_break_env,
     should_kill,
     spawn_and_kill,
     stop_reason_of,
@@ -79,6 +103,7 @@ from tools.soak.faults import (
     terminated_prefix,
     write_sidecars,
 )
+from tools.soak.scenarios.appworld import build_synthetic_observations
 from tools.soak.scenarios.govdoc import AUDIT_LOG_NAME
 
 RUN_ID = "fault-test-0"
@@ -123,6 +148,10 @@ def step_completed(
     status: str | None = "executed",
     parse_ok: bool = True,
     action_status: str | None = "executed",
+    prompt_chars: object = 10,
+    raw_output: str = "x",
+    observation: str = "o",
+    observation_is_synthetic: bool = False,
 ) -> dict[str, object]:
     outcome = (
         None
@@ -142,16 +171,16 @@ def step_completed(
         "action_outcome": outcome,
         "step": {
             "step_idx": step_idx,
-            "raw_output": "x",
+            "raw_output": raw_output,
-            "content_chars": 1,
+            "content_chars": len(raw_output),
             "thinking_chars": 0,
             "action": None,
             "parse_ok": parse_ok,
             "parse_error": None if parse_ok else "解释不了",
-            "observation": "o",
+            "observation": observation,
-            "observation_is_synthetic": False,
+            "observation_is_synthetic": observation_is_synthetic,
             "observation_truncated_chars": 0,
-            "prompt_chars": 10,
+            "prompt_chars": prompt_chars,
             "call_id": "c0",
             "step_wall_ms": 1,
             "tool_name": None,
@@ -163,6 +192,29 @@ def step_completed(
     }
 
 
+#: 造日志时用的观察模板。套一次观察多出 `len("观察：\n") == 4` 个字符，判据算「再走一步的提示词
+#: 会有多大」时要把这四个字符算进去。
+TEMPLATE = "观察：{observation}\n"
+
+
+def run_started(
+    *, max_prompt_chars: int = 100, observation_template: str = TEMPLATE
+) -> dict[str, object]:
+    """运行开始那条记录。参数快照里只放判据会读的两个键。
+
+    真的快照还有十几个键，多放几个不会让任何一条判据的行为改变——它们按键名取值。
+    """
+    return {
+        "record": "run_started",
+        "run_id": RUN_ID,
+        "parameter_snapshot": {
+            "request.max_prompt_chars": str(max_prompt_chars),
+            "request.observation_template": observation_template,
+        },
+        "schema_version": 1,
+    }
+
+
 def run_finished(*, stop_reason: str = "task_completed") -> dict[str, object]:
     return {
         "record": "run_finished",
@@ -1181,3 +1233,497 @@ def test_parser_accepts_repeated_fault_flags() -> None:
         ]
     )
     assert args.fault == ["cancel_model", "cancel_env"]
+
+
+# ---------------------------------------------------------------------------
+# 十五、撞提示词上限
+# ---------------------------------------------------------------------------
+
+
+def test_run_finished_present_passes_and_breaches() -> None:
+    """结束记录在不在与它带的取值对不对是两条判据，成因不同。"""
+    assert check_run_finished_present(as_read(run_finished())).status is CriterionStatus.PASSED
+    assert check_run_finished_present(as_read(step_completed())).status is CriterionStatus.BREACHED
+
+
+def test_stop_reason_context_overflow_passes_and_breaches() -> None:
+    good = as_read(step_completed(), run_finished(stop_reason="context_overflow"))
+    bad = as_read(step_completed(), run_finished(stop_reason="step_budget"))
+    assert check_stop_reason(good, "context_overflow").status is CriterionStatus.PASSED
+    breach = check_stop_reason(bad, "context_overflow")
+    assert breach.status is CriterionStatus.BREACHED
+    assert "step_budget" in breach.evidence
+
+
+def test_at_least_one_step_passes() -> None:
+    assert check_at_least_one_step(as_read(step_completed())).status is CriterionStatus.PASSED
+
+
+def test_at_least_one_step_is_undetermined_without_steps() -> None:
+    """第一步就撞上限的话这一类什么都没验到，报「无法判定」不报「击穿」。"""
+    outcome = check_at_least_one_step(as_read(run_started(), run_finished()))
+    assert outcome.status is CriterionStatus.UNDETERMINED
+    assert "一条步记录都没有" in outcome.evidence
+
+
+def test_prompt_chars_monotonic_passes() -> None:
+    read = as_read(
+        step_completed(step_idx=0, prompt_chars=100),
+        step_completed(step_idx=1, prompt_chars=100),
+        step_completed(step_idx=2, prompt_chars=250),
+    )
+    outcome = check_prompt_chars_monotonic(read)
+    assert outcome.status is CriterionStatus.PASSED
+    assert "100" in outcome.evidence and "250" in outcome.evidence
+
+
+def test_prompt_chars_monotonic_breaches_on_a_drop() -> None:
+    """一次下降就是历史被截断过——那是这件事在轨迹里唯一看得见的痕迹。"""
+    read = as_read(
+        step_completed(step_idx=0, prompt_chars=100),
+        step_completed(step_idx=1, prompt_chars=300),
+        step_completed(step_idx=2, prompt_chars=120),
+    )
+    outcome = check_prompt_chars_monotonic(read)
+    assert outcome.status is CriterionStatus.BREACHED
+    assert "第 2 步从 300 掉到 120" in outcome.evidence
+
+
+def test_prompt_chars_monotonic_breaches_on_a_non_integer() -> None:
+    read = as_read(step_completed(prompt_chars="很多"))
+    assert check_prompt_chars_monotonic(read).status is CriterionStatus.BREACHED
+
+
+def test_prompt_chars_monotonic_is_undetermined_without_steps() -> None:
+    assert check_prompt_chars_monotonic(as_read(run_started())).status is (
+        CriterionStatus.UNDETERMINED
+    )
+
+
+def test_prompt_reached_max_prompt_chars_passes() -> None:
+    """最后一步自己装得下，再走一步就装不下——那才是「撑到了上限那条线上」。
+
+    上限 100；最后一步的提示词 90，它的输出 5 个字符、观察 10 个字符再加模板的 4 个，
+    下一次装配是 109 字符。
+    """
+    read = as_read(
+        run_started(max_prompt_chars=100),
+        step_completed(prompt_chars=90, raw_output="x" * 5, observation="y" * 10),
+        run_finished(stop_reason="context_overflow"),
+    )
+    outcome = check_prompt_reached_max_prompt_chars(read)
+    assert outcome.status is CriterionStatus.PASSED
+    assert "109" in outcome.evidence
+
+
+def test_prompt_reached_max_prompt_chars_breaches_when_it_still_fits() -> None:
+    """还装得下就报 context_overflow：这次运行根本没撞到上限。"""
+    read = as_read(
+        run_started(max_prompt_chars=100),
+        step_completed(prompt_chars=50, raw_output="x", observation="y"),
+        run_finished(stop_reason="context_overflow"),
+    )
+    outcome = check_prompt_reached_max_prompt_chars(read)
+    assert outcome.status is CriterionStatus.BREACHED
+    assert "仍不超过上限 100" in outcome.evidence
+
+
+def test_prompt_reached_max_prompt_chars_breaches_when_an_oversized_prompt_was_admitted() -> None:
+    """步记录自己的提示词就超了上限，说明有一次超限的装配被放行去调了模型。"""
+    read = as_read(
+        run_started(max_prompt_chars=100),
+        step_completed(prompt_chars=120, raw_output="x", observation="y"),
+        run_finished(stop_reason="context_overflow"),
+    )
+    outcome = check_prompt_reached_max_prompt_chars(read)
+    assert outcome.status is CriterionStatus.BREACHED
+    assert "被放行去调了模型" in outcome.evidence
+
+
+def test_prompt_reached_max_prompt_chars_reads_the_limit_from_the_snapshot() -> None:
+    """上限从参数快照读，不硬编码：同一条步记录换个上限就该翻面。"""
+    step = step_completed(prompt_chars=90, raw_output="x" * 5, observation="y" * 10)
+    tight = as_read(run_started(max_prompt_chars=100), step)
+    loose = as_read(run_started(max_prompt_chars=100_000), step)
+    assert check_prompt_reached_max_prompt_chars(tight).status is CriterionStatus.PASSED
+    assert check_prompt_reached_max_prompt_chars(loose).status is CriterionStatus.BREACHED
+
+
+def test_prompt_reached_max_prompt_chars_is_undetermined_without_run_started() -> None:
+    read = as_read(step_completed(), run_finished(stop_reason="context_overflow"))
+    outcome = check_prompt_reached_max_prompt_chars(read)
+    assert outcome.status is CriterionStatus.UNDETERMINED
+    assert "run_started" in outcome.evidence
+
+
+def test_prompt_reached_max_prompt_chars_is_undetermined_without_steps() -> None:
+    read = as_read(run_started(), run_finished(stop_reason="context_overflow"))
+    assert check_prompt_reached_max_prompt_chars(read).status is CriterionStatus.UNDETERMINED
+
+
+def test_prompt_reached_max_prompt_chars_is_undetermined_on_a_bad_snapshot() -> None:
+    read = as_read(
+        {
+            "record": "run_started",
+            "run_id": RUN_ID,
+            "parameter_snapshot": {
+                "request.max_prompt_chars": "很多",
+                "request.observation_template": TEMPLATE,
+            },
+            "schema_version": 1,
+        },
+        step_completed(),
+    )
+    assert check_prompt_reached_max_prompt_chars(read).status is CriterionStatus.UNDETERMINED
+
+
+# ---------------------------------------------------------------------------
+# 十六、按上下文现算提示词上限
+# ---------------------------------------------------------------------------
+
+
+def a_request(
+    *,
+    run_level: tuple[str, ...] = ("系统段",),
+    goal_level: tuple[str, ...] = ("题面",),
+    injections: dict[str, tuple[Injection, ...]] | None = None,
+) -> RunRequest:
+    """一份只填了装配用得着那几样的请求。执行器是个不派生自注册表的普通对象，构造期不比对。"""
+
+    class _Executor:
+        def parameters(self) -> dict[str, str]:
+            return {"kind": "fake"}
+
+        async def execute(self, action: object) -> ActionOutcome:  # pragma: no cover - 用不到
+            raise AssertionError("这份请求只用来算提示词规模")
+
+    return RunRequest(
+        run_id=RUN_ID,
+        budget=Budget(
+            max_steps=1, max_actions=1, max_consecutive_parse_failures=1, max_prompt_chars=1
+        ),
+        action_executor=_Executor(),  # type: ignore[arg-type]
+        tools=ToolRegistry(),
+        context=Context(
+            run_level=tuple(
+                Message(role=Role.SYSTEM, content=(TextBlock(text=text),)) for text in run_level
+            ),
+            goal_level=tuple(
+                Message(role=Role.USER, content=(TextBlock(text=text),)) for text in goal_level
+            ),
+        ),
+        injections=injections or {},
+        model_binding={},
+        model_replay_policy=ReplayPolicy.NEVER,
+        observation_template=TEMPLATE,
+        cancel_grace_seconds=1.0,
+    )
+
+
+def test_initial_prompt_chars_counts_every_segment() -> None:
+    request = a_request(
+        run_level=("a" * 10, "b" * 5),
+        goal_level=("c" * 7,),
+        injections={"skill": (Injection(entry_id="e0", content="d" * 3),)},
+    )
+    assert initial_prompt_chars(request) == 25
+
+
+def test_context_overflow_budget_leaves_room_for_exactly_the_first_step() -> None:
+    """上限 = 初始提示词 + 余量。第一步刚好装得下，第二步靠一步的增长撑过去。"""
+    request = a_request(run_level=("x" * 40,), goal_level=("y" * 60,))
+    budget = build_context_overflow_budget(request)
+    assert budget.max_prompt_chars == 100 + CONTEXT_OVERFLOW_SLACK_CHARS
+
+
+def test_initial_prompt_chars_rejects_an_unknown_block_type() -> None:
+    """认不得的块当成 0 会让上限算小，于是第一步就撞上限、这一类什么都验不到。"""
+
+    class _Weird:
+        pass
+
+    request = a_request()
+    broken = replace_context_block(request, _Weird())
+    with pytest.raises(Exception, match="认不得的内容块类型"):
+        initial_prompt_chars(broken)
+
+
+def replace_context_block(request: RunRequest, block: object) -> RunRequest:
+    """把上下文里那条消息的内容块换成给定的东西。造非法输入用。"""
+    import dataclasses
+
+    context = Context(
+        run_level=(Message(role=Role.SYSTEM, content=(block,)),),  # type: ignore[arg-type]
+        goal_level=(),
+    )
+    return dataclasses.replace(request, context=context)
+
+
+# ---------------------------------------------------------------------------
+# 十七、环境故障：判据
+# ---------------------------------------------------------------------------
+
+
+def test_stop_reason_env_error_passes_and_breaches() -> None:
+    good = as_read(step_completed(), run_finished(stop_reason="env_error"))
+    bad = as_read(step_completed(), run_finished(stop_reason="task_completed"))
+    assert check_stop_reason(good, "env_error").status is CriterionStatus.PASSED
+    assert check_stop_reason(bad, "env_error").status is CriterionStatus.BREACHED
+
+
+def test_last_step_action_status_passes() -> None:
+    read = as_read(
+        step_completed(step_idx=0, action_status="executed"),
+        step_completed(step_idx=1, action_status="env_error"),
+    )
+    assert check_last_step_action_status(read, expected="env_error").status is (
+        CriterionStatus.PASSED
+    )
+
+
+def test_last_step_action_status_breaches() -> None:
+    read = as_read(
+        step_completed(step_idx=0, action_status="env_error"),
+        step_completed(step_idx=1, action_status="executed"),
+    )
+    outcome = check_last_step_action_status(read, expected="env_error")
+    assert outcome.status is CriterionStatus.BREACHED
+    assert "executed" in outcome.evidence
+
+
+def test_last_step_action_status_is_undetermined_without_steps() -> None:
+    assert check_last_step_action_status(as_read(run_started()), expected="env_error").status is (
+        CriterionStatus.UNDETERMINED
+    )
+
+
+def test_env_failed_observation_passes_with_the_value_from_the_scenario() -> None:
+    """期望值从场景那份合成观察取，判据这边不抄一份字面量。"""
+    expected = build_synthetic_observations().env_failed
+    read = as_read(
+        step_completed(
+            action_status="env_error", observation=expected, observation_is_synthetic=True
+        )
+    )
+    assert check_last_observation_is_synthetic(read, expected=expected, name="x").status is (
+        CriterionStatus.PASSED
+    )
+
+
+def test_env_failed_observation_breaches_when_the_flag_is_false() -> None:
+    expected = build_synthetic_observations().env_failed
+    read = as_read(
+        step_completed(
+            action_status="env_error", observation=expected, observation_is_synthetic=False
+        )
+    )
+    outcome = check_last_observation_is_synthetic(read, expected=expected, name="x")
+    assert outcome.status is CriterionStatus.BREACHED
+    assert "observation_is_synthetic" in outcome.evidence
+
+
+def test_env_failed_observation_breaches_when_the_executor_text_survived() -> None:
+    """替换没发生的话，历史里躺着的是与环境通信那一层给的原文——那正是这一条要抓的。"""
+    expected = build_synthetic_observations().env_failed
+    read = as_read(
+        step_completed(
+            action_status="env_error",
+            observation="ConnectError: All connection attempts failed",
+            observation_is_synthetic=True,
+        )
+    )
+    outcome = check_last_observation_is_synthetic(read, expected=expected, name="x")
+    assert outcome.status is CriterionStatus.BREACHED
+    assert "ConnectError" not in outcome.evidence
+
+
+def test_env_failed_observation_is_undetermined_without_steps() -> None:
+    assert (
+        check_last_observation_is_synthetic(as_read(run_started()), expected="x", name="x").status
+        is CriterionStatus.UNDETERMINED
+    )
+
+
+def test_steps_before_last_all_executed_passes() -> None:
+    read = as_read(
+        step_completed(step_idx=0, action_status="executed"),
+        step_completed(step_idx=1, action_status="executed"),
+        step_completed(step_idx=2, action_status="env_error"),
+    )
+    assert check_steps_before_last_all_executed(read).status is CriterionStatus.PASSED
+
+
+def test_steps_before_last_all_executed_breaches() -> None:
+    """环境坏掉之前的步被改写或补上别的状态，说明一次局部故障扩散到了已经落地的轨迹上。"""
+    read = as_read(
+        step_completed(step_idx=0, action_status="executed"),
+        step_completed(step_idx=1, action_status="env_error"),
+        step_completed(step_idx=2, action_status="env_error"),
+    )
+    outcome = check_steps_before_last_all_executed(read)
+    assert outcome.status is CriterionStatus.BREACHED
+    assert "[1]" in outcome.evidence
+
+
+def test_steps_before_last_all_executed_is_undetermined_with_a_single_step() -> None:
+    """只有一步的话这一条真空成立，那种「通过」什么都没验。"""
+    outcome = check_steps_before_last_all_executed(as_read(step_completed()))
+    assert outcome.status is CriterionStatus.UNDETERMINED
+    assert "真空成立" in outcome.evidence
+
+
+def test_env_broken_after_a_full_step_passes() -> None:
+    outcome = check_env_broken_after_a_full_step(broken=True, executions_before_break=1)
+    assert outcome.status is CriterionStatus.PASSED
+
+
+def test_env_broken_after_a_full_step_is_undetermined_when_it_never_broke() -> None:
+    outcome = check_env_broken_after_a_full_step(broken=False, executions_before_break=0)
+    assert outcome.status is CriterionStatus.UNDETERMINED
+    assert "一次都没被弄坏" in outcome.evidence
+
+
+def test_env_broken_after_a_full_step_is_undetermined_when_it_broke_too_early() -> None:
+    """第一次执行之前就打死容器的话，压到的是初始化而不是动作执行接缝。"""
+    outcome = check_env_broken_after_a_full_step(broken=True, executions_before_break=0)
+    assert outcome.status is CriterionStatus.UNDETERMINED
+    assert "初始化" in outcome.evidence
+
+
+# ---------------------------------------------------------------------------
+# 十八、环境故障：杀容器那段编排里的纯函数与执行器包装
+# ---------------------------------------------------------------------------
+
+
+def test_container_name_for_port() -> None:
+    assert container_name_for_port(8201) == "polyloop-soak-appworld-8201"
+
+
+def test_should_break_env_waits_for_a_full_execution() -> None:
+    assert (
+        should_break_env(executions_done=0, break_after_executions=1, already_broken=False) is False
+    )
+    assert (
+        should_break_env(executions_done=1, break_after_executions=1, already_broken=False) is True
+    )
+
+
+def test_should_break_env_never_breaks_twice() -> None:
+    """容器已经没了，再发一次 docker kill 只会拿到一个「没有这个容器」的错误。"""
+    assert (
+        should_break_env(executions_done=5, break_after_executions=1, already_broken=True) is False
+    )
+
+
+class _FakeExecutor:
+    """按剧本一次次返回结果或抛异常的假执行器。"""
+
+    def __init__(self, script: list[object]) -> None:
+        self.script = script
+        self.calls = 0
+
+    def parameters(self) -> dict[str, str]:
+        return {"kind": "fake_appworld"}
+
+    async def execute(self, action: object) -> ActionOutcome:
+        item = self.script[self.calls]
+        self.calls += 1
+        if isinstance(item, BaseException):
+            raise item
+        return item  # type: ignore[return-value]
+
+
+def an_outcome(status: ActionStatus = ActionStatus.EXECUTED) -> ActionOutcome:
+    return ActionOutcome(
+        status=status,
+        observation="环境返回",
+        observation_is_synthetic=False,
+        env_reported_completion=False,
+        observation_truncated_chars=0,
+    )
+
+
+async def test_env_breaking_executor_breaks_after_one_full_execution() -> None:
+    """第一次执行之前不动手，第二次执行之前动手。"""
+    broke_at: list[int] = []
+    inner = _FakeExecutor([an_outcome(), httpx.ConnectError("All connection attempts failed")])
+
+    async def break_env() -> str:
+        broke_at.append(inner.calls)
+        return "容器已被打死"
+
+    executor = EnvBreakingExecutor(inner=inner, break_after_executions=1, break_env=break_env)
+    first = await executor.execute(object())
+    assert first.status is ActionStatus.EXECUTED
+    assert broke_at == []
+
+    second = await executor.execute(object())
+    assert broke_at == [1]
+    assert executor.broken is True
+    assert executor.executions_before_break == 1
+    assert executor.break_note == "容器已被打死"
+    assert second.status is ActionStatus.ENV_ERROR
+
+
+async def test_env_breaking_executor_translates_a_transport_error() -> None:
+    """场景那侧只接 AppWorldError，容器没了时抛的是 httpx.ConnectError，不翻译就整个抛出去。"""
+    inner = _FakeExecutor([httpx.ConnectError("All connection attempts failed")])
+
+    async def break_env() -> str:
+        return "打死了"
+
+    executor = EnvBreakingExecutor(inner=inner, break_after_executions=0, break_env=break_env)
+    outcome = await executor.execute(object())
+    assert outcome.status is ActionStatus.ENV_ERROR
+    assert outcome.observation_is_synthetic is False
+    assert "ConnectError" in outcome.observation
+
+
+async def test_env_breaking_executor_passes_an_inner_env_error_through() -> None:
+    """内层自己判出环境故障时原样透传：场景那边哪天补上转换，这里不用跟着改。"""
+    inner = _FakeExecutor([an_outcome(ActionStatus.ENV_ERROR)])
+
+    async def break_env() -> str:  # pragma: no cover - 这条用例不动手
+        raise AssertionError("不该动手")
+
+    executor = EnvBreakingExecutor(inner=inner, break_after_executions=99, break_env=break_env)
+    outcome = await executor.execute(object())
+    assert outcome.status is ActionStatus.ENV_ERROR
+    assert outcome.observation == "环境返回"
+
+
+async def test_env_breaking_executor_reports_its_probe_in_parameters() -> None:
+    inner = _FakeExecutor([])
+
+    async def break_env() -> str:  # pragma: no cover - 这条用例不执行动作
+        raise AssertionError("不该动手")
+
+    executor = EnvBreakingExecutor(inner=inner, break_after_executions=1, break_env=break_env)
+    assert executor.parameters() == {"kind": "fake_appworld", "env_break_probe": "docker_kill"}
+
+
+# ---------------------------------------------------------------------------
+# 十九、两类新故障接进命令行
+# ---------------------------------------------------------------------------
+
+
+def test_new_faults_are_in_the_default_selection() -> None:
+    """不给 --fault 时按 FAULT_NAMES 全跑，两类都得在里面。"""
+    assert "context_overflow" in FAULT_NAMES
+    assert "env_error" in FAULT_NAMES
+
+
+def test_parser_accepts_the_new_faults() -> None:
+    args = build_parser().parse_args(
+        [
+            "--runs-dir",
+            "x",
+            "--budget-calls",
+            "10",
+            "--fault",
+            "context_overflow",
+            "--fault",
+            "env_error",
+        ]
+    )
+    assert args.fault == ["context_overflow", "env_error"]