From ff594574821a8230935efd551ae3950c5690af13 Mon Sep 17 00:00:00 2001 From: iomgaa Date: Tue, 11 Aug 2026 11:53:56 -0400 Subject: [PATCH] =?UTF-8?q?fix(soak):=20=E8=AE=B0=E5=88=86=E6=9D=BF?= =?UTF-8?q?=E7=9A=84=E7=9C=9F=E7=A9=BA=E6=88=90=E7=AB=8B=E4=B8=8E=E5=81=9C?= =?UTF-8?q?=E6=AD=A2=E5=8E=9F=E5=9B=A0=E8=A6=86=E7=9B=96=EF=BC=8C=E4=B8=A4?= =?UTF-8?q?=E5=A4=84=E9=83=BD=E4=BC=9A=E6=8A=8A=E3=80=8C=E4=BB=80=E4=B9=88?= =?UTF-8?q?=E9=83=BD=E6=B2=A1=E9=AA=8C=E3=80=8D=E6=98=BE=E7=A4=BA=E6=88=90?= =?UTF-8?q?=E7=BB=BF?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Codex 报了两条数据不足时真空成立的不变量，顺着同类找齐了七条：零步的「步号连续」与 「动作结果与步记录一致」、不足两步的「提示词字符数单调不减」、零意图的「意图都有归宿」、 零载荷的「步记录的内部不变量」、零记录的「不串台」、零行的「日志能被读回来」。同一类 缺陷改一半，剩下那一半照样会在某天把一次什么都没验的跑显示成绿。 各条的数据下限不一样，反直觉的三处写进了说明：「步记录的内部不变量」数的是打着标签的行 不是解出来的记录（违反配对的行本来就解不出记录，按记录数当下限会把它最该判的对象数漏）； 「动作结果与步记录一致」不要求那条步记录带动作结果；「不串台」两半各判各的，合成一个的话 一半的真空会被另一半的绿盖住。 「停止原因与轨迹自洽」原本只覆盖四个取值、另外六个直接放行——不是数据不足，是判据本来就 该覆盖而没覆盖，后果和真空成立一样。六个都补了规矩，llm_error 那条按库自己的判据写 （解析失败必定带说明，模型调用失败那一步压根没走到解释器，只看有没有动作结果分不开这两者）。 另加一条断言十个取值一个不漏，将来加了取值而这里没跟上会显式报「还没有规矩」。 「提示词字符数单调不减」的说明原本承诺「历史只追加」，实现只比较库自己记录的数——承诺了 它，读者看见绿就以为截断被排除了。改成只承诺它验得到的，真正的对账在故障注入那侧。 拿 193 次真实运行重跑：十一条仍然全过，而这次那 8 次解析失败连击、1 次模型调用失败、 1 次撞步数上限是被真规矩判过的。 Co-Authored-By: Claude Opus 5 (1M context) --- tools/soak/scoreboard.py | 502 ++++++++++++++++++--- tools/soak/tests/test_run_soak.py | 18 +- tools/soak/tests/test_scoreboard.py | 659 +++++++++++++++++++++++++--- 3 files changed, 1059 insertions(+), 120 deletions(-) diff --git a/tools/soak/scoreboard.py b/tools/soak/scoreboard.py index 085ecac..6ead4ed 100644 --- a/tools/soak/scoreboard.py +++ b/tools/soak/scoreboard.py @@ -61,10 +61,12 @@ from polyloop.types import ( Intent, IntentKind, ModelCallResult, + ReplayPolicy, RunFinished, RunResult, RunStarted, StepCompleted, + StepRecord, StopReason, ) @@ -73,9 +75,12 @@ RESULT_SUFFIX = ".result.json" EVENTS_SUFFIX = ".events.jsonl" META_SUFFIX = ".meta.json" -#: 步数上限在参数快照里的键名。快照由 `RunRequest.parameter_snapshot()` 拼出来， +#: 四个预算上限在参数快照里的键名。快照由 `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 条」。 #: 一份四十步的坏日志能刷出四十行同样的证据，那种报告没人会读到底。 @@ -524,9 +529,28 @@ def _cap(items: Sequence[Evidence], run_id: str) -> list[Evidence]: 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, @@ -624,7 +648,22 @@ def _check_result_matches_log(facts: RunFacts, config: CheckConfig) -> CheckOutc 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: @@ -639,13 +678,16 @@ def _check_step_indices(facts: RunFacts, config: CheckConfig) -> CheckOutcome: return CheckOutcome(breaches=tuple(_cap(breaches, facts.run_id))) -def _check_intents_resolved(facts: RunFacts, config: CheckConfig) -> CheckOutcome: - del config +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} - intents = facts.intents() dangling: list[tuple[int, int, Intent]] = [] - for position, (number, intent) in enumerate(intents): + for position, (number, intent) in enumerate(facts.intents()): resolved = ( intent.result_id in model_ids if intent.kind is IntentKind.MODEL_CALL @@ -653,6 +695,28 @@ def _check_intents_resolved(facts: RunFacts, config: CheckConfig) -> CheckOutcom ) 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 @@ -682,8 +746,24 @@ def _check_step_pairing(facts: RunFacts, config: CheckConfig) -> CheckOutcome: 判据落在**原始载荷**上，不在解出来的记录上：`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"): @@ -723,8 +803,21 @@ def _check_outcome_agrees_with_step(facts: RunFacts, config: CheckConfig) -> Che **观察那三列只在 `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 @@ -807,7 +900,28 @@ def _check_outcome_agrees_with_step(facts: RunFacts, config: CheckConfig) -> Che 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(): @@ -922,9 +1036,23 @@ def _check_delivery_failures(facts: RunFacts, config: CheckConfig) -> CheckOutco 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: @@ -946,6 +1074,13 @@ def _check_no_crosstalk(facts: RunFacts, config: CheckConfig) -> CheckOutcome: 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: @@ -962,81 +1097,268 @@ def _check_no_crosstalk(facts: RunFacts, config: CheckConfig) -> CheckOutcome: ) -def _snapshot_max_steps(facts: RunFacts) -> int: +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(MAX_STEPS_SNAPSHOT_KEY) + raw = started.parameter_snapshot.get(key) if raw is None: - raise _MetaError(f"参数快照里没有 {MAX_STEPS_SNAPSHOT_KEY}") + raise _MetaError(f"参数快照里没有 {key}") try: return int(raw) except ValueError as exc: - raise _MetaError( - f"参数快照里的 {MAX_STEPS_SNAPSHOT_KEY} 不是整数：{_safe_id(raw)}" - ) from 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 CheckOutcome( - undetermined=( - Evidence( - run_id=facts.run_id, - reason="既没有 run_finished 也没有 .result.json，读不到停止原因", - ), - ) - ) + return _blocked(facts, "既没有 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() + 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: @@ -1099,7 +1421,11 @@ class Invariant: INVARIANTS: tuple[Invariant, ...] = ( Invariant( name="日志能被读回来", - description="日志里每一条被换行终结的行都解得出记录；撕裂只允许出现在最后一行。", + description=( + "日志里每一条被换行终结的行都解得出记录；撕裂只允许出现在最后一行。" + "一条被终结的行都没有的 run 报无法判定——空文件和「只写了半行就被杀」都落在这里，" + "没有任何一行被读回来过。" + ), check=_check_log_readable, ), Invariant( @@ -1111,7 +1437,10 @@ INVARIANTS: tuple[Invariant, ...] = ( ), Invariant( name="步号连续", - description="step_completed 的 step_idx 从 0 开始逐 1 递增，不重复不跳号。", + description=( + "step_completed 的 step_idx 从 0 开始逐 1 递增，不重复不跳号。" + "一条步记录都没有的 run 报无法判定，不报通过——那种 run 上这条根本没被验过。" + ), check=_check_step_indices, ), Invariant( @@ -1119,12 +1448,22 @@ INVARIANTS: tuple[Invariant, ...] = ( description=( "每条意图的 result_id 都能在对应的结果记录里找到；" "至多一条悬空，且必须是日志里最后一条意图。" + "一条意图都没有的 run 报无法判定；有意图而没有任何结果记录的 run 照判，" + "那正是这条要判的那种。" ), check=_check_intents_resolved, ), Invariant( name="步记录的内部不变量", - description="step_completed 的 result_id 为空当且仅当 action_outcome 为空。", + description=( + "step_completed 的 result_id 为空当且仅当 action_outcome 为空。" + "**它发现得了的击穿，「日志能被读回来」也都发现得了**——StepCompleted 在构造期就" + "守这条，所以违反它的那一行本来就解不出记录。留着这一条不是为了多一层覆盖，" + "是为了让证据直接指向那两个字段：上一条只会说「第 N 行解不出来」，拿着那句话还得" + "回去翻文件猜是哪儿对不上。它省的是排查时间，不是漏判风险。" + "一条 step_completed 载荷都没有的 run 报无法判定，下限数的是打着这个标签的行、" + "不是解出来的记录——解不出来的那些行正是它最该判的对象。" + ), check=_check_step_pairing, ), Invariant( @@ -1137,12 +1476,21 @@ INVARIANTS: tuple[Invariant, ...] = ( "同一份配置跑出来的两次运行在模型看来其实不同。执行器的原文没有丢，它就留在同一" "条记录的动作结果里。所以这两档改判另一件事：库既然替换了观察，就必须把步记录的 " "observation_is_synthetic 立起来，不立才是真出了问题。" + "一条步记录都没有的 run 报无法判定。下限是一条步记录，不要求它带动作结果——" + "没有动作结果的那一档也在这条的判定范围里（那时步记录的 action_status 必须为空），" + "所以一份全是解析失败的日志确实验到了这条的一部分。" ), check=_check_outcome_agrees_with_step, ), Invariant( name="提示词字符数单调不减", - description="同一个 run 里 prompt_chars 随步号不减——历史只追加。", + description=( + "同一个 run 里步记录的 prompt_chars 随步号不回退。**它验的是库自己记下来的那个" + "数，不是历史真的没被截断过**：库要是截断了历史、却接着记一串不下降的 " + "prompt_chars，这条照样通过。记分板手上只有日志，而日志里没有真正发出去的那串" + "消息，所以那件事这一层验不到，得由压测那边包一层模型客户端、拿真实发出去的消息" + "长度对账。不足两步的 run 报无法判定——一步和零步都凑不出相邻的两个值。" + ), check=_check_prompt_chars_monotonic, ), Invariant( @@ -1157,14 +1505,28 @@ INVARIANTS: tuple[Invariant, ...] = ( ), Invariant( name="不串台", - description="日志与事件里每一条记录的 run_id 都等于文件名去掉后缀那部分。", + description=( + "日志与事件里每一条记录的 run_id 都等于文件名去掉后缀那部分。" + "两半各有各的下限：日志那半要至少一条解得出来的记录，事件那半要至少一条事件，" + "任一半没东西可验就报无法判定。一份有记录、事件却是空的产物（零步的 run，或者" + "续跑时本进程一步没走完）只判得了日志那半。" + ), check=_check_no_crosstalk, ), Invariant( name="停止原因与轨迹自洽", description=( - "task_completed 有完成证据；step_budget 时步数等于上限；" - "agent_finished 时最终回答非空；cancelled 时日志里有结束记录。" + "十个停止原因各有一条规矩。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, ), diff --git a/tools/soak/tests/test_run_soak.py b/tools/soak/tests/test_run_soak.py index 4968822..48ae405 100644 --- a/tools/soak/tests/test_run_soak.py +++ b/tools/soak/tests/test_run_soak.py @@ -85,6 +85,16 @@ _SUBMIT = json.dumps( #: 两个阶段没有结束通路（提交工具只在 summarize 那一阶段），不这样它们会一路走到 50 步。 _GARBAGE = "我先想一想这道题。" +#: summarize 阶段的第一步：先读一段再提交。 +#: +#: **不是可有可无的一步。** 记分板的「提示词字符数单调不减」要两步才凑得出相邻一对， +#: 一步就提交的运行在它那里是「无从判起」而不是「通过」。这条测试断言的是整批全绿， +#: 所以替身必须走够两步——否则它测到的是记分板在数据不足时的行为，不是产物合不合约定。 +_READ = json.dumps( + {"tool": "read_document", "arguments": {"path": "tender.md", "start_line": 1, "end_line": 2}}, + ensure_ascii=False, +) + def _audit_task(index: int) -> govdoc_scenario.AuditTask: checkpoint = govdoc_scenario.Checkpoint( @@ -376,7 +386,13 @@ async def test_artifacts_pass_the_scoreboard(tmp_path: Path) -> None: 两边的字段约定没有任何机器约束把它们钉在一起，这条测试就是那个约束。 """ - guard = _guard(render=lambda call: _SUBMIT if "summarize" in call.run_id else _GARBAGE) + + def render(call: object) -> str: + if "summarize" not in call.run_id: # type: ignore[attr-defined] + return _GARBAGE + return _READ if call.call_index == 0 else _SUBMIT # type: ignore[attr-defined] + + guard = _guard(render=render) for index in range(2): await run_govdoc_task( task=_audit_task(index), diff --git a/tools/soak/tests/test_scoreboard.py b/tools/soak/tests/test_scoreboard.py index 6fcb8d6..690d62a 100644 --- a/tools/soak/tests/test_scoreboard.py +++ b/tools/soak/tests/test_scoreboard.py @@ -34,6 +34,7 @@ from polyloop.types import ( # noqa: E402 StopReason, ) from tools.soak.scoreboard import ( # noqa: E402 + _STOP_REASON_RULES, EXIT_BREACHED, EXIT_UNDETERMINED, Verdict, @@ -59,93 +60,172 @@ _TAGS = { # --------------------------------------------------------------------------- +#: 一步可以长成的几种样子，逐条对着库里造那条步记录的那个函数。 +#: `action` 走 `_action_step`，`parse_failure` 走 `_parse_failure_step`， +#: `call_failure` 走 `_failed_call_step`，`final_answer` 走 `_final_answer_step`。 +STEP_KINDS = ( + "action", + "not_executed", + "env_error", + "parse_failure", + "call_failure", + "final_answer", +) + + def build_records( run_id: str, *, steps: int = 2, + step_kinds: Sequence[str] | None = None, stop_reason: StopReason = StopReason.TASK_COMPLETED, max_steps: int = 5, + max_actions: int = 20, + max_parse_failures: int = 2, + max_prompt_chars: int = 100000, + model_replay_policy: ReplayPolicy = ReplayPolicy.NEVER, final_answer: str | None = None, complete_on_last: bool = True, tool_name: str | None = "run_code", sink_failures: int = 0, observation: str = "普通观察", + prompt_chars_at: Callable[[int], int] = lambda index: 100 + index * 10, ) -> tuple[list[object], RunResult]: - """造一份自洽的记录序列：一步一组「模型意图 / 模型结果 / 动作意图 / 逐步结果」。""" + """造一份自洽的记录序列。 + + 默认每一步都是「模型意图 / 模型结果 / 动作意图 / 逐步结果」那四条，动作执行成功。 + `step_kinds` 给出的话就按它逐步造，取值见 `STEP_KINDS`——那几种步在库里由不同的函数 + 产出，字段形状各不相同，停止原因的自洽判据分的正是这些形状。 + """ + kinds = list(step_kinds) if step_kinds is not None else ["action"] * steps + for kind in kinds: + assert kind in STEP_KINDS, kind records: list[object] = [ RunStarted( run_id=run_id, parameter_snapshot={ "request.max_steps": str(max_steps), + "request.max_actions": str(max_actions), + "request.max_consecutive_parse_failures": str(max_parse_failures), + "request.max_prompt_chars": str(max_prompt_chars), "store.kind": "jsonl", }, ) ] step_records: list[StepRecord] = [] - for index in range(steps): - completed = complete_on_last and index == steps - 1 + for index, kind in enumerate(kinds): + completed = complete_on_last and index == len(kinds) - 1 text = f"{observation}#{index}" + chars = prompt_chars_at(index) records.append( Intent( run_id=run_id, kind=IntentKind.MODEL_CALL, call_index=index, result_id=f"m{index}", - replay_policy=ReplayPolicy.NEVER, + replay_policy=model_replay_policy, ) ) + reply = ModelReply(call_id=f"c{index}", content="决策文本", thinking="") records.append( ModelCallResult( run_id=run_id, result_id=f"m{index}", - reply=ModelReply(call_id=f"c{index}", content="决策文本", thinking=""), - failure=None, + # 模型调用失败那一步：结果记录在，但它记的是失败。 + reply=None if kind == "call_failure" else reply, + failure="TimeoutError: 网关没回" if kind == "call_failure" else None, ) ) - records.append( - Intent( - run_id=run_id, - kind=IntentKind.ACTION, - call_index=index, - result_id=f"a{index}", - replay_policy=ReplayPolicy.NEVER, + common = { + "step_idx": index, + "content_chars": 4, + "thinking_chars": 0, + "observation_truncated_chars": 0, + "prompt_chars": chars, + "step_wall_ms": 7, + } + if kind == "call_failure": + step = StepRecord( + **common, # type: ignore[arg-type] + raw_output="", + action=None, + parse_ok=False, + # 这一步压根没走到解释器，所以没有回喂给模型的说明——这正是库用来把它和 + # 解析失败分开的那一对字段。 + parse_error=None, + observation="[模型调用失败]", + observation_is_synthetic=True, + call_id=None, ) - ) - 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, - ) + elif kind == "parse_failure": + step = StepRecord( + **common, # type: ignore[arg-type] + raw_output="决策文本", + action=None, + parse_ok=False, + parse_error="解释不出有效决策，请重新输出一个 JSON 对象。", + observation="解释不出有效决策，请重新输出一个 JSON 对象。", + observation_is_synthetic=True, + call_id=f"c{index}", + ) + elif kind == "final_answer": + step = StepRecord( + **common, # type: ignore[arg-type] + raw_output="决策文本", + action=None, + parse_ok=True, + parse_error=None, + observation="", + observation_is_synthetic=False, + call_id=f"c{index}", + ) + else: + status = { + "action": ActionStatus.EXECUTED, + "not_executed": ActionStatus.NOT_EXECUTED, + "env_error": ActionStatus.ENV_ERROR, + }[kind] + passthrough = status is ActionStatus.EXECUTED + 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=status, + observation=text, + observation_is_synthetic=status is ActionStatus.NOT_EXECUTED, + env_reported_completion=completed, + observation_truncated_chars=0, + ) + step = StepRecord( + **common, # type: ignore[arg-type] + raw_output="决策文本", + action="run_code", + parse_ok=True, + parse_error=None, + # 未执行与环境故障两档，库换掉回填进历史的那段观察并把合成标记立起来。 + observation=text if passthrough else "[动作没有执行]", + observation_is_synthetic=not passthrough, + call_id=f"c{index}", + tool_name=tool_name, + tool_arguments="{}", + action_status=status, + 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 + ) + ) + continue step_records.append(step) - records.append( - StepCompleted( - run_id=run_id, - result_id=f"a{index}", - action_outcome=outcome, - step=step, - ) - ) + records.append(StepCompleted(run_id=run_id, result_id=None, action_outcome=None, step=step)) result = RunResult( run_id=run_id, stop_reason=stop_reason, @@ -256,6 +336,46 @@ def reshape_step( edit_log(path, edit) +def edit_result_steps(runs_dir: Path, run_id: str, mutate: Callable[[list[dict]], None]) -> None: + """同时改 run_finished 内嵌的那份结果与 `.result.json` 里的步。 + + 两边一起改，「跨进程的结果与内存里的一致」才不会跟着一起红——这里要看的是停止原因那条， + 不是那条。日志里独立的 step_completed 行不动：没有任何判据拿它和结果里的步对比。 + """ + + def edit(items: list[dict]) -> list[dict]: + for item in items: + if item["record"] == "run_finished": + mutate(item["result"]["steps"]) + return items + + edit_log(runs_dir / f"{run_id}.jsonl", edit) + path = runs_dir / f"{run_id}.result.json" + payload = json.loads(path.read_text(encoding="utf-8")) + mutate(payload["steps"]) + path.write_text(json.dumps(payload, ensure_ascii=False), encoding="utf-8") + + +def append_dangling_intent( + runs_dir: Path, run_id: str, *, replay_policy: str = "never", kind: str = "model_call" +) -> None: + """在日志末尾补一条没有归宿的意图，模拟「意图写了、结果没写」那个断点。""" + edit_log( + runs_dir / f"{run_id}.jsonl", + lambda items: [ + *items, + { + "record": "intent", + "run_id": run_id, + "kind": kind, + "call_index": 99, + "result_id": "dangling-99", + "replay_policy": replay_policy, + }, + ], + ) + + def verdict_of(scoreboard, name: str) -> Verdict: for item in scoreboard.invariants: if item.name == name: @@ -302,7 +422,8 @@ def test_several_legal_runs_pass(tmp_path: Path) -> None: materialize( tmp_path, "soak-0003", - steps=1, + # 两步，不是一步：一步凑不出相邻的两个 prompt_chars，那条会报无法判定。 + steps=2, stop_reason=StopReason.AGENT_FINISHED, complete_on_last=False, final_answer="给出的答案", @@ -695,6 +816,227 @@ def test_cancelled_without_run_finished_is_a_breach(tmp_path: Path) -> None: assert_breached(evaluate(tmp_path), "停止原因与轨迹自洽") +RULE = "停止原因与轨迹自洽" + + +def test_parse_failed_repeatedly_tail_matches_the_limit(tmp_path: Path) -> None: + materialize( + tmp_path, + step_kinds=["action", "parse_failure", "parse_failure"], + stop_reason=StopReason.PARSE_FAILED_REPEATEDLY, + max_parse_failures=2, + complete_on_last=False, + ) + assert verdict_of(evaluate(tmp_path), RULE) is Verdict.PASSED + + +def test_parse_failed_repeatedly_with_a_short_tail_is_a_breach(tmp_path: Path) -> None: + """末尾只有两步解析失败，上限却是三——那个计数撞线时不可能停在两步。""" + materialize( + tmp_path, + step_kinds=["action", "parse_failure", "parse_failure"], + stop_reason=StopReason.PARSE_FAILED_REPEATEDLY, + max_parse_failures=3, + complete_on_last=False, + ) + assert_breached(evaluate(tmp_path), RULE) + + +def test_parse_failed_repeatedly_tail_touching_the_env_is_a_breach(tmp_path: Path) -> None: + """解析失败那一支根本不碰环境，末尾那几步不该有动作状态。""" + materialize( + tmp_path, + step_kinds=["action", "parse_failure"], + stop_reason=StopReason.PARSE_FAILED_REPEATEDLY, + max_parse_failures=1, + complete_on_last=False, + ) + + def touch_env(steps: list[dict]) -> None: + steps[-1]["action_status"] = "executed" + + edit_result_steps(tmp_path, "soak-0001", touch_env) + assert_breached(evaluate(tmp_path), RULE) + + +def test_context_overflow_with_every_step_inside_the_limit_passes(tmp_path: Path) -> None: + materialize( + tmp_path, + steps=2, + stop_reason=StopReason.CONTEXT_OVERFLOW, + max_prompt_chars=1000, + complete_on_last=False, + ) + assert verdict_of(evaluate(tmp_path), RULE) is Verdict.PASSED + + +def test_context_overflow_with_a_step_over_the_limit_is_a_breach(tmp_path: Path) -> None: + """超限的那次装配根本不产生步记录，所以落盘的每一步必定在线内。""" + materialize( + tmp_path, + steps=2, + stop_reason=StopReason.CONTEXT_OVERFLOW, + max_prompt_chars=105, + complete_on_last=False, + ) + assert_breached(evaluate(tmp_path), RULE) + + +def test_context_overflow_without_any_step_is_undetermined(tmp_path: Path) -> None: + """首次装配就超限的运行一步都没落盘，那是这个原因最典型的形态，可确实没东西可验。""" + materialize( + tmp_path, + steps=0, + stop_reason=StopReason.CONTEXT_OVERFLOW, + complete_on_last=False, + ) + assert verdict_of(evaluate(tmp_path), RULE) is Verdict.UNDETERMINED + + +def test_action_budget_counts_only_executed_steps(tmp_path: Path) -> None: + """未执行的那一步不加已执行动作计数，所以两步里只有一步算数。""" + materialize( + tmp_path, + step_kinds=["action", "not_executed"], + stop_reason=StopReason.ACTION_BUDGET, + max_actions=1, + complete_on_last=False, + ) + assert verdict_of(evaluate(tmp_path), RULE) is Verdict.PASSED + + +def test_action_budget_not_matching_max_actions_is_a_breach(tmp_path: Path) -> None: + materialize( + tmp_path, + steps=2, + stop_reason=StopReason.ACTION_BUDGET, + max_actions=5, + complete_on_last=False, + ) + assert_breached(evaluate(tmp_path), RULE) + + +def test_env_error_with_a_broken_last_step_passes(tmp_path: Path) -> None: + materialize( + tmp_path, + step_kinds=["action", "env_error"], + stop_reason=StopReason.ENV_ERROR, + complete_on_last=False, + ) + assert verdict_of(evaluate(tmp_path), RULE) is Verdict.PASSED + + +def test_env_error_without_a_broken_last_step_is_a_breach(tmp_path: Path) -> None: + materialize( + tmp_path, + step_kinds=["env_error", "action"], + stop_reason=StopReason.ENV_ERROR, + complete_on_last=False, + ) + assert_breached(evaluate(tmp_path), RULE) + + +def test_llm_error_with_a_failed_call_step_passes(tmp_path: Path) -> None: + materialize( + tmp_path, + step_kinds=["action", "call_failure"], + stop_reason=StopReason.LLM_ERROR, + complete_on_last=False, + ) + assert verdict_of(evaluate(tmp_path), RULE) is Verdict.PASSED + + +def test_llm_error_ending_on_a_parse_failure_is_a_breach(tmp_path: Path) -> None: + """解析失败那一步也没有动作结果，两者只能靠 parse_error 分开——它必须为空。""" + materialize( + tmp_path, + step_kinds=["action", "parse_failure"], + stop_reason=StopReason.LLM_ERROR, + complete_on_last=False, + ) + scoreboard = evaluate(tmp_path) + assert_breached(scoreboard, RULE) + rule = next(item for item in scoreboard.invariants if item.name == RULE) + assert any("parse_error 有值" in e.describe() for e in rule.breaches) + + +def test_llm_error_ending_on_an_action_is_a_breach(tmp_path: Path) -> None: + materialize( + tmp_path, + steps=2, + stop_reason=StopReason.LLM_ERROR, + complete_on_last=False, + ) + assert_breached(evaluate(tmp_path), RULE) + + +def test_resume_state_unknown_with_a_never_intent_dangling_passes(tmp_path: Path) -> None: + materialize( + tmp_path, + steps=1, + stop_reason=StopReason.RESUME_STATE_UNKNOWN, + complete_on_last=False, + ) + append_dangling_intent(tmp_path, "soak-0001", replay_policy="never") + assert verdict_of(evaluate(tmp_path), RULE) is Verdict.PASSED + + +def test_resume_state_unknown_with_a_safe_intent_is_a_breach(tmp_path: Path) -> None: + """声明可安全重放的意图会被直接重放，不会停在这一档。""" + materialize( + tmp_path, + steps=1, + stop_reason=StopReason.RESUME_STATE_UNKNOWN, + complete_on_last=False, + ) + append_dangling_intent(tmp_path, "soak-0001", replay_policy="safe") + assert_breached(evaluate(tmp_path), RULE) + + +def test_resume_state_unknown_without_a_dangling_intent_is_a_breach(tmp_path: Path) -> None: + materialize( + tmp_path, + steps=1, + stop_reason=StopReason.RESUME_STATE_UNKNOWN, + complete_on_last=False, + ) + assert_breached(evaluate(tmp_path), RULE) + + +def test_missing_budget_in_the_snapshot_is_undetermined(tmp_path: Path) -> None: + """规矩要的上限不在参数快照里就照实说缺什么，不硬编一个默认值。""" + materialize( + tmp_path, + steps=2, + stop_reason=StopReason.ACTION_BUDGET, + max_actions=2, + complete_on_last=False, + ) + + def drop(items: list[dict]) -> list[dict]: + for item in items: + if item["record"] == "run_started": + del item["parameter_snapshot"]["request.max_actions"] + return items + + edit_log(tmp_path / "soak-0001.jsonl", drop) + scoreboard = evaluate(tmp_path) + assert verdict_of(scoreboard, RULE) is Verdict.UNDETERMINED + rule = next(item for item in scoreboard.invariants if item.name == RULE) + assert any("request.max_actions" in e.describe() for e in rule.undetermined) + + +def test_every_stop_reason_has_a_rule(tmp_path: Path) -> None: + """十个取值一个都不许落在「没有规矩」那条兜底路径上。 + + 兜底路径本身留着，是给将来给 StopReason 加取值的人：那时它显式地报无法判定， + 而不是静默地给一条绿。 + """ + del tmp_path + covered = set(_STOP_REASON_RULES) | {StopReason.TASK_COMPLETED} + assert covered == set(StopReason) + + 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 @@ -741,6 +1083,225 @@ def test_sigkilled_run_keeps_only_the_log(tmp_path: Path) -> None: assert scoreboard.verdict is Verdict.UNDETERMINED +def materialize_stepless(runs_dir: Path, run_id: str = "soak-0001") -> None: + """造一个零步的 run：只有 run_started 与 run_finished，没有意图、没有步、没有事件。 + + 停止原因取 `cancelled`，因为十个原因里只有它的规矩不约束轨迹——它只要求日志里有结束 + 记录，而这个夹具本来就有。换成别的会顺带撞出那条规矩的击穿（`task_completed` 撞 + 「零步不可能完成」，`llm_error` 撞「至少有一步」），把这里要看的东西盖住。 + """ + materialize( + runs_dir, + run_id, + steps=0, + stop_reason=StopReason.CANCELLED, + complete_on_last=False, + ) + + +def test_empty_log_cannot_be_judged_readable(tmp_path: Path) -> None: + """一条被换行终结的行都没有：没有任何一行被读回来过，说「读得回来」没有依据。 + + 造的是真实形态：`write_run_started` 先建文件、再写那一行，杀在两者之间就只剩一个空 + 文件，另外三个文件根本来不及写。 + """ + materialize(tmp_path, write_result=False, write_events=False, write_meta=False) + (tmp_path / "soak-0001.jsonl").write_text("", encoding="utf-8") + scoreboard = evaluate(tmp_path) + assert verdict_of(scoreboard, "日志能被读回来") is Verdict.UNDETERMINED + assert not scoreboard.breached + + +def test_log_with_only_a_torn_half_line_cannot_be_judged_readable(tmp_path: Path) -> None: + materialize(tmp_path) + (tmp_path / "soak-0001.jsonl").write_text('{"record": "run_star', encoding="utf-8") + scoreboard = evaluate(tmp_path) + assert verdict_of(scoreboard, "日志能被读回来") is Verdict.UNDETERMINED + assert any("撕裂尾行" in note for note in scoreboard.notes) + + +def test_one_line_is_enough_to_judge_readability(tmp_path: Path) -> None: + """下限是一行，别为了整齐往上抬：一行就足以判它解不解得开。""" + materialize(tmp_path) + path = tmp_path / "soak-0001.jsonl" + first = path.read_text(encoding="utf-8").splitlines()[0] + path.write_text(first + "\n", encoding="utf-8") + assert verdict_of(evaluate(tmp_path), "日志能被读回来") is Verdict.PASSED + + path.write_text("{这一行解不开\n", encoding="utf-8") + assert_breached(evaluate(tmp_path), "日志能被读回来") + + +def test_run_without_intents_cannot_judge_their_homes(tmp_path: Path) -> None: + materialize_stepless(tmp_path) + scoreboard = evaluate(tmp_path) + assert verdict_of(scoreboard, "意图都有归宿") is Verdict.UNDETERMINED + assert not scoreboard.breached + + +def test_intents_without_any_result_records_are_still_judged(tmp_path: Path) -> None: + """有意图、没有任何结果记录，正是这条要判的那种，不许赖成判不了。""" + materialize(tmp_path, steps=2) + edit_log( + tmp_path / "soak-0001.jsonl", + lambda items: [item for item in items if item["record"] in {"run_started", "intent"}], + ) + assert_breached(evaluate(tmp_path), "意图都有归宿") + + +def test_one_intent_is_enough_to_judge_its_home(tmp_path: Path) -> None: + """下限是一条意图：一条就足以判它悬不悬空、以及悬空的是不是最后一条。""" + materialize(tmp_path, steps=1, stop_reason=StopReason.CANCELLED, complete_on_last=False) + edit_log( + tmp_path / "soak-0001.jsonl", + lambda items: [items[0], items[1]], + ) + # 唯一那条意图悬空，而它就是最后一条——那是崩溃点，判通过，不是判不了。 + assert verdict_of(evaluate(tmp_path), "意图都有归宿") is Verdict.PASSED + + +def test_run_without_step_payloads_cannot_judge_pairing(tmp_path: Path) -> None: + materialize_stepless(tmp_path) + scoreboard = evaluate(tmp_path) + assert verdict_of(scoreboard, "步记录的内部不变量") is Verdict.UNDETERMINED + assert not scoreboard.breached + + +def test_pairing_counts_tagged_lines_not_decoded_records(tmp_path: Path) -> None: + """下限数的是打着标签的行，不是解出来的记录。 + + 唯一那条 `step_completed` 因为违反配对而解不出记录——按记录数当下限的话这条会报「判不 + 了」，可它要判的对象恰恰就是这一行。 + """ + materialize(tmp_path, steps=1, stop_reason=StopReason.CANCELLED, complete_on_last=False) + + def unpair(items: list[dict]) -> list[dict]: + for item in items: + if item["record"] == "step_completed": + item["result_id"] = None + return items + + edit_log(tmp_path / "soak-0001.jsonl", unpair) + assert_breached(evaluate(tmp_path), "步记录的内部不变量") + + +def test_run_without_step_records_cannot_judge_agreement(tmp_path: Path) -> None: + materialize_stepless(tmp_path) + scoreboard = evaluate(tmp_path) + assert verdict_of(scoreboard, "动作结果与步记录一致") is Verdict.UNDETERMINED + assert not scoreboard.breached + + +def test_step_without_an_action_outcome_still_counts_as_data(tmp_path: Path) -> None: + """下限是一条步记录，不要求它带动作结果。 + + 没有动作结果的那一档也在这条的判定范围里——那时步记录的 `action_status` 必须为空。 + 所以一份全是解析失败的日志确实验到了这条的一部分，报判不了反而是假的。 + """ + materialize(tmp_path, steps=1, stop_reason=StopReason.CANCELLED, complete_on_last=False) + + def strip_outcome(items: list[dict]) -> list[dict]: + for item in items: + if item["record"] == "step_completed": + item["result_id"] = None + item["action_outcome"] = None + item["step"]["action_status"] = None + return items + + edit_log(tmp_path / "soak-0001.jsonl", strip_outcome) + assert verdict_of(evaluate(tmp_path), "动作结果与步记录一致") is Verdict.PASSED + + def relabel(items: list[dict]) -> list[dict]: + for item in items: + if item["record"] == "step_completed": + item["step"]["action_status"] = "executed" + return items + + edit_log(tmp_path / "soak-0001.jsonl", relabel) + assert_breached(evaluate(tmp_path), "动作结果与步记录一致") + + +def test_empty_events_file_leaves_the_event_half_unjudged(tmp_path: Path) -> None: + """两半各判各的：日志那半判过了，也不能替事件那半的真空背书。""" + materialize_stepless(tmp_path) + scoreboard = evaluate(tmp_path) + assert verdict_of(scoreboard, "不串台") is Verdict.UNDETERMINED + assert not scoreboard.breached + + # 日志那半仍然是真判的：改掉一条记录的 run_id 照样击穿。 + edit_log( + tmp_path / "soak-0001.jsonl", + lambda items: [ + {**item, "run_id": "soak-9999"} if item["record"] == "run_finished" else item + for item in items + ], + ) + assert_breached(evaluate(tmp_path), "不串台") + + +def test_log_without_records_leaves_the_log_half_unjudged(tmp_path: Path) -> None: + """事件那半判过了（两条事件的 run_id 都对），也不能替日志那半的真空背书。""" + materialize(tmp_path, steps=2) + (tmp_path / "soak-0001.jsonl").write_text("", encoding="utf-8") + scoreboard = evaluate(tmp_path) + crosstalk = next(item for item in scoreboard.invariants if item.name == "不串台") + assert crosstalk.verdict is Verdict.UNDETERMINED + assert crosstalk.breaches == () + + +def test_run_without_steps_cannot_judge_step_indices(tmp_path: Path) -> None: + """零步的 run 上「步号从 0 开始逐 1 递增」根本没被验过，所以不许报通过。 + + 崩溃注入那两类产物里真的会出现零步的 run——进程在第一步落盘之前就被杀了。 + """ + materialize( + tmp_path, + steps=0, + stop_reason=StopReason.CANCELLED, + complete_on_last=False, + ) + scoreboard = evaluate(tmp_path) + assert verdict_of(scoreboard, "步号连续") is Verdict.UNDETERMINED + assert not scoreboard.breached + + +@pytest.mark.parametrize("steps", [0, 1]) +def test_too_few_steps_cannot_judge_prompt_monotonicity(tmp_path: Path, steps: int) -> None: + """零步和一步都凑不出相邻的两个值，一次比较都没发生过。""" + materialize( + tmp_path, + steps=steps, + stop_reason=StopReason.CANCELLED, + complete_on_last=False, + ) + scoreboard = evaluate(tmp_path) + assert verdict_of(scoreboard, "提示词字符数单调不减") is Verdict.UNDETERMINED + assert not scoreboard.breached + + +def test_one_step_still_judges_the_step_index(tmp_path: Path) -> None: + """一步凑不出单调性，但「从 0 开始」验得了——两条的数据下限不一样。""" + materialize(tmp_path, steps=1, stop_reason=StopReason.CANCELLED, complete_on_last=False) + assert verdict_of(evaluate(tmp_path), "步号连续") is Verdict.PASSED + + def shift(items: list[dict]) -> list[dict]: + for item in items: + if item["record"] == "step_completed": + item["step"]["step_idx"] = 3 + return items + + edit_log(tmp_path / "soak-0001.jsonl", shift) + assert_breached(evaluate(tmp_path), "步号连续") + + +def test_two_steps_are_enough_for_both(tmp_path: Path) -> None: + """数据够了就必须真的判，不许赖着报无法判定。""" + materialize(tmp_path, steps=2) + scoreboard = evaluate(tmp_path) + assert verdict_of(scoreboard, "步号连续") is Verdict.PASSED + assert verdict_of(scoreboard, "提示词字符数单调不减") is Verdict.PASSED + + def test_empty_directory_is_undetermined_not_passed(tmp_path: Path) -> None: scoreboard = evaluate(tmp_path) assert scoreboard.verdict is Verdict.UNDETERMINED