diff --git a/tools/soak/faults.py b/tools/soak/faults.py index 2bb5e28..8df933c 100644 --- a/tools/soak/faults.py +++ b/tools/soak/faults.py @@ -16,6 +16,13 @@ 只会随机变红。这里的判据全是结构不变量：字节前缀、步号连续性、意图有没有归宿、审计账去重 前后的条数、停止原因的取值、步数与上限的关系、环境侧的执行计数。 +**崩溃是子进程自己在指定的写入点上 `os._exit`，不是父进程从外面抢窗口发信号。** 第一版靠 +父进程轮询日志尾部再 `SIGKILL`，实测三次全都没能命中「一步完整落地之后」那个时机——库写完 +步记录紧接着就写下一步的模型调用意图，中间只有内存里的装配计算，窗口窄到信号挤不进去。 +**那次实测本身是一条要记住的事实**：自然发生的崩溃几乎总是落在「有意图没结果」那一态上， +而不是落在两步之间的干净边界上。但要验那个边界就不能靠碰运气。做法见 `SelfKillingStore`； +外部 SIGKILL 那条路留着兜底，没有删。 + **产物照 `tools/soak/scoreboard.py` 的 sidecar 约定写出**（`.result.json` / `.events.jsonl` / `.meta.json`），`meta.fault` 填故障名，这样记分板能把它们和正常批次一起判。崩溃那两类天然 缺 sidecar——被 SIGKILL 的子进程来不及写，记分板会报「无法判定」，那是对的，不为了让它变绿 @@ -33,20 +40,31 @@ from __future__ import annotations import argparse import asyncio import json +import os import sys import time -from collections.abc import Awaitable, Mapping, Sequence +from collections.abc import Awaitable, Callable, Mapping, Sequence from dataclasses import dataclass, replace from enum import StrEnum from pathlib import Path from polyloop import session from polyloop.adapters import GatewayModelClient -from polyloop.ports import Event, InvalidDecision, ParsedReply +from polyloop.ports import Event, InvalidDecision, ParsedReply, RunLog, RunStore from polyloop.serialization import encode from polyloop.session import AgentDefinition, ParameterDriftError, RunRequest from polyloop.stores import RECORD_KEY, JsonlRunStore -from polyloop.types import Budget, ModelReply, RunResult +from polyloop.types import ( + Budget, + Intent, + ModelCallResult, + ModelReply, + ReplayPolicy, + RunFinished, + RunResult, + RunStarted, + StepCompleted, +) from tools.soak.appworld import AppWorldPool from tools.soak.scenarios import appworld as appworld_scenario from tools.soak.scenarios import govdoc as govdoc_scenario @@ -596,12 +614,17 @@ def check_lease_returned(*, borrowed: bool, timeout_s: float, pool_size: int) -> # --------------------------------------------------------------------------- -# 五、子进程编排：轮询日志尾部，按时机 SIGKILL +# 五、子进程编排：子进程按时机自杀，外部 SIGKILL 兜底 # --------------------------------------------------------------------------- +#: 子进程按时机自杀时用的退出码。父进程靠它区分「按预期崩在时机上」与「因为别的原因退出」。 +#: +#: 取 137 是照 128+9 那个惯例（外部 SIGKILL 的等价形态），让两条路径在日志里读起来是同一件事。 +CRASH_EXIT_CODE = 137 + class KillTiming(StrEnum): - """在哪个时机把子进程杀掉。两种时机的判据不同，不许混成一个用例。""" + """在哪个时机让子进程崩掉。两种时机的判据不同，不许混成一个用例。""" #: 时机 A：最后一条是 `step_completed`，一步完整落地之后崩。续跑该真的接着往下跑。 AFTER_STEP = "after_step" @@ -611,7 +634,9 @@ class KillTiming(StrEnum): def should_kill(read: LogRead, *, timing: KillTiming, after_steps: int) -> bool: - """现在这份日志尾部是不是要等的那个时机。纯函数，轮询循环每次拿它问一句。 + """现在这份日志尾部是不是要等的那个时机。纯函数。 + + 子进程自杀之后父进程拿它复核一遍尾部形态；外部 SIGKILL 那条兜底路径每次轮询也问它一句。 时机 B 额外要求那条意图的重放策略是 `never`。**这比「最后一条是意图」更严**，而且必须 更严：GovDoc 的 `read_document` 与 `grep_document` 声明的是 `safe`，悬在那种意图上续跑 @@ -628,20 +653,127 @@ def should_kill(read: LogRead, *, timing: KillTiming, after_steps: int) -> bool: return last.get(RECORD_KEY) == "intent" and last.get("replay_policy") == "never" +class SelfKillingStore: + """包一层 `RunStore`：某一次写入落盘返回之后，按时机让子进程当场死掉。 + + **为什么不再靠外部 SIGKILL 抢窗口。** 实测下来时机 A 一次都没命中过，三次全都报「发信号 + 与子进程停笔之间又写进了记录」：库写完 `step_completed` 紧接着就写下一步的模型调用意图， + 中间只有内存里的装配计算，那个窗口窄到外面的信号挤不进去。**这个观察本身值得记住**—— + 它说明自然发生的崩溃几乎总是落在「有意图没结果」那一态上，而不是落在两步之间的干净边界 + 上。但要验时机 A 就不能靠碰运气，得让子进程自己在那一点上死。 + + **`os._exit` 与 SIGKILL 对磁盘的效果等价。** 它不跑 `finally`、不跑 `atexit`、不 flush + 任何缓冲，直接进 `_exit(2)` 系统调用。`JsonlRunStore` 每次写完自己 `fsync`（意图与运行 + 开始、运行结束三处）或者靠同一文件的追加序兜（`0005` 决策五那条前缀持久性），本来就没有 + 未刷缓冲要指望进程退出时替它写下去，所以「有没有机会清理」在这里不影响磁盘上留下什么。 + + **`parameters()` 原样转发内层的**，不加自己的键：它进参数快照，而父进程续跑时用的是一个 + 没包过的 `JsonlRunStore`。加一个键，续跑就报参数漂移，而那是一次假故障。 + + **自杀条件带上「审计账已经非空」**：要验的最硬那条判据是「声明绝不重放的动作没有被执行 + 两次」，它数的是工作区审计账。崩在模型还没调过 `write_note` 的时候，账是空的，那条判据 + 真空成立——报出来是「通过」，实际什么都没验。 + """ + + __slots__ = ("_exit_now", "_inner", "_timing", "_workspace") + + def __init__( + self, + *, + inner: RunStore, + timing: KillTiming, + workspace: Path, + exit_now: Callable[[int], object] = os._exit, + ) -> None: + """Args: + inner: 真正写盘的那个存储。 + timing: 在哪一次写入之后死。 + workspace: 数审计账用的工作区目录。 + exit_now: 怎么死。**做成参数是为了能测**——默认的 `os._exit` 在测试里会把 pytest + 自己一起带走，测试注入一个抛哨兵异常的替身。 + """ + self._inner = inner + self._timing = timing + self._workspace = Path(workspace) + self._exit_now = exit_now + + def parameters(self) -> Mapping[str, str]: + return self._inner.parameters() + + def _audit_is_not_empty(self) -> bool: + return bool(read_audit_lines(self._workspace)) + + async def read_log(self, run_id: str) -> RunLog: + return await self._inner.read_log(run_id) + + async def write_run_started(self, record: RunStarted) -> None: + await self._inner.write_run_started(record) + + async def write_intent(self, record: Intent) -> None: + await self._inner.write_intent(record) + if ( + self._timing is KillTiming.AT_INTENT + and record.replay_policy is ReplayPolicy.NEVER + and self._audit_is_not_empty() + ): + self._exit_now(CRASH_EXIT_CODE) + + async def write_model_call_result(self, record: ModelCallResult) -> None: + await self._inner.write_model_call_result(record) + + async def write_step_completed(self, record: StepCompleted) -> None: + await self._inner.write_step_completed(record) + if self._timing is KillTiming.AFTER_STEP and self._audit_is_not_empty(): + self._exit_now(CRASH_EXIT_CODE) + + async def write_run_finished(self, record: RunFinished) -> None: + await self._inner.write_run_finished(record) + + @dataclass(frozen=True, slots=True, kw_only=True) class KillOutcome: - """一次「起子进程、等时机、杀掉」的结果。""" + """一次「起子进程、等它崩在时机上」的结果。""" #: 时机命中了没有。没命中要重试，重试若干次仍不命中要报出来——悄悄降级成另一种时机会让 #: 报告显示「验过了」而其实验的是另一件事。 hit: bool reason: str - #: 杀掉之后重读日志、截到最后一个换行为止的那一段。字节比对拿它当基准。 + #: 崩溃之后重读日志、截到最后一个换行为止的那一段。字节比对拿它当基准。 snapshot: bytes = b"" #: 崩溃时已经完整落地的步数。 steps: int = 0 - #: 这次子进程发起过几次模型调用（含被杀在半路的那次），记账用。 + #: 这次子进程发起过几次模型调用（含崩在半路的那次），记账用。 model_calls: int = 0 + #: 子进程的退出码。按预期自杀是 `CRASH_EXIT_CODE`，被外部 SIGKILL 兜底掉是 -9。 + exit_code: int | None = None + + +def _judge_crash( + *, log_path: Path, timing: KillTiming, after_steps: int, exit_code: int | None, how: str +) -> KillOutcome: + """崩溃之后重读一次日志，判尾部形态是不是要的那个时机。 + + **不管子进程是自杀的还是被外部信号杀的，都要重判一次。** 自杀那条路上判的是「自杀条件与 + 时机判据说的是不是同一件事」；外部信号那条路上判的是「读日志与发信号之间它有没有又写进 + 一条」——不重判的话，一次「本想在时机 A 杀、实际杀在时机 B」会被当成时机 A 判下去。 + """ + crashed = log_path.read_bytes() if log_path.is_file() else b"" + after = parse_terminated(crashed) + if not should_kill(after, timing=timing, after_steps=after_steps): + return KillOutcome( + hit=False, + reason=f"{how}之后重读日志，尾部不是时机 {timing.value} 要的形态", + model_calls=count_model_calls(after), + exit_code=exit_code, + ) + return KillOutcome( + hit=True, + reason=f"{how}，命中时机 {timing.value}", + snapshot=terminated_prefix(crashed), + steps=len(tagged(after, "step_completed")), + model_calls=count_model_calls(after), + exit_code=exit_code, + ) async def spawn_and_kill( @@ -650,61 +782,74 @@ async def spawn_and_kill( log_path: Path, timing: KillTiming, after_steps: int, + child_log_path: Path | None = None, poll_interval_s: float = 0.002, timeout_s: float = 600.0, ) -> KillOutcome: - """起一个子进程，轮询它的日志，命中时机就 `SIGKILL`。 + """起一个子进程，等它崩在时机上。 - **用 SIGKILL 不用 SIGTERM**：要的是没有任何清理机会的死法。SIGTERM 会走 Python 的信号 - 处理，`finally` 有机会跑完，那验的是优雅退出而不是崩溃。 + **主路径是子进程自己在时机上 `os._exit`**（见 `SelfKillingStore`），父进程只负责认领： + 退出码等于 `CRASH_EXIT_CODE` 且日志尾部形态对得上，就算命中。 - 杀掉之后**重读一次日志再判时机是不是还成立**：读日志与发信号之间子进程还在写，读到的 - 尾部可能已经不是杀掉那一刻的尾部了。不重判的话，一次「本想在时机 A 杀、实际杀在时机 B」 - 会被当成时机 A 判下去。 + **外部 SIGKILL 那条路留着兜底**，没有删掉：子进程那侧的自杀条件万一因为别的原因没触发 + （包装漏了一处、审计账一直是空的），轮询仍然会在尾部形态对上的那一刻把它杀掉。**用 + SIGKILL 不用 SIGTERM**，要的是没有任何清理机会的死法——SIGTERM 会走 Python 的信号处理， + `finally` 有机会跑完，那验的是优雅退出而不是崩溃。 + + 子进程的输出**写进文件而不是管道**：管道缓冲区满了子进程会阻塞在写上，而表现是「它卡住 + 不动了」，从外面区分不出是卡在模型调用上还是卡在一行日志上。 """ - process = await asyncio.create_subprocess_exec( - *argv, - cwd=str(REPO_ROOT), - stdout=asyncio.subprocess.PIPE, - stderr=asyncio.subprocess.PIPE, - ) + if child_log_path is None: + sink: object = asyncio.subprocess.DEVNULL + handle = None + else: + child_log_path.parent.mkdir(parents=True, exist_ok=True) + handle = child_log_path.open("ab") + sink = handle + try: + process = await asyncio.create_subprocess_exec( + *argv, cwd=str(REPO_ROOT), stdout=sink, stderr=asyncio.subprocess.STDOUT + ) + finally: + if handle is not None: + handle.close() + deadline = time.monotonic() + timeout_s try: while True: if process.returncode is not None: - stdout, stderr = await process.communicate() - tail = stderr.decode("utf-8", errors="replace").strip().splitlines()[-3:] - del stdout + code = process.returncode + if code == CRASH_EXIT_CODE: + return _judge_crash( + log_path=log_path, + timing=timing, + after_steps=after_steps, + exit_code=code, + how=f"子进程按时机自杀（退出码 {code}）", + ) return KillOutcome( hit=False, reason=( - f"子进程在命中时机之前就退出了（退出码 {process.returncode}）" - + (f"，stderr 末尾：{' / '.join(tail)}" if tail else "") + f"子进程以退出码 {code} 结束，不是按时机自杀的 {CRASH_EXIT_CODE}" + + (f"，输出见 {child_log_path.name}" if child_log_path else "") ), model_calls=count_model_calls(read_log(log_path)), + exit_code=code, ) if should_kill(read_log(log_path), timing=timing, after_steps=after_steps): process.kill() await process.wait() - crashed = log_path.read_bytes() if log_path.is_file() else b"" - after = parse_terminated(crashed) - if not should_kill(after, timing=timing, after_steps=after_steps): - return KillOutcome( - hit=False, - reason="发信号与子进程停笔之间又写进了记录，杀掉之后时机已经不成立", - model_calls=count_model_calls(after), - ) - return KillOutcome( - hit=True, - reason=f"命中时机 {timing.value}", - snapshot=terminated_prefix(crashed), - steps=len(tagged(after, "step_completed")), - model_calls=count_model_calls(after), + return _judge_crash( + log_path=log_path, + timing=timing, + after_steps=after_steps, + exit_code=process.returncode, + how="父进程兜底发了 SIGKILL", ) if time.monotonic() > deadline: return KillOutcome( hit=False, - reason=f"等了 {timeout_s} 秒仍没命中时机 {timing.value}", + reason=f"等了 {timeout_s} 秒仍没崩在时机 {timing.value} 上", model_calls=count_model_calls(read_log(log_path)), ) await asyncio.sleep(poll_interval_s) @@ -910,20 +1055,31 @@ CRASH_PHASE = "execute" #: **由父进程用普通文件写入放进去，不走 `write_note`**：走工具的话审计账里会先躺一条记录， #: 而那条记录不是这次运行产生的，会把去重比对的基数弄脏。 #: -#: 正文全是自造的，一个字都不取自真实文书。 +#: **写得极其直白，只留一处候选、只留一条待办**：两种崩溃时机的触发条件都要求审计账已经非空， +#: 也就是模型必须先真的调过一次 `write_note`。计划写得散一点，模型会先检索几轮再动笔，六步的 +#: 预算跑完都还没写过笔记，那一类就只能报「无法判定」。这里不改 `govdoc.py` 的阶段提示词 +#: （那是场景层的东西），只改这份由本文件自己造的计划正文。 +#: +#: 行号是随手挑的一段正文，正文内容不在这里出现，也没有任何一个字取自真实文书。 SEEDED_PLAN_NAME = "plan.md" SEEDED_PLAN_TEXT = """# 审核计划（故障注入脚本预置） -候选证据两条，逐条核实： +候选证据只有一处，已经定位好了：**tender.md 第 880 到第 910 行**。 -1. tender.md 第 1 到第 60 行：项目基本信息与采购人信息，用来确认主体与项目编号。 -2. tender.md 第 200 到第 260 行：供应商资格条件，审核点多半落在这一段。 +只做两件事，做完就停： -核实完成后把逐条摘录写进 evidence.md，写完就停下。 +1. 用 read_document 读 tender.md 的第 880 到第 910 行。 +2. 立刻用 write_note 把上一步读到的原文逐字摘录写进 evidence.md，注明文档名与行号区间。 + +不要再检索别的关键词，不要再读别的段落——其余部分与本审核点无关。evidence.md 写完就停下。 """ #: 崩溃续跑这一路的预算。**比 GovDoc 场景自己那份（50 步）小得多**：这里要验的是崩溃与续跑 #: 的接缝，接缝在头几步就压得到，而每多一步都是一次真实的模型调用。 +#: +#: 六步是留给「读一段 + 写一次笔记」的余量：崩溃的触发条件要求模型先真的调过一次 `write_note` +#: （见 `SEEDED_PLAN_TEXT`），照计划走是第 2 步就写，留到六步是给它两三次走弯路的机会。跑到 +#: 预算用完仍然一次笔记都没写过，那一类报「无法判定」并说明原因，不假装验过。 CRASH_BUDGET = Budget( max_steps=6, max_actions=6, @@ -934,9 +1090,12 @@ CRASH_BUDGET = Budget( #: 模型绑定必须在父子两侧逐字相同——它整个进参数快照，差一个键续跑就报参数漂移。 CRASH_MODEL_BINDING: Mapping[str, str] = {"scenario": "govdoc", "phase": CRASH_PHASE} -#: 时机判定之前先等几步落地。等到第 2 步是为了让模型有机会走到 `write_note` 那一步，否则 -#: 审计账全程为空，最硬那条判据就只能报「无法判定」。 -CRASH_AFTER_STEPS = 2 +#: 时机判据里「至少已经落地几步」这一项。 +#: +#: **取 1，不再拿它当「等模型写笔记」的代理**：等笔记这件事现在由自杀条件本身负责（审计账 +#: 非空），而自杀点可能落在第 1 步的 `step_completed` 上。这里再要求两步的话，一次本来完全 +#: 正确的崩溃会被判成没命中。留着 1 是为了保证崩溃前至少有一步完整轨迹可供字节比对与续跑。 +CRASH_AFTER_STEPS = 1 APPWORLD_MODEL_BINDING: Mapping[str, str] = {"scenario": "appworld"} @@ -1029,7 +1188,7 @@ async def run_crash_fault( guard: CallGuard, attempts: int, ) -> FaultReport: - """一类崩溃续跑：起子进程 → 按时机 SIGKILL → 拷字节 → 同一个 run_id 续跑 → 逐条判。""" + """一类崩溃续跑：起子进程 → 它按时机自杀 → 拷字节 → 同一个 run_id 续跑 → 逐条判。""" task = load_govdoc_task(govdoc_db=govdoc_db, govdoc_corpus=govdoc_corpus) notes: list[str] = [] for attempt in range(1, attempts + 1): @@ -1055,18 +1214,31 @@ async def run_crash_fault( run_id, "--workspace", str(workspace), + "--timing", + timing.value, "--govdoc-db", str(govdoc_db), "--govdoc-corpus", str(govdoc_corpus), ] outcome = await spawn_and_kill( - argv=argv, log_path=log_path, timing=timing, after_steps=CRASH_AFTER_STEPS + argv=argv, + log_path=log_path, + timing=timing, + after_steps=CRASH_AFTER_STEPS, + child_log_path=runs_dir / f"{run_id}.child.log", ) guard.charge(outcome.model_calls) if not outcome.hit: + audit = read_audit_lines(workspace) + why_empty = ( + "；这次跑到结束都没调过一次 write_note，审计账是空的，所以自杀条件从来没满足过" + if not audit + else "" + ) notes.append( - f"第 {attempt} 次没命中时机：{outcome.reason}（花了 {outcome.model_calls} 次调用）" + f"第 {attempt} 次没崩在时机上：{outcome.reason}" + f"（花了 {outcome.model_calls} 次调用）{why_empty}" ) write_sidecars( runs_dir=runs_dir, @@ -1084,7 +1256,10 @@ async def run_crash_fault( ) continue - notes.append(f"第 {attempt} 次命中时机 {timing.value}，崩溃时 {outcome.steps} 步") + notes.append( + f"第 {attempt} 次{outcome.reason}，崩溃时 {outcome.steps} 步、" + f"审计账 {len(read_audit_lines(workspace))} 条" + ) return await _resume_and_judge( fault=fault, timing=timing, @@ -1103,8 +1278,8 @@ async def run_crash_fault( fault=fault, criteria=( undetermined( - "kill_timing_hit", - f"{attempts} 次都没能把子进程杀在时机 {timing.value} 上，这一类什么都没验成", + "crash_timing_hit", + f"{attempts} 次都没能让子进程崩在时机 {timing.value} 上，这一类什么都没验成", ), ), notes=tuple(notes), @@ -1468,9 +1643,13 @@ async def run_parse_failure_fault( async def run_child(args: argparse.Namespace) -> int: - """子进程：跑一次 GovDoc execute 阶段，等着被父进程杀掉。 + """子进程：跑一次 GovDoc execute 阶段，跑到指定时机就把自己打死。 - 正常跑完也是允许的——那时父进程会报「没命中时机」并重试，不会把它当成命中。 + **打死自己这件事由 `SelfKillingStore` 在写入落盘之后做**，不在这里判——判据要贴着那次写 + 才准，隔一层就又变成抢窗口了。它调的是 `os._exit`，所以这个函数在命中时机时根本不返回， + 下面的 `client.aclose()` 与 `finally` 都不会跑。**那正是要的**：崩溃就是不给清理机会。 + + 正常跑完也是允许的（时机一次都没触发），那时父进程会报「没崩在时机上」并重试。 **事件写在 `.child.events.jsonl`，不写记分板认的那个名字。** 事件只在「本进程里 真的走完」的步上发，而记分板按 `meta.resumed_from_step` 把续跑跳过的那一段减掉之后与 @@ -1483,7 +1662,11 @@ async def run_child(args: argparse.Namespace) -> int: runs_dir = Path(args.runs_dir) workspace = Path(args.workspace) task = load_govdoc_task(govdoc_db=Path(args.govdoc_db), govdoc_corpus=Path(args.govdoc_corpus)) - store = JsonlRunStore(directory=runs_dir) + store = SelfKillingStore( + inner=JsonlRunStore(directory=runs_dir), + timing=KillTiming(args.timing), + workspace=workspace, + ) sink = JsonlEventSink(runs_dir / f"{args.run_id}.child.events.jsonl") client = GatewayClient.from_env() try: @@ -1533,6 +1716,11 @@ def build_parser() -> argparse.ArgumentParser: ) parser.add_argument("--run-id", help="内部用：子进程的运行标识") parser.add_argument("--workspace", help="内部用：子进程的工作区目录") + parser.add_argument( + "--timing", + choices=[item.value for item in KillTiming], + help="内部用：子进程在哪一次写入之后把自己打死", + ) return parser @@ -1712,7 +1900,7 @@ def main(argv: Sequence[str] | None = None) -> int: if args.child: missing = [ name - for name in ("run_id", "workspace", "govdoc_db", "govdoc_corpus") + for name in ("run_id", "workspace", "timing", "govdoc_db", "govdoc_corpus") if not getattr(args, name) ] if missing: @@ -1732,6 +1920,7 @@ if __name__ == "__main__": __all__ = [ "APPWORLD_FAULTS", "CRASH_BUDGET", + "CRASH_EXIT_CODE", "FAULT_NAMES", "GOVDOC_FAULTS", "AlwaysInvalidParser", @@ -1744,6 +1933,7 @@ __all__ = [ "KillOutcome", "KillTiming", "LogRead", + "SelfKillingStore", "check_all_steps_parse_failed", "check_audit_unchanged", "check_cancelled_raised", diff --git a/tools/soak/tests/test_faults.py b/tools/soak/tests/test_faults.py index 396d815..9d55e24 100644 --- a/tools/soak/tests/test_faults.py +++ b/tools/soak/tests/test_faults.py @@ -6,8 +6,12 @@ 违反的输入验它说击穿。 子进程编排那部分拆出了两个纯函数（`should_kill` 判时机到没到、`terminated_prefix` 截已终结 -前缀），它们不碰进程也不碰模型，直接单独测。真起子进程那两条用的是一个只会往文件里写几行 +前缀），它们不碰进程也不碰模型，直接单独测。真起子进程那几条用的是一个只会往文件里写几行 JSON 的假子进程，跑完不到两秒。 + +确定性自杀那条路（`SelfKillingStore`）的测试**把 `os._exit` 换成一个抛哨兵异常的替身**：真调 +`os._exit` 会把跑测试的 pytest 进程一起带走，整次收集连一行结果都留不下。替身同时让「死之前 +那条记录有没有先写进内层存储」变得可断言——顺序反了的话，崩溃现场就少一条本该已经落地的记录。 """ from __future__ import annotations @@ -19,9 +23,24 @@ from pathlib import Path import pytest -from polyloop.ports import EventKind, InvalidDecision -from polyloop.types import ModelReply, RunResult, StopReason +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, @@ -30,6 +49,7 @@ from tools.soak.faults import ( JsonlEventSink, KillTiming, LogRead, + SelfKillingStore, build_parser, check_all_steps_parse_failed, check_audit_unchanged, @@ -59,6 +79,7 @@ from tools.soak.faults import ( terminated_prefix, write_sidecars, ) +from tools.soak.scenarios.govdoc import AUDIT_LOG_NAME RUN_ID = "fault-test-0" @@ -596,11 +617,195 @@ def test_should_kill_on_empty_log() -> None: # --------------------------------------------------------------------------- -# 十二、子进程编排：真起一个假子进程 +# 十二、确定性自杀：写入落盘之后按时机把自己打死 +# --------------------------------------------------------------------------- + + +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: + (workspace / AUDIT_LOG_NAME).write_text("write_note\tevidence.md\taaa\n", encoding="utf-8") + 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] @@ -613,6 +818,7 @@ with open(path, "a", encoding="utf-8") as handle: time.sleep(30) """ +#: 写完就正常退出，一次都不自杀。 _FAKE_CHILD_EXITS = """ import json, sys path = sys.argv[1] @@ -622,8 +828,72 @@ with open(path, "a", encoding="utf-8") as handle: handle.write(json.dumps(record) + "\\n") """ +#: 写完就按崩溃退出码把自己打死，模拟 `SelfKillingStore` 那条主路径。 +_FAKE_CHILD_SELF_KILL = """ +import json, os, 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") + handle.flush() +print("子进程说了句话") +sys.stdout.flush() +os._exit(int(sys.argv[3])) +""" -async def test_spawn_and_kill_hits_the_after_step_timing(tmp_path: Path) -> None: + +async def test_spawn_and_kill_accepts_a_self_killed_child(tmp_path: Path) -> None: + """主路径：子进程按崩溃退出码自杀，且日志尾部形态对得上，算命中。""" + log_path = tmp_path / f"{RUN_ID}.jsonl" + records = [intent(), model_result(), step_completed()] + outcome = await spawn_and_kill( + argv=[ + sys.executable, + "-c", + _FAKE_CHILD_SELF_KILL, + str(log_path), + json.dumps(records), + str(CRASH_EXIT_CODE), + ], + log_path=log_path, + 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_rejects_a_self_kill_at_the_wrong_tail(tmp_path: Path) -> None: + """自杀了但尾部形态不对：报没命中，不许因为退出码对就当成命中。""" + log_path = tmp_path / f"{RUN_ID}.jsonl" + records = [intent(), model_result()] + outcome = await spawn_and_kill( + argv=[ + sys.executable, + "-c", + _FAKE_CHILD_SELF_KILL, + str(log_path), + json.dumps(records), + str(CRASH_EXIT_CODE), + ], + log_path=log_path, + 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_still_falls_back_to_sigkill(tmp_path: Path) -> None: + """兜底路径没删：子进程一直不自杀时，父进程仍然会在尾部形态对上的那一刻杀掉它。""" log_path = tmp_path / f"{RUN_ID}.jsonl" records = [intent(), model_result(), step_completed()] outcome = await spawn_and_kill( @@ -634,13 +904,14 @@ async def test_spawn_and_kill_hits_the_after_step_timing(tmp_path: Path) -> None timeout_s=20.0, ) assert outcome.hit is True + assert "兜底" in outcome.reason assert outcome.steps == 1 assert outcome.model_calls == 1 assert outcome.snapshot == log_path.read_bytes() -async def test_spawn_and_kill_reports_a_miss_when_the_child_exits(tmp_path: Path) -> None: - """子进程在命中时机之前就退出了要报出来，不许悄悄当成命中。""" +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" records = [intent(), model_result()] outcome = await spawn_and_kill( @@ -651,7 +922,8 @@ async def test_spawn_and_kill_reports_a_miss_when_the_child_exits(tmp_path: Path timeout_s=20.0, ) assert outcome.hit is False - assert "退出" in outcome.reason + 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: @@ -664,11 +936,11 @@ async def test_spawn_and_kill_reports_a_miss_on_timeout(tmp_path: Path) -> None: timeout_s=0.5, ) assert outcome.hit is False - assert "没命中时机" in outcome.reason + assert "仍没崩在时机" in outcome.reason # --------------------------------------------------------------------------- -# 十三、事件出口、护栏、sidecar、命令行 +# 十四、事件出口、护栏、sidecar、命令行 # ---------------------------------------------------------------------------