iomgaa/PolyLoop
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases 3 Wiki Activity

fix(soak): 记分板的真空成立与停止原因覆盖，两处都会把「什么都没验」显示成绿

Browse Source 
Codex 报了两条数据不足时真空成立的不变量，顺着同类找齐了七条：零步的「步号连续」与
「动作结果与步记录一致」、不足两步的「提示词字符数单调不减」、零意图的「意图都有归宿」、
零载荷的「步记录的内部不变量」、零记录的「不串台」、零行的「日志能被读回来」。同一类
缺陷改一半，剩下那一半照样会在某天把一次什么都没验的跑显示成绿。

各条的数据下限不一样，反直觉的三处写进了说明：「步记录的内部不变量」数的是打着标签的行
不是解出来的记录（违反配对的行本来就解不出记录，按记录数当下限会把它最该判的对象数漏）；
「动作结果与步记录一致」不要求那条步记录带动作结果；「不串台」两半各判各的，合成一个的话
一半的真空会被另一半的绿盖住。

「停止原因与轨迹自洽」原本只覆盖四个取值、另外六个直接放行——不是数据不足，是判据本来就
该覆盖而没覆盖，后果和真空成立一样。六个都补了规矩，llm_error 那条按库自己的判据写
（解析失败必定带说明，模型调用失败那一步压根没走到解释器，只看有没有动作结果分不开这两者）。
另加一条断言十个取值一个不漏，将来加了取值而这里没跟上会显式报「还没有规矩」。

「提示词字符数单调不减」的说明原本承诺「历史只追加」，实现只比较库自己记录的数——承诺了
它，读者看见绿就以为截断被排除了。改成只承诺它验得到的，真正的对账在故障注入那侧。

拿 193 次真实运行重跑：十一条仍然全过，而这次那 8 次解析失败连击、1 次模型调用失败、
1 次撞步数上限是被真规矩判过的。

Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
This commit is contained in:
iomgaa
2026-08-11 11:53:56 -04:00
parent 5a4d13b0f2
commit ff59457482
3 changed files with 1059 additions and 120 deletions
Download Patch File Download Diff File Expand all files Collapse all files
tools/soak/tests/test_scoreboard.py
+610 -49
View File
@@ -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