iomgaa/PolyLoop
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases 3 Wiki Activity
Files
5d5371792d8732624194242d6da675b4a0f8f6ff
PolyLoop/tools/soak/tests/test_faults.py
T
iomgaa 5d5371792d fix(soak): 父子两侧的崩溃条件对齐，并给自杀留一个窗口 
上一轮实跑里确定性自杀一次都没走到——父进程一看见日志尾部形态对上就发信号，而自杀条件
比它严一档（还要求审计账非空），于是外部信号永远抢先，自杀路径成了死代码，崩溃点又变回
碰运气。表现是崩得太早：动作还没执行过，最硬的那条判据没有实料可判。

两处对齐：父进程的命中条件也要求审计账非空（做成必传参数，给默认值等于让某个调用点静默
跳过这一条，而这正是这次出问题的方式）；兜底 SIGKILL 之前先等两秒看子进程是不是自己以
137 退出。

实跑结果：两档都走确定性自杀路径，都崩在「写入执行过之后」，绝不重放那条判据在两档都有
实料——审计账 1 条、去重后仍 1 条，续跑前后也都是 1 条。整套七类击穿 0 条、无法判定 0 条。

Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
2026-08-11 10:41:27 -04:00

1184 lines
43 KiB
Python
Raw 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 pytest
from polyloop.ports import EventKind, InvalidDecision, RunLog
from polyloop.types import (
ActionOutcome,
ActionStatus,
Intent,
IntentKind,
ModelCallResult,
ModelReply,
ReplayPolicy,
RunFinished,
RunResult,
RunStarted,
StepCompleted,
StepRecord,
StopReason,
)
from tools.soak.faults import (
CRASH_EXIT_CODE,
AlwaysInvalidParser,
CallGuard,
Criterion,
CriterionStatus,
FaultReport,
JsonlEventSink,
KillTiming,
LogRead,
SelfKillingStore,
build_parser,
check_all_steps_parse_failed,
check_audit_unchanged,
check_cancelled_raised,
check_crash_prefix_preserved,
check_env_untouched,
check_executed_action_count,
check_intents_settled,
check_lease_returned,
check_log_readable,
check_never_action_not_replayed,
check_no_env_error_step,
check_resume_made_progress,
check_step_count,
check_step_indices_dense,
check_stop_reason,
count_model_calls,
guarded,
main,
parse_audit_line,
parse_terminated,
read_log,
should_kill,
spawn_and_kill,
stop_reason_of,
tagged,
terminated_prefix,
write_sidecars,
)
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",
) -> 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": "x",
"content_chars": 1,
"thinking_chars": 0,
"action": None,
"parse_ok": parse_ok,
"parse_error": None if parse_ok else "解释不了",
"observation": "o",
"observation_is_synthetic": False,
"observation_truncated_chars": 0,
"prompt_chars": 10,
"call_id": "c0",
"step_wall_ms": 1,
"tool_name": None,
"tool_arguments": None,
"action_status": action_status,
"env_reported_completion": False,
"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"]
Reference in New Issue View Git Blame Copy Permalink