From 43971573b71df4ae24bb0bdcaf39ff386398c70f Mon Sep 17 00:00:00 2001 From: iomgaa Date: Tue, 11 Aug 2026 09:05:05 -0400 Subject: [PATCH] =?UTF-8?q?feat(soak):=20=E6=95=85=E9=9A=9C=E6=B3=A8?= =?UTF-8?q?=E5=85=A5=E2=80=94=E2=80=94=E5=B4=A9=E6=BA=83=E7=BB=AD=E8=B7=91?= =?UTF-8?q?=E3=80=81=E5=8F=96=E6=B6=88=E3=80=81=E6=92=9E=E9=A2=84=E7=AE=97?= =?UTF-8?q?=E3=80=81=E8=A7=A3=E6=9E=90=E5=A4=B1=E8=B4=A5=E8=BF=9E=E5=87=BB?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 这七类是压测真正要看的东西：一百个任务顺利跑完什么都证明不了，能证明东西的是这些没有 失败现场的路径上库有没有守住承诺。 判据一条都不依赖模型的确定性，全是结构不变量。最硬的那条是「声明绝不重放的动作没有被 执行两次」——证据取自环境侧自己记的账（工作区的审计文件），不取库报的步数或动作数， 后者是库对自己行为的陈述，拿它验库的行为就是我们和我们自己对账。 崩溃是真 SIGKILL 子进程，不是模拟的注入点，而且分两种时机：一步完整落地之后崩（续跑应 该真的接着跑），以及意图落盘、结果还没落盘时崩（那条意图声明绝不重放，库应该判定状态 未知、干净停下）。第二种的命中条件是「最后一条意图的重放策略是 never」而不是「最后一条 是意图」——只读工具声明的是 safe，悬在那种意图上续跑会重放接着走，判据会时对时错，而错 的那几次看起来只像模型走了别的路。 判定和记分板一样分三档，「无法判定」不折算成通过：审计账为空时「去重前后条数相等」是真 空成立的，报成通过等于把「什么都没验到」显示成绿。命中不了时机也报无法判定，不降级成 另一种时机假装验过。 调用数护栏按每类故障的边界拦，不在模型客户端里抛异常——在那里抛的话库会把它记成模型调用 失败、合成观察接着跑，护栏本身就成了一次注入进来的故障，把要验的停止原因搅乱了。 Co-Authored-By: Claude Opus 5 (1M context) --- tools/soak/faults.py | 1771 +++++++++++++++++++++++++++++++ tools/soak/tests/test_faults.py | 810 ++++++++++++++ 2 files changed, 2581 insertions(+) create mode 100644 tools/soak/faults.py create mode 100644 tools/soak/tests/test_faults.py diff --git a/tools/soak/faults.py b/tools/soak/faults.py new file mode 100644 index 0000000..2bb5e28 --- /dev/null +++ b/tools/soak/faults.py @@ -0,0 +1,1771 @@ +"""压测的故障注入：在崩溃、取消、撞预算、解析连击四条路径上验库有没有守住承诺。 + +一百个任务顺利跑完什么都证明不了——顺利那条路上库只要不崩就算过。能证明东西的是这四条： +进程被杀在半路、调用方中途取消、目标根本完不成、模型的输出一句都解析不了。库在这些地方给 +下游的承诺（崩溃前的轨迹逐字节不变、声明绝不重放的动作不会执行两次、取消原样穿透且资源 +归还、撞上限时以正确的停止原因干净收尾、解析失败不碰环境）只有在这里才检验得到。 + +**崩溃续跑用 GovDoc 场景，不用 AppWorld。** GovDoc 的「环境」是一个工作区目录加一份审计 +日志（`tools/soak/scenarios/govdoc.py` 里 `GovDocTools._append_audit` 写的那份），它天然活过 +进程的死亡，而且给得出**环境侧自己数的实际执行次数**——那是「声明绝不重放的动作有没有被 +执行两次」唯一可信的证据。AppWorld 的环境活在容器里：子进程被 SIGKILL 之后容器的归属与 +清理会变复杂，而它的执行计数存在会话对象里，会随进程一起消失，于是只剩「库自己报的步数」 +可用，那等于我们和我们自己对账。 + +**判据一律不依赖模型的确定性。** 活模型下「两次运行产出同样的轨迹」本来就不成立，断言它 +只会随机变红。这里的判据全是结构不变量：字节前缀、步号连续性、意图有没有归宿、审计账去重 +前后的条数、停止原因的取值、步数与上限的关系、环境侧的执行计数。 + +**产物照 `tools/soak/scoreboard.py` 的 sidecar 约定写出**（`.result.json` / `.events.jsonl` / +`.meta.json`），`meta.fault` 填故障名，这样记分板能把它们和正常批次一起判。崩溃那两类天然 +缺 sidecar——被 SIGKILL 的子进程来不及写，记分板会报「无法判定」，那是对的，不为了让它变绿 +去补假数据。 + +跑法（会打真实模型、会花钱）:: + + PYTHONUNBUFFERED=1 conda run --live-stream -n PolyLoop python -m tools.soak.faults \\ + --runs-dir <目录> --budget-calls 60 \\ + --data-root --govdoc-db --govdoc-corpus <语料目录> +""" + +from __future__ import annotations + +import argparse +import asyncio +import json +import sys +import time +from collections.abc import Awaitable, Mapping, Sequence +from dataclasses import dataclass, replace +from enum import StrEnum +from pathlib import Path + +from polyloop import session +from polyloop.adapters import GatewayModelClient +from polyloop.ports import Event, InvalidDecision, ParsedReply +from polyloop.serialization import encode +from polyloop.session import AgentDefinition, ParameterDriftError, RunRequest +from polyloop.stores import RECORD_KEY, JsonlRunStore +from polyloop.types import Budget, ModelReply, RunResult +from tools.soak.appworld import 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 +from tools.soak.scenarios.govdoc import ( + AuditTask, + GovDocParser, + Redactor, + load_checkpoints, + load_documents, + read_audit_lines, +) + +#: 仓库根目录。子进程要在这里启动，否则 `tools.soak` 这个命名空间包 import 不到。 +REPO_ROOT = Path(__file__).resolve().parents[2] + +FAULT_NAMES: tuple[str, ...] = ( + "crash_resume_a", + "crash_resume_b", + "cancel_model", + "cancel_env", + "step_budget", + "action_budget", + "parse_failures", +) + +#: 用 GovDoc 场景做的那两类。其余用 AppWorld。 +GOVDOC_FAULTS = frozenset({"crash_resume_a", "crash_resume_b"}) +APPWORLD_FAULTS = frozenset(FAULT_NAMES) - GOVDOC_FAULTS + + +class FaultInjectionError(RuntimeError): + """故障注入这一层装配不出来或跑不下去：缺数据、缺参数、子进程起不来。 + + 它不表示某条判据被击穿——那种事由 `Criterion` 记录，不抛异常。 + """ + + +# --------------------------------------------------------------------------- +# 一、判据与报告 +# --------------------------------------------------------------------------- + + +class CriterionStatus(StrEnum): + """一条判据的判定。三档，不是两档。 + + 第三档独立存在，不许折算成前两档中的任何一个。判不了和判过了是两回事——压成一档的话， + 一次什么都没验成的跑会显示成全绿，而那正是最需要被看见的情况。理由与 + `tools/soak/scoreboard.py` 的三档判定相同。 + """ + + PASSED = "passed" + BREACHED = "breached" + UNDETERMINED = "undetermined" + + +@dataclass(frozen=True, slots=True, kw_only=True) +class Criterion: + """一条判据的判定加它的证据。 + + **证据是必填的**，由构造期校验守住：一条只会说「有问题」或者只会说「没问题」的判据等于 + 没有判据。通过也要给证据，因为「通过」最常见的坏法是判据根本没跑到该判的东西上——比了 + 零个字节、数了零条审计、看了一个空列表，那些情况下证据文本会当场露馅。 + """ + + name: str + status: CriterionStatus + evidence: str + + def __post_init__(self) -> None: + if not self.name.strip(): + raise ValueError("判据必须有名字") + if not self.evidence.strip(): + raise ValueError(f"判据 {self.name!r} 没有给证据") + + def describe(self) -> str: + return f"[{self.status.value}] {self.name}：{self.evidence}" + + +def passed(name: str, evidence: str) -> Criterion: + return Criterion(name=name, status=CriterionStatus.PASSED, evidence=evidence) + + +def breached(name: str, evidence: str) -> Criterion: + return Criterion(name=name, status=CriterionStatus.BREACHED, evidence=evidence) + + +def undetermined(name: str, evidence: str) -> Criterion: + return Criterion(name=name, status=CriterionStatus.UNDETERMINED, evidence=evidence) + + +@dataclass(frozen=True, slots=True, kw_only=True) +class FaultReport: + """一类故障跑完之后的全部结论。""" + + fault: str + criteria: tuple[Criterion, ...] + #: 不构成判定、但必须被人看见的事实：重试了几次、撕裂尾行、跳过的原因。 + notes: tuple[str, ...] = () + + @property + def breaches(self) -> tuple[Criterion, ...]: + return tuple(item for item in self.criteria if item.status is CriterionStatus.BREACHED) + + @property + def undetermineds(self) -> tuple[Criterion, ...]: + return tuple(item for item in self.criteria if item.status is CriterionStatus.UNDETERMINED) + + def render(self) -> str: + lines = [f"## {self.fault}"] + lines += [f"- {item.describe()}" for item in self.criteria] + lines += [f"- (note) {note}" for note in self.notes] + return "\n".join(lines) + + +# --------------------------------------------------------------------------- +# 二、日志读取：逐行读，撕裂尾行按「没发生过」算 +# --------------------------------------------------------------------------- + + +@dataclass(frozen=True, slots=True, kw_only=True) +class LogRead: + """一份 `.jsonl` 日志按行读回来的结果。 + + `torn` 说的是末尾有没有一段没被换行终结的字节。判据照存储那边的规矩：**看有没有被换行 + 终结，不看能不能解析**（`polyloop.stores` 的 `_parse`）。那段字节对应的那次写从来没有被 + 确认过，按契约它就是没发生。 + """ + + payloads: tuple[Mapping[str, object], ...] + torn: bool + #: 已经被换行终结、却解不出 JSON 对象的行号。它们是真问题，不是撕裂尾行。 + bad_lines: tuple[int, ...] + + +def terminated_prefix(raw: bytes) -> bytes: + """截到最后一个换行为止的那一段。没有换行就是空。 + + 崩溃快照与最终日志的字节比对只能比这一段：末尾那段没有换行的字节没被确认过，而续跑会 + 直接追加在它后面（`JsonlRunStore` 用 `O_APPEND`，不回退文件指针），于是那一行在最终 + 日志里长得和崩溃时不一样——那不是承诺被破坏，是那次写从来没算数。 + """ + cut = raw.rfind(b"\n") + return b"" if cut < 0 else raw[: cut + 1] + + +def parse_terminated(raw: bytes) -> LogRead: + """把日志字节切成一条条记录载荷。""" + chunks = raw.split(b"\n") + torn = bool(chunks) and bool(chunks[-1].strip()) + body = chunks[:-1] if torn else chunks + payloads: list[Mapping[str, object]] = [] + bad: list[int] = [] + for number, chunk in enumerate(body, start=1): + if not chunk.strip(): + continue + try: + value = json.loads(chunk) + except (UnicodeDecodeError, json.JSONDecodeError): + bad.append(number) + continue + if not isinstance(value, dict) or RECORD_KEY not in value: + bad.append(number) + continue + payloads.append(value) + return LogRead(payloads=tuple(payloads), torn=torn, bad_lines=tuple(bad)) + + +def read_log(path: Path) -> LogRead: + if not path.is_file(): + return LogRead(payloads=(), torn=False, bad_lines=()) + return parse_terminated(path.read_bytes()) + + +def tagged(read: LogRead, tag: str) -> tuple[Mapping[str, object], ...]: + return tuple(item for item in read.payloads if item.get(RECORD_KEY) == tag) + + +def count_model_calls(read: LogRead) -> int: + """这份日志里发起过几次模型调用。 + + 数的是模型调用意图，不是步数：意图在调用发出去**之前**落盘，所以被杀在调用中途的那次 + 也算，而它已经花过钱了。用步数会把那次漏掉。 + """ + return sum(1 for item in tagged(read, "intent") if item.get("kind") == "model_call") + + +def stop_reason_of(read: LogRead) -> str | None: + """日志里结束记录带的停止原因。没有结束记录就是 None。""" + finished = tagged(read, "run_finished") + if not finished: + return None + result = finished[-1].get("result") + if not isinstance(result, Mapping): + return None + value = result.get("stop_reason") + return value if isinstance(value, str) else None + + +def _step_payloads(read: LogRead) -> tuple[Mapping[str, object], ...]: + steps: list[Mapping[str, object]] = [] + for item in tagged(read, "step_completed"): + step = item.get("step") + if isinstance(step, Mapping): + steps.append(step) + return tuple(steps) + + +# --------------------------------------------------------------------------- +# 三、判据：崩溃续跑 +# --------------------------------------------------------------------------- + + +def check_log_readable(read: LogRead) -> Criterion: + """每一条被换行终结的行都解得出一个带类型标签的 JSON 对象。 + + 解不出来的行让下面几条判据判的东西缺一块，所以它先判——不先判的话，一份少了半截的日志 + 会让「步号连续」这类判据在残缺数据上给出「通过」。 + """ + if read.bad_lines: + return breached( + "log_lines_readable", + f"第 {list(read.bad_lines)} 行已被换行终结却解不出带 {RECORD_KEY!r} 标签的 JSON 对象", + ) + return passed("log_lines_readable", f"{len(read.payloads)} 条记录全部解得开") + + +def check_crash_prefix_preserved(*, crashed: bytes, final: bytes) -> Criterion: + """崩溃前已经落地的那些字节，续跑之后逐字节没变。 + + **比的是字节串，不是解析出来的对象。** 比对象只能说明「语义等价」，而承诺是更硬的那一 + 条：已经写下去的记录不许被重写、不许被改格式、不许被补字段。这两者的差别只有在字节层面 + 看得见。 + """ + prefix = terminated_prefix(crashed) + if not prefix: + return undetermined( + "crash_prefix_preserved", + "崩溃快照里没有任何被换行终结的记录，前缀比对无从谈起", + ) + if len(final) < len(prefix): + return breached( + "crash_prefix_preserved", + f"最终日志只有 {len(final)} 字节，比崩溃时已终结的 {len(prefix)} 字节还短", + ) + head = final[: len(prefix)] + if head != prefix: + offset = next( + ( + index + for index, pair in enumerate(zip(prefix, head, strict=True)) + if pair[0] != pair[1] + ), + len(prefix), + ) + return breached( + "crash_prefix_preserved", + f"崩溃时已终结的 {len(prefix)} 字节里，第 {offset} 字节起与最终日志不同", + ) + return passed( + "crash_prefix_preserved", + f"崩溃时已终结的 {len(prefix)} 字节在最终日志里逐字节相同", + ) + + +def check_step_indices_dense(read: LogRead) -> Criterion: + """`step_completed` 的步号从 0 开始、逐 1 递增、不重不跳。 + + 重号意味着同一步被写了两遍（续跑把已经落地的一步又跑了一次），跳号意味着中间某一步的 + 原子写整个丢了。两者都会让重建出来的历史与不中断跑完时不一样，而那件事在轨迹里看不出来。 + """ + indices: list[object] = [step.get("step_idx") for step in _step_payloads(read)] + if not indices: + return undetermined("step_idx_dense", "日志里一条步记录都没有，步号连续性无从判起") + bad = [value for value in indices if not isinstance(value, int) or isinstance(value, bool)] + if bad: + return breached("step_idx_dense", f"有 {len(bad)} 条步记录的 step_idx 不是整数") + expected = list(range(len(indices))) + if indices != expected: + return breached("step_idx_dense", f"步号按文件顺序是 {indices}，期望 {expected}") + return passed("step_idx_dense", f"{len(indices)} 条步记录的步号是 0 到 {len(indices) - 1}") + + +def check_intents_settled(read: LogRead) -> Criterion: + """每条意图都有归宿；至多一条悬空，且必须是最后一条意图。 + + 模型调用意图的归宿是一条同 `result_id` 的模型调用结果，动作意图的归宿是一条同 `result_id` + 的步记录。**允许最后一条悬空**：崩溃点上那条意图写了、结果没写，续跑判成状态未知干净 + 停下之后它就永远悬在那儿，那是合法终态。中间悬空则不同——它说明有一步的执行状态被跳过 + 去了，而后面的步是建立在「那一步到底做没做」这个没有答案的问题上的。 + """ + intents = tagged(read, "intent") + if not intents: + return undetermined("intents_settled", "日志里一条意图都没有，归宿无从判起") + model_results = {item.get("result_id") for item in tagged(read, "model_call_result")} + action_results = {item.get("result_id") for item in tagged(read, "step_completed")} + dangling: list[int] = [] + for position, intent in enumerate(intents): + settled = ( + intent.get("result_id") in model_results + if intent.get("kind") == "model_call" + else intent.get("result_id") in action_results + ) + if not settled: + dangling.append(position) + if not dangling: + return passed("intents_settled", f"{len(intents)} 条意图全部有归宿，没有悬空") + last = len(intents) - 1 + if dangling == [last]: + kind = intents[last].get("kind") + return passed( + "intents_settled", + f"{len(intents)} 条意图里只有最后一条（kind={kind}）悬空，那是崩溃点，合法", + ) + return breached( + "intents_settled", + f"{len(intents)} 条意图里第 {dangling} 条悬空（按意图出现的次序计），" + f"只有第 {last} 条允许悬空", + ) + + +def parse_audit_line(line: str) -> tuple[str, str, str] | None: + """把审计账的一行切成 `(工具名, 文件名, 内容摘要)`。切不出来返回 None。 + + 格式来自 `GovDocTools._append_audit`：三段用制表符分隔。 + """ + parts = line.split("\t") + if len(parts) != 3 or not all(part for part in parts): + return None + return parts[0], parts[1], parts[2] + + +def check_never_action_not_replayed(audit_lines: Sequence[str]) -> Criterion: + """声明绝不重放的动作没有被执行两次。 + + **证据取自环境侧自己记的账**（工作区里的 `.write_audit.log`），不取库报的步数或动作数—— + 后者是库对自己行为的陈述，用它来验库的行为就是我们和我们自己对账。真正的重放长这样： + 库在状态未知时把一次已经落过盘的写又执行了一遍，于是环境的账上多出一条一模一样的记录， + 而轨迹里看不出任何异常。 + + 判据是「按 (工具, 文件名, 内容摘要) 去重前后的条数相等」。**它有一种已知的假阳性**： + 模型自己把同一份内容原样写了两次，账上也会出现两条一样的记录。GovDoc 的 execute 阶段 + 提示词明确要求写完 evidence.md 就停下，所以这件事很少发生；真发生了要看的是轨迹里那两 + 步的步号——重放来自续跑，两条记录会分属崩溃前后。 + """ + entries = [parse_audit_line(line) for line in audit_lines] + broken = [index for index, entry in enumerate(entries, start=1) if entry is None] + if broken: + return undetermined( + "never_action_not_replayed", + f"审计账第 {broken} 行不是「工具\\t文件名\\t摘要」三段，数不出实际执行次数", + ) + if not entries: + return undetermined( + "never_action_not_replayed", + "审计账是空的：这次运行里没有任何有副作用的动作被执行过，去重比对无从谈起", + ) + kept = [entry for entry in entries if entry is not None] + unique = set(kept) + if len(unique) == len(kept): + return passed( + "never_action_not_replayed", + f"审计账 {len(kept)} 条，按 (工具, 文件名, 内容摘要) 去重后仍是 {len(unique)} 条", + ) + counts: dict[tuple[str, str, str], int] = {} + for entry in kept: + counts[entry] = counts.get(entry, 0) + 1 + repeated = sorted( + ( + f"{tool} 摘要 {digest} 出现 {times} 次" + for (tool, _name, digest), times in counts.items() + if times > 1 + ) + ) + return breached( + "never_action_not_replayed", + f"审计账 {len(kept)} 条，去重后只剩 {len(unique)} 条：{'；'.join(repeated)}", + ) + + +def check_audit_unchanged(*, before: Sequence[str], after: Sequence[str]) -> Criterion: + """续跑之后环境的账一条都没多。 + + 时机 B 下库判定状态未知、干净停下，那就不该把被打断的那个动作再执行一遍。多出一条就是 + 重放，少一条或者变了内容说明账被改写过。 + """ + if list(after[: len(before)]) != list(before): + return breached( + "audit_unchanged_after_resume", + f"续跑前 {len(before)} 条审计记录在续跑后不再是原来那几条", + ) + if len(after) != len(before): + return breached( + "audit_unchanged_after_resume", + f"续跑前审计账 {len(before)} 条，续跑后 {len(after)} 条，多出 {len(after) - len(before)} 条", + ) + return passed("audit_unchanged_after_resume", f"续跑前后审计账都是 {len(before)} 条") + + +def check_resume_made_progress(*, crashed_steps: int, final_steps: int) -> Criterion: + """时机 A 下续跑真的接着往下跑了。 + + 一步完整落地之后崩，日志处在「上一步是完整的」这个状态，续跑该从下一步的开头接着走。 + 步数没长说明它没接着跑——那种「续跑」只是把旧结果读回来又写了一遍结束记录。 + """ + if final_steps > crashed_steps: + return passed( + "resume_made_progress", + f"崩溃时 {crashed_steps} 步，续跑之后 {final_steps} 步", + ) + return breached( + "resume_made_progress", + f"崩溃时 {crashed_steps} 步，续跑之后仍是 {final_steps} 步，没有接着往下跑", + ) + + +# --------------------------------------------------------------------------- +# 四、判据：停止原因、步数、取消、环境 +# --------------------------------------------------------------------------- + + +def check_stop_reason(read: LogRead, expected: str) -> Criterion: + """结束记录在场，且它带的停止原因是期望的那个。 + + 判据取自**日志里的结束记录**而不是 `run` 的返回值：取消那条路径上 `run` 原样重抛 + `CancelledError`、根本不返回结果，返回值那侧什么都拿不到，而结束记录仍然在。 + """ + actual = stop_reason_of(read) + if actual is None: + return breached( + f"stop_reason_is_{expected}", "日志里没有 run_finished 记录，读不到停止原因" + ) + if actual != expected: + return breached(f"stop_reason_is_{expected}", f"停止原因是 {actual}，期望 {expected}") + return passed(f"stop_reason_is_{expected}", f"日志的 run_finished 里停止原因是 {actual}") + + +def check_step_count(read: LogRead, *, expected: int, name: str) -> Criterion: + actual = len(_step_payloads(read)) + if actual != expected: + return breached(name, f"日志里 {actual} 条步记录，期望正好 {expected} 条") + return passed(name, f"日志里正好 {expected} 条步记录") + + +def check_executed_action_count(read: LogRead, *, expected: int) -> Criterion: + """真正执行成功的动作数正好等于动作上限。 + + 数的是步记录里 `action_outcome.status == "executed"` 的条数，与 `_stopping` 那侧 + `actions_executed` 的口径一致：被拒绝与环境故障都不计入动作预算。 + """ + executed = 0 + for item in tagged(read, "step_completed"): + outcome = item.get("action_outcome") + if isinstance(outcome, Mapping) and outcome.get("status") == "executed": + executed += 1 + if executed != expected: + return breached( + "executed_actions_equal_max_actions", + f"执行成功的动作 {executed} 次，期望正好 {expected} 次", + ) + return passed("executed_actions_equal_max_actions", f"执行成功的动作正好 {expected} 次") + + +def check_no_env_error_step(read: LogRead) -> Criterion: + """轨迹里没有环境故障的步。 + + 撞预算那两条要的是「干净地撞上限」。中途出过环境故障的话，步数与动作数的账仍然对得上， + 但这次跑压到的已经不是预算这条路径了。 + """ + entries = tagged(read, "step_completed") + bad: list[int] = [] + for index, item in enumerate(entries): + outcome = item.get("action_outcome") + if isinstance(outcome, Mapping) and outcome.get("status") == "env_error": + bad.append(index) + if bad: + return breached("no_env_error_step", f"第 {bad} 步（按步记录次序计）的动作状态是 env_error") + return passed("no_env_error_step", f"{len(entries)} 步里没有 env_error") + + +def check_all_steps_parse_failed(read: LogRead) -> Criterion: + """这几步全部解析失败，且一个动作都没被分发。 + + `action_status` 为空是「这一步压根没走到动作那一档」的形态：解析失败那一支直接跳过完成 + 判定与动作执行（`polyloop.session` 的 D 档）。它要是有值，说明有动作被分发过。 + """ + steps = _step_payloads(read) + if not steps: + return undetermined("all_steps_parse_failed", "日志里一条步记录都没有") + bad_parse = [index for index, step in enumerate(steps) if step.get("parse_ok") is not False] + bad_action = [ + index for index, step in enumerate(steps) if step.get("action_status") is not None + ] + if bad_parse or bad_action: + return breached( + "all_steps_parse_failed", + f"第 {bad_parse} 步的 parse_ok 不是 False，第 {bad_action} 步的 action_status 不为空", + ) + return passed( + "all_steps_parse_failed", + f"{len(steps)} 步全部 parse_ok=False 且 action_status 为空", + ) + + +def check_env_untouched(*, executions: int, source: str) -> Criterion: + """环境侧一次都没被执行过。 + + 解析失败不该碰环境，这是契约。计数来自环境侧（AppWorld 的 `n_executions` 或 GovDoc 的 + 审计条数），不来自轨迹。 + """ + if executions != 0: + return breached("env_untouched", f"{source} 报环境被执行了 {executions} 次，期望 0 次") + return passed("env_untouched", f"{source} 报环境一次都没被执行") + + +def check_cancelled_raised(*, raised: BaseException | None) -> Criterion: + """`await task` 抛的是 `CancelledError`，原样抛出。 + + 吞掉之后返回一个结果，调用方的结构化并发就断了：它以为这次运行正常结束，而它其实是被 + 自己叫停的。 + """ + if raised is None: + return breached("cancelled_error_propagated", "取消之后 await 正常返回了结果，没有抛异常") + if not isinstance(raised, asyncio.CancelledError): + return breached( + "cancelled_error_propagated", + f"取消之后 await 抛的是 {type(raised).__name__}，不是 CancelledError", + ) + return passed("cancelled_error_propagated", "取消之后 await 原样抛出了 CancelledError") + + +def check_lease_returned(*, borrowed: bool, timeout_s: float, pool_size: int) -> Criterion: + """容器租约被归还：取消结束之后还借得到。 + + 池满时 `lease()` 会一直阻塞，所以「借得到」等价于「空闲名额回到了满」。池的大小是 1， + 于是这一借要么立刻成功、要么永远等下去，中间没有含糊地带。 + """ + if not borrowed: + return breached( + "container_lease_returned", + f"取消之后再借一个容器，{timeout_s} 秒内没借到（池大小 {pool_size}），租约没还回来", + ) + return passed( + "container_lease_returned", + f"取消之后在 {timeout_s} 秒内又借到了容器（池大小 {pool_size}），租约还回来了", + ) + + +# --------------------------------------------------------------------------- +# 五、子进程编排：轮询日志尾部，按时机 SIGKILL +# --------------------------------------------------------------------------- + + +class KillTiming(StrEnum): + """在哪个时机把子进程杀掉。两种时机的判据不同，不许混成一个用例。""" + + #: 时机 A：最后一条是 `step_completed`，一步完整落地之后崩。续跑该真的接着往下跑。 + AFTER_STEP = "after_step" + #: 时机 B：最后一条是一条声明绝不重放的意图，意图写了、结果还没写。续跑该判状态未知、 + #: 干净停下。 + AT_INTENT = "at_intent" + + +def should_kill(read: LogRead, *, timing: KillTiming, after_steps: int) -> bool: + """现在这份日志尾部是不是要等的那个时机。纯函数，轮询循环每次拿它问一句。 + + 时机 B 额外要求那条意图的重放策略是 `never`。**这比「最后一条是意图」更严**，而且必须 + 更严：GovDoc 的 `read_document` 与 `grep_document` 声明的是 `safe`，悬在那种意图上续跑 + 会重放动作接着跑，停止原因不是 `resume_state_unknown`。放宽这一条，判据就会时对时错， + 而错的那些次看起来只是「模型这次走了别的路」。 + """ + if len(tagged(read, "step_completed")) < after_steps: + return False + if not read.payloads: + return False + last = read.payloads[-1] + if timing is KillTiming.AFTER_STEP: + return last.get(RECORD_KEY) == "step_completed" + return last.get(RECORD_KEY) == "intent" and last.get("replay_policy") == "never" + + +@dataclass(frozen=True, slots=True, kw_only=True) +class KillOutcome: + """一次「起子进程、等时机、杀掉」的结果。""" + + #: 时机命中了没有。没命中要重试，重试若干次仍不命中要报出来——悄悄降级成另一种时机会让 + #: 报告显示「验过了」而其实验的是另一件事。 + hit: bool + reason: str + #: 杀掉之后重读日志、截到最后一个换行为止的那一段。字节比对拿它当基准。 + snapshot: bytes = b"" + #: 崩溃时已经完整落地的步数。 + steps: int = 0 + #: 这次子进程发起过几次模型调用（含被杀在半路的那次），记账用。 + model_calls: int = 0 + + +async def spawn_and_kill( + *, + argv: Sequence[str], + log_path: Path, + timing: KillTiming, + after_steps: int, + poll_interval_s: float = 0.002, + timeout_s: float = 600.0, +) -> KillOutcome: + """起一个子进程，轮询它的日志，命中时机就 `SIGKILL`。 + + **用 SIGKILL 不用 SIGTERM**：要的是没有任何清理机会的死法。SIGTERM 会走 Python 的信号 + 处理，`finally` 有机会跑完，那验的是优雅退出而不是崩溃。 + + 杀掉之后**重读一次日志再判时机是不是还成立**：读日志与发信号之间子进程还在写，读到的 + 尾部可能已经不是杀掉那一刻的尾部了。不重判的话，一次「本想在时机 A 杀、实际杀在时机 B」 + 会被当成时机 A 判下去。 + """ + process = await asyncio.create_subprocess_exec( + *argv, + cwd=str(REPO_ROOT), + stdout=asyncio.subprocess.PIPE, + stderr=asyncio.subprocess.PIPE, + ) + deadline = time.monotonic() + timeout_s + try: + while True: + if process.returncode is not None: + stdout, stderr = await process.communicate() + tail = stderr.decode("utf-8", errors="replace").strip().splitlines()[-3:] + del stdout + return KillOutcome( + hit=False, + reason=( + f"子进程在命中时机之前就退出了（退出码 {process.returncode}）" + + (f"，stderr 末尾：{' / '.join(tail)}" if tail else "") + ), + model_calls=count_model_calls(read_log(log_path)), + ) + if should_kill(read_log(log_path), timing=timing, after_steps=after_steps): + process.kill() + await process.wait() + crashed = log_path.read_bytes() if log_path.is_file() else b"" + after = parse_terminated(crashed) + if not should_kill(after, timing=timing, after_steps=after_steps): + return KillOutcome( + hit=False, + reason="发信号与子进程停笔之间又写进了记录，杀掉之后时机已经不成立", + model_calls=count_model_calls(after), + ) + return KillOutcome( + hit=True, + reason=f"命中时机 {timing.value}", + snapshot=terminated_prefix(crashed), + steps=len(tagged(after, "step_completed")), + model_calls=count_model_calls(after), + ) + if time.monotonic() > deadline: + return KillOutcome( + hit=False, + reason=f"等了 {timeout_s} 秒仍没命中时机 {timing.value}", + model_calls=count_model_calls(read_log(log_path)), + ) + await asyncio.sleep(poll_interval_s) + finally: + if process.returncode is None: + process.kill() + await process.wait() + + +# --------------------------------------------------------------------------- +# 六、接缝：事件出口、计数、故意坏掉的解释器、取消触发器 +# --------------------------------------------------------------------------- + + +class JsonlEventSink: + """把事件逐行写进一个 `.jsonl`。满足 `polyloop.ports.EventSink`。 + + `parameters()` 里**不放路径**：它进参数快照，而续跑时父进程写的是另一个文件（子进程那 + 份事件随 SIGKILL 留在原地），路径进快照会报一次假的参数漂移。 + """ + + __slots__ = ("_path", "delivered", "failures") + + def __init__(self, path: Path | str) -> None: + self._path = Path(path) + self.delivered = 0 + self.failures = 0 + + def parameters(self) -> Mapping[str, str]: + return {"kind": "jsonl_events"} + + async def emit(self, event: Event) -> None: + line = json.dumps( + { + "kind": event.kind.value, + "run_id": event.run_id, + "step_idx": event.step.step_idx, + }, + ensure_ascii=False, + ) + await asyncio.to_thread(self._append, line + "\n") + self.delivered += 1 + + def _append(self, line: str) -> None: + self._path.parent.mkdir(parents=True, exist_ok=True) + with self._path.open("a", encoding="utf-8") as handle: + handle.write(line) + + +class AlwaysInvalidParser: + """对任何模型输出都返回无效决策。解析失败连击那一类用它。 + + 它不是「解析不出来」，是**声明这一步解释不了**——契约要求 `parse` 同步、不抛异常、解释 + 不出来时返回 `InvalidDecision`（`tests/contract/test_decision_parser.py`），这份实现照做。 + """ + + def parameters(self) -> Mapping[str, str]: + return {"kind": "always_invalid"} + + def parse(self, reply: ModelReply) -> ParsedReply: + return ParsedReply( + history_text=reply.content, + decision=InvalidDecision( + explanation="故障注入：这一路的决策解释器对任何输出都判无效，用来压解析失败连击。" + ), + ) + + +class TriggeringModelClient: + """转发模型调用，并在进入第 N 次调用时通知父协程。父协程收到通知立刻取消。 + + 这样取消落在模型调用**中途**——那条路径上在飞的是网关连接与一次已经计过费的请求。 + """ + + __slots__ = ("_at_call_index", "_inner", "_trigger", "calls") + + def __init__(self, *, inner: object, trigger: asyncio.Event, at_call_index: int) -> None: + self._inner = inner + self._trigger = trigger + self._at_call_index = at_call_index + self.calls = 0 + + def parameters(self) -> Mapping[str, str]: + return dict(self._inner.parameters()) | {"cancel_probe": "model_call"} # type: ignore[attr-defined] + + async def call(self, call: object) -> ModelReply: + self.calls += 1 + if call.call_index >= self._at_call_index: # type: ignore[attr-defined] + self._trigger.set() + return await self._inner.call(call) # type: ignore[attr-defined] + + +class TriggeringExecutor: + """转发动作执行，并在进入第一次执行时通知父协程。 + + 取消落在环境执行中途——那条路径上在飞的是容器租约与一次已经发出去的 HTTP 请求，与模型 + 调用那条路上的资源完全不同，所以两条各做一次。 + """ + + __slots__ = ("_inner", "_trigger", "executions") + + def __init__(self, *, inner: object, trigger: asyncio.Event) -> None: + self._inner = inner + self._trigger = trigger + self.executions = 0 + + def parameters(self) -> Mapping[str, str]: + return dict(self._inner.parameters()) | {"cancel_probe": "env_execute"} # type: ignore[attr-defined] + + async def execute(self, action: object) -> object: + self.executions += 1 + self._trigger.set() + return await self._inner.execute(action) # type: ignore[attr-defined] + + +@dataclass(slots=True) +class CallGuard: + """调用数护栏：花掉多少次真实模型调用，还剩多少。 + + **它按每类故障的边界拦，不在模型客户端里拦。** 在客户端里抛异常的话，库会把那次调用记成 + 一次模型调用失败、合成一段观察接着跑，于是护栏本身变成了一次注入进来的故障，把要验的 + 停止原因搅乱。按边界拦的代价是可能超出一点点，收益是判据看到的运行是干净的。 + """ + + limit: int + spent: int = 0 + + @property + def remaining(self) -> int: + return max(0, self.limit - self.spent) + + def charge(self, calls: int) -> None: + self.spent += calls + + def affordable(self, worst_case: int) -> bool: + return self.spent + worst_case <= self.limit + + +# --------------------------------------------------------------------------- +# 七、sidecar +# --------------------------------------------------------------------------- + + +def write_sidecars( + *, + runs_dir: Path, + run_id: str, + scenario: str, + fault: str, + result: RunResult | None, + wall_ms: int, + model_calls: int, + sink_failures: int, + env_executions: int, + task_id: str | None, + phase: str | None, + resumed_from_step: int | None, +) -> None: + """按记分板的约定写 `.result.json` 与 `.meta.json`。 + + `.events.jsonl` 不在这里写——它由 `JsonlEventSink` 边跑边写，那才是它的真实形态。 + + **结果为空时不写 `.result.json`。** 取消那条路径上 `run` 不返回结果，硬造一份是伪造； + 记分板会退回日志里的结束记录，那份是真的。 + + `success` 恒为 `None`：故障注入的运行没有「任务做没做成」这回事，填一个布尔值会让记分板 + 的成功率统计混进一批语义不同的行。 + """ + runs_dir.mkdir(parents=True, exist_ok=True) + if result is not None: + (runs_dir / f"{run_id}.result.json").write_text( + json.dumps(encode(result), ensure_ascii=False), encoding="utf-8" + ) + meta: dict[str, object] = { + "scenario": scenario, + "task_id": task_id, + "phase": phase, + "wall_ms": wall_ms, + "model_calls": model_calls, + "sink_failures": sink_failures, + "env_executions": env_executions, + "fault": fault, + "success": None, + "resumed_from_step": resumed_from_step, + } + (runs_dir / f"{run_id}.meta.json").write_text( + json.dumps(meta, ensure_ascii=False), encoding="utf-8" + ) + + +# --------------------------------------------------------------------------- +# 八、GovDoc 崩溃续跑 +# --------------------------------------------------------------------------- + +#: 崩溃续跑跑的是 execute 阶段。**选它是因为它最快走到一次有副作用的动作**：提示词让模型先 +#: 读 plan.md、再按行号核实、然后 `write_note` 写 evidence.md，而 `write_note` 的重放策略是 +#: `never`——那正是最硬那条判据要数的东西。plan 阶段要先自己检索出候选行号才写得成笔记， +#: 多烧好几次调用。 +CRASH_PHASE = "execute" + +#: 预置在工作区里的计划文件。真实场景里它由 plan 阶段写出来，这里直接写盘省掉那一整次运行。 +#: +#: **由父进程用普通文件写入放进去，不走 `write_note`**：走工具的话审计账里会先躺一条记录， +#: 而那条记录不是这次运行产生的，会把去重比对的基数弄脏。 +#: +#: 正文全是自造的，一个字都不取自真实文书。 +SEEDED_PLAN_NAME = "plan.md" +SEEDED_PLAN_TEXT = """# 审核计划（故障注入脚本预置） + +候选证据两条，逐条核实： + +1. tender.md 第 1 到第 60 行：项目基本信息与采购人信息，用来确认主体与项目编号。 +2. tender.md 第 200 到第 260 行：供应商资格条件，审核点多半落在这一段。 + +核实完成后把逐条摘录写进 evidence.md，写完就停下。 +""" + +#: 崩溃续跑这一路的预算。**比 GovDoc 场景自己那份（50 步）小得多**：这里要验的是崩溃与续跑 +#: 的接缝，接缝在头几步就压得到，而每多一步都是一次真实的模型调用。 +CRASH_BUDGET = Budget( + max_steps=6, + max_actions=6, + max_consecutive_parse_failures=3, + max_prompt_chars=400_000, +) + +#: 模型绑定必须在父子两侧逐字相同——它整个进参数快照，差一个键续跑就报参数漂移。 +CRASH_MODEL_BINDING: Mapping[str, str] = {"scenario": "govdoc", "phase": CRASH_PHASE} + +#: 时机判定之前先等几步落地。等到第 2 步是为了让模型有机会走到 `write_note` 那一步，否则 +#: 审计账全程为空，最硬那条判据就只能报「无法判定」。 +CRASH_AFTER_STEPS = 2 + +APPWORLD_MODEL_BINDING: Mapping[str, str] = {"scenario": "appworld"} + +#: 崩溃那一刻的日志快照落盘时用的后缀。 +CRASH_SNAPSHOT_SUFFIX = ".crash-snapshot" + + +def load_govdoc_task(*, govdoc_db: Path, govdoc_corpus: Path) -> AuditTask: + """装配崩溃续跑用的那一个审核任务：第一条已批准的审核点配脱敏后的招标文书。 + + 父子两个进程各装配一次，装出来的必须一致。**它确实一致**：审核点按 id 排序取第一条， + 语料是同一个文件过同一份脱敏器。就算不一致也不会静默出错——上下文与语料不进参数快照， + 进快照的是预算、工具集、绑定与四个接缝上报的参数，那些是常量。 + """ + checkpoints = load_checkpoints(db_path=govdoc_db, limit=1) + loaded = load_documents(prepared_dir=govdoc_corpus, redactor=Redactor()) + return AuditTask( + index=0, + checkpoint=checkpoints[0], + documents=tuple(document for document, _report in loaded), + ) + + +def build_crash_request(*, task: AuditTask, run_id: str, workspace: Path) -> RunRequest: + """装配崩溃续跑那次运行的请求。父子两侧调的是同一个函数，不各写一遍。""" + request = govdoc_scenario.build_run_request( + task=task, + phase=CRASH_PHASE, + run_id=run_id, + workspace=workspace, + model_binding=CRASH_MODEL_BINDING, + ) + return replace(request, budget=CRASH_BUDGET) + + +def build_crash_definition( + *, model_client: object, store: JsonlRunStore, sink: JsonlEventSink +) -> AgentDefinition: + return AgentDefinition( + model_client=model_client, # type: ignore[arg-type] + decision_parser=GovDocParser(), + store=store, + event_sink=sink, + synthetic_observations=govdoc_scenario.SYNTHETIC_OBSERVATIONS, + ) + + +async def check_parameter_drift_detected( + *, definition: AgentDefinition, request: RunRequest +) -> Criterion: + """故意改一个预算数字续跑，验库确实拒绝。 + + 这条守的是「续跑不能顺便换配置」。它不成立的后果很具体：前几步与后几步来自两份不同的 + 配置，而两段轨迹在文件里看起来是同一次运行，事后分不出来。 + + 改的是 `max_steps` **加一**而不是减：万一这道守卫失效、`resume` 真的跑起来了，加一只是 + 多给一步余量，减到一会当场以预算耗尽收尾、把日志封死，后面真正的续跑就没得做了。 + """ + drifted = replace( + request, budget=replace(request.budget, max_steps=request.budget.max_steps + 1) + ) + try: + await session.resume(definition, drifted) + except ParameterDriftError as exc: + detail = str(exc).split("：", 1)[0] + return passed( + "parameter_drift_detected", + f"改一个预算数字之后续跑报了 ParameterDriftError（{detail}）", + ) + except Exception as exc: # noqa: BLE001 - 任何别的异常都说明守卫走的不是这条路 + return breached( + "parameter_drift_detected", + f"改一个预算数字之后续跑抛的是 {type(exc).__name__}，期望 ParameterDriftError", + ) + return breached( + "parameter_drift_detected", + "改一个预算数字之后续跑正常返回了结果，参数漂移这道守卫没拦住", + ) + + +async def run_crash_fault( + *, + fault: str, + timing: KillTiming, + runs_dir: Path, + workspace_root: Path, + govdoc_db: Path, + govdoc_corpus: Path, + model_client: object, + guard: CallGuard, + attempts: int, +) -> FaultReport: + """一类崩溃续跑：起子进程 → 按时机 SIGKILL → 拷字节 → 同一个 run_id 续跑 → 逐条判。""" + task = load_govdoc_task(govdoc_db=govdoc_db, govdoc_corpus=govdoc_corpus) + notes: list[str] = [] + for attempt in range(1, attempts + 1): + if not guard.affordable(CRASH_BUDGET.max_steps + CRASH_AFTER_STEPS + 1): + notes.append( + f"调用数护栏只剩 {guard.remaining} 次，不够再试一轮，停在第 {attempt} 次之前" + ) + break + run_id = f"fault-{fault}-{attempt}" + workspace = workspace_root / run_id + workspace.mkdir(parents=True, exist_ok=True) + (workspace / SEEDED_PLAN_NAME).write_text(SEEDED_PLAN_TEXT, encoding="utf-8") + log_path = runs_dir / f"{run_id}.jsonl" + + argv = [ + sys.executable, + "-m", + "tools.soak.faults", + "--child", + "--runs-dir", + str(runs_dir), + "--run-id", + run_id, + "--workspace", + str(workspace), + "--govdoc-db", + str(govdoc_db), + "--govdoc-corpus", + str(govdoc_corpus), + ] + outcome = await spawn_and_kill( + argv=argv, log_path=log_path, timing=timing, after_steps=CRASH_AFTER_STEPS + ) + guard.charge(outcome.model_calls) + if not outcome.hit: + notes.append( + f"第 {attempt} 次没命中时机：{outcome.reason}（花了 {outcome.model_calls} 次调用）" + ) + write_sidecars( + runs_dir=runs_dir, + run_id=run_id, + scenario="govdoc", + fault=f"{fault}_missed", + result=None, + wall_ms=0, + model_calls=outcome.model_calls, + sink_failures=0, + env_executions=len(read_audit_lines(workspace)), + task_id=task.checkpoint.checkpoint_id, + phase=CRASH_PHASE, + resumed_from_step=None, + ) + continue + + notes.append(f"第 {attempt} 次命中时机 {timing.value}，崩溃时 {outcome.steps} 步") + return await _resume_and_judge( + fault=fault, + timing=timing, + task=task, + run_id=run_id, + runs_dir=runs_dir, + workspace=workspace, + log_path=log_path, + outcome=outcome, + model_client=model_client, + guard=guard, + notes=notes, + ) + + return FaultReport( + fault=fault, + criteria=( + undetermined( + "kill_timing_hit", + f"{attempts} 次都没能把子进程杀在时机 {timing.value} 上，这一类什么都没验成", + ), + ), + notes=tuple(notes), + ) + + +async def _resume_and_judge( + *, + fault: str, + timing: KillTiming, + task: AuditTask, + run_id: str, + runs_dir: Path, + workspace: Path, + log_path: Path, + outcome: KillOutcome, + model_client: object, + guard: CallGuard, + notes: list[str], +) -> FaultReport: + audit_before = read_audit_lines(workspace) + # 崩溃那一刻的日志原样留一份在盘上，给人事后自己比。**后缀不是 `.jsonl`**：记分板按 + # `*.jsonl` 枚举 run，叫那个名字的话这份快照会被当成另一次运行。 + (runs_dir / f"{run_id}{CRASH_SNAPSHOT_SUFFIX}").write_bytes(outcome.snapshot) + 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) + request = build_crash_request(task=task, run_id=run_id, workspace=workspace) + + criteria: list[Criterion] = [ + await check_parameter_drift_detected(definition=definition, request=request) + ] + + started = time.monotonic() + result: RunResult | None = None + try: + result = await session.resume(definition, request) + except Exception as exc: # noqa: BLE001 - 续跑本身炸了也是一条要报告的判定 + criteria.append(breached("resume_completed", f"续跑抛了 {type(exc).__name__}：{exc}")) + else: + criteria.append( + passed("resume_completed", f"续跑正常结束，停止原因 {result.stop_reason.value}") + ) + wall_ms = int((time.monotonic() - started) * 1000) + + final_bytes = log_path.read_bytes() if log_path.is_file() else b"" + read = parse_terminated(final_bytes) + audit_after = read_audit_lines(workspace) + if read.torn: + notes.append("最终日志末尾有一段没被换行终结的字节：那次写没有被确认过，按契约不算数") + + criteria.append(check_log_readable(read)) + criteria.append(check_crash_prefix_preserved(crashed=outcome.snapshot, final=final_bytes)) + criteria.append(check_step_indices_dense(read)) + criteria.append(check_intents_settled(read)) + criteria.append(check_never_action_not_replayed(audit_after)) + if timing is KillTiming.AFTER_STEP: + criteria.append( + check_resume_made_progress( + crashed_steps=outcome.steps, final_steps=len(tagged(read, "step_completed")) + ) + ) + else: + criteria.append(check_stop_reason(read, "resume_state_unknown")) + criteria.append(check_audit_unchanged(before=audit_before, after=audit_after)) + + resumed_calls = count_model_calls(read) - outcome.model_calls + guard.charge(max(0, resumed_calls)) + 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(audit_after), + task_id=task.checkpoint.checkpoint_id, + phase=CRASH_PHASE, + resumed_from_step=outcome.steps, + ) + return FaultReport(fault=fault, criteria=tuple(criteria), notes=tuple(notes)) + + +# --------------------------------------------------------------------------- +# 九、AppWorld：取消、预算、解析连击 +# --------------------------------------------------------------------------- + +#: 撞步数上限那一路的预算。**动作上限必须比步数上限宽**，否则先撞上的是动作那一维，停止 +#: 原因就成了 `action_budget`，这一条什么都没验到。 +STEP_BUDGET_OVERRIDE = Budget( + max_steps=3, max_actions=40, max_consecutive_parse_failures=3, max_prompt_chars=400_000 +) + +#: 撞动作上限那一路：动作上限压到比步数上限小，让动作那一维先耗尽。 +ACTION_BUDGET_OVERRIDE = Budget( + max_steps=10, max_actions=2, max_consecutive_parse_failures=3, max_prompt_chars=400_000 +) + +#: 解析失败连击那一路。步数上限留得比连击上限宽，让连击那一维先命中。 +PARSE_FAILURE_BUDGET = Budget( + max_steps=10, max_actions=10, max_consecutive_parse_failures=3, max_prompt_chars=400_000 +) + +#: 取消之后再借一个容器的等待上限。池满时 `lease()` 会一直阻塞，所以超时就是击穿。 +LEASE_TIMEOUT_S = 120.0 + +#: 等取消触发器的上限。等不到说明这次运行在触发点之前就结束了。 +TRIGGER_TIMEOUT_S = 300.0 + + +def _appworld_definition( + *, model_client: object, store: JsonlRunStore, sink: JsonlEventSink, parser: object +) -> AgentDefinition: + return AgentDefinition( + model_client=model_client, # type: ignore[arg-type] + decision_parser=parser, # type: ignore[arg-type] + store=store, + event_sink=sink, + synthetic_observations=appworld_scenario.build_synthetic_observations(), + ) + + +async def _can_borrow(pool: AppWorldPool, task_id: str, timeout_s: float) -> bool: + """再借一次容器，借得到返回 True。借不到（超时）说明上一次的租约没还回来。""" + try: + async with asyncio.timeout(timeout_s), pool.session(task_id): + return True + except TimeoutError: + return False + + +async def run_cancel_fault( + *, + fault: str, + pool: AppWorldPool, + task_id: str, + app_descriptions: str, + runs_dir: Path, + model_client: object, + guard: CallGuard, +) -> FaultReport: + """取消：跑到中途 `task.cancel()`，验异常穿透、结束记录留痕、容器租约归还。""" + 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") + trigger = asyncio.Event() + notes: list[str] = [] + raised: BaseException | None = None + env_executions = 0 + started = time.monotonic() + + async with pool.session(task_id) as handle: + request = appworld_scenario.build_run_request( + run_id=run_id, + session=handle, + app_descriptions=app_descriptions, + model_binding=APPWORLD_MODEL_BINDING, + ) + if fault == "cancel_env": + request = replace( + request, + action_executor=TriggeringExecutor(inner=request.action_executor, trigger=trigger), # type: ignore[arg-type] + ) + client: object = model_client + else: + # 第 1 次（从 0 数起的第二次）调用时触发：让日志里先有一步完整的记录，取消才落在 + # 「跑到中途」而不是「刚起步」。 + client = TriggeringModelClient(inner=model_client, trigger=trigger, at_call_index=1) + definition = _appworld_definition( + model_client=client, store=store, sink=sink, parser=AppWorldParser() + ) + task = asyncio.create_task(session.run(definition, request)) + try: + async with asyncio.timeout(TRIGGER_TIMEOUT_S): + await trigger.wait() + except TimeoutError: + notes.append(f"等了 {TRIGGER_TIMEOUT_S} 秒也没等到触发点，这次运行可能在触发前就结束了") + task.cancel() + # 这里接住的是**被等待的那个任务**抛出来的取消，不是本协程自己的取消——本协程从头到尾 + # 没有被 cancel 过。接住它正是这一类要验的那条判据（CLAUDE.md §1.6 禁的是把自己的取消 + # 吞掉）。 + try: + await task + except asyncio.CancelledError as exc: + raised = exc + env_executions = handle.n_executions + + wall_ms = int((time.monotonic() - started) * 1000) + borrowed = await _can_borrow(pool, task_id, LEASE_TIMEOUT_S) + read = read_log(log_path) + guard.charge(count_model_calls(read)) + + criteria = ( + check_log_readable(read), + check_cancelled_raised(raised=raised), + check_stop_reason(read, "cancelled"), + check_lease_returned(borrowed=borrowed, timeout_s=LEASE_TIMEOUT_S, pool_size=1), + ) + write_sidecars( + runs_dir=runs_dir, + run_id=run_id, + scenario="appworld", + fault=fault, + # 取消那条路上 `run` 不返回结果，`.result.json` 天然缺；记分板会退回日志里的结束记录。 + result=None, + wall_ms=wall_ms, + model_calls=count_model_calls(read), + sink_failures=sink.failures, + env_executions=env_executions, + task_id=task_id, + phase=None, + resumed_from_step=None, + ) + return FaultReport(fault=fault, criteria=criteria, notes=tuple(notes)) + + +async def run_budget_fault( + *, + fault: str, + pool: AppWorldPool, + task_id: str, + app_descriptions: str, + runs_dir: Path, + model_client: object, + guard: CallGuard, +) -> FaultReport: + """不可能完成的目标：验它干净地撞预算上限，而不是以别的原因结束。""" + 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") + budget = STEP_BUDGET_OVERRIDE if fault == "step_budget" else ACTION_BUDGET_OVERRIDE + notes: list[str] = [] + result: RunResult | None = None + env_executions = 0 + started = time.monotonic() + + 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=budget, + ) + 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) + read = read_log(log_path) + guard.charge(count_model_calls(read)) + + criteria = [check_log_readable(read), check_stop_reason(read, fault)] + if fault == "step_budget": + criteria.append( + check_step_count(read, expected=budget.max_steps, name="steps_equal_max_steps") + ) + else: + criteria.append(check_executed_action_count(read, expected=budget.max_actions)) + criteria.append(check_no_env_error_step(read)) + + 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, + env_executions=env_executions, + task_id=task_id, + phase=None, + resumed_from_step=None, + ) + return FaultReport(fault=fault, criteria=tuple(criteria), notes=tuple(notes)) + + +async def run_parse_failure_fault( + *, + pool: AppWorldPool, + task_id: str, + app_descriptions: str, + runs_dir: Path, + model_client: object, + guard: CallGuard, +) -> FaultReport: + """解析失败连击：包一层永远判无效的解释器，验它按连击上限收尾且一次都没碰环境。 + + 这一类只花三次调用，但它守的是「解析失败不调环境」这条契约——那条错了的表现是环境状态 + 被一批根本没解释出来的动作改掉，而轨迹里每一步都写着解析失败。 + """ + fault = "parse_failures" + 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") + result: RunResult | None = None + env_executions = 0 + started = time.monotonic() + + 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=PARSE_FAILURE_BUDGET, + ) + definition = _appworld_definition( + model_client=model_client, store=store, sink=sink, parser=AlwaysInvalidParser() + ) + result = await session.run(definition, request) + env_executions = handle.n_executions + + wall_ms = int((time.monotonic() - started) * 1000) + read = read_log(log_path) + guard.charge(count_model_calls(read)) + + criteria = ( + check_log_readable(read), + check_stop_reason(read, "parse_failed_repeatedly"), + check_step_count( + read, + expected=PARSE_FAILURE_BUDGET.max_consecutive_parse_failures, + name="steps_equal_max_consecutive_parse_failures", + ), + check_all_steps_parse_failed(read), + check_env_untouched(executions=env_executions, source="AppWorld 的 n_executions"), + ) + 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, + env_executions=env_executions, + task_id=task_id, + phase=None, + resumed_from_step=None, + ) + return FaultReport(fault=fault, criteria=criteria, notes=()) + + +# --------------------------------------------------------------------------- +# 十、子进程入口 +# --------------------------------------------------------------------------- + + +async def run_child(args: argparse.Namespace) -> int: + """子进程：跑一次 GovDoc execute 阶段，等着被父进程杀掉。 + + 正常跑完也是允许的——那时父进程会报「没命中时机」并重试，不会把它当成命中。 + + **事件写在 `.child.events.jsonl`，不写记分板认的那个名字。** 事件只在「本进程里 + 真的走完」的步上发，而记分板按 `meta.resumed_from_step` 把续跑跳过的那一段减掉之后与 + `.events.jsonl` 的条数对账。子进程的事件混进同一个文件的话，那笔账永远对不上，而对不上 + 的原因是我们把两个进程的事件叠在了一起，不是库出了问题。这个文件名以 `.events.jsonl` + 结尾，所以记分板枚举 run 时会把它排掉，不会被当成另一次运行。 + """ + from polygateway import GatewayClient, GatewaySettings + + runs_dir = Path(args.runs_dir) + workspace = Path(args.workspace) + task = load_govdoc_task(govdoc_db=Path(args.govdoc_db), govdoc_corpus=Path(args.govdoc_corpus)) + store = JsonlRunStore(directory=runs_dir) + sink = JsonlEventSink(runs_dir / f"{args.run_id}.child.events.jsonl") + client = GatewayClient.from_env() + try: + definition = build_crash_definition( + model_client=GatewayModelClient(client=client, settings=GatewaySettings.from_env()), + store=store, + sink=sink, + ) + request = build_crash_request(task=task, run_id=args.run_id, workspace=workspace) + await session.run(definition, request) + finally: + await client.aclose() + return 0 + + +# --------------------------------------------------------------------------- +# 十一、入口 +# --------------------------------------------------------------------------- + + +def build_parser() -> argparse.ArgumentParser: + parser = argparse.ArgumentParser( + prog="python -m tools.soak.faults", + description="PolyLoop 压测的故障注入：崩溃续跑、取消、撞预算、解析失败连击", + ) + parser.add_argument("--runs-dir", required=True, help="产物目录：日志与三种 sidecar 都写在这里") + parser.add_argument( + "--fault", + action="append", + choices=FAULT_NAMES, + help="要跑哪一类，可给多次。不给就全跑", + ) + parser.add_argument( + "--budget-calls", + type=int, + help="真实模型调用数的护栏。每类故障开跑前检查一次，不够就跳过（父进程必填）", + ) + parser.add_argument("--data-root", help="AppWorld 数据根目录（跑 AppWorld 那几类时必填）") + parser.add_argument("--govdoc-db", help="GovDoc 审核点 sqlite（跑崩溃续跑时必填）") + parser.add_argument("--govdoc-corpus", help="GovDoc 脱敏前语料目录（跑崩溃续跑时必填）") + parser.add_argument("--split", default="train", help="AppWorld 的任务划分，默认 train") + parser.add_argument( + "--attempts", type=int, default=3, help="崩溃续跑每类最多重试几次去命中时机，默认 3" + ) + parser.add_argument( + "--child", action="store_true", help="内部用：以子进程身份跑一次 GovDoc 运行" + ) + parser.add_argument("--run-id", help="内部用：子进程的运行标识") + parser.add_argument("--workspace", help="内部用：子进程的工作区目录") + return parser + + +def _require(args: argparse.Namespace, name: str, why: str) -> Path: + value = getattr(args, name.replace("-", "_")) + if not value: + raise FaultInjectionError(f"缺 --{name}：{why}") + path = Path(value) + if not path.exists(): + raise FaultInjectionError(f"--{name} 指向的东西不在：{path}") + return path + + +async def guarded(fault: str, coroutine: Awaitable[FaultReport]) -> FaultReport: + """跑一类故障，装配或环境本身炸了就记成「无法判定」接着跑下一类。 + + **这不是吞错误**：异常的类名与文本原样进了那条判据的证据里，退出码那侧也看得见它不是 + 「通过」。不这么做的话，第五类故障在起容器时崩一下，前面四类已经花钱跑出来的结论会一起 + 丢掉，而那些结论正是这次跑的产出。 + + `CancelledError` 继承 `BaseException`，不在 `except Exception` 的范围里，所以人按 + Ctrl-C 时整个跑照样立刻停下（CLAUDE.md §1.6）。 + """ + try: + return await coroutine + except Exception as exc: # noqa: BLE001 - 见 docstring：记下来，不吞掉 + return FaultReport( + fault=fault, + criteria=( + undetermined( + "fault_ran_to_completion", + f"这一类跑到一半抛了 {type(exc).__name__}：{exc}", + ), + ), + ) + + +async def run_all(args: argparse.Namespace) -> int: + from polygateway import GatewayClient, GatewaySettings + + selected = tuple(args.fault) if args.fault else FAULT_NAMES + runs_dir = Path(args.runs_dir) + runs_dir.mkdir(parents=True, exist_ok=True) + guard = CallGuard(limit=args.budget_calls) + reports: list[FaultReport] = [] + + client = GatewayClient.from_env() + try: + model_client = GatewayModelClient(client=client, settings=GatewaySettings.from_env()) + + 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", "崩溃续跑要读语料") + 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, + ), + ) + ) + + appworld_selected = [name for name in selected if name in APPWORLD_FAULTS] + if appworld_selected: + data_root = _require(args, "data-root", "AppWorld 那几类要起容器") + reports.extend( + await _run_appworld_faults( + names=appworld_selected, + data_root=data_root, + split=args.split, + runs_dir=runs_dir, + model_client=model_client, + guard=guard, + ) + ) + finally: + await client.aclose() + + print("\n".join(report.render() for report in reports)) + print(f"\n真实模型调用：{guard.spent} / {guard.limit}") + breaches = [item for report in reports for item in report.breaches] + unknowns = [item for report in reports for item in report.undetermineds] + print(f"击穿 {len(breaches)} 条，无法判定 {len(unknowns)} 条") + return 1 if breaches else 0 + + +async def _run_appworld_faults( + *, + names: Sequence[str], + data_root: Path, + split: str, + runs_dir: Path, + model_client: object, + guard: CallGuard, +) -> list[FaultReport]: + """AppWorld 那几类共用一个池。 + + **池大小固定为 1**：容器租约那条判据要的是「取消之后空闲名额回到满」，而池里有第二个 + 容器的话，借得到只说明还剩别的名额，什么都证明不了。 + """ + 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) + 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] + if not guard.affordable(worst_case): + reports.append( + FaultReport( + fault=name, + criteria=( + undetermined( + "call_budget_available", + f"调用数护栏只剩 {guard.remaining} 次，最坏要 {worst_case} 次，这一类没跑", + ), + ), + ) + ) + continue + coroutine: Awaitable[FaultReport] + if name in {"cancel_model", "cancel_env"}: + coroutine = run_cancel_fault( + fault=name, + pool=pool, + task_id=task_id, + app_descriptions=app_descriptions, + runs_dir=runs_dir, + model_client=model_client, + guard=guard, + ) + elif name == "parse_failures": + coroutine = run_parse_failure_fault( + pool=pool, + task_id=task_id, + app_descriptions=app_descriptions, + runs_dir=runs_dir, + model_client=model_client, + guard=guard, + ) + else: + coroutine = run_budget_fault( + fault=name, + pool=pool, + task_id=task_id, + app_descriptions=app_descriptions, + runs_dir=runs_dir, + model_client=model_client, + guard=guard, + ) + reports.append(await guarded(name, coroutine)) + return reports + + +def main(argv: Sequence[str] | None = None) -> int: + parser = build_parser() + args = parser.parse_args(argv) + if args.child: + missing = [ + name + for name in ("run_id", "workspace", "govdoc_db", "govdoc_corpus") + if not getattr(args, name) + ] + if missing: + parser.error(f"--child 模式还缺 {missing}") + return asyncio.run(run_child(args)) + if args.budget_calls is None: + parser.error("--budget-calls 是必填的：没有它这一跑可以无上限地花钱") + if args.budget_calls < 1: + parser.error("--budget-calls 至少是 1") + return asyncio.run(run_all(args)) + + +if __name__ == "__main__": + raise SystemExit(main()) + + +__all__ = [ + "APPWORLD_FAULTS", + "CRASH_BUDGET", + "FAULT_NAMES", + "GOVDOC_FAULTS", + "AlwaysInvalidParser", + "CallGuard", + "Criterion", + "CriterionStatus", + "FaultInjectionError", + "FaultReport", + "JsonlEventSink", + "KillOutcome", + "KillTiming", + "LogRead", + "check_all_steps_parse_failed", + "check_audit_unchanged", + "check_cancelled_raised", + "check_crash_prefix_preserved", + "check_env_untouched", + "check_executed_action_count", + "check_intents_settled", + "check_lease_returned", + "check_log_readable", + "check_never_action_not_replayed", + "check_no_env_error_step", + "check_resume_made_progress", + "check_step_count", + "check_step_indices_dense", + "check_stop_reason", + "count_model_calls", + "main", + "parse_audit_line", + "parse_terminated", + "read_log", + "should_kill", + "stop_reason_of", + "tagged", + "terminated_prefix", +] diff --git a/tools/soak/tests/test_faults.py b/tools/soak/tests/test_faults.py new file mode 100644 index 0000000..396d815 --- /dev/null +++ b/tools/soak/tests/test_faults.py @@ -0,0 +1,810 @@ +"""故障注入判据的测试。不打真实模型、不起容器，全部用构造出来的输入。 + +**每一条判据都要有一个「构造出违反它的输入 → 判据确实报击穿」的用例。** 一个永远返回通过的 +判据比没有判据更糟：它会让所有人以为这些不变量被守着，而它什么都没守，而且这件事在压测报告 +上看起来是一整片绿。所以下面每条判据都成对出现——一条喂它合规的输入验它说通过，一条喂它明确 +违反的输入验它说击穿。 + +子进程编排那部分拆出了两个纯函数（`should_kill` 判时机到没到、`terminated_prefix` 截已终结 +前缀），它们不碰进程也不碰模型，直接单独测。真起子进程那两条用的是一个只会往文件里写几行 +JSON 的假子进程，跑完不到两秒。 +""" + +from __future__ import annotations + +import asyncio +import json +import sys +from pathlib import Path + +import pytest + +from polyloop.ports import EventKind, InvalidDecision +from polyloop.types import ModelReply, RunResult, StopReason +from tools.soak.faults import ( + AlwaysInvalidParser, + CallGuard, + Criterion, + CriterionStatus, + FaultReport, + JsonlEventSink, + KillTiming, + LogRead, + build_parser, + check_all_steps_parse_failed, + check_audit_unchanged, + check_cancelled_raised, + check_crash_prefix_preserved, + check_env_untouched, + check_executed_action_count, + check_intents_settled, + check_lease_returned, + check_log_readable, + check_never_action_not_replayed, + check_no_env_error_step, + check_resume_made_progress, + check_step_count, + check_step_indices_dense, + check_stop_reason, + count_model_calls, + guarded, + main, + parse_audit_line, + parse_terminated, + read_log, + should_kill, + spawn_and_kill, + stop_reason_of, + tagged, + terminated_prefix, + write_sidecars, +) + +RUN_ID = "fault-test-0" + + +# --------------------------------------------------------------------------- +# 造记录：日志行的形状照 `polyloop.serialization.encode` 加 `polyloop.stores` 的类型标签 +# --------------------------------------------------------------------------- + + +def intent( + *, + kind: str = "model_call", + call_index: int = 0, + result_id: str = "r0", + replay_policy: str = "never", +) -> dict[str, object]: + return { + "record": "intent", + "run_id": RUN_ID, + "kind": kind, + "call_index": call_index, + "result_id": result_id, + "replay_policy": replay_policy, + } + + +def model_result(*, result_id: str = "r0") -> dict[str, object]: + return { + "record": "model_call_result", + "run_id": RUN_ID, + "result_id": result_id, + "reply": {"call_id": "c0", "content": "x", "thinking": ""}, + "failure": None, + } + + +def step_completed( + *, + step_idx: int = 0, + result_id: str | None = "a0", + status: str | None = "executed", + parse_ok: bool = True, + action_status: str | None = "executed", +) -> dict[str, object]: + outcome = ( + None + if status is None + else { + "status": status, + "observation": "o", + "observation_is_synthetic": False, + "env_reported_completion": False, + "observation_truncated_chars": 0, + } + ) + return { + "record": "step_completed", + "run_id": RUN_ID, + "result_id": result_id, + "action_outcome": outcome, + "step": { + "step_idx": step_idx, + "raw_output": "x", + "content_chars": 1, + "thinking_chars": 0, + "action": None, + "parse_ok": parse_ok, + "parse_error": None if parse_ok else "解释不了", + "observation": "o", + "observation_is_synthetic": False, + "observation_truncated_chars": 0, + "prompt_chars": 10, + "call_id": "c0", + "step_wall_ms": 1, + "tool_name": None, + "tool_arguments": None, + "action_status": action_status, + "env_reported_completion": False, + "schema_version": 1, + }, + } + + +def run_finished(*, stop_reason: str = "task_completed") -> dict[str, object]: + return { + "record": "run_finished", + "run_id": RUN_ID, + "result": { + "run_id": RUN_ID, + "stop_reason": stop_reason, + "final_answer": None, + "steps": [], + "schema_version": 1, + "event_delivery_failures": 0, + }, + } + + +def as_bytes(*payloads: dict[str, object]) -> bytes: + return "".join(json.dumps(item, ensure_ascii=False) + "\n" for item in payloads).encode("utf-8") + + +def as_read(*payloads: dict[str, object]) -> LogRead: + return parse_terminated(as_bytes(*payloads)) + + +# --------------------------------------------------------------------------- +# 一、判据本身的形状 +# --------------------------------------------------------------------------- + + +def test_criterion_requires_evidence() -> None: + """没有证据的判据构造不出来。一条只会说「有问题」的判据等于没有判据。""" + with pytest.raises(ValueError, match="没有给证据"): + Criterion(name="x", status=CriterionStatus.PASSED, evidence=" ") + + +def test_criterion_requires_name() -> None: + with pytest.raises(ValueError, match="必须有名字"): + Criterion(name=" ", status=CriterionStatus.PASSED, evidence="有证据") + + +# --------------------------------------------------------------------------- +# 二、日志读取 +# --------------------------------------------------------------------------- + + +def test_terminated_prefix_cuts_at_last_newline() -> None: + assert terminated_prefix(b'{"a":1}\n{"b":2}\n') == b'{"a":1}\n{"b":2}\n' + assert terminated_prefix(b'{"a":1}\n{"b":') == b'{"a":1}\n' + assert terminated_prefix(b'{"a":') == b"" + + +def test_parse_terminated_drops_torn_tail() -> None: + raw = as_bytes(intent()) + b'{"record":"inte' + read = parse_terminated(raw) + assert read.torn is True + assert len(read.payloads) == 1 + assert read.bad_lines == () + + +def test_parse_terminated_flags_bad_terminated_line() -> None: + raw = as_bytes(intent()) + b"not json at all\n" + read = parse_terminated(raw) + assert read.torn is False + assert read.bad_lines == (2,) + + +def test_check_log_readable_breaches_on_bad_line() -> None: + read = parse_terminated(as_bytes(intent()) + b"{}\n") + outcome = check_log_readable(read) + assert outcome.status is CriterionStatus.BREACHED + + +def test_check_log_readable_passes_on_clean_log() -> None: + assert check_log_readable(as_read(intent())).status is CriterionStatus.PASSED + + +def test_read_log_of_missing_file_is_empty() -> None: + assert read_log(Path("/nonexistent/nope.jsonl")).payloads == () + + +def test_tagged_and_counters() -> None: + read = as_read( + intent(kind="model_call", result_id="r0"), + intent(kind="action", result_id="a0"), + run_finished(stop_reason="step_budget"), + ) + assert len(tagged(read, "intent")) == 2 + assert count_model_calls(read) == 1 + assert stop_reason_of(read) == "step_budget" + + +def test_stop_reason_of_without_finished_record() -> None: + assert stop_reason_of(as_read(intent())) is None + + +# --------------------------------------------------------------------------- +# 三、崩溃续跑：字节前缀 +# --------------------------------------------------------------------------- + + +def test_prefix_preserved_passes_when_resume_only_appends() -> None: + crashed = as_bytes(intent(), model_result()) + final = crashed + as_bytes(step_completed()) + outcome = check_crash_prefix_preserved(crashed=crashed, final=final) + assert outcome.status is CriterionStatus.PASSED + + +def test_prefix_preserved_breaches_when_one_byte_changed() -> None: + """前缀里改一个字节就必须报击穿。 + + 这是这条判据存在的全部理由：解析出来的对象可能仍然等价（改的是空格、字段顺序、数字的 + 表示法），而承诺是更硬的那一条——已经写下去的字节不许再动。 + """ + crashed = as_bytes(intent(), model_result()) + mutated = bytearray(crashed) + mutated[5] = mutated[5] ^ 0x01 + final = bytes(mutated) + as_bytes(step_completed()) + outcome = check_crash_prefix_preserved(crashed=crashed, final=final) + assert outcome.status is CriterionStatus.BREACHED + assert "第 5 字节" in outcome.evidence + + +def test_prefix_preserved_breaches_when_final_is_shorter() -> None: + crashed = as_bytes(intent(), model_result()) + outcome = check_crash_prefix_preserved(crashed=crashed, final=crashed[:10]) + assert outcome.status is CriterionStatus.BREACHED + + +def test_prefix_preserved_ignores_torn_tail() -> None: + """崩溃快照末尾那段没被换行终结的字节不参与比对：那次写从来没算数。""" + crashed = as_bytes(intent()) + b'{"record":"model_ca' + final = as_bytes(intent()) + b'{"record":"model_ca' + as_bytes(model_result()) + assert check_crash_prefix_preserved(crashed=crashed, final=final).status is ( + CriterionStatus.PASSED + ) + + +def test_prefix_preserved_undetermined_without_any_terminated_record() -> None: + outcome = check_crash_prefix_preserved(crashed=b'{"half', final=b'{"half"}\n') + assert outcome.status is CriterionStatus.UNDETERMINED + + +# --------------------------------------------------------------------------- +# 四、崩溃续跑：步号 +# --------------------------------------------------------------------------- + + +def test_step_indices_dense_passes() -> None: + read = as_read(*(step_completed(step_idx=index) for index in range(4))) + assert check_step_indices_dense(read).status is CriterionStatus.PASSED + + +def test_step_indices_dense_breaches_on_gap() -> None: + read = as_read(step_completed(step_idx=0), step_completed(step_idx=2)) + assert check_step_indices_dense(read).status is CriterionStatus.BREACHED + + +def test_step_indices_dense_breaches_on_repeat() -> None: + """同一个步号出现两次：续跑把已经落地的那一步又跑了一遍。""" + read = as_read(step_completed(step_idx=0), step_completed(step_idx=0)) + assert check_step_indices_dense(read).status is CriterionStatus.BREACHED + + +def test_step_indices_dense_undetermined_without_steps() -> None: + assert check_step_indices_dense(as_read(intent())).status is CriterionStatus.UNDETERMINED + + +# --------------------------------------------------------------------------- +# 五、崩溃续跑：意图有没有归宿 +# --------------------------------------------------------------------------- + + +def test_intents_settled_passes_when_all_have_results() -> None: + read = as_read( + intent(kind="model_call", result_id="r0"), + model_result(result_id="r0"), + intent(kind="action", result_id="a0"), + step_completed(result_id="a0"), + ) + assert check_intents_settled(read).status is CriterionStatus.PASSED + + +def test_intents_settled_allows_one_dangling_at_the_end() -> None: + """崩溃点上那条悬空的意图是合法的：时机 B 下它就是崩溃点本身。""" + read = as_read( + intent(kind="model_call", result_id="r0"), + model_result(result_id="r0"), + intent(kind="action", result_id="a0"), + step_completed(result_id="a0"), + intent(kind="model_call", call_index=1, result_id="r1"), + run_finished(stop_reason="resume_state_unknown"), + ) + assert check_intents_settled(read).status is CriterionStatus.PASSED + + +def test_intents_settled_breaches_on_dangling_in_the_middle() -> None: + """中间悬空说明有一步的执行状态被跳过去了，后面的步建立在一个没有答案的问题上。""" + read = as_read( + intent(kind="model_call", result_id="r0"), + intent(kind="model_call", call_index=1, result_id="r1"), + model_result(result_id="r1"), + ) + outcome = check_intents_settled(read) + assert outcome.status is CriterionStatus.BREACHED + assert "[0]" in outcome.evidence + + +def test_intents_settled_undetermined_without_intents() -> None: + assert check_intents_settled(as_read(run_finished())).status is CriterionStatus.UNDETERMINED + + +# --------------------------------------------------------------------------- +# 六、崩溃续跑：审计账 +# --------------------------------------------------------------------------- + + +def test_parse_audit_line() -> None: + assert parse_audit_line("write_note\tplan.md\tabc123") == ("write_note", "plan.md", "abc123") + assert parse_audit_line("write_note\tplan.md") is None + assert parse_audit_line("write_note\t\tabc123") is None + + +def test_never_action_not_replayed_passes_on_distinct_writes() -> None: + audit = ["write_note\tplan.md\taaa", "write_note\tevidence.md\tbbb"] + assert check_never_action_not_replayed(audit).status is CriterionStatus.PASSED + + +def test_never_action_not_replayed_breaches_on_duplicate() -> None: + """同一个 (文件名, 内容摘要) 出现两次就是重放。""" + audit = [ + "write_note\tevidence.md\tbbb", + "write_note\tplan.md\taaa", + "write_note\tevidence.md\tbbb", + ] + outcome = check_never_action_not_replayed(audit) + assert outcome.status is CriterionStatus.BREACHED + assert "bbb" in outcome.evidence + + +def test_never_action_not_replayed_undetermined_on_empty_audit() -> None: + """一条副作用都没执行过时不许报「通过」——那次跑根本没验到这条不变量。""" + assert check_never_action_not_replayed([]).status is CriterionStatus.UNDETERMINED + + +def test_never_action_not_replayed_undetermined_on_malformed_audit() -> None: + assert check_never_action_not_replayed(["坏行"]).status is CriterionStatus.UNDETERMINED + + +def test_audit_unchanged_passes() -> None: + audit = ["write_note\tplan.md\taaa"] + assert check_audit_unchanged(before=audit, after=list(audit)).status is CriterionStatus.PASSED + + +def test_audit_unchanged_breaches_when_resume_adds_a_line() -> None: + before = ["write_note\tplan.md\taaa"] + after = [*before, "write_note\tplan.md\taaa"] + outcome = check_audit_unchanged(before=before, after=after) + assert outcome.status is CriterionStatus.BREACHED + assert "多出 1 条" in outcome.evidence + + +def test_audit_unchanged_breaches_when_prefix_rewritten() -> None: + outcome = check_audit_unchanged( + before=["write_note\tplan.md\taaa"], after=["write_note\tplan.md\tzzz"] + ) + assert outcome.status is CriterionStatus.BREACHED + + +# --------------------------------------------------------------------------- +# 七、崩溃续跑：续跑有没有往下走 +# --------------------------------------------------------------------------- + + +def test_resume_made_progress_passes() -> None: + assert check_resume_made_progress(crashed_steps=2, final_steps=5).status is ( + CriterionStatus.PASSED + ) + + +def test_resume_made_progress_breaches_when_stalled() -> None: + assert check_resume_made_progress(crashed_steps=2, final_steps=2).status is ( + CriterionStatus.BREACHED + ) + + +# --------------------------------------------------------------------------- +# 八、停止原因与步数 +# --------------------------------------------------------------------------- + + +def test_stop_reason_passes() -> None: + read = as_read(run_finished(stop_reason="step_budget")) + assert check_stop_reason(read, "step_budget").status is CriterionStatus.PASSED + + +def test_stop_reason_breaches_on_wrong_value() -> None: + read = as_read(run_finished(stop_reason="env_error")) + outcome = check_stop_reason(read, "step_budget") + assert outcome.status is CriterionStatus.BREACHED + assert "env_error" in outcome.evidence + + +def test_stop_reason_breaches_without_finished_record() -> None: + """取消也要留结束记录，不然恢复会把它当成可以续跑。缺记录是击穿，不是判不了。""" + outcome = check_stop_reason(as_read(step_completed()), "cancelled") + assert outcome.status is CriterionStatus.BREACHED + + +def test_step_count_passes_and_breaches() -> None: + read = as_read(*(step_completed(step_idx=index) for index in range(3))) + assert check_step_count(read, expected=3, name="n").status is CriterionStatus.PASSED + assert check_step_count(read, expected=2, name="n").status is CriterionStatus.BREACHED + + +def test_executed_action_count_passes_and_breaches() -> None: + read = as_read( + step_completed(step_idx=0, status="executed"), + step_completed(step_idx=1, status="executed"), + step_completed(step_idx=2, status="not_executed"), + ) + assert check_executed_action_count(read, expected=2).status is CriterionStatus.PASSED + assert check_executed_action_count(read, expected=3).status is CriterionStatus.BREACHED + + +def test_no_env_error_step_passes_and_breaches() -> None: + clean = as_read(step_completed(step_idx=0, status="executed")) + assert check_no_env_error_step(clean).status is CriterionStatus.PASSED + dirty = as_read( + step_completed(step_idx=0, status="executed"), + step_completed(step_idx=1, status="env_error"), + ) + outcome = check_no_env_error_step(dirty) + assert outcome.status is CriterionStatus.BREACHED + assert "[1]" in outcome.evidence + + +# --------------------------------------------------------------------------- +# 九、解析失败连击 +# --------------------------------------------------------------------------- + + +def test_all_steps_parse_failed_passes() -> None: + read = as_read( + *( + step_completed( + step_idx=index, result_id=None, status=None, parse_ok=False, action_status=None + ) + for index in range(3) + ) + ) + assert check_all_steps_parse_failed(read).status is CriterionStatus.PASSED + + +def test_all_steps_parse_failed_breaches_when_a_step_parsed() -> None: + read = as_read( + step_completed(step_idx=0, result_id=None, status=None, parse_ok=False, action_status=None), + step_completed(step_idx=1, parse_ok=True), + ) + assert check_all_steps_parse_failed(read).status is CriterionStatus.BREACHED + + +def test_all_steps_parse_failed_breaches_when_an_action_was_dispatched() -> None: + """解析失败那一步不许带动作状态——带了就说明有动作被分发过。""" + read = as_read( + step_completed(step_idx=0, parse_ok=False, action_status="executed"), + ) + assert check_all_steps_parse_failed(read).status is CriterionStatus.BREACHED + + +def test_all_steps_parse_failed_undetermined_without_steps() -> None: + assert check_all_steps_parse_failed(as_read(intent())).status is CriterionStatus.UNDETERMINED + + +def test_env_untouched_passes_and_breaches() -> None: + assert check_env_untouched(executions=0, source="假环境").status is CriterionStatus.PASSED + outcome = check_env_untouched(executions=1, source="假环境") + assert outcome.status is CriterionStatus.BREACHED + + +def test_always_invalid_parser_is_sync_and_never_raises() -> None: + parser = AlwaysInvalidParser() + parsed = parser.parse(ModelReply(call_id=None, content="```python\nprint(1)\n```", thinking="")) + assert isinstance(parsed.decision, InvalidDecision) + assert parsed.decision.explanation.strip() + # 契约：回填历史的那段不许比模型原文长。 + assert len(parsed.history_text) <= len("```python\nprint(1)\n```") + assert parser.parameters()["kind"] == "always_invalid" + + +# --------------------------------------------------------------------------- +# 十、取消 +# --------------------------------------------------------------------------- + + +def test_cancelled_raised_passes() -> None: + outcome = check_cancelled_raised(raised=asyncio.CancelledError()) + assert outcome.status is CriterionStatus.PASSED + + +def test_cancelled_raised_breaches_when_swallowed() -> None: + """取消被吞掉之后返回一个结果，调用方的结构化并发就断了。""" + assert check_cancelled_raised(raised=None).status is CriterionStatus.BREACHED + + +def test_cancelled_raised_breaches_on_other_exception() -> None: + outcome = check_cancelled_raised(raised=RuntimeError("别的错")) + assert outcome.status is CriterionStatus.BREACHED + assert "RuntimeError" in outcome.evidence + + +def test_lease_returned_passes_and_breaches() -> None: + assert check_lease_returned(borrowed=True, timeout_s=1.0, pool_size=1).status is ( + CriterionStatus.PASSED + ) + assert check_lease_returned(borrowed=False, timeout_s=1.0, pool_size=1).status is ( + CriterionStatus.BREACHED + ) + + +# --------------------------------------------------------------------------- +# 十一、子进程编排：时机判定这个纯函数 +# --------------------------------------------------------------------------- + + +def test_should_kill_after_step_waits_for_enough_steps() -> None: + read = as_read(intent(), model_result(), step_completed()) + assert should_kill(read, timing=KillTiming.AFTER_STEP, after_steps=2) is False + assert should_kill(read, timing=KillTiming.AFTER_STEP, after_steps=1) is True + + +def test_should_kill_after_step_rejects_trailing_intent() -> None: + read = as_read(step_completed(), intent(call_index=1, result_id="r1")) + assert should_kill(read, timing=KillTiming.AFTER_STEP, after_steps=1) is False + + +def test_should_kill_at_intent_requires_never_policy() -> None: + """`safe` 那条意图不算命中：悬在它上面续跑会重放动作接着跑，停止原因不是状态未知。""" + never = as_read(step_completed(), intent(call_index=1, result_id="r1", replay_policy="never")) + safe = as_read(step_completed(), intent(call_index=1, result_id="r1", replay_policy="safe")) + assert should_kill(never, timing=KillTiming.AT_INTENT, after_steps=1) is True + assert should_kill(safe, timing=KillTiming.AT_INTENT, after_steps=1) is False + + +def test_should_kill_at_intent_rejects_trailing_step() -> None: + read = as_read(step_completed()) + assert should_kill(read, timing=KillTiming.AT_INTENT, after_steps=0) is False + + +def test_should_kill_on_empty_log() -> None: + empty = LogRead(payloads=(), torn=False, bad_lines=()) + assert should_kill(empty, timing=KillTiming.AFTER_STEP, after_steps=0) is False + assert should_kill(empty, timing=KillTiming.AT_INTENT, after_steps=0) is False + + +# --------------------------------------------------------------------------- +# 十二、子进程编排：真起一个假子进程 +# --------------------------------------------------------------------------- + +#: 假子进程：往指定文件里逐条追加记录，中间留出足够父进程轮询到的间隔，然后一直睡着不退出。 +#: 它不打模型、不起容器，只是一个会按顺序写文件的东西。 +_FAKE_CHILD = """ +import json, sys, time +path = sys.argv[1] +records = json.loads(sys.argv[2]) +with open(path, "a", encoding="utf-8") as handle: + for record in records: + handle.write(json.dumps(record) + "\\n") + handle.flush() + time.sleep(0.05) +time.sleep(30) +""" + +_FAKE_CHILD_EXITS = """ +import json, sys +path = sys.argv[1] +records = json.loads(sys.argv[2]) +with open(path, "a", encoding="utf-8") as handle: + for record in records: + handle.write(json.dumps(record) + "\\n") +""" + + +async def test_spawn_and_kill_hits_the_after_step_timing(tmp_path: Path) -> None: + log_path = tmp_path / f"{RUN_ID}.jsonl" + records = [intent(), model_result(), step_completed()] + outcome = await spawn_and_kill( + argv=[sys.executable, "-c", _FAKE_CHILD, str(log_path), json.dumps(records)], + log_path=log_path, + timing=KillTiming.AFTER_STEP, + after_steps=1, + timeout_s=20.0, + ) + assert outcome.hit is True + assert outcome.steps == 1 + assert outcome.model_calls == 1 + assert outcome.snapshot == log_path.read_bytes() + + +async def test_spawn_and_kill_reports_a_miss_when_the_child_exits(tmp_path: Path) -> None: + """子进程在命中时机之前就退出了要报出来，不许悄悄当成命中。""" + log_path = tmp_path / f"{RUN_ID}.jsonl" + records = [intent(), model_result()] + outcome = await spawn_and_kill( + argv=[sys.executable, "-c", _FAKE_CHILD_EXITS, str(log_path), json.dumps(records)], + log_path=log_path, + timing=KillTiming.AFTER_STEP, + after_steps=1, + timeout_s=20.0, + ) + assert outcome.hit is False + assert "退出" in outcome.reason + + +async def test_spawn_and_kill_reports_a_miss_on_timeout(tmp_path: Path) -> None: + log_path = tmp_path / f"{RUN_ID}.jsonl" + outcome = await spawn_and_kill( + argv=[sys.executable, "-c", _FAKE_CHILD, str(log_path), json.dumps([intent()])], + log_path=log_path, + timing=KillTiming.AFTER_STEP, + after_steps=1, + timeout_s=0.5, + ) + assert outcome.hit is False + assert "没命中时机" in outcome.reason + + +# --------------------------------------------------------------------------- +# 十三、事件出口、护栏、sidecar、命令行 +# --------------------------------------------------------------------------- + + +class _FakeEvent: + """只带事件出口会读的那三个字段。""" + + def __init__(self, step_idx: int) -> None: + self.kind = EventKind.STEP_FINISHED + self.run_id = RUN_ID + self.step = type("Step", (), {"step_idx": step_idx})() + + +async def test_event_sink_writes_one_line_per_event(tmp_path: Path) -> None: + sink = JsonlEventSink(tmp_path / f"{RUN_ID}.events.jsonl") + await sink.emit(_FakeEvent(0)) # type: ignore[arg-type] + await sink.emit(_FakeEvent(1)) # type: ignore[arg-type] + lines = (tmp_path / f"{RUN_ID}.events.jsonl").read_text(encoding="utf-8").splitlines() + assert [json.loads(line)["step_idx"] for line in lines] == [0, 1] + assert sink.delivered == 2 + + +def test_event_sink_parameters_carry_no_path() -> None: + """路径进参数快照会让续跑报一次假的参数漂移。""" + sink = JsonlEventSink("/tmp/whatever.jsonl") + assert sink.parameters() == {"kind": "jsonl_events"} + + +def test_call_guard_accounting() -> None: + guard = CallGuard(limit=10) + guard.charge(4) + assert guard.remaining == 6 + assert guard.affordable(6) is True + assert guard.affordable(7) is False + + +def test_write_sidecars_shape(tmp_path: Path) -> None: + result = RunResult( + run_id=RUN_ID, stop_reason=StopReason.STEP_BUDGET, final_answer=None, steps=() + ) + write_sidecars( + runs_dir=tmp_path, + run_id=RUN_ID, + scenario="appworld", + fault="step_budget", + result=result, + wall_ms=12, + model_calls=3, + sink_failures=0, + env_executions=3, + task_id="t0", + phase=None, + resumed_from_step=None, + ) + meta = json.loads((tmp_path / f"{RUN_ID}.meta.json").read_text(encoding="utf-8")) + assert meta["fault"] == "step_budget" + assert meta["scenario"] == "appworld" + assert meta["success"] is None + assert set(meta) == { + "scenario", + "task_id", + "phase", + "wall_ms", + "model_calls", + "sink_failures", + "env_executions", + "fault", + "success", + "resumed_from_step", + } + stored = json.loads((tmp_path / f"{RUN_ID}.result.json").read_text(encoding="utf-8")) + assert stored["stop_reason"] == "step_budget" + + +def test_write_sidecars_omits_result_when_there_is_none(tmp_path: Path) -> None: + """取消那条路上 `run` 不返回结果，硬造一份是伪造。""" + write_sidecars( + runs_dir=tmp_path, + run_id=RUN_ID, + scenario="appworld", + fault="cancel_model", + result=None, + wall_ms=1, + model_calls=2, + sink_failures=0, + env_executions=1, + task_id="t0", + phase=None, + resumed_from_step=None, + ) + assert not (tmp_path / f"{RUN_ID}.result.json").exists() + assert (tmp_path / f"{RUN_ID}.meta.json").exists() + + +async def test_guarded_turns_a_crash_into_undetermined() -> None: + """某一类跑到一半炸了，前面几类已经花钱跑出来的结论不能跟着丢。""" + + async def boom() -> FaultReport: + raise RuntimeError("容器起不来") + + report = await guarded("cancel_env", boom()) + assert report.fault == "cancel_env" + assert report.undetermineds + assert "容器起不来" in report.undetermineds[0].evidence + assert not report.breaches + + +async def test_guarded_passes_a_normal_report_through() -> None: + async def fine() -> FaultReport: + return FaultReport(fault="x", criteria=(check_env_untouched(executions=0, source="假"),)) + + report = await guarded("x", fine()) + assert not report.breaches + assert not report.undetermineds + + +def test_main_requires_budget_calls(tmp_path: Path) -> None: + with pytest.raises(SystemExit): + main(["--runs-dir", str(tmp_path)]) + + +def test_main_rejects_unknown_fault(tmp_path: Path) -> None: + with pytest.raises(SystemExit): + main(["--runs-dir", str(tmp_path), "--budget-calls", "10", "--fault", "不存在"]) + + +def test_parser_accepts_repeated_fault_flags() -> None: + args = build_parser().parse_args( + [ + "--runs-dir", + "x", + "--budget-calls", + "10", + "--fault", + "cancel_model", + "--fault", + "cancel_env", + ] + ) + assert args.fault == ["cancel_model", "cancel_env"]