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