iomgaa/PolyLoop
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases 3 Wiki Activity
Files
main
PolyLoop/tools/soak/tests/test_faults.py
T
iomgaa c4e5732587 docs: 回写全仓库对契约套件的指向，以及 README、架构与 CHANGELOG 
套件从 tests/contract/ 搬进 polyloop.testing 之后，全仓库 28 处引用要重新指过。修了 12 处，
其余在 design/（只增不改）与 scratch/（由人清理）里。

**CLAUDE.md 改了四处事实**：§0 权威表里行为契约的权威、§0 那句依赖规则的条数、§5 目录树与
模块数、§1.8 那句「谁断言公共 Protocol 的签名」。§1 的其余硬约束与 §2 的人类门一条没动。

**architecture.md**：分层图第 4 层加一格，装配层从三个变四个；代码地图加一行；第十节按代码
逐项重写——那笔「工具段渲染样式」的欠账**没有被数字对上盖掉**，加了 fingerprints 之后请求
的字段数恰好还是十一，而组成已经换过，所以那一节正面写着它仍然欠着；新增第十条依赖规则
（pytest 只在 testing 那个 extra 里，别处 import 它会让下游的生产环境一 import 本库就
ModuleNotFoundError），带静态与运行时两半；删掉「src/ 下一行代码都没有」那段过期状态说明；
决策索引补齐 0008 到 0016，其中四行原描述说的不是那份文档真正定的东西。

**migrations/dissect.md** 那笔「内存实现不存在」的欠账还掉了。

**压测那边**三条测试守的是一条已经撤销的公共契约，改名并写清它们现在守的是场景自己的选择。
AppWorld 那处刻意的偏离（不补三个反引号）留着不恢复——那条路径要模型输出被 stop 序列截断才
触发，而压测不配 stop 序列，恢复的收益不抵重跑一次压测的成本。但注释的理由改对了：它现在是
一笔有出处的欠账，不是一个决定。

CHANGELOG 攒在「未发布」段，版本号不提前写（§1.10）。
2026-08-27 03:59:39 -04:00

2029 lines
76 KiB
Python
Raw Permalink Blame History

"""故障注入判据的测试。不打真实模型、不起容器，全部用构造出来的输入。
**每一条判据都要有一个「构造出违反它的输入 → 判据确实报击穿」的用例。** 一个永远返回通过的
判据比没有判据更糟：它会让所有人以为这些不变量被守着，而它什么都没守，而且这件事在压测报告
上看起来是一整片绿。所以下面每条判据都成对出现——一条喂它合规的输入验它说通过，一条喂它明确
违反的输入验它说击穿。
子进程编排那部分拆出了两个纯函数（`should_kill` 判时机到没到、`terminated_prefix` 截已终结
前缀），它们不碰进程也不碰模型，直接单独测。真起子进程那几条用的是一个只会往文件里写几行
JSON 的假子进程，跑完不到两秒。
确定性自杀那条路（`SelfKillingStore`）的测试**把 `os._exit` 换成一个抛哨兵异常的替身**：真调
`os._exit` 会把跑测试的 pytest 进程一起带走，整次收集连一行结果都留不下。替身同时让「死之前
那条记录有没有先写进内层存储」变得可断言——顺序反了的话，崩溃现场就少一条本该已经落地的记录。
"""
from __future__ import annotations
import asyncio
import json
import sys
from pathlib import Path
import httpx
import pytest
from polyloop.ports import EventKind, InvalidDecision, RunLog
from polyloop.session import RunRequest
from polyloop.tools import ToolRegistry
from polyloop.types import (
ActionOutcome,
ActionStatus,
Budget,
Context,
Injection,
Intent,
IntentKind,
Message,
ModelCallResult,
ModelReply,
ReplayPolicy,
Role,
RunFinished,
RunResult,
RunStarted,
StepCompleted,
StepRecord,
StopReason,
TextBlock,
)
from tools.soak.faults import (
CONTEXT_OVERFLOW_SLACK_CHARS,
CRASH_EXIT_CODE,
FAULT_NAMES,
AlwaysInvalidParser,
CallGuard,
Criterion,
CriterionStatus,
EnvBreakingExecutor,
FaultReport,
JsonlEventSink,
KillTiming,
LogRead,
PromptSizeRecordingClient,
SelfKillingStore,
build_context_overflow_budget,
build_parser,
check_all_steps_parse_failed,
check_at_least_one_step,
check_audit_prefix_preserved,
check_audit_unchanged,
check_cancelled_raised,
check_crash_prefix_preserved,
check_env_broken_after_a_full_step,
check_env_quiet_after_cancel,
check_env_untouched,
check_events_file_intact,
check_executed_action_count,
check_intents_settled,
check_last_observation_is_synthetic,
check_last_step_action_status,
check_lease_returned,
check_log_readable,
check_never_action_not_replayed,
check_no_cross_segment_replay,
check_no_env_error_step,
check_prompt_chars_match_what_was_sent,
check_prompt_chars_monotonic,
check_prompt_reached_max_prompt_chars,
check_resume_made_progress,
check_run_finished_present,
check_step_count,
check_step_indices_dense,
check_steps_before_last_all_executed,
check_stop_reason,
container_name_for_port,
count_model_calls,
guarded,
initial_prompt_chars,
main,
parse_audit_line,
parse_terminated,
read_log,
should_break_env,
should_kill,
spawn_and_kill,
stop_reason_of,
tagged,
terminated_prefix,
write_sidecars,
)
from tools.soak.scenarios.appworld import build_synthetic_observations
from tools.soak.scenarios.govdoc import AUDIT_LOG_NAME
RUN_ID = "fault-test-0"
# ---------------------------------------------------------------------------
# 造记录：日志行的形状照 `polyloop.serialization.encode` 加 `polyloop.stores` 的类型标签
# ---------------------------------------------------------------------------
def intent(
*,
kind: str = "model_call",
call_index: int = 0,
result_id: str = "r0",
replay_policy: str = "never",
) -> dict[str, object]:
return {
"record": "intent",
"run_id": RUN_ID,
"kind": kind,
"call_index": call_index,
"result_id": result_id,
"replay_policy": replay_policy,
}
def model_result(*, result_id: str = "r0") -> dict[str, object]:
return {
"record": "model_call_result",
"run_id": RUN_ID,
"result_id": result_id,
"reply": {"call_id": "c0", "content": "x", "thinking": ""},
"failure": None,
}
def step_completed(
*,
step_idx: int = 0,
result_id: str | None = "a0",
status: str | None = "executed",
parse_ok: bool = True,
action_status: str | None = "executed",
prompt_chars: object = 10,
raw_output: str = "x",
observation: str = "o",
observation_is_synthetic: bool = False,
) -> dict[str, object]:
outcome = (
None
if status is None
else {
"status": status,
"observation": "o",
"observation_is_synthetic": False,
"env_reported_completion": False,
"observation_truncated_chars": 0,
}
)
return {
"record": "step_completed",
"run_id": RUN_ID,
"result_id": result_id,
"action_outcome": outcome,
"step": {
"step_idx": step_idx,
"raw_output": raw_output,
"content_chars": len(raw_output),
"thinking_chars": 0,
"action": None,
"parse_ok": parse_ok,
"parse_error": None if parse_ok else "解释不了",
"observation": observation,
"observation_is_synthetic": observation_is_synthetic,
"observation_truncated_chars": 0,
"prompt_chars": prompt_chars,
"call_id": "c0",
"step_wall_ms": 1,
"tool_name": None,
"tool_arguments": None,
"action_status": action_status,
"env_reported_completion": False,
"schema_version": 1,
},
}
#: 造日志时用的观察模板。套一次观察多出 `len("观察：\n") == 4` 个字符，判据算「再走一步的提示词
#: 会有多大」时要把这四个字符算进去。
TEMPLATE = "观察：{observation}\n"
def run_started(
*, max_prompt_chars: int = 100, observation_template: str = TEMPLATE
) -> dict[str, object]:
"""运行开始那条记录。参数快照里只放判据会读的两个键。
真的快照还有十几个键，多放几个不会让任何一条判据的行为改变——它们按键名取值。
"""
return {
"record": "run_started",
"run_id": RUN_ID,
"parameter_snapshot": {
"request.max_prompt_chars": str(max_prompt_chars),
"request.observation_template": observation_template,
},
"schema_version": 1,
}
def run_finished(*, stop_reason: str = "task_completed") -> dict[str, object]:
return {
"record": "run_finished",
"run_id": RUN_ID,
"result": {
"run_id": RUN_ID,
"stop_reason": stop_reason,
"final_answer": None,
"steps": [],
"schema_version": 1,
"event_delivery_failures": 0,
},
}
def as_bytes(*payloads: dict[str, object]) -> bytes:
return "".join(json.dumps(item, ensure_ascii=False) + "\n" for item in payloads).encode("utf-8")
def as_read(*payloads: dict[str, object]) -> LogRead:
return parse_terminated(as_bytes(*payloads))
# ---------------------------------------------------------------------------
# 一、判据本身的形状
# ---------------------------------------------------------------------------
def test_criterion_requires_evidence() -> None:
"""没有证据的判据构造不出来。一条只会说「有问题」的判据等于没有判据。"""
with pytest.raises(ValueError, match="没有给证据"):
Criterion(name="x", status=CriterionStatus.PASSED, evidence=" ")
def test_criterion_requires_name() -> None:
with pytest.raises(ValueError, match="必须有名字"):
Criterion(name=" ", status=CriterionStatus.PASSED, evidence="有证据")
# ---------------------------------------------------------------------------
# 二、日志读取
# ---------------------------------------------------------------------------
def test_terminated_prefix_cuts_at_last_newline() -> None:
assert terminated_prefix(b'{"a":1}\n{"b":2}\n') == b'{"a":1}\n{"b":2}\n'
assert terminated_prefix(b'{"a":1}\n{"b":') == b'{"a":1}\n'
assert terminated_prefix(b'{"a":') == b""
def test_parse_terminated_drops_torn_tail() -> None:
raw = as_bytes(intent()) + b'{"record":"inte'
read = parse_terminated(raw)
assert read.torn is True
assert len(read.payloads) == 1
assert read.bad_lines == ()
def test_parse_terminated_flags_bad_terminated_line() -> None:
raw = as_bytes(intent()) + b"not json at all\n"
read = parse_terminated(raw)
assert read.torn is False
assert read.bad_lines == (2,)
def test_check_log_readable_breaches_on_bad_line() -> None:
read = parse_terminated(as_bytes(intent()) + b"{}\n")
outcome = check_log_readable(read)
assert outcome.status is CriterionStatus.BREACHED
def test_check_log_readable_passes_on_clean_log() -> None:
assert check_log_readable(as_read(intent())).status is CriterionStatus.PASSED
def test_read_log_of_missing_file_is_empty() -> None:
assert read_log(Path("/nonexistent/nope.jsonl")).payloads == ()
def test_tagged_and_counters() -> None:
read = as_read(
intent(kind="model_call", result_id="r0"),
intent(kind="action", result_id="a0"),
run_finished(stop_reason="step_budget"),
)
assert len(tagged(read, "intent")) == 2
assert count_model_calls(read) == 1
assert stop_reason_of(read) == "step_budget"
def test_stop_reason_of_without_finished_record() -> None:
assert stop_reason_of(as_read(intent())) is None
# ---------------------------------------------------------------------------
# 三、崩溃续跑：字节前缀
# ---------------------------------------------------------------------------
def test_prefix_preserved_passes_when_resume_only_appends() -> None:
crashed = as_bytes(intent(), model_result())
final = crashed + as_bytes(step_completed())
outcome = check_crash_prefix_preserved(crashed=crashed, final=final)
assert outcome.status is CriterionStatus.PASSED
def test_prefix_preserved_breaches_when_one_byte_changed() -> None:
"""前缀里改一个字节就必须报击穿。
这是这条判据存在的全部理由：解析出来的对象可能仍然等价（改的是空格、字段顺序、数字的
表示法），而承诺是更硬的那一条——已经写下去的字节不许再动。
"""
crashed = as_bytes(intent(), model_result())
mutated = bytearray(crashed)
mutated[5] = mutated[5] ^ 0x01
final = bytes(mutated) + as_bytes(step_completed())
outcome = check_crash_prefix_preserved(crashed=crashed, final=final)
assert outcome.status is CriterionStatus.BREACHED
assert "第 5 字节" in outcome.evidence
def test_prefix_preserved_breaches_when_final_is_shorter() -> None:
crashed = as_bytes(intent(), model_result())
outcome = check_crash_prefix_preserved(crashed=crashed, final=crashed[:10])
assert outcome.status is CriterionStatus.BREACHED
def test_prefix_preserved_ignores_torn_tail() -> None:
"""崩溃快照末尾那段没被换行终结的字节不参与比对：那次写从来没算数。"""
crashed = as_bytes(intent()) + b'{"record":"model_ca'
final = as_bytes(intent()) + b'{"record":"model_ca' + as_bytes(model_result())
assert check_crash_prefix_preserved(crashed=crashed, final=final).status is (
CriterionStatus.PASSED
)
def test_prefix_preserved_undetermined_without_any_terminated_record() -> None:
outcome = check_crash_prefix_preserved(crashed=b'{"half', final=b'{"half"}\n')
assert outcome.status is CriterionStatus.UNDETERMINED
# ---------------------------------------------------------------------------
# 四、崩溃续跑：步号
# ---------------------------------------------------------------------------
def test_step_indices_dense_passes() -> None:
read = as_read(*(step_completed(step_idx=index) for index in range(4)))
assert check_step_indices_dense(read).status is CriterionStatus.PASSED
def test_step_indices_dense_breaches_on_gap() -> None:
read = as_read(step_completed(step_idx=0), step_completed(step_idx=2))
assert check_step_indices_dense(read).status is CriterionStatus.BREACHED
def test_step_indices_dense_breaches_on_repeat() -> None:
"""同一个步号出现两次：续跑把已经落地的那一步又跑了一遍。"""
read = as_read(step_completed(step_idx=0), step_completed(step_idx=0))
assert check_step_indices_dense(read).status is CriterionStatus.BREACHED
def test_step_indices_dense_undetermined_without_steps() -> None:
assert check_step_indices_dense(as_read(intent())).status is CriterionStatus.UNDETERMINED
# ---------------------------------------------------------------------------
# 五、崩溃续跑：意图有没有归宿
# ---------------------------------------------------------------------------
def test_intents_settled_passes_when_all_have_results() -> None:
read = as_read(
intent(kind="model_call", result_id="r0"),
model_result(result_id="r0"),
intent(kind="action", result_id="a0"),
step_completed(result_id="a0"),
)
assert check_intents_settled(read).status is CriterionStatus.PASSED
def test_intents_settled_allows_one_dangling_at_the_end() -> None:
"""崩溃点上那条悬空的意图是合法的：时机 B 下它就是崩溃点本身。"""
read = as_read(
intent(kind="model_call", result_id="r0"),
model_result(result_id="r0"),
intent(kind="action", result_id="a0"),
step_completed(result_id="a0"),
intent(kind="model_call", call_index=1, result_id="r1"),
run_finished(stop_reason="resume_state_unknown"),
)
assert check_intents_settled(read).status is CriterionStatus.PASSED
def test_intents_settled_breaches_on_dangling_in_the_middle() -> None:
"""中间悬空说明有一步的执行状态被跳过去了，后面的步建立在一个没有答案的问题上。"""
read = as_read(
intent(kind="model_call", result_id="r0"),
intent(kind="model_call", call_index=1, result_id="r1"),
model_result(result_id="r1"),
)
outcome = check_intents_settled(read)
assert outcome.status is CriterionStatus.BREACHED
assert "[0]" in outcome.evidence
def test_intents_settled_undetermined_without_intents() -> None:
assert check_intents_settled(as_read(run_finished())).status is CriterionStatus.UNDETERMINED
# ---------------------------------------------------------------------------
# 六、崩溃续跑：审计账
# ---------------------------------------------------------------------------
def test_parse_audit_line() -> None:
assert parse_audit_line("write_note\tplan.md\tabc123") == ("write_note", "plan.md", "abc123")
assert parse_audit_line("write_note\tplan.md") is None
assert parse_audit_line("write_note\t\tabc123") is None
def test_never_action_not_replayed_passes_on_distinct_writes() -> None:
audit = ["write_note\tplan.md\taaa", "write_note\tevidence.md\tbbb"]
assert check_never_action_not_replayed(audit).status is CriterionStatus.PASSED
def test_never_action_not_replayed_breaches_on_duplicate() -> None:
"""同一个 (文件名, 内容摘要) 出现两次就是重放。"""
audit = [
"write_note\tevidence.md\tbbb",
"write_note\tplan.md\taaa",
"write_note\tevidence.md\tbbb",
]
outcome = check_never_action_not_replayed(audit)
assert outcome.status is CriterionStatus.BREACHED
assert "bbb" in outcome.evidence
def test_never_action_not_replayed_undetermined_on_empty_audit() -> None:
"""一条副作用都没执行过时不许报「通过」——那次跑根本没验到这条不变量。"""
assert check_never_action_not_replayed([]).status is CriterionStatus.UNDETERMINED
def test_never_action_not_replayed_undetermined_on_malformed_audit() -> None:
assert check_never_action_not_replayed(["坏行"]).status is CriterionStatus.UNDETERMINED
def test_audit_unchanged_passes() -> None:
audit = ["write_note\tplan.md\taaa"]
assert check_audit_unchanged(before=audit, after=list(audit)).status is CriterionStatus.PASSED
def test_audit_unchanged_breaches_when_resume_adds_a_line() -> None:
before = ["write_note\tplan.md\taaa"]
after = [*before, "write_note\tplan.md\taaa"]
outcome = check_audit_unchanged(before=before, after=after)
assert outcome.status is CriterionStatus.BREACHED
assert "多出 1 条" in outcome.evidence
def test_audit_unchanged_breaches_when_prefix_rewritten() -> None:
outcome = check_audit_unchanged(
before=["write_note\tplan.md\taaa"], after=["write_note\tplan.md\tzzz"]
)
assert outcome.status is CriterionStatus.BREACHED
# ---------------------------------------------------------------------------
# 七、崩溃续跑：续跑有没有往下走
# ---------------------------------------------------------------------------
def test_resume_made_progress_passes() -> None:
assert check_resume_made_progress(crashed_steps=2, final_steps=5).status is (
CriterionStatus.PASSED
)
def test_resume_made_progress_breaches_when_stalled() -> None:
assert check_resume_made_progress(crashed_steps=2, final_steps=2).status is (
CriterionStatus.BREACHED
)
# ---------------------------------------------------------------------------
# 八、停止原因与步数
# ---------------------------------------------------------------------------
def test_stop_reason_passes() -> None:
read = as_read(run_finished(stop_reason="step_budget"))
assert check_stop_reason(read, "step_budget").status is CriterionStatus.PASSED
def test_stop_reason_breaches_on_wrong_value() -> None:
read = as_read(run_finished(stop_reason="env_error"))
outcome = check_stop_reason(read, "step_budget")
assert outcome.status is CriterionStatus.BREACHED
assert "env_error" in outcome.evidence
def test_stop_reason_breaches_without_finished_record() -> None:
"""取消也要留结束记录，不然恢复会把它当成可以续跑。缺记录是击穿，不是判不了。"""
outcome = check_stop_reason(as_read(step_completed()), "cancelled")
assert outcome.status is CriterionStatus.BREACHED
def test_step_count_passes_and_breaches() -> None:
read = as_read(*(step_completed(step_idx=index) for index in range(3)))
assert check_step_count(read, expected=3, name="n").status is CriterionStatus.PASSED
assert check_step_count(read, expected=2, name="n").status is CriterionStatus.BREACHED
def test_executed_action_count_passes_and_breaches() -> None:
read = as_read(
step_completed(step_idx=0, status="executed"),
step_completed(step_idx=1, status="executed"),
step_completed(step_idx=2, status="not_executed"),
)
assert check_executed_action_count(read, expected=2).status is CriterionStatus.PASSED
assert check_executed_action_count(read, expected=3).status is CriterionStatus.BREACHED
def test_no_env_error_step_passes_and_breaches() -> None:
clean = as_read(step_completed(step_idx=0, status="executed"))
assert check_no_env_error_step(clean).status is CriterionStatus.PASSED
dirty = as_read(
step_completed(step_idx=0, status="executed"),
step_completed(step_idx=1, status="env_error"),
)
outcome = check_no_env_error_step(dirty)
assert outcome.status is CriterionStatus.BREACHED
assert "[1]" in outcome.evidence
# ---------------------------------------------------------------------------
# 九、解析失败连击
# ---------------------------------------------------------------------------
def test_all_steps_parse_failed_passes() -> None:
read = as_read(
*(
step_completed(
step_idx=index, result_id=None, status=None, parse_ok=False, action_status=None
)
for index in range(3)
)
)
assert check_all_steps_parse_failed(read).status is CriterionStatus.PASSED
def test_all_steps_parse_failed_breaches_when_a_step_parsed() -> None:
read = as_read(
step_completed(step_idx=0, result_id=None, status=None, parse_ok=False, action_status=None),
step_completed(step_idx=1, parse_ok=True),
)
assert check_all_steps_parse_failed(read).status is CriterionStatus.BREACHED
def test_all_steps_parse_failed_breaches_when_an_action_was_dispatched() -> None:
"""解析失败那一步不许带动作状态——带了就说明有动作被分发过。"""
read = as_read(
step_completed(step_idx=0, parse_ok=False, action_status="executed"),
)
assert check_all_steps_parse_failed(read).status is CriterionStatus.BREACHED
def test_all_steps_parse_failed_undetermined_without_steps() -> None:
assert check_all_steps_parse_failed(as_read(intent())).status is CriterionStatus.UNDETERMINED
def test_env_untouched_passes_and_breaches() -> None:
assert check_env_untouched(executions=0, source="假环境").status is CriterionStatus.PASSED
outcome = check_env_untouched(executions=1, source="假环境")
assert outcome.status is CriterionStatus.BREACHED
def test_always_invalid_parser_is_sync_and_never_raises() -> None:
parser = AlwaysInvalidParser()
parsed = parser.parse(ModelReply(call_id=None, content="```python\nprint(1)\n```", thinking=""))
assert isinstance(parsed.decision, InvalidDecision)
assert parsed.decision.explanation.strip()
# 回填历史的那段不比模型原文长。公共契约不断言长度，这守的是这份注入用解析器自己的形状：
# 它原样回填模型原文，不拼接任何东西。
assert len(parsed.history_text) <= len("```python\nprint(1)\n```")
assert parser.parameters()["kind"] == "always_invalid"
# ---------------------------------------------------------------------------
# 十、取消
# ---------------------------------------------------------------------------
def test_cancelled_raised_passes() -> None:
outcome = check_cancelled_raised(raised=asyncio.CancelledError())
assert outcome.status is CriterionStatus.PASSED
def test_cancelled_raised_breaches_when_swallowed() -> None:
"""取消被吞掉之后返回一个结果，调用方的结构化并发就断了。"""
assert check_cancelled_raised(raised=None).status is CriterionStatus.BREACHED
def test_cancelled_raised_breaches_on_other_exception() -> None:
outcome = check_cancelled_raised(raised=RuntimeError("别的错"))
assert outcome.status is CriterionStatus.BREACHED
assert "RuntimeError" in outcome.evidence
def test_lease_returned_passes_and_breaches() -> None:
assert check_lease_returned(borrowed=True, timeout_s=1.0, pool_size=1).status is (
CriterionStatus.PASSED
)
assert check_lease_returned(borrowed=False, timeout_s=1.0, pool_size=1).status is (
CriterionStatus.BREACHED
)
# ---------------------------------------------------------------------------
# 十一、子进程编排：时机判定这个纯函数
# ---------------------------------------------------------------------------
def test_should_kill_after_step_waits_for_enough_steps() -> None:
read = as_read(intent(), model_result(), step_completed())
assert (
should_kill(read, timing=KillTiming.AFTER_STEP, after_steps=2, audit_is_not_empty=True)
is False
)
assert (
should_kill(read, timing=KillTiming.AFTER_STEP, after_steps=1, audit_is_not_empty=True)
is True
)
def test_should_kill_after_step_rejects_trailing_intent() -> None:
read = as_read(step_completed(), intent(call_index=1, result_id="r1"))
assert (
should_kill(read, timing=KillTiming.AFTER_STEP, after_steps=1, audit_is_not_empty=True)
is False
)
def test_should_kill_at_intent_requires_never_policy() -> None:
"""`safe` 那条意图不算命中：悬在它上面续跑会重放动作接着跑，停止原因不是状态未知。"""
never = as_read(step_completed(), intent(call_index=1, result_id="r1", replay_policy="never"))
safe = as_read(step_completed(), intent(call_index=1, result_id="r1", replay_policy="safe"))
assert (
should_kill(never, timing=KillTiming.AT_INTENT, after_steps=1, audit_is_not_empty=True)
is True
)
assert (
should_kill(safe, timing=KillTiming.AT_INTENT, after_steps=1, audit_is_not_empty=True)
is False
)
def test_should_kill_at_intent_rejects_trailing_step() -> None:
read = as_read(step_completed())
assert (
should_kill(read, timing=KillTiming.AT_INTENT, after_steps=0, audit_is_not_empty=True)
is False
)
def test_should_kill_on_empty_log() -> None:
empty = LogRead(payloads=(), torn=False, bad_lines=())
assert (
should_kill(empty, timing=KillTiming.AFTER_STEP, after_steps=0, audit_is_not_empty=True)
is False
)
assert (
should_kill(empty, timing=KillTiming.AT_INTENT, after_steps=0, audit_is_not_empty=True)
is False
)
def test_should_kill_requires_a_non_empty_audit() -> None:
"""父进程的条件必须与子进程的自杀条件逐字对齐，包括审计账那一项。
松一档的后果实测过：父进程每次都在子进程走到自杀那一行之前抢先发信号，自杀路径成了死代码，
而崩溃点落在哪儿又变回碰运气——其中一次就崩在审计账还是空的时候，最硬那条判据只能真空成立。
"""
after_step = as_read(intent(), model_result(), step_completed())
at_intent = as_read(step_completed(), intent(call_index=1, result_id="r1"))
for read, timing in ((after_step, KillTiming.AFTER_STEP), (at_intent, KillTiming.AT_INTENT)):
assert should_kill(read, timing=timing, after_steps=1, audit_is_not_empty=True) is True
assert should_kill(read, timing=timing, after_steps=1, audit_is_not_empty=False) is False
# ---------------------------------------------------------------------------
# 十二、确定性自杀：写入落盘之后按时机把自己打死
# ---------------------------------------------------------------------------
class _ExitCalledError(Exception):
"""`os._exit` 的替身抛的哨兵。真调 `os._exit` 会把 pytest 一起带走。"""
def __init__(self, code: int) -> None:
super().__init__(code)
self.code = code
def _exit_sentinel(code: int) -> None:
raise _ExitCalledError(code)
class _RecordingStore:
"""记下每一次写入的假存储。`parameters()` 报的键与 `JsonlRunStore` 一致。"""
def __init__(self) -> None:
self.calls: list[str] = []
def parameters(self) -> dict[str, str]:
return {"kind": "jsonl"}
async def read_log(self, run_id: str) -> RunLog:
self.calls.append("read_log")
return RunLog()
async def write_run_started(self, record: RunStarted) -> None:
self.calls.append("run_started")
async def write_intent(self, record: Intent) -> None:
self.calls.append("intent")
async def write_model_call_result(self, record: ModelCallResult) -> None:
self.calls.append("model_call_result")
async def write_step_completed(self, record: StepCompleted) -> None:
self.calls.append("step_completed")
async def write_run_finished(self, record: RunFinished) -> None:
self.calls.append("run_finished")
def an_intent(policy: ReplayPolicy = ReplayPolicy.NEVER) -> Intent:
return Intent(
run_id=RUN_ID,
kind=IntentKind.MODEL_CALL,
call_index=0,
result_id="r0",
replay_policy=policy,
)
def a_step_completed() -> StepCompleted:
return StepCompleted(
run_id=RUN_ID,
result_id="a0",
action_outcome=ActionOutcome(
status=ActionStatus.EXECUTED,
observation="o",
observation_is_synthetic=False,
env_reported_completion=False,
observation_truncated_chars=0,
),
step=StepRecord(
step_idx=0,
raw_output="x",
content_chars=1,
thinking_chars=0,
action="a",
parse_ok=True,
parse_error=None,
observation="o",
observation_is_synthetic=False,
observation_truncated_chars=0,
prompt_chars=10,
call_id="c0",
step_wall_ms=1,
),
)
def wrapped(
inner: _RecordingStore, *, timing: KillTiming, workspace: Path, audited: bool
) -> SelfKillingStore:
"""包一层，并按需要让工作区的审计账非空。"""
workspace.mkdir(parents=True, exist_ok=True)
if audited:
seed_audit(workspace)
return SelfKillingStore(
inner=inner, timing=timing, workspace=workspace, exit_now=_exit_sentinel
)
async def test_self_kill_after_step_when_the_audit_is_not_empty(tmp_path: Path) -> None:
"""时机 A：步记录落盘之后当场死。**死之前那条记录必须已经写进内层存储**。"""
inner = _RecordingStore()
store = wrapped(inner, timing=KillTiming.AFTER_STEP, workspace=tmp_path, audited=True)
with pytest.raises(_ExitCalledError) as caught:
await store.write_step_completed(a_step_completed())
assert caught.value.code == CRASH_EXIT_CODE
assert inner.calls == ["step_completed"]
async def test_self_kill_after_step_waits_for_a_real_write_note(tmp_path: Path) -> None:
"""审计账还是空的就不死：那时崩掉，最硬那条判据只能真空成立。"""
inner = _RecordingStore()
store = wrapped(inner, timing=KillTiming.AFTER_STEP, workspace=tmp_path, audited=False)
await store.write_step_completed(a_step_completed())
assert inner.calls == ["step_completed"]
async def test_self_kill_after_step_ignores_intents(tmp_path: Path) -> None:
inner = _RecordingStore()
store = wrapped(inner, timing=KillTiming.AFTER_STEP, workspace=tmp_path, audited=True)
await store.write_intent(an_intent())
assert inner.calls == ["intent"]
async def test_self_kill_at_intent_on_a_never_intent(tmp_path: Path) -> None:
inner = _RecordingStore()
store = wrapped(inner, timing=KillTiming.AT_INTENT, workspace=tmp_path, audited=True)
with pytest.raises(_ExitCalledError) as caught:
await store.write_intent(an_intent(ReplayPolicy.NEVER))
assert caught.value.code == CRASH_EXIT_CODE
assert inner.calls == ["intent"]
async def test_self_kill_at_intent_skips_a_safe_intent(tmp_path: Path) -> None:
"""`safe` 那条意图崩了也没用：续跑会重放动作接着跑，停止原因不是状态未知。"""
inner = _RecordingStore()
store = wrapped(inner, timing=KillTiming.AT_INTENT, workspace=tmp_path, audited=True)
await store.write_intent(an_intent(ReplayPolicy.SAFE))
assert inner.calls == ["intent"]
async def test_self_kill_at_intent_waits_for_a_real_write_note(tmp_path: Path) -> None:
inner = _RecordingStore()
store = wrapped(inner, timing=KillTiming.AT_INTENT, workspace=tmp_path, audited=False)
await store.write_intent(an_intent(ReplayPolicy.NEVER))
assert inner.calls == ["intent"]
async def test_self_kill_at_intent_ignores_step_records(tmp_path: Path) -> None:
inner = _RecordingStore()
store = wrapped(inner, timing=KillTiming.AT_INTENT, workspace=tmp_path, audited=True)
await store.write_step_completed(a_step_completed())
assert inner.calls == ["step_completed"]
async def test_self_kill_never_fires_on_the_other_three_writes(tmp_path: Path) -> None:
"""开始、模型调用结果、结束这三处一律不死：它们不是任何一个时机的定义点。"""
for timing in KillTiming:
inner = _RecordingStore()
store = wrapped(inner, timing=timing, workspace=tmp_path, audited=True)
await store.write_run_started(RunStarted(run_id=RUN_ID, parameter_snapshot={}))
await store.write_model_call_result(
ModelCallResult(run_id=RUN_ID, result_id="r0", reply=None, failure="炸了")
)
await store.write_run_finished(
RunFinished(
run_id=RUN_ID,
result=RunResult(
run_id=RUN_ID,
stop_reason=StopReason.STEP_BUDGET,
final_answer=None,
steps=(),
),
)
)
assert await store.read_log(RUN_ID) == RunLog()
assert inner.calls == ["run_started", "model_call_result", "run_finished", "read_log"]
def test_self_kill_forwards_parameters_verbatim(tmp_path: Path) -> None:
"""包装层不许往参数快照里加自己的键：父进程续跑用的是没包过的存储，加了就报假漂移。"""
inner = _RecordingStore()
store = wrapped(inner, timing=KillTiming.AFTER_STEP, workspace=tmp_path, audited=True)
assert store.parameters() == inner.parameters()
# ---------------------------------------------------------------------------
# 十三、子进程编排：真起一个假子进程
# ---------------------------------------------------------------------------
#: 假子进程：往指定文件里逐条追加记录，中间留出足够父进程轮询到的间隔，然后一直睡着不退出。
#: 它不打模型、不起容器，只是一个会按顺序写文件的东西。走的是外部 SIGKILL 那条兜底路径。
_FAKE_CHILD = """
import json, sys, time
path = sys.argv[1]
records = json.loads(sys.argv[2])
with open(path, "a", encoding="utf-8") as handle:
for record in records:
handle.write(json.dumps(record) + "\\n")
handle.flush()
time.sleep(0.05)
time.sleep(30)
"""
#: 写完就正常退出，一次都不自杀。
_FAKE_CHILD_EXITS = """
import json, sys
path = sys.argv[1]
records = json.loads(sys.argv[2])
with open(path, "a", encoding="utf-8") as handle:
for record in records:
handle.write(json.dumps(record) + "\\n")
"""
#: 写完就按给定的退出码把自己打死，模拟 `SelfKillingStore` 那条主路径。第四个参数是写完之后
#: 先睡多久再死——用来验父进程真的留了自杀窗口，而不是看见条件成立就立刻开枪。
_FAKE_CHILD_SELF_KILL = """
import json, os, sys, time
path = sys.argv[1]
records = json.loads(sys.argv[2])
with open(path, "a", encoding="utf-8") as handle:
for record in records:
handle.write(json.dumps(record) + "\\n")
handle.flush()
print("子进程说了句话")
sys.stdout.flush()
time.sleep(float(sys.argv[4]))
os._exit(int(sys.argv[3]))
"""
def seed_audit(workspace: Path) -> Path:
"""让工作区的审计账非空。父子两侧的命中条件都要求它。"""
workspace.mkdir(parents=True, exist_ok=True)
(workspace / AUDIT_LOG_NAME).write_text("write_note\tevidence.md\taaa\n", encoding="utf-8")
return workspace
def self_kill_argv(log_path: Path, records: list[dict[str, object]], *, code: int, sleep: float):
return [
sys.executable,
"-c",
_FAKE_CHILD_SELF_KILL,
str(log_path),
json.dumps(records),
str(code),
str(sleep),
]
async def test_spawn_and_kill_accepts_a_self_killed_child(tmp_path: Path) -> None:
"""主路径：子进程按崩溃退出码自杀，且日志尾部形态对得上，算命中。"""
log_path = tmp_path / f"{RUN_ID}.jsonl"
workspace = seed_audit(tmp_path / "ws")
records = [intent(), model_result(), step_completed()]
outcome = await spawn_and_kill(
argv=self_kill_argv(log_path, records, code=CRASH_EXIT_CODE, sleep=0.0),
log_path=log_path,
workspace=workspace,
timing=KillTiming.AFTER_STEP,
after_steps=1,
child_log_path=tmp_path / f"{RUN_ID}.child.log",
timeout_s=20.0,
)
assert outcome.hit is True
assert outcome.exit_code == CRASH_EXIT_CODE
assert "自杀" in outcome.reason
assert outcome.steps == 1
assert outcome.snapshot == log_path.read_bytes()
assert "子进程说了句话" in (tmp_path / f"{RUN_ID}.child.log").read_text(encoding="utf-8")
async def test_spawn_and_kill_waits_out_the_self_kill_grace(tmp_path: Path) -> None:
"""条件对上之后子进程还要过一会儿才死：父进程必须等它，不许抢先开枪。
抢先的后果不是「杀错了」，是自杀那条路永远走不到，而崩溃点落在哪儿又变回碰运气。
"""
log_path = tmp_path / f"{RUN_ID}.jsonl"
workspace = seed_audit(tmp_path / "ws")
records = [intent(), model_result(), step_completed()]
outcome = await spawn_and_kill(
argv=self_kill_argv(log_path, records, code=CRASH_EXIT_CODE, sleep=0.4),
log_path=log_path,
workspace=workspace,
timing=KillTiming.AFTER_STEP,
after_steps=1,
self_kill_grace_s=5.0,
timeout_s=20.0,
)
assert outcome.hit is True
assert outcome.exit_code == CRASH_EXIT_CODE
assert "自杀" in outcome.reason
async def test_spawn_and_kill_falls_back_after_the_grace_runs_out(tmp_path: Path) -> None:
"""窗口用完子进程还活着，才轮到兜底 SIGKILL。这条路径没被删。"""
log_path = tmp_path / f"{RUN_ID}.jsonl"
workspace = seed_audit(tmp_path / "ws")
records = [intent(), model_result(), step_completed()]
outcome = await spawn_and_kill(
argv=self_kill_argv(log_path, records, code=CRASH_EXIT_CODE, sleep=30.0),
log_path=log_path,
workspace=workspace,
timing=KillTiming.AFTER_STEP,
after_steps=1,
self_kill_grace_s=0.2,
timeout_s=20.0,
)
assert outcome.hit is True
assert "兜底" in outcome.reason
assert outcome.exit_code == -9
assert outcome.steps == 1
assert outcome.model_calls == 1
assert outcome.snapshot == log_path.read_bytes()
async def test_spawn_and_kill_needs_a_non_empty_audit(tmp_path: Path) -> None:
"""日志尾部形态对了但审计账是空的：不算命中，等到超时为止。
这一条守的正是上一版实测出来的毛病——那次崩在第 1 步、审计账 0 条，最硬那条判据只能报
无法判定，而报告上看起来只是「少了一条」。
"""
log_path = tmp_path / f"{RUN_ID}.jsonl"
workspace = tmp_path / "ws"
workspace.mkdir()
records = [intent(), model_result(), step_completed()]
outcome = await spawn_and_kill(
argv=[sys.executable, "-c", _FAKE_CHILD, str(log_path), json.dumps(records)],
log_path=log_path,
workspace=workspace,
timing=KillTiming.AFTER_STEP,
after_steps=1,
self_kill_grace_s=0.2,
timeout_s=0.6,
)
assert outcome.hit is False
assert "仍没崩在时机" in outcome.reason
async def test_spawn_and_kill_rejects_a_self_kill_at_the_wrong_tail(tmp_path: Path) -> None:
"""自杀了但尾部形态不对：报没命中，不许因为退出码对就当成命中。"""
log_path = tmp_path / f"{RUN_ID}.jsonl"
workspace = seed_audit(tmp_path / "ws")
records = [intent(), model_result()]
outcome = await spawn_and_kill(
argv=self_kill_argv(log_path, records, code=CRASH_EXIT_CODE, sleep=0.0),
log_path=log_path,
workspace=workspace,
timing=KillTiming.AFTER_STEP,
after_steps=1,
timeout_s=20.0,
)
assert outcome.hit is False
assert "尾部不是时机" in outcome.reason
async def test_spawn_and_kill_reports_a_miss_when_the_child_exits_normally(tmp_path: Path) -> None:
"""子进程正常跑完了要报出来，不许悄悄当成命中。"""
log_path = tmp_path / f"{RUN_ID}.jsonl"
workspace = seed_audit(tmp_path / "ws")
records = [intent(), model_result()]
outcome = await spawn_and_kill(
argv=[sys.executable, "-c", _FAKE_CHILD_EXITS, str(log_path), json.dumps(records)],
log_path=log_path,
workspace=workspace,
timing=KillTiming.AFTER_STEP,
after_steps=1,
timeout_s=20.0,
)
assert outcome.hit is False
assert outcome.exit_code == 0
assert f"不是按时机自杀的 {CRASH_EXIT_CODE}" in outcome.reason
async def test_spawn_and_kill_reports_a_miss_on_timeout(tmp_path: Path) -> None:
log_path = tmp_path / f"{RUN_ID}.jsonl"
workspace = seed_audit(tmp_path / "ws")
outcome = await spawn_and_kill(
argv=[sys.executable, "-c", _FAKE_CHILD, str(log_path), json.dumps([intent()])],
log_path=log_path,
workspace=workspace,
timing=KillTiming.AFTER_STEP,
after_steps=1,
timeout_s=0.5,
)
assert outcome.hit is False
assert "仍没崩在时机" in outcome.reason
# ---------------------------------------------------------------------------
# 十四、事件出口、护栏、sidecar、命令行
# ---------------------------------------------------------------------------
class _FakeEvent:
"""只带事件出口会读的那三个字段。"""
def __init__(self, step_idx: int) -> None:
self.kind = EventKind.STEP_FINISHED
self.run_id = RUN_ID
self.step = type("Step", (), {"step_idx": step_idx})()
async def test_event_sink_writes_one_line_per_event(tmp_path: Path) -> None:
sink = JsonlEventSink(tmp_path / f"{RUN_ID}.events.jsonl")
await sink.emit(_FakeEvent(0)) # type: ignore[arg-type]
await sink.emit(_FakeEvent(1)) # type: ignore[arg-type]
lines = (tmp_path / f"{RUN_ID}.events.jsonl").read_text(encoding="utf-8").splitlines()
assert [json.loads(line)["step_idx"] for line in lines] == [0, 1]
assert sink.delivered == 2
def test_event_sink_parameters_carry_no_path() -> None:
"""路径进参数快照会让续跑报一次假的参数漂移。"""
sink = JsonlEventSink("/tmp/whatever.jsonl")
assert sink.parameters() == {"kind": "jsonl_events"}
def test_call_guard_accounting() -> None:
guard = CallGuard(limit=10)
guard.charge(4)
assert guard.remaining == 6
assert guard.affordable(6) is True
assert guard.affordable(7) is False
def test_write_sidecars_shape(tmp_path: Path) -> None:
result = RunResult(
run_id=RUN_ID, stop_reason=StopReason.STEP_BUDGET, final_answer=None, steps=()
)
write_sidecars(
runs_dir=tmp_path,
run_id=RUN_ID,
scenario="appworld",
fault="step_budget",
result=result,
wall_ms=12,
model_calls=3,
sink_failures=0,
env_executions=3,
task_id="t0",
phase=None,
resumed_from_step=None,
)
meta = json.loads((tmp_path / f"{RUN_ID}.meta.json").read_text(encoding="utf-8"))
assert meta["fault"] == "step_budget"
assert meta["scenario"] == "appworld"
assert meta["success"] is None
assert set(meta) == {
"scenario",
"task_id",
"phase",
"wall_ms",
"model_calls",
"sink_failures",
"env_executions",
"fault",
"success",
"resumed_from_step",
}
stored = json.loads((tmp_path / f"{RUN_ID}.result.json").read_text(encoding="utf-8"))
assert stored["stop_reason"] == "step_budget"
def test_write_sidecars_omits_result_when_there_is_none(tmp_path: Path) -> None:
"""取消那条路上 `run` 不返回结果，硬造一份是伪造。"""
write_sidecars(
runs_dir=tmp_path,
run_id=RUN_ID,
scenario="appworld",
fault="cancel_model",
result=None,
wall_ms=1,
model_calls=2,
sink_failures=0,
env_executions=1,
task_id="t0",
phase=None,
resumed_from_step=None,
)
assert not (tmp_path / f"{RUN_ID}.result.json").exists()
assert (tmp_path / f"{RUN_ID}.meta.json").exists()
async def test_guarded_turns_a_crash_into_undetermined() -> None:
"""某一类跑到一半炸了，前面几类已经花钱跑出来的结论不能跟着丢。"""
async def boom() -> FaultReport:
raise RuntimeError("容器起不来")
report = await guarded("cancel_env", boom())
assert report.fault == "cancel_env"
assert report.undetermineds
assert "容器起不来" in report.undetermineds[0].evidence
assert not report.breaches
async def test_guarded_passes_a_normal_report_through() -> None:
async def fine() -> FaultReport:
return FaultReport(fault="x", criteria=(check_env_untouched(executions=0, source="假"),))
report = await guarded("x", fine())
assert not report.breaches
assert not report.undetermineds
def test_main_requires_budget_calls(tmp_path: Path) -> None:
with pytest.raises(SystemExit):
main(["--runs-dir", str(tmp_path)])
def test_main_rejects_unknown_fault(tmp_path: Path) -> None:
with pytest.raises(SystemExit):
main(["--runs-dir", str(tmp_path), "--budget-calls", "10", "--fault", "不存在"])
def test_parser_accepts_repeated_fault_flags() -> None:
args = build_parser().parse_args(
[
"--runs-dir",
"x",
"--budget-calls",
"10",
"--fault",
"cancel_model",
"--fault",
"cancel_env",
]
)
assert args.fault == ["cancel_model", "cancel_env"]
# ---------------------------------------------------------------------------
# 十五、撞提示词上限
# ---------------------------------------------------------------------------
def test_run_finished_present_passes_and_breaches() -> None:
"""结束记录在不在与它带的取值对不对是两条判据，成因不同。"""
assert check_run_finished_present(as_read(run_finished())).status is CriterionStatus.PASSED
assert check_run_finished_present(as_read(step_completed())).status is CriterionStatus.BREACHED
def test_stop_reason_context_overflow_passes_and_breaches() -> None:
good = as_read(step_completed(), run_finished(stop_reason="context_overflow"))
bad = as_read(step_completed(), run_finished(stop_reason="step_budget"))
assert check_stop_reason(good, "context_overflow").status is CriterionStatus.PASSED
breach = check_stop_reason(bad, "context_overflow")
assert breach.status is CriterionStatus.BREACHED
assert "step_budget" in breach.evidence
def test_at_least_one_step_passes() -> None:
assert check_at_least_one_step(as_read(step_completed())).status is CriterionStatus.PASSED
def test_at_least_one_step_is_undetermined_without_steps() -> None:
"""第一步就撞上限的话这一类什么都没验到，报「无法判定」不报「击穿」。"""
outcome = check_at_least_one_step(as_read(run_started(), run_finished()))
assert outcome.status is CriterionStatus.UNDETERMINED
assert "一条步记录都没有" in outcome.evidence
def test_prompt_chars_monotonic_passes() -> None:
read = as_read(
step_completed(step_idx=0, prompt_chars=100),
step_completed(step_idx=1, prompt_chars=100),
step_completed(step_idx=2, prompt_chars=250),
)
outcome = check_prompt_chars_monotonic(read)
assert outcome.status is CriterionStatus.PASSED
assert "100" in outcome.evidence and "250" in outcome.evidence
def test_prompt_chars_monotonic_breaches_on_a_drop() -> None:
"""一次下降就是历史被截断过——那是这件事在轨迹里唯一看得见的痕迹。"""
read = as_read(
step_completed(step_idx=0, prompt_chars=100),
step_completed(step_idx=1, prompt_chars=300),
step_completed(step_idx=2, prompt_chars=120),
)
outcome = check_prompt_chars_monotonic(read)
assert outcome.status is CriterionStatus.BREACHED
assert "第 2 步从 300 掉到 120" in outcome.evidence
def test_prompt_chars_monotonic_breaches_on_a_non_integer() -> None:
read = as_read(step_completed(prompt_chars="很多"))
assert check_prompt_chars_monotonic(read).status is CriterionStatus.BREACHED
def test_prompt_chars_monotonic_is_undetermined_without_steps() -> None:
assert check_prompt_chars_monotonic(as_read(run_started())).status is (
CriterionStatus.UNDETERMINED
)
def test_prompt_reached_max_prompt_chars_passes() -> None:
"""最后一步自己装得下，再走一步就装不下——那才是「撑到了上限那条线上」。
上限 100；最后一步的提示词 90，它的输出 5 个字符、观察 10 个字符再加模板的 4 个，
下一次装配是 109 字符。
"""
read = as_read(
run_started(max_prompt_chars=100),
step_completed(prompt_chars=90, raw_output="x" * 5, observation="y" * 10),
run_finished(stop_reason="context_overflow"),
)
outcome = check_prompt_reached_max_prompt_chars(read)
assert outcome.status is CriterionStatus.PASSED
assert "109" in outcome.evidence
def test_prompt_reached_max_prompt_chars_breaches_when_it_still_fits() -> None:
"""还装得下就报 context_overflow：这次运行根本没撞到上限。"""
read = as_read(
run_started(max_prompt_chars=100),
step_completed(prompt_chars=50, raw_output="x", observation="y"),
run_finished(stop_reason="context_overflow"),
)
outcome = check_prompt_reached_max_prompt_chars(read)
assert outcome.status is CriterionStatus.BREACHED
assert "仍不超过上限 100" in outcome.evidence
def test_prompt_reached_max_prompt_chars_breaches_when_an_oversized_prompt_was_admitted() -> None:
"""步记录自己的提示词就超了上限，说明有一次超限的装配被放行去调了模型。"""
read = as_read(
run_started(max_prompt_chars=100),
step_completed(prompt_chars=120, raw_output="x", observation="y"),
run_finished(stop_reason="context_overflow"),
)
outcome = check_prompt_reached_max_prompt_chars(read)
assert outcome.status is CriterionStatus.BREACHED
assert "被放行去调了模型" in outcome.evidence
def test_prompt_reached_max_prompt_chars_reads_the_limit_from_the_snapshot() -> None:
"""上限从参数快照读，不硬编码：同一条步记录换个上限就该翻面。"""
step = step_completed(prompt_chars=90, raw_output="x" * 5, observation="y" * 10)
tight = as_read(run_started(max_prompt_chars=100), step)
loose = as_read(run_started(max_prompt_chars=100_000), step)
assert check_prompt_reached_max_prompt_chars(tight).status is CriterionStatus.PASSED
assert check_prompt_reached_max_prompt_chars(loose).status is CriterionStatus.BREACHED
def test_prompt_reached_max_prompt_chars_is_undetermined_without_run_started() -> None:
read = as_read(step_completed(), run_finished(stop_reason="context_overflow"))
outcome = check_prompt_reached_max_prompt_chars(read)
assert outcome.status is CriterionStatus.UNDETERMINED
assert "run_started" in outcome.evidence
def test_prompt_reached_max_prompt_chars_is_undetermined_without_steps() -> None:
read = as_read(run_started(), run_finished(stop_reason="context_overflow"))
assert check_prompt_reached_max_prompt_chars(read).status is CriterionStatus.UNDETERMINED
def test_prompt_reached_max_prompt_chars_is_undetermined_on_a_bad_snapshot() -> None:
read = as_read(
{
"record": "run_started",
"run_id": RUN_ID,
"parameter_snapshot": {
"request.max_prompt_chars": "很多",
"request.observation_template": TEMPLATE,
},
"schema_version": 1,
},
step_completed(),
)
assert check_prompt_reached_max_prompt_chars(read).status is CriterionStatus.UNDETERMINED
# ---------------------------------------------------------------------------
# 十六、按上下文现算提示词上限
# ---------------------------------------------------------------------------
def a_request(
*,
run_level: tuple[str, ...] = ("系统段",),
goal_level: tuple[str, ...] = ("题面",),
injections: dict[str, tuple[Injection, ...]] | None = None,
) -> RunRequest:
"""一份只填了装配用得着那几样的请求。执行器是个不派生自注册表的普通对象，构造期不比对。"""
class _Executor:
def parameters(self) -> dict[str, str]:
return {"kind": "fake"}
async def execute(self, action: object) -> ActionOutcome: # pragma: no cover - 用不到
raise AssertionError("这份请求只用来算提示词规模")
return RunRequest(
run_id=RUN_ID,
budget=Budget(
max_steps=1, max_actions=1, max_consecutive_parse_failures=1, max_prompt_chars=1
),
action_executor=_Executor(), # type: ignore[arg-type]
tools=ToolRegistry(),
context=Context(
run_level=tuple(
Message(role=Role.SYSTEM, content=(TextBlock(text=text),)) for text in run_level
),
goal_level=tuple(
Message(role=Role.USER, content=(TextBlock(text=text),)) for text in goal_level
),
),
injections=injections or {},
model_binding={},
model_replay_policy=ReplayPolicy.NEVER,
observation_template=TEMPLATE,
cancel_grace_seconds=1.0,
)
def test_initial_prompt_chars_counts_every_segment() -> None:
request = a_request(
run_level=("a" * 10, "b" * 5),
goal_level=("c" * 7,),
injections={"skill": (Injection(entry_id="e0", content="d" * 3),)},
)
assert initial_prompt_chars(request) == 25
def test_context_overflow_budget_leaves_room_for_exactly_the_first_step() -> None:
"""上限 = 初始提示词 + 余量。第一步刚好装得下，第二步靠一步的增长撑过去。"""
request = a_request(run_level=("x" * 40,), goal_level=("y" * 60,))
budget = build_context_overflow_budget(request)
assert budget.max_prompt_chars == 100 + CONTEXT_OVERFLOW_SLACK_CHARS
def test_initial_prompt_chars_rejects_an_unknown_block_type() -> None:
"""认不得的块当成 0 会让上限算小，于是第一步就撞上限、这一类什么都验不到。"""
class _Weird:
pass
request = a_request()
broken = replace_context_block(request, _Weird())
with pytest.raises(Exception, match="认不得的内容块类型"):
initial_prompt_chars(broken)
def replace_context_block(request: RunRequest, block: object) -> RunRequest:
"""把上下文里那条消息的内容块换成给定的东西。造非法输入用。"""
import dataclasses
context = Context(
run_level=(Message(role=Role.SYSTEM, content=(block,)),), # type: ignore[arg-type]
goal_level=(),
)
return dataclasses.replace(request, context=context)
# ---------------------------------------------------------------------------
# 十七、环境故障：判据
# ---------------------------------------------------------------------------
def test_stop_reason_env_error_passes_and_breaches() -> None:
good = as_read(step_completed(), run_finished(stop_reason="env_error"))
bad = as_read(step_completed(), run_finished(stop_reason="task_completed"))
assert check_stop_reason(good, "env_error").status is CriterionStatus.PASSED
assert check_stop_reason(bad, "env_error").status is CriterionStatus.BREACHED
def test_last_step_action_status_passes() -> None:
read = as_read(
step_completed(step_idx=0, action_status="executed"),
step_completed(step_idx=1, action_status="env_error"),
)
assert check_last_step_action_status(read, expected="env_error").status is (
CriterionStatus.PASSED
)
def test_last_step_action_status_breaches() -> None:
read = as_read(
step_completed(step_idx=0, action_status="env_error"),
step_completed(step_idx=1, action_status="executed"),
)
outcome = check_last_step_action_status(read, expected="env_error")
assert outcome.status is CriterionStatus.BREACHED
assert "executed" in outcome.evidence
def test_last_step_action_status_is_undetermined_without_steps() -> None:
assert check_last_step_action_status(as_read(run_started()), expected="env_error").status is (
CriterionStatus.UNDETERMINED
)
def test_env_failed_observation_passes_with_the_value_from_the_scenario() -> None:
"""期望值从场景那份合成观察取，判据这边不抄一份字面量。"""
expected = build_synthetic_observations().env_failed
read = as_read(
step_completed(
action_status="env_error", observation=expected, observation_is_synthetic=True
)
)
assert check_last_observation_is_synthetic(read, expected=expected, name="x").status is (
CriterionStatus.PASSED
)
def test_env_failed_observation_breaches_when_the_flag_is_false() -> None:
expected = build_synthetic_observations().env_failed
read = as_read(
step_completed(
action_status="env_error", observation=expected, observation_is_synthetic=False
)
)
outcome = check_last_observation_is_synthetic(read, expected=expected, name="x")
assert outcome.status is CriterionStatus.BREACHED
assert "observation_is_synthetic" in outcome.evidence
def test_env_failed_observation_breaches_when_the_executor_text_survived() -> None:
"""替换没发生的话，历史里躺着的是与环境通信那一层给的原文——那正是这一条要抓的。"""
expected = build_synthetic_observations().env_failed
read = as_read(
step_completed(
action_status="env_error",
observation="ConnectError: All connection attempts failed",
observation_is_synthetic=True,
)
)
outcome = check_last_observation_is_synthetic(read, expected=expected, name="x")
assert outcome.status is CriterionStatus.BREACHED
assert "ConnectError" not in outcome.evidence
def test_env_failed_observation_is_undetermined_without_steps() -> None:
assert (
check_last_observation_is_synthetic(as_read(run_started()), expected="x", name="x").status
is CriterionStatus.UNDETERMINED
)
def test_steps_before_last_all_executed_passes() -> None:
read = as_read(
step_completed(step_idx=0, action_status="executed"),
step_completed(step_idx=1, action_status="executed"),
step_completed(step_idx=2, action_status="env_error"),
)
assert check_steps_before_last_all_executed(read).status is CriterionStatus.PASSED
def test_steps_before_last_all_executed_breaches() -> None:
"""环境坏掉之前的步被改写或补上别的状态，说明一次局部故障扩散到了已经落地的轨迹上。"""
read = as_read(
step_completed(step_idx=0, action_status="executed"),
step_completed(step_idx=1, action_status="env_error"),
step_completed(step_idx=2, action_status="env_error"),
)
outcome = check_steps_before_last_all_executed(read)
assert outcome.status is CriterionStatus.BREACHED
assert "[1]" in outcome.evidence
def test_steps_before_last_all_executed_is_undetermined_with_a_single_step() -> None:
"""只有一步的话这一条真空成立，那种「通过」什么都没验。"""
outcome = check_steps_before_last_all_executed(as_read(step_completed()))
assert outcome.status is CriterionStatus.UNDETERMINED
assert "真空成立" in outcome.evidence
def test_env_broken_after_a_full_step_passes() -> None:
outcome = check_env_broken_after_a_full_step(
broken=True, executions_before_break=1, expected_after=1
)
assert outcome.status is CriterionStatus.PASSED
def test_env_broken_after_a_full_step_is_undetermined_when_it_never_broke() -> None:
outcome = check_env_broken_after_a_full_step(
broken=False, executions_before_break=0, expected_after=1
)
assert outcome.status is CriterionStatus.UNDETERMINED
assert "一次都没被弄坏" in outcome.evidence
def test_env_broken_after_a_full_step_breaches_when_it_broke_too_early() -> None:
"""第一次执行之前就打死容器的话，压到的是初始化而不是动作执行接缝。
**这是击穿不是无法判定**：注入器对外宣称在第 1 次执行之后动手，整类故障的结论都建立在
那句话上。真在别的时刻动手的话，这一类压到的是另一件事，而报告上仍然写着它通过了。
"""
outcome = check_env_broken_after_a_full_step(
broken=True, executions_before_break=0, expected_after=1
)
assert outcome.status is CriterionStatus.BREACHED
assert "初始化" in outcome.evidence
def test_env_broken_after_a_full_step_breaches_when_it_broke_too_late() -> None:
"""比声明的晚动手同样是击穿：压到的仍然不是它声称的那个时刻。"""
outcome = check_env_broken_after_a_full_step(
broken=True, executions_before_break=3, expected_after=1
)
assert outcome.status is CriterionStatus.BREACHED
assert "比声明的晚" in outcome.evidence
# ---------------------------------------------------------------------------
# 十八、环境故障：杀容器那段编排里的纯函数与执行器包装
# ---------------------------------------------------------------------------
def test_container_name_for_port() -> None:
assert container_name_for_port(8201) == "polyloop-soak-appworld-8201"
def test_should_break_env_waits_for_a_full_execution() -> None:
assert (
should_break_env(executions_done=0, break_after_executions=1, already_broken=False) is False
)
assert (
should_break_env(executions_done=1, break_after_executions=1, already_broken=False) is True
)
def test_should_break_env_never_breaks_twice() -> None:
"""容器已经没了，再发一次 docker kill 只会拿到一个「没有这个容器」的错误。"""
assert (
should_break_env(executions_done=5, break_after_executions=1, already_broken=True) is False
)
class _FakeExecutor:
"""按剧本一次次返回结果或抛异常的假执行器。"""
def __init__(self, script: list[object]) -> None:
self.script = script
self.calls = 0
def parameters(self) -> dict[str, str]:
return {"kind": "fake_appworld"}
async def execute(self, action: object) -> ActionOutcome:
item = self.script[self.calls]
self.calls += 1
if isinstance(item, BaseException):
raise item
return item # type: ignore[return-value]
def an_outcome(status: ActionStatus = ActionStatus.EXECUTED) -> ActionOutcome:
return ActionOutcome(
status=status,
observation="环境返回",
observation_is_synthetic=False,
env_reported_completion=False,
observation_truncated_chars=0,
)
async def test_env_breaking_executor_breaks_after_one_full_execution() -> None:
"""第一次执行之前不动手，第二次执行之前动手。"""
broke_at: list[int] = []
inner = _FakeExecutor([an_outcome(), httpx.ConnectError("All connection attempts failed")])
async def break_env() -> str:
broke_at.append(inner.calls)
return "容器已被打死"
executor = EnvBreakingExecutor(inner=inner, break_after_executions=1, break_env=break_env)
first = await executor.execute(object())
assert first.status is ActionStatus.EXECUTED
assert broke_at == []
second = await executor.execute(object())
assert broke_at == [1]
assert executor.broken is True
assert executor.executions_before_break == 1
assert executor.break_note == "容器已被打死"
assert second.status is ActionStatus.ENV_ERROR
async def test_env_breaking_executor_translates_a_transport_error() -> None:
"""场景那侧只接 AppWorldError，容器没了时抛的是 httpx.ConnectError，不翻译就整个抛出去。"""
inner = _FakeExecutor([httpx.ConnectError("All connection attempts failed")])
async def break_env() -> str:
return "打死了"
executor = EnvBreakingExecutor(inner=inner, break_after_executions=0, break_env=break_env)
outcome = await executor.execute(object())
assert outcome.status is ActionStatus.ENV_ERROR
assert outcome.observation_is_synthetic is False
assert "ConnectError" in outcome.observation
async def test_env_breaking_executor_passes_an_inner_env_error_through() -> None:
"""内层自己判出环境故障时原样透传：场景那边哪天补上转换，这里不用跟着改。"""
inner = _FakeExecutor([an_outcome(ActionStatus.ENV_ERROR)])
async def break_env() -> str: # pragma: no cover - 这条用例不动手
raise AssertionError("不该动手")
executor = EnvBreakingExecutor(inner=inner, break_after_executions=99, break_env=break_env)
outcome = await executor.execute(object())
assert outcome.status is ActionStatus.ENV_ERROR
assert outcome.observation == "环境返回"
async def test_env_breaking_executor_reports_its_probe_in_parameters() -> None:
inner = _FakeExecutor([])
async def break_env() -> str: # pragma: no cover - 这条用例不执行动作
raise AssertionError("不该动手")
executor = EnvBreakingExecutor(inner=inner, break_after_executions=1, break_env=break_env)
assert executor.parameters() == {"kind": "fake_appworld", "env_break_probe": "docker_kill"}
# ---------------------------------------------------------------------------
# 十九、两类新故障接进命令行
# ---------------------------------------------------------------------------
def test_new_faults_are_in_the_default_selection() -> None:
"""不给 --fault 时按 FAULT_NAMES 全跑，两类都得在里面。"""
assert "context_overflow" in FAULT_NAMES
assert "env_error" in FAULT_NAMES
def test_parser_accepts_the_new_faults() -> None:
args = build_parser().parse_args(
[
"--runs-dir",
"x",
"--budget-calls",
"10",
"--fault",
"context_overflow",
"--fault",
"env_error",
]
)
assert args.fault == ["context_overflow", "env_error"]
# ---------------------------------------------------------------------------
# 二十、真空成立那一类：空数据不许报通过
# ---------------------------------------------------------------------------
def test_log_readable_is_undetermined_on_an_empty_log() -> None:
"""「零条记录全都解得开」是一句真空成立的话，它在报告上和真验过一模一样。"""
outcome = check_log_readable(LogRead(payloads=(), torn=False, bad_lines=()))
assert outcome.status is CriterionStatus.UNDETERMINED
def test_log_readable_is_undetermined_when_the_file_is_missing(tmp_path: Path) -> None:
"""日志文件根本不在时 `read_log` 给的就是空读数，这条链路要连得上。"""
outcome = check_log_readable(read_log(tmp_path / "nope.jsonl"))
assert outcome.status is CriterionStatus.UNDETERMINED
def test_no_env_error_step_is_undetermined_without_steps() -> None:
outcome = check_no_env_error_step(as_read(intent()))
assert outcome.status is CriterionStatus.UNDETERMINED
def test_audit_unchanged_is_undetermined_when_both_sides_are_empty() -> None:
"""上一版实跑里这条正是 0 比 0 通过的：没有副作用就没有「没被重放」可验。"""
outcome = check_audit_unchanged(before=[], after=[])
assert outcome.status is CriterionStatus.UNDETERMINED
def test_audit_unchanged_still_breaches_when_only_after_has_entries() -> None:
"""崩溃时是空的、续跑之后长出条目：那是续跑执行了副作用，仍然是击穿。"""
outcome = check_audit_unchanged(before=[], after=["write_note\tevidence.md\taaa"])
assert outcome.status is CriterionStatus.BREACHED
# ---------------------------------------------------------------------------
# 二十一、环境账的前缀与跨崩溃边界的重放
# ---------------------------------------------------------------------------
def test_audit_prefix_preserved_passes_when_resume_only_appends() -> None:
before = ["write_note\tplan.md\taaa"]
after = [*before, "write_note\tevidence.md\tbbb"]
assert check_audit_prefix_preserved(before=before, after=after).status is (
CriterionStatus.PASSED
)
def test_audit_prefix_preserved_breaches_when_an_old_entry_is_rewritten() -> None:
"""条数不变、去重也不变，只有这条看得见：已经落地的那次副作用被改写了。"""
outcome = check_audit_prefix_preserved(
before=["write_note\tevidence.md\taaa"], after=["write_note\tevidence.md\tbbb"]
)
assert outcome.status is CriterionStatus.BREACHED
assert "第 0 条" in outcome.evidence
def test_audit_prefix_preserved_breaches_when_entries_disappear() -> None:
outcome = check_audit_prefix_preserved(before=["write_note\tplan.md\taaa"], after=[])
assert outcome.status is CriterionStatus.BREACHED
assert "被抹掉" in outcome.evidence
def test_audit_prefix_preserved_is_undetermined_without_a_crash_side_ledger() -> None:
assert check_audit_prefix_preserved(before=[], after=[]).status is (
CriterionStatus.UNDETERMINED
)
def test_no_cross_segment_replay_breaches_on_a_repeat_after_the_crash() -> None:
"""跨越崩溃边界的重复就是重放的签名。"""
before = ["write_note\tevidence.md\taaa"]
after = [*before, "write_note\tevidence.md\taaa"]
outcome = check_no_cross_segment_replay(before=before, after=after)
assert outcome.status is CriterionStatus.BREACHED
assert "aaa" in outcome.evidence
def test_no_cross_segment_replay_allows_new_content_after_the_crash() -> None:
"""续跑写了别的内容不算重放：时机 A 下崩溃点之后本来就该接着跑。"""
before = ["write_note\tevidence.md\taaa"]
after = [*before, "write_note\tevidence.md\tbbb"]
assert check_no_cross_segment_replay(before=before, after=after).status is (
CriterionStatus.PASSED
)
def test_no_cross_segment_replay_passes_when_resume_wrote_nothing() -> None:
before = ["write_note\tevidence.md\taaa"]
assert check_no_cross_segment_replay(before=before, after=list(before)).status is (
CriterionStatus.PASSED
)
def test_no_cross_segment_replay_is_undetermined_without_a_crash_side_ledger() -> None:
outcome = check_no_cross_segment_replay(before=[], after=["write_note\tplan.md\taaa"])
assert outcome.status is CriterionStatus.UNDETERMINED
# ---------------------------------------------------------------------------
# 二十二、提示词：库记的数与它真正发出去的对账
# ---------------------------------------------------------------------------
class _SpyModelClient:
"""记下每次收到的调用，`parameters()` 有自己的取值，用来验包装层原样转发。"""
def __init__(self) -> None:
self.calls: list[object] = []
def parameters(self) -> dict[str, str]:
return {"scope": "llm", "sources": "src:prov:model"}
async def call(self, call: object) -> ModelReply:
self.calls.append(call)
return ModelReply(call_id="c0", content="x", thinking="")
def a_model_call(*, call_index: int, texts: tuple[str, ...]):
from polyloop.ports import ModelCall
return ModelCall(
messages=tuple(Message(role=Role.USER, content=(TextBlock(text=text),)) for text in texts),
call_index=call_index,
run_id=RUN_ID,
result_id=f"r{call_index}",
binding={},
)
async def test_prompt_size_recorder_counts_what_was_sent() -> None:
inner = _SpyModelClient()
recorder = PromptSizeRecordingClient(inner=inner)
await recorder.call(a_model_call(call_index=0, texts=("abc", "de")))
await recorder.call(a_model_call(call_index=1, texts=("abcdef",)))
assert recorder.chars_by_call_index == {0: 5, 1: 6}
assert len(inner.calls) == 2
def test_prompt_size_recorder_forwards_parameters_verbatim() -> None:
"""加一个键会让续跑报假的参数漂移。"""
inner = _SpyModelClient()
assert PromptSizeRecordingClient(inner=inner).parameters() == inner.parameters()
def test_prompt_chars_match_what_was_sent_passes() -> None:
read = as_read(
step_completed(step_idx=0, prompt_chars=100),
step_completed(step_idx=1, prompt_chars=250),
)
outcome = check_prompt_chars_match_what_was_sent(read, observed={0: 100, 1: 250})
assert outcome.status is CriterionStatus.PASSED
def test_prompt_chars_match_what_was_sent_breaches_on_a_fabricated_value() -> None:
"""库静默截断了历史，却仍把 prompt_chars 记成一路不减的自述值。
单调性那条与撞上限那条都只读这一列，两条都会通过；只有和真正发出去的字符数对账才看得见。
"""
read = as_read(
step_completed(step_idx=0, prompt_chars=100),
step_completed(step_idx=1, prompt_chars=250),
)
assert check_prompt_chars_monotonic(read).status is CriterionStatus.PASSED
outcome = check_prompt_chars_match_what_was_sent(read, observed={0: 100, 1: 40})
assert outcome.status is CriterionStatus.BREACHED
assert "第 1 步记的是 250 字符，实际发出去 40 字符" in outcome.evidence
def test_prompt_chars_match_what_was_sent_is_undetermined_without_steps() -> None:
outcome = check_prompt_chars_match_what_was_sent(as_read(intent()), observed={0: 10})
assert outcome.status is CriterionStatus.UNDETERMINED
def test_prompt_chars_match_what_was_sent_is_undetermined_without_calls() -> None:
read = as_read(step_completed(step_idx=0, prompt_chars=100))
assert check_prompt_chars_match_what_was_sent(read, observed={}).status is (
CriterionStatus.UNDETERMINED
)
def test_prompt_chars_match_what_was_sent_is_undetermined_when_nothing_lines_up() -> None:
read = as_read(step_completed(step_idx=0, prompt_chars=100))
outcome = check_prompt_chars_match_what_was_sent(read, observed={7: 100})
assert outcome.status is CriterionStatus.UNDETERMINED
# ---------------------------------------------------------------------------
# 二十三、取消：环境侧安静下来，事件文件完整
# ---------------------------------------------------------------------------
def test_env_quiet_after_cancel_passes_when_both_counters_stand_still() -> None:
outcome = check_env_quiet_after_cancel(
dispatched_before=1, dispatched_after=1, completed_before=0, completed_after=0, waited_s=3.0
)
assert outcome.status is CriterionStatus.PASSED
def test_env_quiet_after_cancel_breaches_when_more_work_is_dispatched() -> None:
"""库在取消之后还往环境派活。"""
outcome = check_env_quiet_after_cancel(
dispatched_before=1, dispatched_after=2, completed_before=0, completed_after=0, waited_s=3.0
)
assert outcome.status is CriterionStatus.BREACHED
assert "还在往环境派活" in outcome.evidence
def test_env_quiet_after_cancel_breaches_when_an_execution_lands_late() -> None:
"""动作已经发到容器，取消之后仍然跑完并改了环境状态。"""
outcome = check_env_quiet_after_cancel(
dispatched_before=1, dispatched_after=1, completed_before=0, completed_after=1, waited_s=3.0
)
assert outcome.status is CriterionStatus.BREACHED
assert "跑完了" in outcome.evidence
def test_env_quiet_after_cancel_breaches_when_a_counter_goes_backwards() -> None:
outcome = check_env_quiet_after_cancel(
dispatched_before=2, dispatched_after=1, completed_before=1, completed_after=1, waited_s=3.0
)
assert outcome.status is CriterionStatus.BREACHED
assert "倒退" in outcome.evidence
def test_events_file_intact_passes(tmp_path: Path) -> None:
path = tmp_path / f"{RUN_ID}.events.jsonl"
path.write_text(
'{"kind":"step_finished","run_id":"r","step_idx":0}\n'
'{"kind":"step_finished","run_id":"r","step_idx":1}\n',
encoding="utf-8",
)
outcome = check_events_file_intact(path)
assert outcome.status is CriterionStatus.PASSED
assert "2 行" in outcome.evidence
def test_events_file_intact_breaches_on_a_torn_tail(tmp_path: Path) -> None:
"""取消穿过了一次事件写入的中途。库有宽限期，所以这算击穿而不是「没发生过」。"""
path = tmp_path / f"{RUN_ID}.events.jsonl"
path.write_text(
'{"kind":"step_finished","run_id":"r","step_idx":0}\n{"kind":"step_fin',
encoding="utf-8",
)
outcome = check_events_file_intact(path)
assert outcome.status is CriterionStatus.BREACHED
assert "没有被换行终结" in outcome.evidence
def test_events_file_intact_breaches_on_a_broken_line(tmp_path: Path) -> None:
path = tmp_path / f"{RUN_ID}.events.jsonl"
path.write_text('{"kind":"step_finished"}\n不是 JSON\n', encoding="utf-8")
outcome = check_events_file_intact(path)
assert outcome.status is CriterionStatus.BREACHED
assert "[2]" in outcome.evidence
def test_events_file_intact_breaches_on_a_non_object_line(tmp_path: Path) -> None:
path = tmp_path / f"{RUN_ID}.events.jsonl"
path.write_text('{"kind":"step_finished"}\n[1,2,3]\n', encoding="utf-8")
assert check_events_file_intact(path).status is CriterionStatus.BREACHED
def test_events_file_intact_is_undetermined_without_a_file(tmp_path: Path) -> None:
outcome = check_events_file_intact(tmp_path / "nope.events.jsonl")
assert outcome.status is CriterionStatus.UNDETERMINED
def test_events_file_intact_is_undetermined_on_an_empty_file(tmp_path: Path) -> None:
path = tmp_path / f"{RUN_ID}.events.jsonl"
path.write_bytes(b"")
assert check_events_file_intact(path).status is CriterionStatus.UNDETERMINED
Reference in New Issue View Git Blame Copy Permalink