From 292a936d42d41f8b62677569f2bbd271db5fc7fc Mon Sep 17 00:00:00 2001 From: iomgaa Date: Tue, 11 Aug 2026 08:27:09 -0400 Subject: [PATCH] =?UTF-8?q?feat(soak):=20=E8=AE=B0=E5=88=86=E6=9D=BF?= =?UTF-8?q?=E2=80=94=E2=80=94=E5=8D=81=E4=B8=80=E6=9D=A1=E4=B8=8D=E5=8F=98?= =?UTF-8?q?=E9=87=8F=E9=80=90=E6=9D=A1=E5=88=A4=E5=AE=9A=EF=BC=8C=E4=BB=BB?= =?UTF-8?q?=E4=B8=80=E8=A2=AB=E5=87=BB=E7=A9=BF=E6=95=B4=E6=89=B9=E5=A4=B1?= =?UTF-8?q?=E8=B4=A5?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 判定分三档：通过 / 击穿 / 无法判定。第三档不许折算成通过——缺文件、缺字段导致判不了， 和判过了是两回事，混起来会让一次什么都没验的跑看起来全绿。退出码 1 是击穿，2 是只有 无法判定，--allow-undetermined 只放行后者。 守的东西：日志读得回来、跨进程的结果与内存里逐字段相等、步号连续、意图都有归宿（至多 一条悬空且必须在末尾，那正是崩溃点）、动作结果与步记录说同一件事、提示词字符数单调不 减、事件条数等于本进程走完的步数、投递失败计数对得上、并发之间不串台、停止原因与轨迹 自洽。 报告里不出现模型原文、观察、工具名与文档片段——压测语料里有第三方的真实文档，而报告 是要贴给人看的。有四条测试拿哨兵字符串验它确实漏不出去，其中一条是拿恶意工具名试出来 的：原本打算按字符白名单放行工具名，白名单恰好把哨兵放了过去。 每条不变量都配一个「构造出违反它的日志、验它确实报击穿」的用例——一个永远返回通过的 判定器比没有判定器更糟。 Co-Authored-By: Claude Opus 5 (1M context) --- tools/soak/scoreboard.py | 1667 +++++++++++++++++++++++++++ tools/soak/tests/test_scoreboard.py | 857 ++++++++++++++ 2 files changed, 2524 insertions(+) create mode 100644 tools/soak/scoreboard.py create mode 100644 tools/soak/tests/test_scoreboard.py diff --git a/tools/soak/scoreboard.py b/tools/soak/scoreboard.py new file mode 100644 index 0000000..caa9bfa --- /dev/null +++ b/tools/soak/scoreboard.py @@ -0,0 +1,1667 @@ +"""压测记分板：把一批压测产物判成「通过 / 击穿 / 无法判定」，再渲染成一份 markdown 报告。 + +它读的是一次批次跑完之后留在磁盘上的四种文件，每个 run 一组： + + /.jsonl polyloop 的 JsonlRunStore 写的日志，一行一条记录 + /.result.json 驱动写的 polyloop.serialization.encode(RunResult) + /.events.jsonl 事件出口写的，一条事件一行 + /.meta.json 驱动写的运行元信息 + +`.jsonl` 一定在（run 是按它枚举出来的），另外三个可能缺——进程被 SIGKILL 时来不及写。 +**缺文件本身是一条要报告的观察，不是崩溃理由，也不是「通过」。** + +## 三档判定，不是两档 + +一条不变量的判定有三个取值：通过、击穿、无法判定。第三档独立存在，不许折算成前两档中的 +任何一个。缺文件、缺字段导致判不了，和判过了是两回事——压成一档的话，一次什么都没验成的 +跑会显示成全绿，而那正是最需要被看见的情况。数据结构上这条是硬的：`InvariantResult` 分开存 +`breaches` 与 `undetermined` 两串证据，判定由它们算出来，没有一个可以手填的「通过」。 + +击穿必须带具体证据（哪个 run、哪一行 / 哪一步、期望什么实际什么），这条由 `Evidence` 的 +构造期校验守住。一个只会说「有问题」的判定器等于没有判定器。 + +## 报告里不出现任何模型原文与文档片段 + +压测语料里有第三方的真实文档，而报告是要贴给人看的。所以报告里只出现计数、枚举取值、 +run_id、行号与步号，别的一律只报长度不报内容——**工具名与 call_id 也算别的**：一次没通过 +校验的工具调用会把模型编的那串原样记进 `tool_name`，而那串可以是任何东西。 + +只有 run_id、场景名、故障名这三样字符串会原样出现，它们过一遍 `_safe_id`（只留字母数字与 +`_.-`，其余换成 `?`，再截长）。剩下唯一一处外来文本是解码失败时库抛的异常消息，它过 +`_scrub` 压成一行并截到 160 字——那些消息是库自己写的，最多引一个枚举取值或一个快照键名， +引不到观察或模型原文（`polyloop.serialization` 的取值函数对字符串字段只报类型不报内容）。 + +## 不用 JsonlRunStore.read_log 读日志 + +那个方法读回来的是一份 `RunLog`，没有行号，也在撞见第一处损坏时就整份抛出。记分板要的 +恰好相反：逐行读、每行单独判、坏了也接着读完，因为「第几行坏了、后面还有没有记录」正是 +要报告的东西。撕裂尾行的判据照抄存储那边：**看有没有被换行终结，不看能不能解析**。 +""" + +import argparse +import json +import math +from collections.abc import Callable, Mapping, Sequence +from dataclasses import dataclass +from enum import StrEnum +from pathlib import Path + +from polyloop.serialization import ( + DecodeError, + decode_intent, + decode_model_call_result, + decode_run_finished, + decode_run_result, + decode_run_started, + decode_step_completed, +) +from polyloop.stores import RECORD_KEY +from polyloop.types import ( + Intent, + IntentKind, + ModelCallResult, + RunFinished, + RunResult, + RunStarted, + StepCompleted, + StopReason, +) + +LOG_SUFFIX = ".jsonl" +RESULT_SUFFIX = ".result.json" +EVENTS_SUFFIX = ".events.jsonl" +META_SUFFIX = ".meta.json" + +#: 步数上限在参数快照里的键名。快照由 `RunRequest.parameter_snapshot()` 拼出来， +#: 值是十进制字符串。 +MAX_STEPS_SNAPSHOT_KEY = "request.max_steps" + +#: 单个 run 单条不变量最多列几条证据。超出的部分折成一条「还有 N 条」。 +#: 一份四十步的坏日志能刷出四十行同样的证据，那种报告没人会读到底。 +MAX_EVIDENCE_PER_RUN = 5 + +#: 日志里每种标签对应的解码器。标签取值与 `polyloop.stores` 那边一致。 +_DECODERS: Mapping[str, Callable[[Mapping[str, object]], object]] = { + "run_started": decode_run_started, + "intent": decode_intent, + "model_call_result": decode_model_call_result, + "step_completed": decode_step_completed, + "run_finished": decode_run_finished, +} + +#: 步记录里内容不进报告的字段，只报长度。 +#: +#: `tool_name` 与 `call_id` 也在里面：它们看着像标识，可最终仍然来自模型输出——一次没通过 +#: 校验的工具调用会把模型编的那串原样记进 `tool_name`，而那串可以是任何东西。 +_STEP_OPAQUE_FIELDS = frozenset( + { + "raw_output", + "action", + "parse_error", + "observation", + "tool_arguments", + "tool_name", + "call_id", + } +) + +_STEP_FIELDS = ( + "step_idx", + "raw_output", + "content_chars", + "thinking_chars", + "action", + "parse_ok", + "parse_error", + "observation", + "observation_is_synthetic", + "observation_truncated_chars", + "prompt_chars", + "call_id", + "step_wall_ms", + "tool_name", + "tool_arguments", + "action_status", + "env_reported_completion", + "schema_version", +) + +_SAFE_NAME_CHARS = frozenset("abcdefghijklmnopqrstuvwxyzABCDEFGHIJKLMNOPQRSTUVWXYZ0123456789_.-") + + +# --------------------------------------------------------------------------- +# 判定与证据 +# --------------------------------------------------------------------------- + + +class Verdict(StrEnum): + """一条不变量在这一批 run 上的判定。""" + + PASSED = "passed" + BREACHED = "breached" + UNDETERMINED = "undetermined" + + +_VERDICT_LABELS: Mapping[Verdict, str] = { + Verdict.PASSED: "通过", + Verdict.BREACHED: "击穿", + Verdict.UNDETERMINED: "无法判定", +} + + +@dataclass(frozen=True, slots=True, kw_only=True) +class Evidence: + """一条击穿证据，或者一条「为什么判不了」。 + + 击穿必须同时写 `expected` 与 `actual`，判不了必须写 `reason`——两者都没有的证据在构造期 + 就报错。理由见模块 docstring：只说「有问题」的判定器等于没有判定器，而这条限制让那种 + 证据根本构造不出来。 + """ + + run_id: str + line_no: int | None = None + step_idx: int | None = None + expected: str | None = None + actual: str | None = None + reason: str | None = None + + def __post_init__(self) -> None: + if self.reason is None and (self.expected is None or self.actual is None): + raise ValueError( + "一条证据要么写清楚期望与实际（击穿），要么写清楚为什么判不了（无法判定）" + ) + + def describe(self) -> str: + """渲染成一行。报告里的证据都长这样。""" + where = f"run `{self.run_id}`" + if self.line_no is not None: + where += f" 第 {self.line_no} 行" + if self.step_idx is not None: + where += f" 第 {self.step_idx} 步" + if self.expected is None or self.actual is None: + return f"{where}：{self.reason}" + body = f"期望 {self.expected}，实际 {self.actual}" + if self.reason is not None: + body += f"（{self.reason}）" + return f"{where}：{body}" + + +@dataclass(frozen=True, slots=True, kw_only=True) +class CheckOutcome: + """一条不变量在**一个** run 上的检查结果。""" + + breaches: tuple[Evidence, ...] = () + undetermined: tuple[Evidence, ...] = () + + +@dataclass(frozen=True, slots=True, kw_only=True) +class InvariantResult: + """一条不变量在**整批** run 上的判定。 + + 判定不是一个字段，是 `verdict` 算出来的：有击穿就是击穿，否则有判不了的就是判不了， + 两串都空才是通过。没有一条路径能把「判不了」写成「通过」。 + """ + + name: str + description: str + breaches: tuple[Evidence, ...] = () + undetermined: tuple[Evidence, ...] = () + + def __post_init__(self) -> None: + for item in self.breaches: + if item.expected is None or item.actual is None: + raise ValueError(f"不变量 {self.name!r} 的击穿证据缺期望或实际值") + + @property + def verdict(self) -> Verdict: + if self.breaches: + return Verdict.BREACHED + if self.undetermined: + return Verdict.UNDETERMINED + return Verdict.PASSED + + +# --------------------------------------------------------------------------- +# 读盘 +# --------------------------------------------------------------------------- + + +@dataclass(frozen=True, slots=True, kw_only=True) +class LoadedLine: + """日志里被换行终结的一行。 + + 解不开的行也在这里，`record` 为空、`failure` 写着为什么——记分板要接着往下读， + 「后面还有没有记录」本身就是要报告的东西。 + """ + + number: int + tag: str + payload: Mapping[str, object] + record: object | None = None + failure: str | None = None + + +@dataclass(frozen=True, slots=True, kw_only=True) +class RunFacts: + """一个 run 在磁盘上留下的全部东西，读进内存之后的样子。 + + 四个 `*_error` 字段区分「文件缺了」与「文件在但读不了」——两者都让相关不变量判不了， + 但要报告给人的话不是一回事。 + """ + + run_id: str + lines: tuple[LoadedLine, ...] = () + torn_tail: bool = False + log_error: str | None = None + result: RunResult | None = None + result_error: str | None = None + #: 每条事件的 run_id，缺键或不是字符串时那一项为空。整个文件缺失时它是 None。 + event_run_ids: tuple[str | None, ...] | None = None + events_torn_tail: bool = False + events_error: str | None = None + meta: Mapping[str, object] | None = None + meta_error: str | None = None + + def records(self) -> tuple[object, ...]: + return tuple(line.record for line in self.lines if line.record is not None) + + def failures(self) -> tuple[LoadedLine, ...]: + return tuple(line for line in self.lines if line.failure is not None) + + def payloads_tagged(self, tag: str) -> tuple[LoadedLine, ...]: + return tuple(line for line in self.lines if line.tag == tag) + + def run_started(self) -> RunStarted | None: + found = [item for item in self.records() if isinstance(item, RunStarted)] + return found[-1] if found else None + + def run_finished(self) -> RunFinished | None: + found = [item for item in self.records() if isinstance(item, RunFinished)] + return found[-1] if found else None + + def intents(self) -> tuple[tuple[int, Intent], ...]: + return tuple( + (line.number, line.record) for line in self.lines if isinstance(line.record, Intent) + ) + + def model_results(self) -> tuple[ModelCallResult, ...]: + return tuple(item for item in self.records() if isinstance(item, ModelCallResult)) + + def steps(self) -> tuple[tuple[int, StepCompleted], ...]: + return tuple( + (line.number, line.record) + for line in self.lines + if isinstance(line.record, StepCompleted) + ) + + def effective_result(self) -> RunResult | None: + """判停止原因用哪一份结果。 + + 先用日志里 `run_finished` 内嵌的那份：取消那条路径上 `session.run` 原样重抛 + `CancelledError`、根本不返回结果，于是驱动写不出 `.result.json`，而日志里的结束 + 记录仍然在。两份都在时它们必须相等，那是不变量二在管的事。 + """ + finished = self.run_finished() + if finished is not None: + return finished.result + return self.result + + +def _read_bytes(path: Path) -> tuple[bytes | None, str | None]: + if not path.exists(): + return None, f"没有 {path.name}" + try: + return path.read_bytes(), None + except OSError as exc: + return None, f"读 {path.name} 失败：{_scrub(str(exc))}" + + +def _split_terminated(raw: bytes) -> tuple[list[tuple[int, bytes]], bool]: + """按存储那边的规矩切行：判据是有没有被换行终结，不是能不能解析。 + + 末尾那段没有换行的字节对应的那次写从来没有被确认过，按契约它就是没发生。返回值第二项 + 说的就是「有没有这么一段」——它是进程被杀在写入中途的证据，要报告出来。 + + 行号按原始文件的行数走，空行也占一号，这样报出来的号和 `sed -n 'Np'` 对得上。 + """ + chunks = raw.split(b"\n") + torn = bool(chunks) and bool(chunks[-1].strip()) + terminated = chunks[:-1] if torn else chunks + numbered = [ + (number, chunk) for number, chunk in enumerate(terminated, start=1) if chunk.strip() + ] + return numbered, torn + + +def _load_log(path: Path) -> tuple[tuple[LoadedLine, ...], bool, str | None]: + raw, error = _read_bytes(path) + if raw is None: + return (), False, error + numbered, torn = _split_terminated(raw) + return tuple(_load_line(number, chunk) for number, chunk in numbered), torn, None + + +def _load_line(number: int, chunk: bytes) -> LoadedLine: + try: + payload = json.loads(chunk) + except (UnicodeDecodeError, json.JSONDecodeError) as exc: + return LoadedLine( + number=number, + tag="", + payload={}, + failure=f"读不了（{type(exc).__name__}）", + ) + if not isinstance(payload, dict): + return LoadedLine(number=number, tag="", payload={}, failure="不是一个 JSON 对象") + if RECORD_KEY not in payload: + return LoadedLine( + number=number, tag="", payload=payload, failure=f"没有 {RECORD_KEY!r} 标签" + ) + tag = payload[RECORD_KEY] + if not isinstance(tag, str) or tag not in _DECODERS: + return LoadedLine( + number=number, + tag="", + payload=payload, + failure=f"记录类型认不得（标签长度 {len(str(tag))}）", + ) + try: + record = _DECODERS[tag](payload) + except (DecodeError, ValueError, TypeError) as exc: + return LoadedLine( + number=number, + tag=tag, + payload=payload, + failure=f"{tag} 解不出来：{_scrub(str(exc))}", + ) + return LoadedLine(number=number, tag=tag, payload=payload, record=record) + + +def _load_result(path: Path) -> tuple[RunResult | None, str | None]: + raw, error = _read_bytes(path) + if raw is None: + return None, error + try: + payload = json.loads(raw) + except (UnicodeDecodeError, json.JSONDecodeError) as exc: + return None, f"{path.name} 不是合法 JSON（{type(exc).__name__}）" + if not isinstance(payload, dict): + return None, f"{path.name} 不是一个 JSON 对象" + try: + return decode_run_result(payload), None + except (DecodeError, ValueError, TypeError) as exc: + return None, f"{path.name} 解不出 RunResult：{_scrub(str(exc))}" + + +def _load_events(path: Path) -> tuple[tuple[str | None, ...] | None, bool, str | None]: + raw, error = _read_bytes(path) + if raw is None: + return None, False, error + numbered, torn = _split_terminated(raw) + run_ids: list[str | None] = [] + for index, chunk in numbered: + try: + payload = json.loads(chunk) + except (UnicodeDecodeError, json.JSONDecodeError) as exc: + return None, torn, f"{path.name} 第 {index} 行不是合法 JSON（{type(exc).__name__}）" + if not isinstance(payload, dict): + return None, torn, f"{path.name} 第 {index} 行不是一个 JSON 对象" + value = payload.get("run_id") + run_ids.append(value if isinstance(value, str) else None) + return tuple(run_ids), torn, None + + +def _load_meta(path: Path) -> tuple[Mapping[str, object] | None, str | None]: + raw, error = _read_bytes(path) + if raw is None: + return None, error + try: + payload = json.loads(raw) + except (UnicodeDecodeError, json.JSONDecodeError) as exc: + return None, f"{path.name} 不是合法 JSON（{type(exc).__name__}）" + if not isinstance(payload, dict): + return None, f"{path.name} 不是一个 JSON 对象" + return payload, None + + +def list_run_ids(runs_dir: Path) -> tuple[str, ...]: + """枚举这个目录里的 run。 + + `.events.jsonl` 也以 `.jsonl` 结尾，所以要显式排掉——不排的话每个 run 会被数成两个， + 其中一个的「日志」是一串事件，然后整批报告都是错的。 + """ + ids = [ + path.name[: -len(LOG_SUFFIX)] + for path in runs_dir.glob(f"*{LOG_SUFFIX}") + if not path.name.endswith(EVENTS_SUFFIX) + ] + return tuple(sorted(ids)) + + +def load_run(runs_dir: Path, run_id: str) -> RunFacts: + """把一个 run 的四个文件读进内存。缺文件不抛异常，记在对应的 `*_error` 上。""" + lines, torn, log_error = _load_log(runs_dir / f"{run_id}{LOG_SUFFIX}") + result, result_error = _load_result(runs_dir / f"{run_id}{RESULT_SUFFIX}") + event_run_ids, events_torn, events_error = _load_events(runs_dir / f"{run_id}{EVENTS_SUFFIX}") + meta, meta_error = _load_meta(runs_dir / f"{run_id}{META_SUFFIX}") + return RunFacts( + run_id=run_id, + lines=lines, + torn_tail=torn, + log_error=log_error, + result=result, + result_error=result_error, + event_run_ids=event_run_ids, + events_torn_tail=events_torn, + events_error=events_error, + meta=meta, + meta_error=meta_error, + ) + + +# --------------------------------------------------------------------------- +# 文本消毒：报告里只出现计数、枚举取值、run_id、行号与步号 +# --------------------------------------------------------------------------- + + +def _scrub(text: str, *, limit: int = 160) -> str: + """把一段来自外部的文本压成一行、截到上限。 + + 它管的是异常消息这类「本来就该短、但来源不完全可控」的串。真正的文本字段不走这里， + 它们根本不进报告——见模块 docstring。 + """ + flat = " ".join(text.split()) + if len(flat) <= limit: + return flat + return flat[:limit] + "…" + + +def _safe_id(name: str, *, limit: int = 64) -> str: + """把一个标识压成安全形态：只留字母数字与 `_.-`，其余换成 `?`，再截长。 + + 只有 run_id、场景名、故障名走这里。它们由驱动写，本来就该长这样；这一层是防着一份被 + 改过的产物把别的东西塞进这几个位置。 + """ + cleaned = "".join(char if char in _SAFE_NAME_CHARS else "?" for char in name) + if len(cleaned) <= limit: + return cleaned + return cleaned[:limit] + "…" + + +def _cell(text: str) -> str: + """markdown 表格单元格：竖线要转义，换行要压掉，否则整张表塌了。""" + return text.replace("|", "\\|").replace("\n", " ") + + +# --------------------------------------------------------------------------- +# 逐个 run 的不变量 +# --------------------------------------------------------------------------- + + +@dataclass(frozen=True, slots=True, kw_only=True) +class CheckConfig: + """判定时需要、但记分板自己定不了的东西。""" + + #: 带完成标记（`ToolSpec.completes_run`）的工具名。**由调用方传进来，不硬编码**—— + #: 哪些工具算完成通路是场景的事实，不是记分板的。传 None 表示没给，于是不变量十 + #: 在只剩这条路可走时报「无法判定」而不是编一个答案。 + completing_tools: frozenset[str] | None = None + + +def _cap(items: Sequence[Evidence], run_id: str) -> list[Evidence]: + if len(items) <= MAX_EVIDENCE_PER_RUN: + return list(items) + kept = list(items[:MAX_EVIDENCE_PER_RUN]) + kept.append( + Evidence( + run_id=run_id, + expected="同类证据全部列出", + actual=f"还有 {len(items) - MAX_EVIDENCE_PER_RUN} 条未列出", + ) + ) + return kept + + +def _check_log_readable(facts: RunFacts, config: CheckConfig) -> CheckOutcome: + del config + if facts.log_error is not None: + return CheckOutcome(undetermined=(Evidence(run_id=facts.run_id, reason=facts.log_error),)) + breaches = [ + Evidence( + run_id=facts.run_id, + line_no=line.number, + expected="这一行被换行终结，所以它必须解得开", + actual=line.failure or "", + ) + for line in facts.failures() + ] + return CheckOutcome(breaches=tuple(_cap(breaches, facts.run_id))) + + +def _diff_run_results(left: RunResult, right: RunResult) -> list[str]: + """逐字段比两份运行结果，返回人能读的差异描述。 + + 文本字段只报长度：`final_answer` 与步记录里那几个装文本的字段可能整段是模型原文或 + 环境返回的文档，它们不许进报告。 + """ + diffs: list[str] = [] + if left.run_id != right.run_id: + diffs.append(f"run_id：{_safe_id(left.run_id)} vs {_safe_id(right.run_id)}") + if left.stop_reason != right.stop_reason: + diffs.append(f"stop_reason：{left.stop_reason.value} vs {right.stop_reason.value}") + if left.final_answer != right.final_answer: + diffs.append( + f"final_answer 文本不同（长度 {_text_len(left.final_answer)} vs " + f"{_text_len(right.final_answer)}）" + ) + if left.schema_version != right.schema_version: + diffs.append(f"schema_version：{left.schema_version} vs {right.schema_version}") + if left.event_delivery_failures != right.event_delivery_failures: + diffs.append( + f"event_delivery_failures：{left.event_delivery_failures} vs " + f"{right.event_delivery_failures}" + ) + if len(left.steps) != len(right.steps): + diffs.append(f"steps 条数：{len(left.steps)} vs {len(right.steps)}") + return diffs + for index, (one, other) in enumerate(zip(left.steps, right.steps, strict=True)): + for name in _STEP_FIELDS: + first = getattr(one, name) + second = getattr(other, name) + if first == second: + continue + if name in _STEP_OPAQUE_FIELDS: + diffs.append( + f"steps[{index}].{name} 文本不同" + f"（长度 {_text_len(first)} vs {_text_len(second)}）" + ) + else: + diffs.append(f"steps[{index}].{name}：{first!r} vs {second!r}") + return diffs + + +def _text_len(value: str | None) -> str: + return "空" if value is None else str(len(value)) + + +def _check_result_matches_log(facts: RunFacts, config: CheckConfig) -> CheckOutcome: + del config + finished = facts.run_finished() + if facts.result is None: + return CheckOutcome( + undetermined=( + Evidence( + run_id=facts.run_id, + reason=facts.result_error or f"没有 {RESULT_SUFFIX}", + ), + ) + ) + if finished is None: + return CheckOutcome( + undetermined=(Evidence(run_id=facts.run_id, reason="日志里没有 run_finished 记录"),) + ) + if facts.result == finished.result: + return CheckOutcome() + diffs = _diff_run_results(facts.result, finished.result) + breaches = [ + Evidence( + run_id=facts.run_id, + expected="跨进程读回来的 RunResult 与日志里那份逐字段相等", + actual=diff, + ) + for diff in diffs + ] + if not breaches: + breaches = [ + Evidence( + run_id=facts.run_id, + expected="跨进程读回来的 RunResult 与日志里那份相等", + actual="两份不相等，但逐字段比不出差异", + ) + ] + return CheckOutcome(breaches=tuple(_cap(breaches, facts.run_id))) + + +def _check_step_indices(facts: RunFacts, config: CheckConfig) -> CheckOutcome: + del config + breaches: list[Evidence] = [] + for position, (number, step) in enumerate(facts.steps()): + if step.step.step_idx != position: + breaches.append( + Evidence( + run_id=facts.run_id, + line_no=number, + expected=f"step_idx = {position}", + actual=f"step_idx = {step.step.step_idx}", + ) + ) + return CheckOutcome(breaches=tuple(_cap(breaches, facts.run_id))) + + +def _check_intents_resolved(facts: RunFacts, config: CheckConfig) -> CheckOutcome: + del config + model_ids = {item.result_id for item in facts.model_results()} + action_ids = {step.result_id for _, step in facts.steps() if step.result_id is not None} + intents = facts.intents() + dangling: list[tuple[int, int, Intent]] = [] + for position, (number, intent) in enumerate(intents): + resolved = ( + intent.result_id in model_ids + if intent.kind is IntentKind.MODEL_CALL + else intent.result_id in action_ids + ) + if not resolved: + dangling.append((position, number, intent)) + if not dangling: + return CheckOutcome() + last_position = len(intents) - 1 + if len(dangling) == 1 and dangling[0][0] == last_position: + # 这正是崩溃点：进程写了意图、还没来得及写结果。恢复要处理的就是它。 + return CheckOutcome() + breaches = [ + Evidence( + run_id=facts.run_id, + line_no=number, + expected=( + f"{intent.kind.value} 意图的 result_id 有归宿，或者它是日志里最后一条意图（崩溃点）" + ), + actual=( + f"悬空，且日志里共有 {len(dangling)} 条悬空意图，" + f"它在第 {position + 1} / {len(intents)} 条" + ), + ) + for position, number, intent in dangling + ] + return CheckOutcome(breaches=tuple(_cap(breaches, facts.run_id))) + + +def _check_step_pairing(facts: RunFacts, config: CheckConfig) -> CheckOutcome: + """`result_id` 为空当且仅当 `action_outcome` 为空。 + + 判据落在**原始载荷**上，不在解出来的记录上：`StepCompleted` 自己在构造期就守这条， + 所以一条违反它的载荷根本解不出记录——那样这条不变量就永远只能是通过，成了摆设。 + 违反它的行同时会被不变量一报出来（它确实读不回来），两条说的是同一处损坏的两个侧面。 + """ + del config + breaches: list[Evidence] = [] + blocked: list[Evidence] = [] + for line in facts.payloads_tagged("step_completed"): + payload = line.payload + if "result_id" not in payload or "action_outcome" not in payload: + blocked.append( + Evidence( + run_id=facts.run_id, + line_no=line.number, + reason="载荷缺 result_id 或 action_outcome，这条判不了", + ) + ) + continue + has_id = payload["result_id"] is not None + has_outcome = payload["action_outcome"] is not None + if has_id != has_outcome: + breaches.append( + Evidence( + run_id=facts.run_id, + line_no=line.number, + expected="result_id 与 action_outcome 同时为空或同时有值", + actual=f"result_id 有值={has_id}，action_outcome 有值={has_outcome}", + ) + ) + return CheckOutcome( + breaches=tuple(_cap(breaches, facts.run_id)), + undetermined=tuple(_cap(blocked, facts.run_id)), + ) + + +def _check_outcome_agrees_with_step(facts: RunFacts, config: CheckConfig) -> CheckOutcome: + """同一条 `step_completed` 里，动作结果与步记录说的必须是同一件事。 + + 两者一次原子落地，所以它们不可能来自两次不同的执行。对不上说明装配那一层把某一侧 + 填错了，而这种错在轨迹里完全看不出来——两个字段各自都合法。 + """ + del config + breaches: list[Evidence] = [] + for number, record in facts.steps(): + outcome = record.action_outcome + if outcome is None: + if record.step.action_status is not None: + breaches.append( + Evidence( + run_id=facts.run_id, + line_no=number, + step_idx=record.step.step_idx, + expected="没有动作结果时步记录的 action_status 为空", + actual=f"action_status = {record.step.action_status.value}", + ) + ) + continue + pairs = ( + ("action_status", outcome.status, record.step.action_status), + ( + "env_reported_completion", + outcome.env_reported_completion, + record.step.env_reported_completion, + ), + ( + "observation_is_synthetic", + outcome.observation_is_synthetic, + record.step.observation_is_synthetic, + ), + ( + "observation_truncated_chars", + outcome.observation_truncated_chars, + record.step.observation_truncated_chars, + ), + ) + for name, from_outcome, from_step in pairs: + if from_outcome != from_step: + breaches.append( + Evidence( + run_id=facts.run_id, + line_no=number, + step_idx=record.step.step_idx, + expected=f"步记录的 {name} 等于动作结果的同名字段", + actual=f"动作结果 {from_outcome!r}，步记录 {from_step!r}", + ) + ) + if outcome.observation != record.step.observation: + breaches.append( + Evidence( + run_id=facts.run_id, + line_no=number, + step_idx=record.step.step_idx, + expected="步记录的 observation 与动作结果的同名字段相同", + actual=( + f"文本不同（长度 {len(outcome.observation)} vs " + f"{len(record.step.observation)}）" + ), + ) + ) + return CheckOutcome(breaches=tuple(_cap(breaches, facts.run_id))) + + +def _check_prompt_chars_monotonic(facts: RunFacts, config: CheckConfig) -> CheckOutcome: + del config + breaches: list[Evidence] = [] + previous: int | None = None + for number, record in facts.steps(): + current = record.step.prompt_chars + if previous is not None and current < previous: + breaches.append( + Evidence( + run_id=facts.run_id, + line_no=number, + step_idx=record.step.step_idx, + expected=f"prompt_chars ≥ 上一步的 {previous}", + actual=f"prompt_chars = {current}", + ) + ) + previous = current + return CheckOutcome(breaches=tuple(_cap(breaches, facts.run_id))) + + +class _MetaError(Exception): + """`.meta.json` 缺字段或字段类型不对。它让相关不变量判不了，不让它们击穿。""" + + +def _meta_field(meta: Mapping[str, object] | None, key: str) -> object: + if meta is None: + raise _MetaError(f"没有 {META_SUFFIX}") + if key not in meta: + raise _MetaError(f"{META_SUFFIX} 缺字段 {key!r}") + return meta[key] + + +def _meta_int(meta: Mapping[str, object] | None, key: str) -> int: + value = _meta_field(meta, key) + if not isinstance(value, int) or isinstance(value, bool): + raise _MetaError(f"{META_SUFFIX} 的 {key} 应当是整数，收到 {type(value).__name__}") + return value + + +def _meta_optional_int(meta: Mapping[str, object] | None, key: str) -> int | None: + value = _meta_field(meta, key) + if value is None: + return None + if not isinstance(value, int) or isinstance(value, bool): + raise _MetaError(f"{META_SUFFIX} 的 {key} 应当是整数或 null，收到 {type(value).__name__}") + return value + + +def _check_events_match_local_steps(facts: RunFacts, config: CheckConfig) -> CheckOutcome: + """事件条数 == 本进程真的走完的步数。 + + 事件只在**这次进程里真的走完**的步上发，从日志读回来直接跳过的步不发 + （`polyloop.ports` 的 `EventKind.STEP_FINISHED`）。所以续跑时要把跳过的那一段减掉。 + """ + del config + if facts.event_run_ids is None or facts.events_error is not None: + return CheckOutcome( + undetermined=( + Evidence( + run_id=facts.run_id, + reason=facts.events_error or f"没有 {EVENTS_SUFFIX}", + ), + ) + ) + try: + resumed = _meta_optional_int(facts.meta, "resumed_from_step") or 0 + except _MetaError as problem: + return CheckOutcome( + undetermined=(Evidence(run_id=facts.run_id, reason=f"{problem}（读不到续跑起点）"),) + ) + expected = len(facts.steps()) - resumed + actual = len(facts.event_run_ids) + if expected == actual: + return CheckOutcome() + return CheckOutcome( + breaches=( + Evidence( + run_id=facts.run_id, + expected=f"{expected} 条事件（日志 {len(facts.steps())} 步 - 续跑跳过 {resumed} 步）", + actual=f"{actual} 条事件", + ), + ) + ) + + +def _check_delivery_failures(facts: RunFacts, config: CheckConfig) -> CheckOutcome: + del config + result = facts.effective_result() + if result is None: + return CheckOutcome( + undetermined=( + Evidence( + run_id=facts.run_id, + reason=(facts.result_error or f"没有 {RESULT_SUFFIX}") + + "，日志里也没有 run_finished", + ), + ) + ) + try: + sink_failures = _meta_int(facts.meta, "sink_failures") + except _MetaError as problem: + return CheckOutcome(undetermined=(Evidence(run_id=facts.run_id, reason=str(problem)),)) + if result.event_delivery_failures == sink_failures: + return CheckOutcome() + return CheckOutcome( + breaches=( + Evidence( + run_id=facts.run_id, + expected=f"event_delivery_failures = 出口自己数的 {sink_failures}", + actual=f"event_delivery_failures = {result.event_delivery_failures}", + ), + ) + ) + + +def _check_no_crosstalk(facts: RunFacts, config: CheckConfig) -> CheckOutcome: + del config + breaches: list[Evidence] = [] + blocked: list[Evidence] = [] + for line in facts.lines: + record = line.record + if record is None: + continue + found = getattr(record, "run_id", None) + if found != facts.run_id: + breaches.append( + Evidence( + run_id=facts.run_id, + line_no=line.number, + expected=f"记录的 run_id = {facts.run_id}", + actual=f"run_id = {_safe_id(str(found))}", + ) + ) + if facts.event_run_ids is None or facts.events_error is not None: + blocked.append( + Evidence( + run_id=facts.run_id, + reason=(facts.events_error or f"没有 {EVENTS_SUFFIX}") + "，事件那一半判不了", + ) + ) + else: + for index, found in enumerate(facts.event_run_ids, start=1): + if found != facts.run_id: + breaches.append( + Evidence( + run_id=facts.run_id, + expected=f"每条事件的 run_id = {facts.run_id}", + actual=f"第 {index} 条事件的 run_id = {_safe_id(str(found))}", + ) + ) + return CheckOutcome( + breaches=tuple(_cap(breaches, facts.run_id)), + undetermined=tuple(blocked), + ) + + +def _snapshot_max_steps(facts: RunFacts) -> int: + started = facts.run_started() + if started is None: + raise _MetaError("日志里没有 run_started，读不到参数快照") + raw = started.parameter_snapshot.get(MAX_STEPS_SNAPSHOT_KEY) + if raw is None: + raise _MetaError(f"参数快照里没有 {MAX_STEPS_SNAPSHOT_KEY}") + try: + return int(raw) + except ValueError as exc: + raise _MetaError( + f"参数快照里的 {MAX_STEPS_SNAPSHOT_KEY} 不是整数：{_safe_id(raw)}" + ) from exc + + +def _check_stop_reason_consistent(facts: RunFacts, config: CheckConfig) -> CheckOutcome: + """停止原因与轨迹自洽。 + + 四条规矩各管一个停止原因，其余取值这里不判——它们的判据要么在库的单元测试里，要么 + 需要记分板拿不到的事实。判不了的就报「无法判定」，不编。 + """ + result = facts.effective_result() + if result is None: + return CheckOutcome( + undetermined=( + Evidence( + run_id=facts.run_id, + reason="既没有 run_finished 也没有 .result.json，读不到停止原因", + ), + ) + ) + reason = result.stop_reason + if reason is StopReason.TASK_COMPLETED: + return _check_task_completed(facts, result, config) + if reason is StopReason.STEP_BUDGET: + try: + max_steps = _snapshot_max_steps(facts) + except _MetaError as problem: + return CheckOutcome(undetermined=(Evidence(run_id=facts.run_id, reason=str(problem)),)) + if len(result.steps) == max_steps: + return CheckOutcome() + return CheckOutcome( + breaches=( + Evidence( + run_id=facts.run_id, + expected=f"step_budget 时步数 = max_steps = {max_steps}", + actual=f"步数 = {len(result.steps)}", + ), + ) + ) + if reason is StopReason.AGENT_FINISHED: + if result.final_answer: + return CheckOutcome() + return CheckOutcome( + breaches=( + Evidence( + run_id=facts.run_id, + expected="agent_finished 时 final_answer 非空", + actual=f"final_answer 长度 {_text_len(result.final_answer)}", + ), + ) + ) + if reason is StopReason.CANCELLED: + if facts.run_finished() is not None: + return CheckOutcome() + return CheckOutcome( + breaches=( + Evidence( + run_id=facts.run_id, + expected="cancelled 时日志里有 run_finished 记录", + actual="日志里没有 run_finished，恢复会把它当成可以续跑", + ), + ) + ) + return CheckOutcome() + + +def _check_task_completed(facts: RunFacts, result: RunResult, config: CheckConfig) -> CheckOutcome: + if not result.steps: + return CheckOutcome( + breaches=( + Evidence( + run_id=facts.run_id, + expected="task_completed 时至少有一步", + actual="步数 = 0", + ), + ) + ) + last = result.steps[-1] + if last.env_reported_completion: + return CheckOutcome() + if config.completing_tools is None: + return CheckOutcome( + undetermined=( + Evidence( + run_id=facts.run_id, + step_idx=last.step_idx, + reason=( + "最后一步没有环境侧完成证据，而带完成标记的工具名没有传进来，" + "另一条通路判不了" + ), + ), + ) + ) + if last.tool_name is not None and last.tool_name in config.completing_tools: + return CheckOutcome() + return CheckOutcome( + breaches=( + Evidence( + run_id=facts.run_id, + step_idx=last.step_idx, + expected=( + "最后一步 env_reported_completion=True，或者 tool_name 在" + f"带完成标记的那 {len(config.completing_tools)} 个工具里" + ), + actual=( + "env_reported_completion=False，tool_name 不在里面" + + ("（那一步没有 tool_name）" if last.tool_name is None else "") + # 工具名不进报告：它最终来自模型输出，见模块 docstring。 + ), + ), + ) + ) + + +@dataclass(frozen=True, slots=True, kw_only=True) +class Invariant: + """一条不变量：名字、一句话说明、以及怎么在一个 run 上判它。""" + + name: str + description: str + check: Callable[[RunFacts, CheckConfig], CheckOutcome] + + +INVARIANTS: tuple[Invariant, ...] = ( + Invariant( + name="日志能被读回来", + description="日志里每一条被换行终结的行都解得出记录；撕裂只允许出现在最后一行。", + check=_check_log_readable, + ), + Invariant( + name="跨进程的结果与内存里的一致", + description=( + f"{RESULT_SUFFIX} 解出来的 RunResult 与日志里 run_finished 内嵌的那份逐字段相等。" + ), + check=_check_result_matches_log, + ), + Invariant( + name="步号连续", + description="step_completed 的 step_idx 从 0 开始逐 1 递增，不重复不跳号。", + check=_check_step_indices, + ), + Invariant( + name="意图都有归宿", + description=( + "每条意图的 result_id 都能在对应的结果记录里找到；" + "至多一条悬空，且必须是日志里最后一条意图。" + ), + check=_check_intents_resolved, + ), + Invariant( + name="步记录的内部不变量", + description="step_completed 的 result_id 为空当且仅当 action_outcome 为空。", + check=_check_step_pairing, + ), + Invariant( + name="动作结果与步记录一致", + description="同一条 step_completed 里，动作结果与步记录的同名字段说的是同一件事。", + check=_check_outcome_agrees_with_step, + ), + Invariant( + name="提示词字符数单调不减", + description="同一个 run 里 prompt_chars 随步号不减——历史只追加。", + check=_check_prompt_chars_monotonic, + ), + Invariant( + name="事件条数等于本进程走完的步数", + description="事件文件的行数 = 日志里的步数 - 续跑跳过的步数。", + check=_check_events_match_local_steps, + ), + Invariant( + name="投递失败计数对得上", + description="结果里的 event_delivery_failures 等于出口自己数的 sink_failures。", + check=_check_delivery_failures, + ), + Invariant( + name="不串台", + description="日志与事件里每一条记录的 run_id 都等于文件名去掉后缀那部分。", + check=_check_no_crosstalk, + ), + Invariant( + name="停止原因与轨迹自洽", + description=( + "task_completed 有完成证据；step_budget 时步数等于上限；" + "agent_finished 时最终回答非空；cancelled 时日志里有结束记录。" + ), + check=_check_stop_reason_consistent, + ), +) + + +# --------------------------------------------------------------------------- +# 统计与基线对照 +# --------------------------------------------------------------------------- + + +@dataclass(frozen=True, slots=True, kw_only=True) +class StepStats: + """一组 run 的步数分布。没有 run 时四项都是空。""" + + runs: int + p50: int | None = None + p90: int | None = None + maximum: int | None = None + mean: float | None = None + + +def _percentile(values: Sequence[int], fraction: float) -> int: + """最近秩分位：排序之后取第 ceil(fraction × n) 个。 + + 分位数有好几种定义，插值那几种会在小样本上给出不存在于数据里的值（比如 13.5 步）。 + 步数是整数计数，报一个半步没有意义，所以取最近秩。 + """ + ordered = sorted(values) + rank = math.ceil(fraction * len(ordered)) + return ordered[max(rank, 1) - 1] + + +def _step_stats(counts: Sequence[int]) -> StepStats: + if not counts: + return StepStats(runs=0) + return StepStats( + runs=len(counts), + p50=_percentile(counts, 0.50), + p90=_percentile(counts, 0.90), + maximum=max(counts), + mean=sum(counts) / len(counts), + ) + + +@dataclass(frozen=True, slots=True, kw_only=True) +class RunSummary: + """逐个 run 那张表的一行。""" + + run_id: str + scenario: str + stop_reason: str + steps: int + success: bool | None + wall_ms: int | None + fault: str | None + + +@dataclass(frozen=True, slots=True, kw_only=True) +class ScenarioStats: + """一个场景（或整批）的统计。""" + + scenario: str + runs: int + stop_reasons: Mapping[str, int] + steps: StepStats + model_calls: int | None + model_calls_from: int + wall_ms: int | None + wall_ms_from: int + scored_runs: int + successes: int + + +@dataclass(frozen=True, slots=True, kw_only=True) +class Baseline: + """对照用的一份历史分布。**它是对照不是判定**——模型不同、任务子集不同，偏了不算击穿。""" + + label: str + stop_reasons: Mapping[str, int] + steps_p50: int + steps_p90: int + steps_max: int + + +#: dissect 那边 937 个真实 rollout 的分布。数字来自那份历史数据，不是本库跑出来的。 +DISSECT_BASELINE = Baseline( + label="dissect，937 个真实 rollout", + stop_reasons={"task_completed": 915, "step_budget": 20, "parse_failed_repeatedly": 2}, + steps_p50=13, + steps_p90=24, + steps_max=40, +) + +UNKNOWN_SCENARIO = "<未知>" +UNKNOWN_STOP_REASON = "<未知>" + + +def _summarize(facts: RunFacts) -> RunSummary: + meta = facts.meta or {} + scenario = meta.get("scenario") + fault = meta.get("fault") + success = meta.get("success") + wall = meta.get("wall_ms") + result = facts.effective_result() + return RunSummary( + run_id=facts.run_id, + scenario=_safe_id(scenario) if isinstance(scenario, str) else UNKNOWN_SCENARIO, + stop_reason=result.stop_reason.value if result is not None else UNKNOWN_STOP_REASON, + steps=len(facts.steps()), + success=success if isinstance(success, bool) else None, + wall_ms=wall if isinstance(wall, int) and not isinstance(wall, bool) else None, + fault=_safe_id(fault) if isinstance(fault, str) else None, + ) + + +def _scenario_stats( + scenario: str, summaries: Sequence[RunSummary], runs: Sequence[RunFacts] +) -> ScenarioStats: + stop_reasons: dict[str, int] = {} + for item in summaries: + stop_reasons[item.stop_reason] = stop_reasons.get(item.stop_reason, 0) + 1 + calls: list[int] = [] + for facts in runs: + value = (facts.meta or {}).get("model_calls") + if isinstance(value, int) and not isinstance(value, bool): + calls.append(value) + walls = [item.wall_ms for item in summaries if item.wall_ms is not None] + scored = [item.success for item in summaries if item.success is not None] + return ScenarioStats( + scenario=scenario, + runs=len(summaries), + stop_reasons=dict(sorted(stop_reasons.items())), + steps=_step_stats([item.steps for item in summaries]), + model_calls=sum(calls) if calls else None, + model_calls_from=len(calls), + wall_ms=sum(walls) if walls else None, + wall_ms_from=len(walls), + scored_runs=len(scored), + successes=sum(1 for item in scored if item), + ) + + +# --------------------------------------------------------------------------- +# 记分板 +# --------------------------------------------------------------------------- + + +@dataclass(frozen=True, slots=True, kw_only=True) +class Scoreboard: + """一批压测产物判完之后的全部结论。渲染报告与决定退出码都只看它。""" + + runs_dir: Path + invariants: tuple[InvariantResult, ...] + summaries: tuple[RunSummary, ...] + overall_stats: ScenarioStats + scenario_stats: tuple[ScenarioStats, ...] + baseline: Baseline + #: 不构成判定、但必须被人看见的事实：撕裂尾行、缺文件。 + notes: tuple[str, ...] = () + + @property + def breached(self) -> tuple[InvariantResult, ...]: + return tuple(item for item in self.invariants if item.verdict is Verdict.BREACHED) + + @property + def undetermined(self) -> tuple[InvariantResult, ...]: + return tuple(item for item in self.invariants if item.verdict is Verdict.UNDETERMINED) + + @property + def verdict(self) -> Verdict: + if self.breached: + return Verdict.BREACHED + if self.undetermined: + return Verdict.UNDETERMINED + return Verdict.PASSED + + +def _notes_for(facts: RunFacts) -> list[str]: + notes: list[str] = [] + if facts.torn_tail: + notes.append( + f"run `{facts.run_id}`：日志最后一行没有被换行终结（撕裂尾行）。" + "按存储契约那次写从未被确认过，它是进程被杀在写入中途的证据。" + ) + if facts.events_torn_tail: + notes.append(f"run `{facts.run_id}`：事件文件最后一行没有被换行终结（撕裂尾行）。") + for error in (facts.log_error, facts.result_error, facts.events_error, facts.meta_error): + if error is not None: + notes.append(f"run `{facts.run_id}`：{error}。") + return notes + + +def evaluate( + runs_dir: Path, + *, + completing_tools: frozenset[str] | None = None, + baseline: Baseline = DISSECT_BASELINE, +) -> Scoreboard: + """读这个目录里的全部 run，逐条判不变量，算统计，返回一份记分板。""" + config = CheckConfig(completing_tools=completing_tools) + runs = [load_run(runs_dir, run_id) for run_id in list_run_ids(runs_dir)] + + results: list[InvariantResult] = [] + for invariant in INVARIANTS: + breaches: list[Evidence] = [] + blocked: list[Evidence] = [] + if not runs: + blocked.append( + Evidence(run_id="<批次>", reason=f"{runs_dir} 里没有任何 run，这条没被验过") + ) + for facts in runs: + outcome = invariant.check(facts, config) + breaches.extend(outcome.breaches) + blocked.extend(outcome.undetermined) + results.append( + InvariantResult( + name=invariant.name, + description=invariant.description, + breaches=tuple(breaches), + undetermined=tuple(blocked), + ) + ) + + summaries = tuple(_summarize(facts) for facts in runs) + scenarios = sorted({item.scenario for item in summaries}) + scenario_stats = tuple( + _scenario_stats( + scenario, + [item for item in summaries if item.scenario == scenario], + [ + facts + for facts, item in zip(runs, summaries, strict=True) + if item.scenario == scenario + ], + ) + for scenario in scenarios + ) + notes: list[str] = [] + for facts in runs: + notes.extend(_notes_for(facts)) + + return Scoreboard( + runs_dir=runs_dir, + invariants=tuple(results), + summaries=summaries, + overall_stats=_scenario_stats("全部场景", summaries, runs), + scenario_stats=scenario_stats, + baseline=baseline, + notes=tuple(notes), + ) + + +# --------------------------------------------------------------------------- +# 报告 +# --------------------------------------------------------------------------- + + +def _share(count: int, total: int) -> str: + if total == 0: + return "—" + return f"{count / total * 100:.1f}%" + + +def _optional(value: object) -> str: + return "—" if value is None else str(value) + + +def _render_stats(stats: ScenarioStats) -> list[str]: + lines: list[str] = [] + lines.append(f"run 数：{stats.runs}") + lines.append("") + lines.append("| 停止原因 | 次数 | 占比 |") + lines.append("|---|---:|---:|") + if stats.stop_reasons: + for reason, count in stats.stop_reasons.items(): + lines.append(f"| {_cell(reason)} | {count} | {_share(count, stats.runs)} |") + else: + lines.append("| — | 0 | — |") + lines.append("") + lines.append("| 步数分布 | 值 |") + lines.append("|---|---:|") + lines.append(f"| p50 | {_optional(stats.steps.p50)} |") + lines.append(f"| p90 | {_optional(stats.steps.p90)} |") + lines.append(f"| max | {_optional(stats.steps.maximum)} |") + mean = "—" if stats.steps.mean is None else f"{stats.steps.mean:.2f}" + lines.append(f"| 均值 | {mean} |") + lines.append("") + lines.append("| 其它 | 值 |") + lines.append("|---|---:|") + lines.append( + f"| 模型调用总数 | {_optional(stats.model_calls)}" + f"（来自 {stats.model_calls_from} / {stats.runs} 个 run） |" + ) + lines.append( + f"| 总耗时(ms) | {_optional(stats.wall_ms)}" + f"（来自 {stats.wall_ms_from} / {stats.runs} 个 run） |" + ) + rate = _share(stats.successes, stats.scored_runs) + lines.append(f"| 成功率 | {stats.successes} / {stats.scored_runs} = {rate} |") + lines.append("") + return lines + + +def _render_baseline(scoreboard: Scoreboard) -> list[str]: + stats = scoreboard.overall_stats + baseline = scoreboard.baseline + baseline_total = sum(baseline.stop_reasons.values()) + lines = [ + "## 与基线的对照", + "", + f"基线是 {baseline.label}。**这是对照不是判定**：模型不同、任务子集不同，" + "偏了不算击穿，但要看得见。", + "", + "| 项目 | 本批次 | 基线 | 偏差 |", + "|---|---:|---:|---:|", + ] + names = sorted(set(stats.stop_reasons) | set(baseline.stop_reasons)) + for name in names: + mine = stats.stop_reasons.get(name, 0) + theirs = baseline.stop_reasons.get(name, 0) + mine_share = mine / stats.runs * 100 if stats.runs else None + theirs_share = theirs / baseline_total * 100 if baseline_total else None + if mine_share is None or theirs_share is None: + delta = "—" + else: + delta = f"{mine_share - theirs_share:+.1f} 个百分点" + lines.append( + f"| 停止原因 {_cell(name)} | {mine}（{_share(mine, stats.runs)}） | " + f"{theirs}（{_share(theirs, baseline_total)}） | {delta} |" + ) + for label, mine_value, theirs_value in ( + ("步数 p50", stats.steps.p50, baseline.steps_p50), + ("步数 p90", stats.steps.p90, baseline.steps_p90), + ("步数 max", stats.steps.maximum, baseline.steps_max), + ): + delta = "—" if mine_value is None else f"{mine_value - theirs_value:+d}" + lines.append(f"| {label} | {_optional(mine_value)} | {theirs_value} | {delta} |") + lines.append("") + return lines + + +def render_report(scoreboard: Scoreboard) -> str: + """把一份记分板渲染成 markdown。 + + 这里是「不许出现模型原文与文档片段」那条规矩的最后一道口子：所有进报告的字符串都已经 + 在判定那一层过过 `_scrub` 或 `_safe_id`，别的字符串从头到尾只以长度出现。 + """ + lines: list[str] = ["# PolyLoop 压测记分板", ""] + lines.append(f"- 批次目录：`{scoreboard.runs_dir}`") + lines.append(f"- run 数：{len(scoreboard.summaries)}") + lines.append(f"- 不变量：{len(scoreboard.invariants)} 条") + lines.append("") + + lines.append("## 总判定") + lines.append("") + breached = scoreboard.breached + blocked = scoreboard.undetermined + if not breached and not blocked: + lines.append(f"**全部通过：{len(scoreboard.invariants)} 条不变量逐条判过，无一击穿。**") + else: + parts = [] + if breached: + parts.append(f"{len(breached)} 条不变量被击穿") + if blocked: + parts.append(f"{len(blocked)} 条无法判定") + lines.append(f"**有 {'，'.join(parts)}。**") + lines.append("") + lines.append( + "「无法判定」是独立的第三档，不折算成通过：缺文件、缺字段导致判不了，和判过了是两回事。" + ) + lines.append("") + + lines.append("## 不变量") + lines.append("") + lines.append("| 不变量 | 判定 | 击穿证据 | 无法判定 |") + lines.append("|---|---|---:|---:|") + for item in scoreboard.invariants: + lines.append( + f"| {_cell(item.name)} | {_VERDICT_LABELS[item.verdict]} | " + f"{len(item.breaches)} | {len(item.undetermined)} |" + ) + lines.append("") + for item in scoreboard.invariants: + if item.verdict is Verdict.PASSED: + continue + lines.append(f"### {item.name} — {_VERDICT_LABELS[item.verdict]}") + lines.append("") + lines.append(item.description) + lines.append("") + if item.breaches: + lines.append("击穿：") + lines.append("") + for evidence in item.breaches: + lines.append(f"- {evidence.describe()}") + lines.append("") + if item.undetermined: + lines.append("无法判定：") + lines.append("") + for evidence in item.undetermined: + lines.append(f"- {evidence.describe()}") + lines.append("") + + lines.append("## 观察") + lines.append("") + if scoreboard.notes: + lines.extend(f"- {note}" for note in scoreboard.notes) + else: + lines.append("没有缺文件，也没有撕裂尾行。") + lines.append("") + + lines.append("## 统计") + lines.append("") + lines.append("### 全部场景") + lines.append("") + lines.extend(_render_stats(scoreboard.overall_stats)) + for stats in scoreboard.scenario_stats: + lines.append(f"### 场景 {stats.scenario}") + lines.append("") + lines.extend(_render_stats(stats)) + + lines.extend(_render_baseline(scoreboard)) + + lines.append("## 逐个 run") + lines.append("") + lines.append("| run_id | 场景 | 停止原因 | 步数 | 成功 | 耗时(ms) | 故障注入 |") + lines.append("|---|---|---|---:|---|---:|---|") + for item in scoreboard.summaries: + success = "未判分" if item.success is None else ("是" if item.success else "否") + lines.append( + f"| `{_cell(item.run_id)}` | {_cell(item.scenario)} | " + f"{_cell(item.stop_reason)} | {item.steps} | {success} | " + f"{_optional(item.wall_ms)} | {_cell(item.fault or '—')} |" + ) + lines.append("") + return "\n".join(lines) + + +def write_report(scoreboard: Scoreboard, path: Path) -> None: + path.parent.mkdir(parents=True, exist_ok=True) + path.write_text(render_report(scoreboard), encoding="utf-8") + + +# --------------------------------------------------------------------------- +# 命令行 +# --------------------------------------------------------------------------- + +#: 有不变量被击穿。 +EXIT_BREACHED = 1 +#: 没有击穿，但有判不了的。默认也算失败，`--allow-undetermined` 放行。 +EXIT_UNDETERMINED = 2 + + +def _parse_stop_reason_pair(text: str) -> tuple[str, int]: + name, _, raw = text.partition("=") + if not name or not raw: + raise argparse.ArgumentTypeError(f"要写成 名字=次数，收到 {text!r}") + try: + count = int(raw) + except ValueError as exc: + raise argparse.ArgumentTypeError(f"次数不是整数：{raw!r}") from exc + return name, count + + +def build_parser() -> argparse.ArgumentParser: + parser = argparse.ArgumentParser( + prog="scoreboard", + description="把一批压测产物判成通过 / 击穿 / 无法判定，并渲染成 markdown 报告。", + ) + parser.add_argument("--runs-dir", required=True, type=Path, help="压测产物所在目录") + parser.add_argument("--report", required=True, type=Path, help="报告写到哪儿") + parser.add_argument( + "--allow-undetermined", + action="store_true", + help="判不了的不算失败。崩溃注入那一类天然会缺文件，只有那时才该开。", + ) + parser.add_argument( + "--completing-tool", + action="append", + default=[], + metavar="NAME", + help="带完成标记的工具名，可重复。不给的话「停止原因与轨迹自洽」在只剩这条路时判不了。", + ) + parser.add_argument( + "--baseline-label", default=DISSECT_BASELINE.label, help="基线的名字，只进报告" + ) + parser.add_argument( + "--baseline-stop-reason", + action="append", + default=[], + type=_parse_stop_reason_pair, + metavar="NAME=COUNT", + help="基线的停止原因分布，可重复。给了任意一条就整份替换默认分布。", + ) + parser.add_argument("--baseline-steps-p50", type=int, default=DISSECT_BASELINE.steps_p50) + parser.add_argument("--baseline-steps-p90", type=int, default=DISSECT_BASELINE.steps_p90) + parser.add_argument("--baseline-steps-max", type=int, default=DISSECT_BASELINE.steps_max) + return parser + + +def main(argv: Sequence[str] | None = None) -> int: + args = build_parser().parse_args(argv) + baseline = Baseline( + label=args.baseline_label, + stop_reasons=( + dict(args.baseline_stop_reason) + if args.baseline_stop_reason + else DISSECT_BASELINE.stop_reasons + ), + steps_p50=args.baseline_steps_p50, + steps_p90=args.baseline_steps_p90, + steps_max=args.baseline_steps_max, + ) + completing = frozenset(args.completing_tool) if args.completing_tool else None + scoreboard = evaluate(args.runs_dir, completing_tools=completing, baseline=baseline) + write_report(scoreboard, args.report) + + print(f"报告写到了 {args.report}") + for item in scoreboard.invariants: + print(f" {_VERDICT_LABELS[item.verdict]}\t{item.name}") + if scoreboard.breached: + print(f"{len(scoreboard.breached)} 条不变量被击穿") + return EXIT_BREACHED + if scoreboard.undetermined: + if args.allow_undetermined: + print(f"{len(scoreboard.undetermined)} 条无法判定，按 --allow-undetermined 放行") + return 0 + print(f"{len(scoreboard.undetermined)} 条无法判定") + return EXIT_UNDETERMINED + return 0 + + +if __name__ == "__main__": + raise SystemExit(main()) diff --git a/tools/soak/tests/test_scoreboard.py b/tools/soak/tests/test_scoreboard.py new file mode 100644 index 0000000..3f19b20 --- /dev/null +++ b/tools/soak/tests/test_scoreboard.py @@ -0,0 +1,857 @@ +"""记分板的测试：全部用合成日志，不跑模型、不起容器。 + +**重点是「构造出违反某条不变量的产物、记分板确实报击穿」那一组。** 一个永远返回通过的 +判定器比没有判定器更糟——它会让一次什么都没验成的跑看起来全绿，而那正是最需要被看见的 +情况。所以每条不变量都配一个反例，合法产物那组只是对照。 +""" + +import json +import re +import sys +from collections.abc import Callable, Sequence +from pathlib import Path + +import pytest + +_REPO_ROOT = Path(__file__).resolve().parents[3] +if str(_REPO_ROOT) not in sys.path: + sys.path.insert(0, str(_REPO_ROOT)) + +from polyloop.serialization import encode # noqa: E402 +from polyloop.types import ( # noqa: E402 + ActionOutcome, + ActionStatus, + Intent, + IntentKind, + ModelCallResult, + ModelReply, + ReplayPolicy, + RunFinished, + RunResult, + RunStarted, + StepCompleted, + StepRecord, + StopReason, +) +from tools.soak.scoreboard import ( # noqa: E402 + EXIT_BREACHED, + EXIT_UNDETERMINED, + Verdict, + evaluate, + main, + render_report, +) + +#: 塞进观察与最终回答里的哨兵。压测语料里有第三方的真实文档，报告里一个字都不许出现。 +SENTINEL = "SENTINEL-ZZZ-第三方文档正文" + +_TAGS = { + RunStarted: "run_started", + Intent: "intent", + ModelCallResult: "model_call_result", + StepCompleted: "step_completed", + RunFinished: "run_finished", +} + + +# --------------------------------------------------------------------------- +# 造合法产物 +# --------------------------------------------------------------------------- + + +def build_records( + run_id: str, + *, + steps: int = 2, + stop_reason: StopReason = StopReason.TASK_COMPLETED, + max_steps: int = 5, + final_answer: str | None = None, + complete_on_last: bool = True, + tool_name: str | None = "run_code", + sink_failures: int = 0, + observation: str = "普通观察", +) -> tuple[list[object], RunResult]: + """造一份自洽的记录序列：一步一组「模型意图 / 模型结果 / 动作意图 / 逐步结果」。""" + records: list[object] = [ + RunStarted( + run_id=run_id, + parameter_snapshot={ + "request.max_steps": str(max_steps), + "store.kind": "jsonl", + }, + ) + ] + step_records: list[StepRecord] = [] + for index in range(steps): + completed = complete_on_last and index == steps - 1 + text = f"{observation}#{index}" + records.append( + Intent( + run_id=run_id, + kind=IntentKind.MODEL_CALL, + call_index=index, + result_id=f"m{index}", + replay_policy=ReplayPolicy.NEVER, + ) + ) + records.append( + ModelCallResult( + run_id=run_id, + result_id=f"m{index}", + reply=ModelReply(call_id=f"c{index}", content="决策文本", thinking=""), + failure=None, + ) + ) + records.append( + Intent( + run_id=run_id, + kind=IntentKind.ACTION, + call_index=index, + result_id=f"a{index}", + replay_policy=ReplayPolicy.NEVER, + ) + ) + outcome = ActionOutcome( + status=ActionStatus.EXECUTED, + observation=text, + observation_is_synthetic=False, + env_reported_completion=completed, + observation_truncated_chars=0, + ) + step = StepRecord( + step_idx=index, + raw_output="决策文本", + content_chars=4, + thinking_chars=0, + action="run_code", + parse_ok=True, + parse_error=None, + observation=text, + observation_is_synthetic=False, + observation_truncated_chars=0, + prompt_chars=100 + index * 10, + call_id=f"c{index}", + step_wall_ms=7, + tool_name=tool_name, + tool_arguments="{}", + action_status=ActionStatus.EXECUTED, + env_reported_completion=completed, + ) + step_records.append(step) + records.append( + StepCompleted( + run_id=run_id, + result_id=f"a{index}", + action_outcome=outcome, + step=step, + ) + ) + result = RunResult( + run_id=run_id, + stop_reason=stop_reason, + final_answer=final_answer, + steps=tuple(step_records), + event_delivery_failures=sink_failures, + ) + records.append(RunFinished(run_id=run_id, result=result)) + return records, result + + +def write_log(path: Path, records: Sequence[object]) -> None: + lines = [ + json.dumps({"record": _TAGS[type(record)], **encode(record)}, ensure_ascii=False) + for record in records + ] + path.write_text("\n".join(lines) + "\n", encoding="utf-8") + + +def materialize( + runs_dir: Path, + run_id: str = "soak-0001", + *, + write_result: bool = True, + write_events: bool = True, + write_meta: bool = True, + events: int | None = None, + meta_overrides: dict[str, object] | None = None, + **kwargs: object, +) -> RunResult: + """把一个自洽的 run 的四个文件都写到磁盘上，返回它的运行结果。""" + runs_dir.mkdir(parents=True, exist_ok=True) + records, result = build_records(run_id, **kwargs) # type: ignore[arg-type] + write_log(runs_dir / f"{run_id}.jsonl", records) + if write_result: + (runs_dir / f"{run_id}.result.json").write_text( + json.dumps(encode(result), ensure_ascii=False), encoding="utf-8" + ) + if write_events: + count = len(result.steps) if events is None else events + payload = "".join( + json.dumps({"kind": "step_finished", "run_id": run_id, "step_idx": index}) + "\n" + for index in range(count) + ) + (runs_dir / f"{run_id}.events.jsonl").write_text(payload, encoding="utf-8") + if write_meta: + meta: dict[str, object] = { + "scenario": "appworld", + "task_id": "82e2fac_1", + "phase": None, + "wall_ms": 1234, + "model_calls": len(result.steps), + "sink_failures": result.event_delivery_failures, + "env_executions": len(result.steps), + "fault": None, + "success": True, + "resumed_from_step": None, + } + meta.update(meta_overrides or {}) + (runs_dir / f"{run_id}.meta.json").write_text( + json.dumps(meta, ensure_ascii=False), encoding="utf-8" + ) + return result + + +def edit_log(path: Path, edit: Callable[[list[dict]], list[dict]]) -> None: + """按行读出来、交给 `edit` 改、再写回去。制造反例用的。""" + payloads = [json.loads(line) for line in path.read_text(encoding="utf-8").splitlines()] + changed = edit(payloads) + path.write_text( + "".join(json.dumps(item, ensure_ascii=False) + "\n" for item in changed), + encoding="utf-8", + ) + + +def verdict_of(scoreboard, name: str) -> Verdict: + for item in scoreboard.invariants: + if item.name == name: + return item.verdict + raise AssertionError(f"记分板里没有名为 {name!r} 的不变量") + + +def assert_breached(scoreboard, name: str) -> None: + assert verdict_of(scoreboard, name) is Verdict.BREACHED, ( + f"{name} 应当被击穿，实际 {verdict_of(scoreboard, name)}" + ) + result = next(item for item in scoreboard.invariants if item.name == name) + for evidence in result.breaches: + assert evidence.expected is not None + assert evidence.actual is not None + + +# --------------------------------------------------------------------------- +# 合法产物：全部通过 +# --------------------------------------------------------------------------- + + +def test_legal_run_passes_every_invariant(tmp_path: Path) -> None: + materialize(tmp_path) + scoreboard = evaluate(tmp_path) + assert scoreboard.verdict is Verdict.PASSED, [ + (item.name, item.verdict, [e.describe() for e in item.breaches + item.undetermined]) + for item in scoreboard.invariants + if item.verdict is not Verdict.PASSED + ] + assert scoreboard.notes == () + + +def test_several_legal_runs_pass(tmp_path: Path) -> None: + materialize(tmp_path, "soak-0001", steps=3) + materialize( + tmp_path, + "soak-0002", + steps=5, + stop_reason=StopReason.STEP_BUDGET, + max_steps=5, + complete_on_last=False, + ) + materialize( + tmp_path, + "soak-0003", + steps=1, + stop_reason=StopReason.AGENT_FINISHED, + complete_on_last=False, + final_answer="给出的答案", + ) + scoreboard = evaluate(tmp_path) + assert scoreboard.verdict is Verdict.PASSED + assert len(scoreboard.summaries) == 3 + assert scoreboard.overall_stats.stop_reasons == { + "agent_finished": 1, + "step_budget": 1, + "task_completed": 1, + } + + +def test_completing_tool_is_supplied_by_the_caller(tmp_path: Path) -> None: + """task_completed 而最后一步没有环境侧证据：给了工具名就判得了，不给就是无法判定。""" + materialize(tmp_path, complete_on_last=False, tool_name="finish_task") + assert verdict_of(evaluate(tmp_path), "停止原因与轨迹自洽") is Verdict.UNDETERMINED + assert ( + verdict_of( + evaluate(tmp_path, completing_tools=frozenset({"finish_task"})), + "停止原因与轨迹自洽", + ) + is Verdict.PASSED + ) + assert ( + verdict_of( + evaluate(tmp_path, completing_tools=frozenset({"别的工具"})), + "停止原因与轨迹自洽", + ) + is Verdict.BREACHED + ) + + +# --------------------------------------------------------------------------- +# 逐条不变量的反例 +# --------------------------------------------------------------------------- + + +def test_unreadable_line_is_a_breach(tmp_path: Path) -> None: + materialize(tmp_path) + path = tmp_path / "soak-0001.jsonl" + lines = path.read_text(encoding="utf-8").splitlines() + lines[2] = "{这不是 JSON" + path.write_text("\n".join(lines) + "\n", encoding="utf-8") + + scoreboard = evaluate(tmp_path) + assert_breached(scoreboard, "日志能被读回来") + assert any("第 3 行" in e.describe() for e in scoreboard.breached[0].breaches) + + +def test_unknown_record_tag_is_a_breach(tmp_path: Path) -> None: + materialize(tmp_path) + edit_log( + tmp_path / "soak-0001.jsonl", + lambda items: [{**items[0], "record": "谁知道这是什么"}] + items[1:], + ) + assert_breached(evaluate(tmp_path), "日志能被读回来") + + +def test_torn_tail_is_reported_but_not_a_breach(tmp_path: Path) -> None: + materialize(tmp_path) + path = tmp_path / "soak-0001.jsonl" + # 最后一行只写了一半、没有换行终结：进程被杀在写入中途的样子。 + path.write_text(path.read_text(encoding="utf-8") + '{"record": "step_comp', encoding="utf-8") + + scoreboard = evaluate(tmp_path) + assert verdict_of(scoreboard, "日志能被读回来") is Verdict.PASSED + assert any("撕裂尾行" in note for note in scoreboard.notes) + assert "撕裂尾行" in render_report(scoreboard) + + +def test_torn_line_in_the_middle_is_a_breach(tmp_path: Path) -> None: + materialize(tmp_path) + path = tmp_path / "soak-0001.jsonl" + lines = path.read_text(encoding="utf-8").splitlines() + # 被换行终结、却是半截 JSON：追加写不会产生这种东西，所以是损坏。 + lines[3] = lines[3][: len(lines[3]) // 2] + path.write_text("\n".join(lines) + "\n", encoding="utf-8") + + scoreboard = evaluate(tmp_path) + assert_breached(scoreboard, "日志能被读回来") + assert not any("撕裂尾行" in note for note in scoreboard.notes) + + +def test_result_file_disagreeing_with_the_log_is_a_breach(tmp_path: Path) -> None: + result = materialize(tmp_path) + payload = encode(result) + payload["final_answer"] = "跨进程读回来的和内存里的不一样" + (tmp_path / "soak-0001.result.json").write_text( + json.dumps(payload, ensure_ascii=False), encoding="utf-8" + ) + assert_breached(evaluate(tmp_path), "跨进程的结果与内存里的一致") + + +def test_result_file_step_field_disagreeing_is_a_breach(tmp_path: Path) -> None: + result = materialize(tmp_path) + payload = encode(result) + payload["steps"][1]["step_wall_ms"] = 99999 # type: ignore[index] + (tmp_path / "soak-0001.result.json").write_text( + json.dumps(payload, ensure_ascii=False), encoding="utf-8" + ) + scoreboard = evaluate(tmp_path) + assert_breached(scoreboard, "跨进程的结果与内存里的一致") + evidence = next( + item for item in scoreboard.invariants if item.name == "跨进程的结果与内存里的一致" + ) + assert any("steps[1].step_wall_ms" in e.describe() for e in evidence.breaches) + + +def test_step_index_gap_is_a_breach(tmp_path: Path) -> None: + materialize(tmp_path, steps=3) + + def bump(items: list[dict]) -> list[dict]: + for item in items: + if item["record"] == "step_completed" and item["step"]["step_idx"] == 1: + item["step"]["step_idx"] = 2 + return items + + edit_log(tmp_path / "soak-0001.jsonl", bump) + assert_breached(evaluate(tmp_path), "步号连续") + + +def test_dangling_intent_in_the_middle_is_a_breach(tmp_path: Path) -> None: + materialize(tmp_path, steps=3) + edit_log( + tmp_path / "soak-0001.jsonl", + lambda items: [ + item + for item in items + if not (item["record"] == "model_call_result" and item["result_id"] == "m0") + ], + ) + assert_breached(evaluate(tmp_path), "意图都有归宿") + + +def test_two_dangling_intents_are_a_breach(tmp_path: Path) -> None: + materialize(tmp_path, steps=3) + edit_log( + tmp_path / "soak-0001.jsonl", + lambda items: [ + item + for item in items + if not (item["record"] == "model_call_result" and item["result_id"] in {"m1", "m2"}) + ], + ) + assert_breached(evaluate(tmp_path), "意图都有归宿") + + +def test_one_dangling_intent_at_the_tail_is_the_crash_point_and_passes(tmp_path: Path) -> None: + materialize(tmp_path, steps=2) + + def crash(items: list[dict]) -> list[dict]: + kept = [item for item in items if item["record"] != "run_finished"] + kept.append( + { + "record": "intent", + "run_id": "soak-0001", + "kind": "model_call", + "call_index": 2, + "result_id": "m2", + "replay_policy": "never", + } + ) + return kept + + edit_log(tmp_path / "soak-0001.jsonl", crash) + assert verdict_of(evaluate(tmp_path), "意图都有归宿") is Verdict.PASSED + + +def test_step_completed_pairing_violation_is_a_breach(tmp_path: Path) -> None: + materialize(tmp_path) + + def unpair(items: list[dict]) -> list[dict]: + for item in items: + if item["record"] == "step_completed" and item["step"]["step_idx"] == 0: + item["result_id"] = None + return items + + edit_log(tmp_path / "soak-0001.jsonl", unpair) + assert_breached(evaluate(tmp_path), "步记录的内部不变量") + + +def test_outcome_disagreeing_with_step_record_is_a_breach(tmp_path: Path) -> None: + materialize(tmp_path) + + def skew(items: list[dict]) -> list[dict]: + for item in items: + if item["record"] == "step_completed" and item["step"]["step_idx"] == 0: + item["step"]["action_status"] = "not_executed" + return items + + edit_log(tmp_path / "soak-0001.jsonl", skew) + assert_breached(evaluate(tmp_path), "动作结果与步记录一致") + + +def test_prompt_chars_going_backwards_is_a_breach(tmp_path: Path) -> None: + materialize(tmp_path, steps=3) + + def shrink(items: list[dict]) -> list[dict]: + for item in items: + if item["record"] == "step_completed" and item["step"]["step_idx"] == 2: + item["step"]["prompt_chars"] = 1 + return items + + edit_log(tmp_path / "soak-0001.jsonl", shrink) + assert_breached(evaluate(tmp_path), "提示词字符数单调不减") + + +def test_event_count_mismatch_is_a_breach(tmp_path: Path) -> None: + materialize(tmp_path, steps=3, events=2) + assert_breached(evaluate(tmp_path), "事件条数等于本进程走完的步数") + + +def test_resumed_run_subtracts_the_skipped_steps(tmp_path: Path) -> None: + """续跑：日志里有四步，本进程只真的走完后两步，所以只该有两条事件。""" + materialize( + tmp_path, steps=4, events=2, meta_overrides={"resumed_from_step": 2, "model_calls": 2} + ) + assert verdict_of(evaluate(tmp_path), "事件条数等于本进程走完的步数") is Verdict.PASSED + + materialize( + tmp_path, + "soak-0002", + steps=4, + events=4, + meta_overrides={"resumed_from_step": 2}, + ) + assert_breached(evaluate(tmp_path), "事件条数等于本进程走完的步数") + + +def test_delivery_failure_count_mismatch_is_a_breach(tmp_path: Path) -> None: + materialize(tmp_path, sink_failures=0, meta_overrides={"sink_failures": 3}) + assert_breached(evaluate(tmp_path), "投递失败计数对得上") + + +def test_crosstalk_in_the_log_is_a_breach(tmp_path: Path) -> None: + materialize(tmp_path) + + def swap(items: list[dict]) -> list[dict]: + for item in items: + if item["record"] == "step_completed" and item["step"]["step_idx"] == 1: + item["run_id"] = "soak-9999" + return items + + edit_log(tmp_path / "soak-0001.jsonl", swap) + assert_breached(evaluate(tmp_path), "不串台") + + +def test_crosstalk_in_the_events_is_a_breach(tmp_path: Path) -> None: + materialize(tmp_path) + (tmp_path / "soak-0001.events.jsonl").write_text( + json.dumps({"kind": "step_finished", "run_id": "soak-0001", "step_idx": 0}) + + "\n" + + json.dumps({"kind": "step_finished", "run_id": "soak-9999", "step_idx": 1}) + + "\n", + encoding="utf-8", + ) + assert_breached(evaluate(tmp_path), "不串台") + + +def test_task_completed_without_evidence_is_a_breach(tmp_path: Path) -> None: + materialize(tmp_path, complete_on_last=False, tool_name="run_code") + scoreboard = evaluate(tmp_path, completing_tools=frozenset()) + assert_breached(scoreboard, "停止原因与轨迹自洽") + + +def test_step_budget_not_matching_max_steps_is_a_breach(tmp_path: Path) -> None: + materialize( + tmp_path, + steps=3, + max_steps=9, + stop_reason=StopReason.STEP_BUDGET, + complete_on_last=False, + ) + assert_breached(evaluate(tmp_path), "停止原因与轨迹自洽") + + +def test_agent_finished_without_final_answer_is_a_breach(tmp_path: Path) -> None: + materialize( + tmp_path, + stop_reason=StopReason.AGENT_FINISHED, + complete_on_last=False, + final_answer="", + ) + assert_breached(evaluate(tmp_path), "停止原因与轨迹自洽") + + +def test_cancelled_without_run_finished_is_a_breach(tmp_path: Path) -> None: + materialize(tmp_path, stop_reason=StopReason.CANCELLED, complete_on_last=False) + edit_log( + tmp_path / "soak-0001.jsonl", + lambda items: [item for item in items if item["record"] != "run_finished"], + ) + assert_breached(evaluate(tmp_path), "停止原因与轨迹自洽") + + +def test_cancelled_with_run_finished_passes(tmp_path: Path) -> None: + materialize(tmp_path, stop_reason=StopReason.CANCELLED, complete_on_last=False) + assert verdict_of(evaluate(tmp_path), "停止原因与轨迹自洽") is Verdict.PASSED + + +# --------------------------------------------------------------------------- +# 缺文件：无法判定，既不是通过也不是击穿 +# --------------------------------------------------------------------------- + + +def test_missing_result_file_is_undetermined(tmp_path: Path) -> None: + materialize(tmp_path, write_result=False) + scoreboard = evaluate(tmp_path) + assert verdict_of(scoreboard, "跨进程的结果与内存里的一致") is Verdict.UNDETERMINED + assert scoreboard.verdict is Verdict.UNDETERMINED + assert any(".result.json" in note for note in scoreboard.notes) + + +def test_missing_meta_file_is_undetermined(tmp_path: Path) -> None: + materialize(tmp_path, write_meta=False) + scoreboard = evaluate(tmp_path) + assert verdict_of(scoreboard, "事件条数等于本进程走完的步数") is Verdict.UNDETERMINED + assert verdict_of(scoreboard, "投递失败计数对得上") is Verdict.UNDETERMINED + assert not scoreboard.breached + + +def test_missing_events_file_is_undetermined(tmp_path: Path) -> None: + materialize(tmp_path, write_events=False) + scoreboard = evaluate(tmp_path) + assert verdict_of(scoreboard, "事件条数等于本进程走完的步数") is Verdict.UNDETERMINED + assert verdict_of(scoreboard, "不串台") is Verdict.UNDETERMINED + assert not scoreboard.breached + + +def test_sigkilled_run_keeps_only_the_log(tmp_path: Path) -> None: + """进程被 SIGKILL：只剩日志，而且尾行撕裂。整批判成无法判定，不是通过也不是击穿。""" + materialize(tmp_path, write_result=False, write_events=False, write_meta=False) + path = tmp_path / "soak-0001.jsonl" + lines = path.read_text(encoding="utf-8").splitlines()[:-1] + path.write_text("\n".join(lines) + '\n{"record": "run_fin', encoding="utf-8") + + scoreboard = evaluate(tmp_path) + assert not scoreboard.breached + assert scoreboard.verdict is Verdict.UNDETERMINED + + +def test_empty_directory_is_undetermined_not_passed(tmp_path: Path) -> None: + scoreboard = evaluate(tmp_path) + assert scoreboard.verdict is Verdict.UNDETERMINED + assert all(item.verdict is Verdict.UNDETERMINED for item in scoreboard.invariants) + + +def test_events_file_is_not_mistaken_for_a_run(tmp_path: Path) -> None: + materialize(tmp_path) + scoreboard = evaluate(tmp_path) + assert [item.run_id for item in scoreboard.summaries] == ["soak-0001"] + + +# --------------------------------------------------------------------------- +# 报告 +# --------------------------------------------------------------------------- + + +def _table_rows(report: str) -> list[list[str]]: + return [re.split(r"(? None: + materialize(tmp_path, "soak-0001", steps=2) + materialize( + tmp_path, + "soak-0002", + steps=4, + stop_reason=StopReason.STEP_BUDGET, + max_steps=4, + complete_on_last=False, + ) + report = render_report(evaluate(tmp_path)) + + assert report.startswith("# PolyLoop 压测记分板") + for heading in ( + "## 总判定", + "## 不变量", + "## 观察", + "## 统计", + "## 与基线的对照", + "## 逐个 run", + ): + assert heading in report + # 表格的每一行都以竖线开头结尾，且同一段里的列数一致。 + widths: set[int] = set() + previous_was_table = False + for line in report.splitlines(): + if line.startswith("|"): + assert line.endswith("|"), line + if not previous_was_table: + widths = set() + widths.add(len(re.split(r"(? None: + materialize(tmp_path, steps=2) + report = render_report(evaluate(tmp_path)) + assert "937" in report + assert "915" in report # 基线的 task_completed 计数 + assert "步数 p90" in report + assert "对照不是判定" in report + + +def test_report_never_contains_document_fragments(tmp_path: Path) -> None: + """观察与最终回答里的哨兵一个字都不许进报告——压测语料里有第三方的真实文档。""" + result = materialize( + tmp_path, + steps=2, + observation=SENTINEL, + stop_reason=StopReason.AGENT_FINISHED, + complete_on_last=False, + final_answer=SENTINEL, + ) + # 再造一处击穿，让证据渲染那条路径也被走到。 + payload = encode(result) + payload["final_answer"] = SENTINEL + "-改过的" + payload["steps"][0]["observation"] = SENTINEL + "-也改过" # type: ignore[index] + payload["steps"][0]["tool_name"] = SENTINEL + "-当成工具名" # type: ignore[index] + payload["steps"][1]["tool_arguments"] = SENTINEL + "-当成参数" # type: ignore[index] + (tmp_path / "soak-0001.result.json").write_text( + json.dumps(payload, ensure_ascii=False), encoding="utf-8" + ) + + scoreboard = evaluate(tmp_path) + assert scoreboard.breached + report = render_report(scoreboard) + assert SENTINEL not in report + assert "SENTINEL" not in report + + +def test_report_does_not_echo_a_hostile_record_tag(tmp_path: Path) -> None: + materialize(tmp_path) + edit_log( + tmp_path / "soak-0001.jsonl", + lambda items: [{**items[0], "record": SENTINEL}] + items[1:], + ) + scoreboard = evaluate(tmp_path) + assert_breached(scoreboard, "日志能被读回来") + report = render_report(scoreboard) + assert SENTINEL not in report + assert "SENTINEL" not in report + + +def test_report_does_not_echo_a_hostile_observation_through_decode_errors( + tmp_path: Path, +) -> None: + """解码失败的消息也不许把观察带出来。库的取值函数对字符串字段只报类型不报内容。""" + materialize(tmp_path, observation=SENTINEL) + + def corrupt(items: list[dict]) -> list[dict]: + for item in items: + if item["record"] == "step_completed": + # 少一个必填字段：整行解不出来，消息里只会有字段名。 + del item["step"]["prompt_chars"] + return items + + edit_log(tmp_path / "soak-0001.jsonl", corrupt) + scoreboard = evaluate(tmp_path) + assert_breached(scoreboard, "日志能被读回来") + report = render_report(scoreboard) + assert SENTINEL not in report + assert "SENTINEL" not in report + + +def test_report_keeps_tool_names_out(tmp_path: Path) -> None: + """工具名也不进报告。 + + 一次没通过校验的工具调用会把模型编的那串原样记进 `tool_name`，所以它和观察同档， + 不是「我们自己的枚举取值」。这条被一次实测撞出来过：先前的实现把工具名过一遍字符 + 白名单就放行，而白名单留下的正是 ASCII 那一段，哨兵原样穿了过去。 + """ + materialize( + tmp_path, + complete_on_last=False, + tool_name="工具|名里有竖线和" + SENTINEL, + ) + scoreboard = evaluate(tmp_path, completing_tools=frozenset({"finish"})) + assert_breached(scoreboard, "停止原因与轨迹自洽") + report = render_report(scoreboard) + assert SENTINEL not in report + assert "SENTINEL" not in report + assert "竖线" not in report + + +# --------------------------------------------------------------------------- +# 命令行与退出码 +# --------------------------------------------------------------------------- + + +def test_main_returns_zero_on_a_clean_batch(tmp_path: Path) -> None: + runs = tmp_path / "runs" + report = tmp_path / "reports" / "batch.md" + materialize(runs) + assert main(["--runs-dir", str(runs), "--report", str(report)]) == 0 + assert report.read_text(encoding="utf-8").startswith("# PolyLoop 压测记分板") + + +def test_main_returns_nonzero_on_a_breach(tmp_path: Path) -> None: + runs = tmp_path / "runs" + materialize(runs, steps=3, events=1) + code = main(["--runs-dir", str(runs), "--report", str(tmp_path / "r.md")]) + assert code == EXIT_BREACHED + + +def test_main_returns_nonzero_on_undetermined(tmp_path: Path) -> None: + runs = tmp_path / "runs" + materialize(runs, write_meta=False) + argv = ["--runs-dir", str(runs), "--report", str(tmp_path / "r.md")] + assert main(argv) == EXIT_UNDETERMINED + assert main([*argv, "--allow-undetermined"]) == 0 + + +def test_allow_undetermined_does_not_forgive_a_breach(tmp_path: Path) -> None: + runs = tmp_path / "runs" + materialize(runs, steps=3, write_meta=False) + + def bump(items: list[dict]) -> list[dict]: + for item in items: + if item["record"] == "step_completed" and item["step"]["step_idx"] == 1: + item["step"]["step_idx"] = 2 + return items + + edit_log(runs / "soak-0001.jsonl", bump) + code = main( + [ + "--runs-dir", + str(runs), + "--report", + str(tmp_path / "r.md"), + "--allow-undetermined", + ] + ) + assert code == EXIT_BREACHED + + +def test_completing_tool_flag_reaches_the_invariant(tmp_path: Path) -> None: + runs = tmp_path / "runs" + materialize(runs, complete_on_last=False, tool_name="finish_task") + argv = ["--runs-dir", str(runs), "--report", str(tmp_path / "r.md")] + assert main(argv) == EXIT_UNDETERMINED + assert main([*argv, "--completing-tool", "finish_task"]) == 0 + + +def test_baseline_flags_override_the_defaults(tmp_path: Path) -> None: + runs = tmp_path / "runs" + report = tmp_path / "r.md" + materialize(runs) + code = main( + [ + "--runs-dir", + str(runs), + "--report", + str(report), + "--baseline-label", + "自造基线", + "--baseline-stop-reason", + "task_completed=7", + "--baseline-steps-p50", + "3", + ] + ) + assert code == 0 + text = report.read_text(encoding="utf-8") + assert "自造基线" in text + assert "915" not in text + + +def test_bad_baseline_pair_is_rejected(tmp_path: Path) -> None: + with pytest.raises(SystemExit): + main( + [ + "--runs-dir", + str(tmp_path), + "--report", + str(tmp_path / "r.md"), + "--baseline-stop-reason", + "没有等号", + ] + )