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

feat(soak): 记分板——十一条不变量逐条判定，任一被击穿整批失败

Browse Source 
判定分三档：通过 / 击穿 / 无法判定。第三档不许折算成通过——缺文件、缺字段导致判不了，
和判过了是两回事，混起来会让一次什么都没验的跑看起来全绿。退出码 1 是击穿，2 是只有
无法判定，--allow-undetermined 只放行后者。

守的东西：日志读得回来、跨进程的结果与内存里逐字段相等、步号连续、意图都有归宿（至多
一条悬空且必须在末尾，那正是崩溃点）、动作结果与步记录说同一件事、提示词字符数单调不
减、事件条数等于本进程走完的步数、投递失败计数对得上、并发之间不串台、停止原因与轨迹
自洽。

报告里不出现模型原文、观察、工具名与文档片段——压测语料里有第三方的真实文档，而报告
是要贴给人看的。有四条测试拿哨兵字符串验它确实漏不出去，其中一条是拿恶意工具名试出来
的：原本打算按字符白名单放行工具名，白名单恰好把哨兵放了过去。

每条不变量都配一个「构造出违反它的日志、验它确实报击穿」的用例——一个永远返回通过的
判定器比没有判定器更糟。

Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
This commit is contained in:
iomgaa
2026-08-11 08:27:09 -04:00
parent 21a9a30ee9
commit 292a936d42
2 changed files with 2524 additions and 0 deletions
Download Patch File Download Diff File Expand all files Collapse all files
tools/soak/scoreboard.py
+1667
View File
File diff suppressed because it is too large Load Diff
tools/soak/tests/test_scoreboard.py
+857
View File
@@ -0,0 +1,857 @@
"""记分板的测试：全部用合成日志，不跑模型、不起容器。
**重点是「构造出违反某条不变量的产物、记分板确实报击穿」那一组。** 一个永远返回通过的
判定器比没有判定器更糟——它会让一次什么都没验成的跑看起来全绿，而那正是最需要被看见的
情况。所以每条不变量都配一个反例，合法产物那组只是对照。
"""
import json
import re
import sys
from collections.abc import Callable, Sequence
from pathlib import Path
import pytest
_REPO_ROOT = Path(__file__).resolve().parents[3]
if str(_REPO_ROOT) not in sys.path:
sys.path.insert(0, str(_REPO_ROOT))
from polyloop.serialization import encode # noqa: E402
from polyloop.types import ( # noqa: E402
ActionOutcome,
ActionStatus,
Intent,
IntentKind,
ModelCallResult,
ModelReply,
ReplayPolicy,
RunFinished,
RunResult,
RunStarted,
StepCompleted,
StepRecord,
StopReason,
)
from tools.soak.scoreboard import ( # noqa: E402
EXIT_BREACHED,
EXIT_UNDETERMINED,
Verdict,
evaluate,
main,
render_report,
)
#: 塞进观察与最终回答里的哨兵。压测语料里有第三方的真实文档，报告里一个字都不许出现。
SENTINEL = "SENTINEL-ZZZ-第三方文档正文"
_TAGS = {
RunStarted: "run_started",
Intent: "intent",
ModelCallResult: "model_call_result",
StepCompleted: "step_completed",
RunFinished: "run_finished",
}
# ---------------------------------------------------------------------------
# 造合法产物
# ---------------------------------------------------------------------------
def build_records(
run_id: str,
*,
steps: int = 2,
stop_reason: StopReason = StopReason.TASK_COMPLETED,
max_steps: int = 5,
final_answer: str | None = None,
complete_on_last: bool = True,
tool_name: str | None = "run_code",
sink_failures: int = 0,
observation: str = "普通观察",
) -> tuple[list[object], RunResult]:
"""造一份自洽的记录序列：一步一组「模型意图 / 模型结果 / 动作意图 / 逐步结果」。"""
records: list[object] = [
RunStarted(
run_id=run_id,
parameter_snapshot={
"request.max_steps": str(max_steps),
"store.kind": "jsonl",
},
)
]
step_records: list[StepRecord] = []
for index in range(steps):
completed = complete_on_last and index == steps - 1
text = f"{observation}#{index}"
records.append(
Intent(
run_id=run_id,
kind=IntentKind.MODEL_CALL,
call_index=index,
result_id=f"m{index}",
replay_policy=ReplayPolicy.NEVER,
)
)
records.append(
ModelCallResult(
run_id=run_id,
result_id=f"m{index}",
reply=ModelReply(call_id=f"c{index}", content="决策文本", thinking=""),
failure=None,
)
)
records.append(
Intent(
run_id=run_id,
kind=IntentKind.ACTION,
call_index=index,
result_id=f"a{index}",
replay_policy=ReplayPolicy.NEVER,
)
)
outcome = ActionOutcome(
status=ActionStatus.EXECUTED,
observation=text,
observation_is_synthetic=False,
env_reported_completion=completed,
observation_truncated_chars=0,
)
step = StepRecord(
step_idx=index,
raw_output="决策文本",
content_chars=4,
thinking_chars=0,
action="run_code",
parse_ok=True,
parse_error=None,
observation=text,
observation_is_synthetic=False,
observation_truncated_chars=0,
prompt_chars=100 + index * 10,
call_id=f"c{index}",
step_wall_ms=7,
tool_name=tool_name,
tool_arguments="{}",
action_status=ActionStatus.EXECUTED,
env_reported_completion=completed,
)
step_records.append(step)
records.append(
StepCompleted(
run_id=run_id,
result_id=f"a{index}",
action_outcome=outcome,
step=step,
)
)
result = RunResult(
run_id=run_id,
stop_reason=stop_reason,
final_answer=final_answer,
steps=tuple(step_records),
event_delivery_failures=sink_failures,
)
records.append(RunFinished(run_id=run_id, result=result))
return records, result
def write_log(path: Path, records: Sequence[object]) -> None:
lines = [
json.dumps({"record": _TAGS[type(record)], **encode(record)}, ensure_ascii=False)
for record in records
]
path.write_text("\n".join(lines) + "\n", encoding="utf-8")
def materialize(
runs_dir: Path,
run_id: str = "soak-0001",
*,
write_result: bool = True,
write_events: bool = True,
write_meta: bool = True,
events: int | None = None,
meta_overrides: dict[str, object] | None = None,
**kwargs: object,
) -> RunResult:
"""把一个自洽的 run 的四个文件都写到磁盘上，返回它的运行结果。"""
runs_dir.mkdir(parents=True, exist_ok=True)
records, result = build_records(run_id, **kwargs) # type: ignore[arg-type]
write_log(runs_dir / f"{run_id}.jsonl", records)
if write_result:
(runs_dir / f"{run_id}.result.json").write_text(
json.dumps(encode(result), ensure_ascii=False), encoding="utf-8"
)
if write_events:
count = len(result.steps) if events is None else events
payload = "".join(
json.dumps({"kind": "step_finished", "run_id": run_id, "step_idx": index}) + "\n"
for index in range(count)
)
(runs_dir / f"{run_id}.events.jsonl").write_text(payload, encoding="utf-8")
if write_meta:
meta: dict[str, object] = {
"scenario": "appworld",
"task_id": "82e2fac_1",
"phase": None,
"wall_ms": 1234,
"model_calls": len(result.steps),
"sink_failures": result.event_delivery_failures,
"env_executions": len(result.steps),
"fault": None,
"success": True,
"resumed_from_step": None,
}
meta.update(meta_overrides or {})
(runs_dir / f"{run_id}.meta.json").write_text(
json.dumps(meta, ensure_ascii=False), encoding="utf-8"
)
return result
def edit_log(path: Path, edit: Callable[[list[dict]], list[dict]]) -> None:
"""按行读出来、交给 `edit` 改、再写回去。制造反例用的。"""
payloads = [json.loads(line) for line in path.read_text(encoding="utf-8").splitlines()]
changed = edit(payloads)
path.write_text(
"".join(json.dumps(item, ensure_ascii=False) + "\n" for item in changed),
encoding="utf-8",
)
def verdict_of(scoreboard, name: str) -> Verdict:
for item in scoreboard.invariants:
if item.name == name:
return item.verdict
raise AssertionError(f"记分板里没有名为 {name!r} 的不变量")
def assert_breached(scoreboard, name: str) -> None:
assert verdict_of(scoreboard, name) is Verdict.BREACHED, (
f"{name} 应当被击穿，实际 {verdict_of(scoreboard, name)}"
)
result = next(item for item in scoreboard.invariants if item.name == name)
for evidence in result.breaches:
assert evidence.expected is not None
assert evidence.actual is not None
# ---------------------------------------------------------------------------
# 合法产物：全部通过
# ---------------------------------------------------------------------------
def test_legal_run_passes_every_invariant(tmp_path: Path) -> None:
materialize(tmp_path)
scoreboard = evaluate(tmp_path)
assert scoreboard.verdict is Verdict.PASSED, [
(item.name, item.verdict, [e.describe() for e in item.breaches + item.undetermined])
for item in scoreboard.invariants
if item.verdict is not Verdict.PASSED
]
assert scoreboard.notes == ()
def test_several_legal_runs_pass(tmp_path: Path) -> None:
materialize(tmp_path, "soak-0001", steps=3)
materialize(
tmp_path,
"soak-0002",
steps=5,
stop_reason=StopReason.STEP_BUDGET,
max_steps=5,
complete_on_last=False,
)
materialize(
tmp_path,
"soak-0003",
steps=1,
stop_reason=StopReason.AGENT_FINISHED,
complete_on_last=False,
final_answer="给出的答案",
)
scoreboard = evaluate(tmp_path)
assert scoreboard.verdict is Verdict.PASSED
assert len(scoreboard.summaries) == 3
assert scoreboard.overall_stats.stop_reasons == {
"agent_finished": 1,
"step_budget": 1,
"task_completed": 1,
}
def test_completing_tool_is_supplied_by_the_caller(tmp_path: Path) -> None:
"""task_completed 而最后一步没有环境侧证据：给了工具名就判得了，不给就是无法判定。"""
materialize(tmp_path, complete_on_last=False, tool_name="finish_task")
assert verdict_of(evaluate(tmp_path), "停止原因与轨迹自洽") is Verdict.UNDETERMINED
assert (
verdict_of(
evaluate(tmp_path, completing_tools=frozenset({"finish_task"})),
"停止原因与轨迹自洽",
)
is Verdict.PASSED
)
assert (
verdict_of(
evaluate(tmp_path, completing_tools=frozenset({"别的工具"})),
"停止原因与轨迹自洽",
)
is Verdict.BREACHED
)
# ---------------------------------------------------------------------------
# 逐条不变量的反例
# ---------------------------------------------------------------------------
def test_unreadable_line_is_a_breach(tmp_path: Path) -> None:
materialize(tmp_path)
path = tmp_path / "soak-0001.jsonl"
lines = path.read_text(encoding="utf-8").splitlines()
lines[2] = "{这不是 JSON"
path.write_text("\n".join(lines) + "\n", encoding="utf-8")
scoreboard = evaluate(tmp_path)
assert_breached(scoreboard, "日志能被读回来")
assert any("第 3 行" in e.describe() for e in scoreboard.breached[0].breaches)
def test_unknown_record_tag_is_a_breach(tmp_path: Path) -> None:
materialize(tmp_path)
edit_log(
tmp_path / "soak-0001.jsonl",
lambda items: [{**items[0], "record": "谁知道这是什么"}] + items[1:],
)
assert_breached(evaluate(tmp_path), "日志能被读回来")
def test_torn_tail_is_reported_but_not_a_breach(tmp_path: Path) -> None:
materialize(tmp_path)
path = tmp_path / "soak-0001.jsonl"
# 最后一行只写了一半、没有换行终结：进程被杀在写入中途的样子。
path.write_text(path.read_text(encoding="utf-8") + '{"record": "step_comp', encoding="utf-8")
scoreboard = evaluate(tmp_path)
assert verdict_of(scoreboard, "日志能被读回来") is Verdict.PASSED
assert any("撕裂尾行" in note for note in scoreboard.notes)
assert "撕裂尾行" in render_report(scoreboard)
def test_torn_line_in_the_middle_is_a_breach(tmp_path: Path) -> None:
materialize(tmp_path)
path = tmp_path / "soak-0001.jsonl"
lines = path.read_text(encoding="utf-8").splitlines()
# 被换行终结、却是半截 JSON：追加写不会产生这种东西，所以是损坏。
lines[3] = lines[3][: len(lines[3]) // 2]
path.write_text("\n".join(lines) + "\n", encoding="utf-8")
scoreboard = evaluate(tmp_path)
assert_breached(scoreboard, "日志能被读回来")
assert not any("撕裂尾行" in note for note in scoreboard.notes)
def test_result_file_disagreeing_with_the_log_is_a_breach(tmp_path: Path) -> None:
result = materialize(tmp_path)
payload = encode(result)
payload["final_answer"] = "跨进程读回来的和内存里的不一样"
(tmp_path / "soak-0001.result.json").write_text(
json.dumps(payload, ensure_ascii=False), encoding="utf-8"
)
assert_breached(evaluate(tmp_path), "跨进程的结果与内存里的一致")
def test_result_file_step_field_disagreeing_is_a_breach(tmp_path: Path) -> None:
result = materialize(tmp_path)
payload = encode(result)
payload["steps"][1]["step_wall_ms"] = 99999 # type: ignore[index]
(tmp_path / "soak-0001.result.json").write_text(
json.dumps(payload, ensure_ascii=False), encoding="utf-8"
)
scoreboard = evaluate(tmp_path)
assert_breached(scoreboard, "跨进程的结果与内存里的一致")
evidence = next(
item for item in scoreboard.invariants if item.name == "跨进程的结果与内存里的一致"
)
assert any("steps[1].step_wall_ms" in e.describe() for e in evidence.breaches)
def test_step_index_gap_is_a_breach(tmp_path: Path) -> None:
materialize(tmp_path, steps=3)
def bump(items: list[dict]) -> list[dict]:
for item in items:
if item["record"] == "step_completed" and item["step"]["step_idx"] == 1:
item["step"]["step_idx"] = 2
return items
edit_log(tmp_path / "soak-0001.jsonl", bump)
assert_breached(evaluate(tmp_path), "步号连续")
def test_dangling_intent_in_the_middle_is_a_breach(tmp_path: Path) -> None:
materialize(tmp_path, steps=3)
edit_log(
tmp_path / "soak-0001.jsonl",
lambda items: [
item
for item in items
if not (item["record"] == "model_call_result" and item["result_id"] == "m0")
],
)
assert_breached(evaluate(tmp_path), "意图都有归宿")
def test_two_dangling_intents_are_a_breach(tmp_path: Path) -> None:
materialize(tmp_path, steps=3)
edit_log(
tmp_path / "soak-0001.jsonl",
lambda items: [
item
for item in items
if not (item["record"] == "model_call_result" and item["result_id"] in {"m1", "m2"})
],
)
assert_breached(evaluate(tmp_path), "意图都有归宿")
def test_one_dangling_intent_at_the_tail_is_the_crash_point_and_passes(tmp_path: Path) -> None:
materialize(tmp_path, steps=2)
def crash(items: list[dict]) -> list[dict]:
kept = [item for item in items if item["record"] != "run_finished"]
kept.append(
{
"record": "intent",
"run_id": "soak-0001",
"kind": "model_call",
"call_index": 2,
"result_id": "m2",
"replay_policy": "never",
}
)
return kept
edit_log(tmp_path / "soak-0001.jsonl", crash)
assert verdict_of(evaluate(tmp_path), "意图都有归宿") is Verdict.PASSED
def test_step_completed_pairing_violation_is_a_breach(tmp_path: Path) -> None:
materialize(tmp_path)
def unpair(items: list[dict]) -> list[dict]:
for item in items:
if item["record"] == "step_completed" and item["step"]["step_idx"] == 0:
item["result_id"] = None
return items
edit_log(tmp_path / "soak-0001.jsonl", unpair)
assert_breached(evaluate(tmp_path), "步记录的内部不变量")
def test_outcome_disagreeing_with_step_record_is_a_breach(tmp_path: Path) -> None:
materialize(tmp_path)
def skew(items: list[dict]) -> list[dict]:
for item in items:
if item["record"] == "step_completed" and item["step"]["step_idx"] == 0:
item["step"]["action_status"] = "not_executed"
return items
edit_log(tmp_path / "soak-0001.jsonl", skew)
assert_breached(evaluate(tmp_path), "动作结果与步记录一致")
def test_prompt_chars_going_backwards_is_a_breach(tmp_path: Path) -> None:
materialize(tmp_path, steps=3)
def shrink(items: list[dict]) -> list[dict]:
for item in items:
if item["record"] == "step_completed" and item["step"]["step_idx"] == 2:
item["step"]["prompt_chars"] = 1
return items
edit_log(tmp_path / "soak-0001.jsonl", shrink)
assert_breached(evaluate(tmp_path), "提示词字符数单调不减")
def test_event_count_mismatch_is_a_breach(tmp_path: Path) -> None:
materialize(tmp_path, steps=3, events=2)
assert_breached(evaluate(tmp_path), "事件条数等于本进程走完的步数")
def test_resumed_run_subtracts_the_skipped_steps(tmp_path: Path) -> None:
"""续跑：日志里有四步，本进程只真的走完后两步，所以只该有两条事件。"""
materialize(
tmp_path, steps=4, events=2, meta_overrides={"resumed_from_step": 2, "model_calls": 2}
)
assert verdict_of(evaluate(tmp_path), "事件条数等于本进程走完的步数") is Verdict.PASSED
materialize(
tmp_path,
"soak-0002",
steps=4,
events=4,
meta_overrides={"resumed_from_step": 2},
)
assert_breached(evaluate(tmp_path), "事件条数等于本进程走完的步数")
def test_delivery_failure_count_mismatch_is_a_breach(tmp_path: Path) -> None:
materialize(tmp_path, sink_failures=0, meta_overrides={"sink_failures": 3})
assert_breached(evaluate(tmp_path), "投递失败计数对得上")
def test_crosstalk_in_the_log_is_a_breach(tmp_path: Path) -> None:
materialize(tmp_path)
def swap(items: list[dict]) -> list[dict]:
for item in items:
if item["record"] == "step_completed" and item["step"]["step_idx"] == 1:
item["run_id"] = "soak-9999"
return items
edit_log(tmp_path / "soak-0001.jsonl", swap)
assert_breached(evaluate(tmp_path), "不串台")
def test_crosstalk_in_the_events_is_a_breach(tmp_path: Path) -> None:
materialize(tmp_path)
(tmp_path / "soak-0001.events.jsonl").write_text(
json.dumps({"kind": "step_finished", "run_id": "soak-0001", "step_idx": 0})
+ "\n"
+ json.dumps({"kind": "step_finished", "run_id": "soak-9999", "step_idx": 1})
+ "\n",
encoding="utf-8",
)
assert_breached(evaluate(tmp_path), "不串台")
def test_task_completed_without_evidence_is_a_breach(tmp_path: Path) -> None:
materialize(tmp_path, complete_on_last=False, tool_name="run_code")
scoreboard = evaluate(tmp_path, completing_tools=frozenset())
assert_breached(scoreboard, "停止原因与轨迹自洽")
def test_step_budget_not_matching_max_steps_is_a_breach(tmp_path: Path) -> None:
materialize(
tmp_path,
steps=3,
max_steps=9,
stop_reason=StopReason.STEP_BUDGET,
complete_on_last=False,
)
assert_breached(evaluate(tmp_path), "停止原因与轨迹自洽")
def test_agent_finished_without_final_answer_is_a_breach(tmp_path: Path) -> None:
materialize(
tmp_path,
stop_reason=StopReason.AGENT_FINISHED,
complete_on_last=False,
final_answer="",
)
assert_breached(evaluate(tmp_path), "停止原因与轨迹自洽")
def test_cancelled_without_run_finished_is_a_breach(tmp_path: Path) -> None:
materialize(tmp_path, stop_reason=StopReason.CANCELLED, complete_on_last=False)
edit_log(
tmp_path / "soak-0001.jsonl",
lambda items: [item for item in items if item["record"] != "run_finished"],
)
assert_breached(evaluate(tmp_path), "停止原因与轨迹自洽")
def test_cancelled_with_run_finished_passes(tmp_path: Path) -> None:
materialize(tmp_path, stop_reason=StopReason.CANCELLED, complete_on_last=False)
assert verdict_of(evaluate(tmp_path), "停止原因与轨迹自洽") is Verdict.PASSED
# ---------------------------------------------------------------------------
# 缺文件：无法判定，既不是通过也不是击穿
# ---------------------------------------------------------------------------
def test_missing_result_file_is_undetermined(tmp_path: Path) -> None:
materialize(tmp_path, write_result=False)
scoreboard = evaluate(tmp_path)
assert verdict_of(scoreboard, "跨进程的结果与内存里的一致") is Verdict.UNDETERMINED
assert scoreboard.verdict is Verdict.UNDETERMINED
assert any(".result.json" in note for note in scoreboard.notes)
def test_missing_meta_file_is_undetermined(tmp_path: Path) -> None:
materialize(tmp_path, write_meta=False)
scoreboard = evaluate(tmp_path)
assert verdict_of(scoreboard, "事件条数等于本进程走完的步数") is Verdict.UNDETERMINED
assert verdict_of(scoreboard, "投递失败计数对得上") is Verdict.UNDETERMINED
assert not scoreboard.breached
def test_missing_events_file_is_undetermined(tmp_path: Path) -> None:
materialize(tmp_path, write_events=False)
scoreboard = evaluate(tmp_path)
assert verdict_of(scoreboard, "事件条数等于本进程走完的步数") is Verdict.UNDETERMINED
assert verdict_of(scoreboard, "不串台") is Verdict.UNDETERMINED
assert not scoreboard.breached
def test_sigkilled_run_keeps_only_the_log(tmp_path: Path) -> None:
"""进程被 SIGKILL：只剩日志，而且尾行撕裂。整批判成无法判定，不是通过也不是击穿。"""
materialize(tmp_path, write_result=False, write_events=False, write_meta=False)
path = tmp_path / "soak-0001.jsonl"
lines = path.read_text(encoding="utf-8").splitlines()[:-1]
path.write_text("\n".join(lines) + '\n{"record": "run_fin', encoding="utf-8")
scoreboard = evaluate(tmp_path)
assert not scoreboard.breached
assert scoreboard.verdict is Verdict.UNDETERMINED
def test_empty_directory_is_undetermined_not_passed(tmp_path: Path) -> None:
scoreboard = evaluate(tmp_path)
assert scoreboard.verdict is Verdict.UNDETERMINED
assert all(item.verdict is Verdict.UNDETERMINED for item in scoreboard.invariants)
def test_events_file_is_not_mistaken_for_a_run(tmp_path: Path) -> None:
materialize(tmp_path)
scoreboard = evaluate(tmp_path)
assert [item.run_id for item in scoreboard.summaries] == ["soak-0001"]
# ---------------------------------------------------------------------------
# 报告
# ---------------------------------------------------------------------------
def _table_rows(report: str) -> list[list[str]]:
return [re.split(r"(?<!\\)\|", line) for line in report.splitlines() if line.startswith("|")]
def test_report_is_legal_markdown(tmp_path: Path) -> None:
materialize(tmp_path, "soak-0001", steps=2)
materialize(
tmp_path,
"soak-0002",
steps=4,
stop_reason=StopReason.STEP_BUDGET,
max_steps=4,
complete_on_last=False,
)
report = render_report(evaluate(tmp_path))
assert report.startswith("# PolyLoop 压测记分板")
for heading in (
"## 总判定",
"## 不变量",
"## 观察",
"## 统计",
"## 与基线的对照",
"## 逐个 run",
):
assert heading in report
# 表格的每一行都以竖线开头结尾，且同一段里的列数一致。
widths: set[int] = set()
previous_was_table = False
for line in report.splitlines():
if line.startswith("|"):
assert line.endswith("|"), line
if not previous_was_table:
widths = set()
widths.add(len(re.split(r"(?<!\\)\|", line)))
assert len(widths) == 1, f"表格列数不一致：{line}"
previous_was_table = True
else:
previous_was_table = False
assert _table_rows(report)
def test_report_carries_the_baseline_comparison(tmp_path: Path) -> None:
materialize(tmp_path, steps=2)
report = render_report(evaluate(tmp_path))
assert "937" in report
assert "915" in report # 基线的 task_completed 计数
assert "步数 p90" in report
assert "对照不是判定" in report
def test_report_never_contains_document_fragments(tmp_path: Path) -> None:
"""观察与最终回答里的哨兵一个字都不许进报告——压测语料里有第三方的真实文档。"""
result = materialize(
tmp_path,
steps=2,
observation=SENTINEL,
stop_reason=StopReason.AGENT_FINISHED,
complete_on_last=False,
final_answer=SENTINEL,
)
# 再造一处击穿，让证据渲染那条路径也被走到。
payload = encode(result)
payload["final_answer"] = SENTINEL + "-改过的"
payload["steps"][0]["observation"] = SENTINEL + "-也改过" # type: ignore[index]
payload["steps"][0]["tool_name"] = SENTINEL + "-当成工具名" # type: ignore[index]
payload["steps"][1]["tool_arguments"] = SENTINEL + "-当成参数" # type: ignore[index]
(tmp_path / "soak-0001.result.json").write_text(
json.dumps(payload, ensure_ascii=False), encoding="utf-8"
)
scoreboard = evaluate(tmp_path)
assert scoreboard.breached
report = render_report(scoreboard)
assert SENTINEL not in report
assert "SENTINEL" not in report
def test_report_does_not_echo_a_hostile_record_tag(tmp_path: Path) -> None:
materialize(tmp_path)
edit_log(
tmp_path / "soak-0001.jsonl",
lambda items: [{**items[0], "record": SENTINEL}] + items[1:],
)
scoreboard = evaluate(tmp_path)
assert_breached(scoreboard, "日志能被读回来")
report = render_report(scoreboard)
assert SENTINEL not in report
assert "SENTINEL" not in report
def test_report_does_not_echo_a_hostile_observation_through_decode_errors(
tmp_path: Path,
) -> None:
"""解码失败的消息也不许把观察带出来。库的取值函数对字符串字段只报类型不报内容。"""
materialize(tmp_path, observation=SENTINEL)
def corrupt(items: list[dict]) -> list[dict]:
for item in items:
if item["record"] == "step_completed":
# 少一个必填字段：整行解不出来，消息里只会有字段名。
del item["step"]["prompt_chars"]
return items
edit_log(tmp_path / "soak-0001.jsonl", corrupt)
scoreboard = evaluate(tmp_path)
assert_breached(scoreboard, "日志能被读回来")
report = render_report(scoreboard)
assert SENTINEL not in report
assert "SENTINEL" not in report
def test_report_keeps_tool_names_out(tmp_path: Path) -> None:
"""工具名也不进报告。
一次没通过校验的工具调用会把模型编的那串原样记进 `tool_name`，所以它和观察同档，
不是「我们自己的枚举取值」。这条被一次实测撞出来过：先前的实现把工具名过一遍字符
白名单就放行，而白名单留下的正是 ASCII 那一段，哨兵原样穿了过去。
"""
materialize(
tmp_path,
complete_on_last=False,
tool_name="工具|名里有竖线和" + SENTINEL,
)
scoreboard = evaluate(tmp_path, completing_tools=frozenset({"finish"}))
assert_breached(scoreboard, "停止原因与轨迹自洽")
report = render_report(scoreboard)
assert SENTINEL not in report
assert "SENTINEL" not in report
assert "竖线" not in report
# ---------------------------------------------------------------------------
# 命令行与退出码
# ---------------------------------------------------------------------------
def test_main_returns_zero_on_a_clean_batch(tmp_path: Path) -> None:
runs = tmp_path / "runs"
report = tmp_path / "reports" / "batch.md"
materialize(runs)
assert main(["--runs-dir", str(runs), "--report", str(report)]) == 0
assert report.read_text(encoding="utf-8").startswith("# PolyLoop 压测记分板")
def test_main_returns_nonzero_on_a_breach(tmp_path: Path) -> None:
runs = tmp_path / "runs"
materialize(runs, steps=3, events=1)
code = main(["--runs-dir", str(runs), "--report", str(tmp_path / "r.md")])
assert code == EXIT_BREACHED
def test_main_returns_nonzero_on_undetermined(tmp_path: Path) -> None:
runs = tmp_path / "runs"
materialize(runs, write_meta=False)
argv = ["--runs-dir", str(runs), "--report", str(tmp_path / "r.md")]
assert main(argv) == EXIT_UNDETERMINED
assert main([*argv, "--allow-undetermined"]) == 0
def test_allow_undetermined_does_not_forgive_a_breach(tmp_path: Path) -> None:
runs = tmp_path / "runs"
materialize(runs, steps=3, write_meta=False)
def bump(items: list[dict]) -> list[dict]:
for item in items:
if item["record"] == "step_completed" and item["step"]["step_idx"] == 1:
item["step"]["step_idx"] = 2
return items
edit_log(runs / "soak-0001.jsonl", bump)
code = main(
[
"--runs-dir",
str(runs),
"--report",
str(tmp_path / "r.md"),
"--allow-undetermined",
]
)
assert code == EXIT_BREACHED
def test_completing_tool_flag_reaches_the_invariant(tmp_path: Path) -> None:
runs = tmp_path / "runs"
materialize(runs, complete_on_last=False, tool_name="finish_task")
argv = ["--runs-dir", str(runs), "--report", str(tmp_path / "r.md")]
assert main(argv) == EXIT_UNDETERMINED
assert main([*argv, "--completing-tool", "finish_task"]) == 0
def test_baseline_flags_override_the_defaults(tmp_path: Path) -> None:
runs = tmp_path / "runs"
report = tmp_path / "r.md"
materialize(runs)
code = main(
[
"--runs-dir",
str(runs),
"--report",
str(report),
"--baseline-label",
"自造基线",
"--baseline-stop-reason",
"task_completed=7",
"--baseline-steps-p50",
"3",
]
)
assert code == 0
text = report.read_text(encoding="utf-8")
assert "自造基线" in text
assert "915" not in text
def test_bad_baseline_pair_is_rejected(tmp_path: Path) -> None:
with pytest.raises(SystemExit):
main(
[
"--runs-dir",
str(tmp_path),
"--report",
str(tmp_path / "r.md"),
"--baseline-stop-reason",
"没有等号",
]
)