iomgaa
ff59457482
Codex 报了两条数据不足时真空成立的不变量，顺着同类找齐了七条：零步的「步号连续」与 「动作结果与步记录一致」、不足两步的「提示词字符数单调不减」、零意图的「意图都有归宿」、 零载荷的「步记录的内部不变量」、零记录的「不串台」、零行的「日志能被读回来」。同一类 缺陷改一半，剩下那一半照样会在某天把一次什么都没验的跑显示成绿。 各条的数据下限不一样，反直觉的三处写进了说明：「步记录的内部不变量」数的是打着标签的行 不是解出来的记录（违反配对的行本来就解不出记录，按记录数当下限会把它最该判的对象数漏）； 「动作结果与步记录一致」不要求那条步记录带动作结果；「不串台」两半各判各的，合成一个的话 一半的真空会被另一半的绿盖住。 「停止原因与轨迹自洽」原本只覆盖四个取值、另外六个直接放行——不是数据不足，是判据本来就 该覆盖而没覆盖，后果和真空成立一样。六个都补了规矩，llm_error 那条按库自己的判据写 （解析失败必定带说明，模型调用失败那一步压根没走到解释器，只看有没有动作结果分不开这两者）。 另加一条断言十个取值一个不漏，将来加了取值而这里没跟上会显式报「还没有规矩」。 「提示词字符数单调不减」的说明原本承诺「历史只追加」，实现只比较库自己记录的数——承诺了 它，读者看见绿就以为截断被排除了。改成只承诺它验得到的，真正的对账在故障注入那侧。 拿 193 次真实运行重跑：十一条仍然全过，而这次那 8 次解析失败连击、1 次模型调用失败、 1 次撞步数上限是被真规矩判过的。 Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
2064 lines
82 KiB
Python
2064 lines
82 KiB
Python
"""压测记分板：把一批压测产物判成「通过 / 击穿 / 无法判定」，再渲染成一份 markdown 报告。
|
||
|
||
它读的是一次批次跑完之后留在磁盘上的四种文件，每个 run 一组：
|
||
|
||
<runs_dir>/<run_id>.jsonl polyloop 的 JsonlRunStore 写的日志，一行一条记录
|
||
<runs_dir>/<run_id>.result.json 驱动写的 polyloop.serialization.encode(RunResult)
|
||
<runs_dir>/<run_id>.events.jsonl 事件出口写的，一条事件一行
|
||
<runs_dir>/<run_id>.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 (
|
||
ActionStatus,
|
||
Intent,
|
||
IntentKind,
|
||
ModelCallResult,
|
||
ReplayPolicy,
|
||
RunFinished,
|
||
RunResult,
|
||
RunStarted,
|
||
StepCompleted,
|
||
StepRecord,
|
||
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"
|
||
MAX_ACTIONS_SNAPSHOT_KEY = "request.max_actions"
|
||
MAX_PARSE_FAILURES_SNAPSHOT_KEY = "request.max_consecutive_parse_failures"
|
||
MAX_PROMPT_CHARS_SNAPSHOT_KEY = "request.max_prompt_chars"
|
||
|
||
#: 单个 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:
|
||
"""每一条被换行终结的行都解得出记录。
|
||
|
||
**一条被终结的行都没有时报「无法判定」。** 空文件与「只写了半行就被杀」都落在这里：
|
||
没有任何一行被读回来过，说「日志读得回来」是没有依据的。这一档在崩溃注入那批里是
|
||
真实存在的——`write_run_started` 先建文件再写那一行，杀在两者之间就留下这种产物。
|
||
"""
|
||
del config
|
||
if facts.log_error is not None:
|
||
return CheckOutcome(undetermined=(Evidence(run_id=facts.run_id, reason=facts.log_error),))
|
||
if not facts.lines:
|
||
return CheckOutcome(
|
||
undetermined=(
|
||
Evidence(
|
||
run_id=facts.run_id,
|
||
reason=(
|
||
"日志里没有任何一条被换行终结的行"
|
||
+ ("（末尾有撕裂的半行）" if facts.torn_tail else "（文件是空的）")
|
||
+ "，没东西可读回来"
|
||
),
|
||
),
|
||
)
|
||
)
|
||
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:
|
||
"""步号从 0 开始逐 1 递增。
|
||
|
||
**一条步记录都没有时报「无法判定」，不报通过。** 循环零次也会返回一个没有击穿的结果，
|
||
而那不是「验过了、对的」，是「没东西可验」——把它算成通过，一批全是崩在第一步之前的
|
||
产物会显示成绿的。崩溃注入那两类产物里真的会出现零步的 run。
|
||
"""
|
||
del config
|
||
if not facts.steps():
|
||
return CheckOutcome(
|
||
undetermined=(
|
||
Evidence(
|
||
run_id=facts.run_id,
|
||
reason="日志里一条 step_completed 都没有，步号这条没东西可验",
|
||
),
|
||
)
|
||
)
|
||
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 _dangling_intents(facts: RunFacts) -> list[tuple[int, int, Intent]]:
|
||
"""挑出没有归宿的意图，每条带上「它是第几条」与「在日志第几行」。
|
||
|
||
「意图都有归宿」与「停止原因与轨迹自洽」的 resume_state_unknown 那一档读的是同一件事，
|
||
所以只算一次：一处改了另一处没改的话，两条会对同一份日志给出互相矛盾的判定。
|
||
"""
|
||
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}
|
||
dangling: list[tuple[int, int, Intent]] = []
|
||
for position, (number, intent) in enumerate(facts.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))
|
||
return dangling
|
||
|
||
|
||
def _check_intents_resolved(facts: RunFacts, config: CheckConfig) -> CheckOutcome:
|
||
"""每条意图的 result_id 都能找到归宿，悬空至多一条且必须在末尾。
|
||
|
||
**一条意图都没有时报「无法判定」。** 下限就是一条：一条意图足够验出它悬不悬空，也足够
|
||
验出「悬空的那条是不是最后一条」——那两问在只有一条意图时都有确定的答案。不要求日志里
|
||
同时有结果记录，一条有意图没结果的日志正是这条要判的那种。
|
||
"""
|
||
del config
|
||
intents = facts.intents()
|
||
if not intents:
|
||
return CheckOutcome(
|
||
undetermined=(
|
||
Evidence(
|
||
run_id=facts.run_id,
|
||
reason="日志里一条意图都没有，归宿这条没东西可验",
|
||
),
|
||
)
|
||
)
|
||
dangling = _dangling_intents(facts)
|
||
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` 自己在构造期就守这条，
|
||
所以一条违反它的载荷根本解不出记录——那样这条不变量就永远只能是通过，成了摆设。
|
||
违反它的行同时会被不变量一报出来（它确实读不回来），两条说的是同一处损坏的两个侧面。
|
||
|
||
**所以这一条不增加发现击穿的覆盖，它增加的是证据的精度**：上一条只能说「第 N 行解不
|
||
出来」，这一条直接指出是 `result_id` 与 `action_outcome` 对不上。留着它省的是排查时间，
|
||
不是漏判风险。
|
||
|
||
**一条 `step_completed` 载荷都没有时报「无法判定」。** 下限数的是**打着这个标签的行**，
|
||
不是解出来的记录——解不出来的那些行正是这条最该判的对象，按记录数当下限会把它们数漏。
|
||
"""
|
||
del config
|
||
if not facts.payloads_tagged("step_completed"):
|
||
return CheckOutcome(
|
||
undetermined=(
|
||
Evidence(
|
||
run_id=facts.run_id,
|
||
reason="日志里一条 step_completed 载荷都没有，这条配对没东西可验",
|
||
),
|
||
)
|
||
)
|
||
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` 里，动作结果与步记录说的必须是同一件事。
|
||
|
||
两者一次原子落地，所以它们不可能来自两次不同的执行。对不上说明装配那一层把某一侧
|
||
填错了，而这种错在轨迹里完全看不出来——两个字段各自都合法。
|
||
|
||
**观察那三列只在 `executed` 一档上是原样透传，另外两档不是**，判据必须跟着分档，
|
||
理由写在 `Invariant` 的说明里（那段会进报告）。这条一开始按「三档都逐字相同」写，
|
||
在 193 次真实运行上报了 9 处击穿，核下来全是判据错、不是库错。
|
||
|
||
**一条步记录都没有时报「无法判定」。下限是一条步记录，不要求它带动作结果**：没有动作
|
||
结果的那一档也在这条的判定范围里（那时步记录的 `action_status` 必须为空），所以一份
|
||
全是解析失败的日志确实验到了这条的一部分，报「判不了」反而是假的。
|
||
"""
|
||
del config
|
||
if not facts.steps():
|
||
return CheckOutcome(
|
||
undetermined=(
|
||
Evidence(
|
||
run_id=facts.run_id,
|
||
reason="日志里一条 step_completed 记录都没有，两侧一致这条没东西可验",
|
||
),
|
||
)
|
||
)
|
||
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
|
||
passthrough = outcome.status is ActionStatus.EXECUTED
|
||
pairs = [
|
||
("action_status", outcome.status, record.step.action_status),
|
||
(
|
||
"env_reported_completion",
|
||
outcome.env_reported_completion,
|
||
record.step.env_reported_completion,
|
||
),
|
||
]
|
||
if passthrough:
|
||
# 只有这一档，步记录上的观察三列是执行器返回值的原样透传。
|
||
pairs.append(
|
||
(
|
||
"observation_is_synthetic",
|
||
outcome.observation_is_synthetic,
|
||
record.step.observation_is_synthetic,
|
||
)
|
||
)
|
||
pairs.append(
|
||
(
|
||
"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 passthrough:
|
||
if outcome.observation != record.step.observation:
|
||
breaches.append(
|
||
Evidence(
|
||
run_id=facts.run_id,
|
||
line_no=number,
|
||
step_idx=record.step.step_idx,
|
||
expected="executed 档下步记录的 observation 与动作结果的同名字段逐字相同",
|
||
actual=(
|
||
f"文本不同（长度 {len(outcome.observation)} vs "
|
||
f"{len(record.step.observation)}）"
|
||
),
|
||
)
|
||
)
|
||
elif not record.step.observation_is_synthetic:
|
||
breaches.append(
|
||
Evidence(
|
||
run_id=facts.run_id,
|
||
line_no=number,
|
||
step_idx=record.step.step_idx,
|
||
expected=(
|
||
f"{outcome.status.value} 档下库替换了回填进历史的观察，"
|
||
"所以步记录的 observation_is_synthetic 为 True"
|
||
),
|
||
actual="observation_is_synthetic = False，那段观察没有被标成合成的",
|
||
)
|
||
)
|
||
return CheckOutcome(breaches=tuple(_cap(breaches, facts.run_id)))
|
||
|
||
|
||
def _check_prompt_chars_monotonic(facts: RunFacts, config: CheckConfig) -> CheckOutcome:
|
||
"""步记录里的 `prompt_chars` 随步号不回退。
|
||
|
||
**它验的是库自己记下来的那个数，不是历史真的没被截断过。** 两者不是一回事：库要是
|
||
截断了历史、却接着记一串不下降的 `prompt_chars`，这条照样通过。记分板手上只有日志，
|
||
日志里没有真正发出去的那串消息，这个缺口只能由压测那边包一层模型客户端、拿真实发出去的
|
||
消息长度对账来补。名字和说明都不承诺那件事——承诺了它，读者看见绿就以为截断已经被排除。
|
||
|
||
**不足两步时报「无法判定」**：一步和零步都凑不出相邻的两个值，没有任何比较发生过。
|
||
"""
|
||
del config
|
||
if len(facts.steps()) < 2:
|
||
return CheckOutcome(
|
||
undetermined=(
|
||
Evidence(
|
||
run_id=facts.run_id,
|
||
reason=(
|
||
f"日志里只有 {len(facts.steps())} 条步记录，凑不出相邻两步，"
|
||
"单调性没东西可验"
|
||
),
|
||
),
|
||
)
|
||
)
|
||
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:
|
||
"""日志与事件里每一条的 run_id 都等于文件名去掉后缀那部分。
|
||
|
||
**两半各有各的下限，分开判。** 日志那半要至少一条解得出来的记录，事件那半要至少一条
|
||
事件。一份有记录、事件却是空的产物（零步的 run，或者续跑时本进程一步没走完），日志那半
|
||
是真判过的，事件那半没东西可验——两半合成一个判定的话，其中一半的真空会被另一半的绿盖住，
|
||
而那正是这条要防的那种「看着验过了、其实没验」。
|
||
"""
|
||
del config
|
||
breaches: list[Evidence] = []
|
||
blocked: list[Evidence] = []
|
||
if not facts.records():
|
||
blocked.append(
|
||
Evidence(
|
||
run_id=facts.run_id,
|
||
reason="日志里没有一条解得出来的记录，日志那一半判不了",
|
||
)
|
||
)
|
||
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}") + "，事件那一半判不了",
|
||
)
|
||
)
|
||
elif not facts.event_run_ids:
|
||
blocked.append(
|
||
Evidence(
|
||
run_id=facts.run_id,
|
||
reason=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_int(facts: RunFacts, key: str) -> int:
|
||
"""从运行开始记录的参数快照里取一个整数上限。取不到就抛 `_MetaError`，让上层报判不了。"""
|
||
started = facts.run_started()
|
||
if started is None:
|
||
raise _MetaError("日志里没有 run_started，读不到参数快照")
|
||
raw = started.parameter_snapshot.get(key)
|
||
if raw is None:
|
||
raise _MetaError(f"参数快照里没有 {key}")
|
||
try:
|
||
return int(raw)
|
||
except ValueError as exc:
|
||
raise _MetaError(f"参数快照里的 {key} 不是整数：{_safe_id(raw)}") from exc
|
||
|
||
|
||
def _breach(
|
||
facts: RunFacts, expected: str, actual: str, *, step_idx: int | None = None
|
||
) -> CheckOutcome:
|
||
return CheckOutcome(
|
||
breaches=(
|
||
Evidence(run_id=facts.run_id, step_idx=step_idx, expected=expected, actual=actual),
|
||
)
|
||
)
|
||
|
||
|
||
def _blocked(facts: RunFacts, reason: str) -> CheckOutcome:
|
||
return CheckOutcome(undetermined=(Evidence(run_id=facts.run_id, reason=reason),))
|
||
|
||
|
||
def _status_name(step: StepRecord) -> str:
|
||
return "空" if step.action_status is None else step.action_status.value
|
||
|
||
|
||
def _rule_step_budget(facts: RunFacts, result: RunResult) -> CheckOutcome:
|
||
max_steps = _snapshot_int(facts, MAX_STEPS_SNAPSHOT_KEY)
|
||
if len(result.steps) == max_steps:
|
||
return CheckOutcome()
|
||
return _breach(
|
||
facts, f"step_budget 时步数 = max_steps = {max_steps}", f"步数 = {len(result.steps)}"
|
||
)
|
||
|
||
|
||
def _rule_agent_finished(facts: RunFacts, result: RunResult) -> CheckOutcome:
|
||
if result.final_answer:
|
||
return CheckOutcome()
|
||
return _breach(
|
||
facts,
|
||
"agent_finished 时 final_answer 非空",
|
||
f"final_answer 长度 {_text_len(result.final_answer)}",
|
||
)
|
||
|
||
|
||
def _rule_cancelled(facts: RunFacts, result: RunResult) -> CheckOutcome:
|
||
del result
|
||
if facts.run_finished() is not None:
|
||
return CheckOutcome()
|
||
return _breach(
|
||
facts,
|
||
"cancelled 时日志里有 run_finished 记录",
|
||
"日志里没有 run_finished，恢复会把它当成可以续跑",
|
||
)
|
||
|
||
|
||
def _rule_parse_failed_repeatedly(facts: RunFacts, result: RunResult) -> CheckOutcome:
|
||
"""轨迹末尾连续解析失败的步数，恰好等于连续失败上限。
|
||
|
||
恰好相等而不是「至少」：那个计数每次解析失败加一、加完立刻问，达到上限就停，所以它撞线
|
||
时不可能超过上限；而任何一个有效决策会把它清零，所以那一段必定连续、必定贴着末尾。
|
||
这几步还必须没有动作状态——解析失败那一支根本不碰环境。
|
||
"""
|
||
limit = _snapshot_int(facts, MAX_PARSE_FAILURES_SNAPSHOT_KEY)
|
||
tail = 0
|
||
for step in reversed(result.steps):
|
||
if step.parse_ok:
|
||
break
|
||
tail += 1
|
||
breaches: list[Evidence] = []
|
||
if tail != limit:
|
||
breaches.append(
|
||
Evidence(
|
||
run_id=facts.run_id,
|
||
expected=(f"轨迹末尾连续 {limit} 步解析失败（= max_consecutive_parse_failures）"),
|
||
actual=f"末尾连续 {tail} 步解析失败",
|
||
)
|
||
)
|
||
for step in result.steps[len(result.steps) - tail :]:
|
||
if step.action_status is not None:
|
||
breaches.append(
|
||
Evidence(
|
||
run_id=facts.run_id,
|
||
step_idx=step.step_idx,
|
||
expected="解析失败的步不碰环境，action_status 为空",
|
||
actual=f"action_status = {step.action_status.value}",
|
||
)
|
||
)
|
||
return CheckOutcome(breaches=tuple(_cap(breaches, facts.run_id)))
|
||
|
||
|
||
def _rule_context_overflow(facts: RunFacts, result: RunResult) -> CheckOutcome:
|
||
"""落盘的每一步的 `prompt_chars` 都不超过上限。
|
||
|
||
超限的那次装配根本不产生步记录（那一档在调模型之前就终止，没花钱、没有调用标识要对账），
|
||
所以落盘的每一步必定在线内。哪一步超了，说明有一次超限装配被放行去调模型了。
|
||
压测那侧的同名判据换个角度验同一件事：它包一层模型客户端，量真实发出去的消息有多长。
|
||
|
||
**一步都没落盘时报无法判定。** 首次装配就超限的运行正是这样，那是这个停止原因最典型的
|
||
形态，可它确实一步都没验到。
|
||
"""
|
||
limit = _snapshot_int(facts, MAX_PROMPT_CHARS_SNAPSHOT_KEY)
|
||
if not result.steps:
|
||
return _blocked(
|
||
facts,
|
||
"这次运行一步都没落盘（首次装配就超限的运行正是这样），"
|
||
"没有任何一步的 prompt_chars 可验",
|
||
)
|
||
breaches = [
|
||
Evidence(
|
||
run_id=facts.run_id,
|
||
step_idx=step.step_idx,
|
||
expected=f"prompt_chars ≤ max_prompt_chars = {limit}",
|
||
actual=f"prompt_chars = {step.prompt_chars}",
|
||
)
|
||
for step in result.steps
|
||
if step.prompt_chars > limit
|
||
]
|
||
return CheckOutcome(breaches=tuple(_cap(breaches, facts.run_id)))
|
||
|
||
|
||
def _rule_action_budget(facts: RunFacts, result: RunResult) -> CheckOutcome:
|
||
"""动作状态为 executed 的步数，恰好等于已执行动作上限。
|
||
|
||
数的是 executed 这一档，不是全部动作：未执行与环境故障都不加那个计数
|
||
（`polyloop._stopping.RunCounters.with_action_executed`）。
|
||
"""
|
||
limit = _snapshot_int(facts, MAX_ACTIONS_SNAPSHOT_KEY)
|
||
executed = sum(1 for step in result.steps if step.action_status is ActionStatus.EXECUTED)
|
||
if executed == limit:
|
||
return CheckOutcome()
|
||
return _breach(
|
||
facts,
|
||
f"action_budget 时已执行动作数 = max_actions = {limit}",
|
||
f"action_status 为 executed 的步有 {executed} 条",
|
||
)
|
||
|
||
|
||
def _rule_env_error(facts: RunFacts, result: RunResult) -> CheckOutcome:
|
||
if not result.steps:
|
||
return _breach(facts, "env_error 时至少有一步（这个原因由一次动作结算撞出来）", "步数 = 0")
|
||
last = result.steps[-1]
|
||
if last.action_status is ActionStatus.ENV_ERROR:
|
||
return CheckOutcome()
|
||
return _breach(
|
||
facts,
|
||
"env_error 时最后一步的 action_status = env_error",
|
||
f"action_status = {_status_name(last)}",
|
||
step_idx=last.step_idx,
|
||
)
|
||
|
||
|
||
def _rule_llm_error(facts: RunFacts, result: RunResult) -> CheckOutcome:
|
||
"""最后一步长得像「模型调用失败」那种步。
|
||
|
||
判据是 `parse_ok=False` 且 `parse_error` 为空，**这一对正是库自己用来把它和解析失败分开
|
||
的东西**（`polyloop.session` 的续跑结算读的就是这两个字段）：那一步压根没走到解释器，
|
||
所以没有回喂给模型的说明，而解析失败必定带着一段。只看「没有动作结果」分不开这两者，
|
||
只看 `call_id` 为空也分不开——解释器拿到的回复本来就允许不带调用标识。
|
||
"""
|
||
if not result.steps:
|
||
return _breach(facts, "llm_error 时至少有一步（那一步记的就是这次失败）", "步数 = 0")
|
||
last = result.steps[-1]
|
||
expected = "llm_error 的最后一步：没有动作状态、parse_ok=False、parse_error 为空、call_id 为空"
|
||
problems: list[str] = []
|
||
if last.action_status is not None:
|
||
problems.append(f"action_status = {last.action_status.value}")
|
||
if last.parse_ok:
|
||
problems.append("parse_ok = True")
|
||
if last.parse_error is not None:
|
||
problems.append(f"parse_error 有值（长度 {len(last.parse_error)}），那是解析失败的样子")
|
||
if last.call_id is not None:
|
||
problems.append("call_id 有值")
|
||
if not problems:
|
||
return CheckOutcome()
|
||
return CheckOutcome(
|
||
breaches=tuple(
|
||
_cap(
|
||
[
|
||
Evidence(
|
||
run_id=facts.run_id,
|
||
step_idx=last.step_idx,
|
||
expected=expected,
|
||
actual=problem,
|
||
)
|
||
for problem in problems
|
||
],
|
||
facts.run_id,
|
||
)
|
||
)
|
||
)
|
||
|
||
|
||
def _rule_resume_state_unknown(facts: RunFacts, result: RunResult) -> CheckOutcome:
|
||
"""日志里最后一条意图悬空，且它声明绝不重放。
|
||
|
||
这个停止原因只有两条来路，两条都长这样：模型调用意图写了、结果没写，而这次运行的模型
|
||
重放策略是「绝不」；或者动作意图写了、步记录没写，而那个工具声明绝不重放。撞上这一档
|
||
时库不再追加步、也不再写意图，所以那条悬空的意图仍然是日志里的最后一条。
|
||
"""
|
||
del result
|
||
intents = facts.intents()
|
||
if not intents:
|
||
return _blocked(facts, "日志里一条意图都没有，这个停止原因的自洽条件判不了")
|
||
dangling = _dangling_intents(facts)
|
||
last_position = len(intents) - 1
|
||
if not dangling or dangling[-1][0] != last_position:
|
||
return _breach(
|
||
facts,
|
||
"resume_state_unknown 时日志里最后一条意图是悬空的",
|
||
"没有悬空的意图" if not dangling else "最后一条意图有归宿",
|
||
)
|
||
intent = dangling[-1][2]
|
||
if intent.replay_policy is ReplayPolicy.NEVER:
|
||
return CheckOutcome()
|
||
return _breach(
|
||
facts,
|
||
"悬空那条意图的 replay_policy = never（声明可安全重放的会被直接重放，不会停在这里）",
|
||
f"replay_policy = {intent.replay_policy.value}",
|
||
)
|
||
|
||
|
||
#: 十个停止原因各自的自洽规矩。`task_completed` 另走一条，它要调用方传进来的工具名。
|
||
_STOP_REASON_RULES: Mapping[StopReason, Callable[[RunFacts, RunResult], CheckOutcome]] = {
|
||
StopReason.STEP_BUDGET: _rule_step_budget,
|
||
StopReason.AGENT_FINISHED: _rule_agent_finished,
|
||
StopReason.CANCELLED: _rule_cancelled,
|
||
StopReason.PARSE_FAILED_REPEATEDLY: _rule_parse_failed_repeatedly,
|
||
StopReason.CONTEXT_OVERFLOW: _rule_context_overflow,
|
||
StopReason.ACTION_BUDGET: _rule_action_budget,
|
||
StopReason.ENV_ERROR: _rule_env_error,
|
||
StopReason.LLM_ERROR: _rule_llm_error,
|
||
StopReason.RESUME_STATE_UNKNOWN: _rule_resume_state_unknown,
|
||
}
|
||
|
||
|
||
def _check_stop_reason_consistent(facts: RunFacts, config: CheckConfig) -> CheckOutcome:
|
||
"""停止原因与轨迹自洽。十个取值各有一条规矩。
|
||
|
||
读不到停止原因、或者某条规矩要的东西不在日志里（参数快照缺了某个上限），报「无法判定」，
|
||
不编一个答案。将来 `StopReason` 加了取值而这里没跟上，也走同一条路——那时它是显式的
|
||
「这个取值还没有规矩」，不是一条静默的绿。
|
||
"""
|
||
result = facts.effective_result()
|
||
if result is None:
|
||
return _blocked(facts, "既没有 run_finished 也没有 .result.json，读不到停止原因")
|
||
reason = result.stop_reason
|
||
if reason is StopReason.TASK_COMPLETED:
|
||
return _check_task_completed(facts, result, config)
|
||
rule = _STOP_REASON_RULES.get(reason)
|
||
if rule is None:
|
||
return _blocked(facts, f"停止原因 {reason.value} 还没有自洽规矩")
|
||
try:
|
||
return rule(facts, result)
|
||
except _MetaError as problem:
|
||
return _blocked(facts, str(problem))
|
||
|
||
|
||
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=(
|
||
"日志里每一条被换行终结的行都解得出记录；撕裂只允许出现在最后一行。"
|
||
"一条被终结的行都没有的 run 报无法判定——空文件和「只写了半行就被杀」都落在这里，"
|
||
"没有任何一行被读回来过。"
|
||
),
|
||
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 递增，不重复不跳号。"
|
||
"一条步记录都没有的 run 报无法判定，不报通过——那种 run 上这条根本没被验过。"
|
||
),
|
||
check=_check_step_indices,
|
||
),
|
||
Invariant(
|
||
name="意图都有归宿",
|
||
description=(
|
||
"每条意图的 result_id 都能在对应的结果记录里找到；"
|
||
"至多一条悬空，且必须是日志里最后一条意图。"
|
||
"一条意图都没有的 run 报无法判定；有意图而没有任何结果记录的 run 照判，"
|
||
"那正是这条要判的那种。"
|
||
),
|
||
check=_check_intents_resolved,
|
||
),
|
||
Invariant(
|
||
name="步记录的内部不变量",
|
||
description=(
|
||
"step_completed 的 result_id 为空当且仅当 action_outcome 为空。"
|
||
"**它发现得了的击穿，「日志能被读回来」也都发现得了**——StepCompleted 在构造期就"
|
||
"守这条，所以违反它的那一行本来就解不出记录。留着这一条不是为了多一层覆盖，"
|
||
"是为了让证据直接指向那两个字段：上一条只会说「第 N 行解不出来」，拿着那句话还得"
|
||
"回去翻文件猜是哪儿对不上。它省的是排查时间，不是漏判风险。"
|
||
"一条 step_completed 载荷都没有的 run 报无法判定，下限数的是打着这个标签的行、"
|
||
"不是解出来的记录——解不出来的那些行正是它最该判的对象。"
|
||
),
|
||
check=_check_step_pairing,
|
||
),
|
||
Invariant(
|
||
name="动作结果与步记录一致",
|
||
description=(
|
||
"同一条 step_completed 里，动作结果与步记录说的是同一件事。状态与完成标记这两项"
|
||
"在三档下都必须相同。观察那几列只在 executed 一档下逐字相同，因为只有那一档是"
|
||
"执行器返回值的原样透传：动作被拒绝与环境故障这两档，库刻意不把执行器给的观察"
|
||
"回填进历史，换成合成的那一段——执行器那段是「模型看得见的东西」，每次现造的话"
|
||
"同一份配置跑出来的两次运行在模型看来其实不同。执行器的原文没有丢，它就留在同一"
|
||
"条记录的动作结果里。所以这两档改判另一件事：库既然替换了观察，就必须把步记录的 "
|
||
"observation_is_synthetic 立起来，不立才是真出了问题。"
|
||
"一条步记录都没有的 run 报无法判定。下限是一条步记录，不要求它带动作结果——"
|
||
"没有动作结果的那一档也在这条的判定范围里（那时步记录的 action_status 必须为空），"
|
||
"所以一份全是解析失败的日志确实验到了这条的一部分。"
|
||
),
|
||
check=_check_outcome_agrees_with_step,
|
||
),
|
||
Invariant(
|
||
name="提示词字符数单调不减",
|
||
description=(
|
||
"同一个 run 里步记录的 prompt_chars 随步号不回退。**它验的是库自己记下来的那个"
|
||
"数，不是历史真的没被截断过**：库要是截断了历史、却接着记一串不下降的 "
|
||
"prompt_chars，这条照样通过。记分板手上只有日志，而日志里没有真正发出去的那串"
|
||
"消息，所以那件事这一层验不到，得由压测那边包一层模型客户端、拿真实发出去的消息"
|
||
"长度对账。不足两步的 run 报无法判定——一步和零步都凑不出相邻的两个值。"
|
||
),
|
||
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 都等于文件名去掉后缀那部分。"
|
||
"两半各有各的下限：日志那半要至少一条解得出来的记录，事件那半要至少一条事件，"
|
||
"任一半没东西可验就报无法判定。一份有记录、事件却是空的产物（零步的 run，或者"
|
||
"续跑时本进程一步没走完）只判得了日志那半。"
|
||
),
|
||
check=_check_no_crosstalk,
|
||
),
|
||
Invariant(
|
||
name="停止原因与轨迹自洽",
|
||
description=(
|
||
"十个停止原因各有一条规矩。task_completed 要环境侧完成证据或带完成标记的工具；"
|
||
"step_budget 时步数等于 max_steps；action_budget 时 executed 的步数等于 "
|
||
"max_actions；parse_failed_repeatedly 时轨迹末尾连续解析失败的步数等于 "
|
||
"max_consecutive_parse_failures，且那几步不碰环境；context_overflow 时落盘的每一"
|
||
"步的 prompt_chars 都不超过 max_prompt_chars（超限的那次装配根本不产生步记录）；"
|
||
"env_error 时最后一步的 action_status 是 env_error；llm_error 时最后一步没有动作"
|
||
"状态、parse_ok 为假且 parse_error 为空（这一对是库自己用来把它和解析失败分开的）；"
|
||
"agent_finished 时最终回答非空；cancelled 时日志里有结束记录；"
|
||
"resume_state_unknown 时最后一条意图悬空且声明绝不重放。"
|
||
"两种情况报无法判定：参数快照里缺某个上限；以及 context_overflow 而一步都没落盘"
|
||
"——首次装配就超限的运行正是这样，那时确实一步都没验到。"
|
||
),
|
||
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())
|