From 67f0d355d0fbd52e869ebe5161609d1863824f4e Mon Sep 17 00:00:00 2001 From: iomgaa Date: Tue, 11 Aug 2026 11:14:04 -0400 Subject: [PATCH] =?UTF-8?q?feat(soak):=20=E8=A1=A5=E4=B8=8A=20context=5Fov?= =?UTF-8?q?erflow=20=E4=B8=8E=20env=5Ferror=20=E4=B8=A4=E7=B1=BB=EF=BC=8C?= =?UTF-8?q?=E4=B8=83=E7=B1=BB=E5=8F=98=E4=B9=9D=E7=B1=BB?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 一次 193 个运行的全量跑完之后，停止原因的十个取值里有两个一次都没出现过。没出现不等于 它们是对的，只等于没验过——这正是「全绿要先怀疑负载」该指向的地方。 context_overflow 的判据不能照字面写成「最后一步的提示词超过上限」：规模判定在调模型之前 做，命中时不产生步记录，所以落盘的每条步记录必定不超上限，那样断言等于断言契约的反面。 改成判「再走一步会有多大」，公式拿全量里 865 对相邻步验过，0 处不符。 env_error 是真把容器 docker kill 掉，不是用测试替身。它自己起一个池、用另一个端口—— 共用那个 size=1 的池的话，排在它后面的每一类都会跑在一个不存在的环境上。 实跑：两类都通过，击穿 0、无法判定 0。 Co-Authored-By: Claude Opus 5 (1M context) --- tools/soak/faults.py | 835 +++++++++++++++++++++++++++++--- tools/soak/tests/test_faults.py | 556 ++++++++++++++++++++- 2 files changed, 1330 insertions(+), 61 deletions(-) diff --git a/tools/soak/faults.py b/tools/soak/faults.py index 5c47916..0a9d6a2 100644 --- a/tools/soak/faults.py +++ b/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_FAULTS = frozenset({"crash_resume_a", "crash_resume_b"}) +#: 用 GovDoc 场景做的那三类。其余用 AppWorld。 +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,27 +2492,44 @@ 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_corpus = _require(args, "govdoc-corpus", "崩溃续跑要读语料") + govdoc_db = _require(args, "govdoc-db", "GovDoc 那几类要读审核点") + 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 - reports.append( - await guarded( - name, - run_crash_fault( - fault=name, - timing=timing, - runs_dir=runs_dir, - workspace_root=runs_dir / "workspaces", - govdoc_db=govdoc_db, - govdoc_corpus=govdoc_corpus, - model_client=model_client, - guard=guard, - attempts=args.attempts, - self_kill_grace_s=args.self_kill_grace, - ), + coroutine: Awaitable[FaultReport] + if name == "context_overflow": + if not guard.affordable(CONTEXT_OVERFLOW_BUDGET_STEPS): + reports.append( + _unaffordable( + name, guard=guard, worst_case=CONTEXT_OVERFLOW_BUDGET_STEPS + ) + ) + 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, + workspace_root=runs_dir / "workspaces", + govdoc_db=govdoc_db, + govdoc_corpus=govdoc_corpus, + model_client=model_client, + 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: - 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) + 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 = { - "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] + worst_case = _APPWORLD_WORST_CASE_CALLS[name] if not guard.affordable(worst_case): - reports.append( - FaultReport( - fault=name, - criteria=( - undetermined( - "call_budget_available", - f"调用数护栏只剩 {guard.remaining} 次，最坏要 {worst_case} 次，这一类没跑", - ), - ), - ) - ) + reports.append(_unaffordable(name, guard=guard, worst_case=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", diff --git a/tools/soak/tests/test_faults.py b/tools/soak/tests/test_faults.py index 3d845f5..78f2c48 100644 --- a/tools/soak/tests/test_faults.py +++ b/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", - "content_chars": 1, + "raw_output": raw_output, + "content_chars": len(raw_output), "thinking_chars": 0, "action": None, "parse_ok": parse_ok, "parse_error": None if parse_ok else "解释不了", - "observation": "o", - "observation_is_synthetic": False, + "observation": observation, + "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"]