From 5a4d13b0f2464d3c1fffe90a38043ca2b425181f Mon Sep 17 00:00:00 2001 From: iomgaa Date: Tue, 11 Aug 2026 11:53:00 -0400 Subject: [PATCH] =?UTF-8?q?fix(soak):=20=E6=8C=89=20Codex=20=E5=AF=B9?= =?UTF-8?q?=E6=8A=97=E5=AE=A1=E6=9F=A5=E5=8A=A0=E5=9B=BA=E6=95=85=E9=9A=9C?= =?UTF-8?q?=E6=B3=A8=E5=85=A5=E7=9A=84=E5=88=A4=E6=8D=AE?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 九类实跑全过之后让 Codex 专门找「判据其实验不到它声称要验的东西」，它报的每条都带具体 失败场景。这次的绿不是假的——实跑数据里这些判据都有料可判——但它们在别的输入下会假绿。 真空成立三处改成无法判定：空日志的「全都解得开」、零步的「没有 env_error」、前后都空的 「审计账没变」。上一版实跑里最后那条正是 0 vs 0 通过的。 「环境是在完整一步之后坏的」原本只有通过和无法判定两档，没有击穿分支——那条判的是注入器 自己，它对外宣称在第 N 次执行之后动手，整类故障的结论都建立在这句话上。 提示词那两条原本只读库写进日志的数，而这套东西反复强调不能拿库的自述验库的行为——它在 自己的核心主张上破了例。现在包一层模型客户端记下每次真正发出去的消息字符数，跑完逐步 对账。字符数口径与库的算法对拍过，不然会全程假击穿。 绝不重放那条补了两个角度：崩溃时账上那几条续跑之后要逐行原样还在（旧条目被改写、被抹掉 原来一路绿灯），以及续跑段里不许出现与崩溃前完全相同的条目。Codex 提的「同一文件不同 内容」在 b 档已被条数判据挡住，a 档不能加同样的规则——那一档续跑本来就该接着跑，模型 再写一次是正常行为，加了会变成随机红。 取消补了环境侧的静默判据与事件文件完整性。容器里已经发出去的执行在客户端计数上不留痕， 这个盲区如实写进已知缺口，没假装验到；崩溃形态只能落在写入调用返回之后，同样记下来。 Co-Authored-By: Claude Opus 5 (1M context) --- tools/soak/faults.py | 342 ++++++++++++++++++++++++++++++-- tools/soak/tests/test_faults.py | 310 ++++++++++++++++++++++++++++- 2 files changed, 633 insertions(+), 19 deletions(-) diff --git a/tools/soak/faults.py b/tools/soak/faults.py index 0a9d6a2..20fec4f 100644 --- a/tools/soak/faults.py +++ b/tools/soak/faults.py @@ -39,6 +39,26 @@ GovDoc 那一路把 `max_prompt_chars` 压到刚好等于装配出来的初始 缺 sidecar——被 SIGKILL 的子进程来不及写，记分板会报「无法判定」，那是对的，不为了让它变绿 去补假数据。 +## 已知缺口 + +这两条是这套注入器**造不出来**的情形。写在这里是因为「没被验到」与「验过了没问题」在报告上 +长得一样，而报告只列跑过的判据，列不出没跑的那些。 + +**一、崩溃形态只有「写入调用返回之后」这一种。** `os._exit` 是一次系统调用，它只能发生在两条 +Python 语句之间；进程内的自杀不可能停在 `write(2)` 的中途。所以「一行只写了半截就断电」 +「记录进了页缓存但 `fsync` 之前机器崩了」这两种形态在这里造不出来——真要造得靠外部手段（掐 +电、`dm-flakey` 这类会掉写的块设备、或者在文件系统层注入故障），那是另一套东西。现在靠的是 +记分板那侧的撕裂行判据：它按「有没有被换行终结」切行，末尾那段没终结的字节按契约算没发生过 +（`tools/soak/scoreboard.py` 的 `_split_terminated`）。也就是说这种日志**读得对**是验过的， +**写出来会怎样**没验过。 + +**二、取消之后容器里那次执行的下落看不见。** `AppWorldSession.n_executions` 是客户端侧的 +计数，HTTP 响应返回之后才加一，而取消把那个协程掐断了。于是「代码已经发到容器、容器仍然把它 +跑完并改了环境状态」这种情形留不下痕迹。容器那侧没有可查的执行计数接口（只有 `/execute`、 +`/task_completed`、`/evaluate`、`/close`），补上它要改 `tools/soak/appworld.py` 那一层。 +`check_env_quiet_after_cancel` 覆盖的是另外两种坏法：库在取消之后还往环境派活，或者有一次 +执行在取消之后才落地。 + 跑法（会打真实模型、会花钱）:: PYTHONUNBUFFERED=1 conda run --live-stream -n PolyLoop python -m tools.soak.faults \\ @@ -315,12 +335,21 @@ def check_log_readable(read: LogRead) -> Criterion: 解不出来的行让下面几条判据判的东西缺一块，所以它先判——不先判的话，一份少了半截的日志 会让「步号连续」这类判据在残缺数据上给出「通过」。 + + **一条记录都没有时报「无法判定」，不报通过。** 日志文件根本不存在时 `read_log()` 返回的 + 就是这种空读数，而「零条全都解得开」是一句真空成立的话：它读起来像验过了，实际那次跑 + 连一个字节都没落盘，后面每一条判据都建立在一份不存在的日志上。 """ if read.bad_lines: return breached( "log_lines_readable", f"第 {list(read.bad_lines)} 行已被换行终结却解不出带 {RECORD_KEY!r} 标签的 JSON 对象", ) + if not read.payloads: + return undetermined( + "log_lines_readable", + "日志里一条记录都没有（文件不存在，或者一个字节都没落盘），行的完整性无从判起", + ) return passed("log_lines_readable", f"{len(read.payloads)} 条记录全部解得开") @@ -482,7 +511,16 @@ def check_audit_unchanged(*, before: Sequence[str], after: Sequence[str]) -> Cri 时机 B 下库判定状态未知、干净停下，那就不该把被打断的那个动作再执行一遍。多出一条就是 重放，少一条或者变了内容说明账被改写过。 + + **两边都是空的时候报「无法判定」。** 那代表续跑前后都没有任何副作用被记录过，「一条都没多」 + 就成了一句真空成立的话——而这一档要验的恰恰是「已经落过盘的那次副作用没有再发生一次」， + 没有那次副作用就没有要验的东西。上一版实跑里这一条正是 0 比 0 通过的。 """ + if not before and not after: + return undetermined( + "audit_unchanged_after_resume", + "续跑前后审计账都是空的：这次运行里没有任何副作用被记录过，「没多出来」真空成立", + ) if list(after[: len(before)]) != list(before): return breached( "audit_unchanged_after_resume", @@ -496,6 +534,69 @@ def check_audit_unchanged(*, before: Sequence[str], after: Sequence[str]) -> Cri return passed("audit_unchanged_after_resume", f"续跑前后审计账都是 {len(before)} 条") +def check_audit_prefix_preserved(*, before: Sequence[str], after: Sequence[str]) -> Criterion: + """崩溃时环境账上那几条，续跑之后逐字节还在原处。 + + 这是日志那侧 `check_crash_prefix_preserved` 在**环境账**上的对应物，而且两者不可互相 + 替代：日志是库自己写的，账是环境写的。库把已经落地的一段轨迹重写掉，日志那条会报；库让 + 环境把已经发生过的副作用重做或抹掉，只有这条会报。 + + **时机 A 之前没有这条判据。** 那一档只跑「去重前后条数相等」，而一次把 `evidence.md` 的 + 旧条目改写成别的内容的续跑，条数不变、去重也不变，一路绿灯。 + """ + name = "audit_prefix_preserved" + if not before: + return undetermined(name, "崩溃时审计账是空的：没有已经落地的副作用可比，前缀比对无从谈起") + if len(after) < len(before): + return breached( + name, + f"崩溃时审计账 {len(before)} 条，续跑之后只剩 {len(after)} 条，已经落地的记录被抹掉了", + ) + for index, (old, new) in enumerate(zip(before, after, strict=False)): + if old != new: + return breached( + name, f"崩溃时审计账的第 {index} 条在续跑之后变了内容（按行序计，正文不进报告）" + ) + return passed(name, f"崩溃时审计账的 {len(before)} 条在续跑之后逐字节还在原处") + + +def check_no_cross_segment_replay(*, before: Sequence[str], after: Sequence[str]) -> Criterion: + """崩溃前已经落地的那次副作用，续跑之后没有再发生一次。 + + 与 `check_never_action_not_replayed` 的区别在**位置**：那一条在整份账上找重复，找到了也 + 说不出两条分别在崩溃的哪一侧；这一条只看续跑段里有没有哪一条与崩溃前的某一条完全相同， + 而跨越崩溃边界的重复正是重放的签名——库在状态未知时把一次已经落过盘的写又执行了一遍。 + + **它与那一条共享同一个已知假阳性**：模型自己在续跑之后把同一份内容原样又写了一次，看起来 + 一模一样。GovDoc 的 execute 阶段提示词要求写完就停，所以很少发生；真发生了要看轨迹里那两步 + 的步号。这个代价是被接受的，理由与那一条相同。 + """ + name = "no_cross_segment_replay" + if not before: + return undetermined( + name, "崩溃时审计账是空的：没有已经落地的副作用可能被重放，这一条无从判起" + ) + resumed = list(after[len(before) :]) + if not resumed: + return passed(name, f"续跑段里一条审计记录都没有，崩溃前那 {len(before)} 条不可能被重放过") + known = {parse_audit_line(line) for line in before} + repeats: list[str] = [] + for line in resumed: + entry = parse_audit_line(line) + if entry is not None and entry in known: + repeats.append(f"{entry[0]} 摘要 {entry[2]}") + if repeats: + return breached( + name, + f"续跑段的 {len(resumed)} 条记录里有 {len(repeats)} 条与崩溃前的某一条完全相同：" + f"{'；'.join(sorted(set(repeats)))}", + ) + return passed( + name, + f"续跑段新增 {len(resumed)} 条审计记录，没有一条与崩溃前那 {len(before)} 条中的任何一条相同", + ) + + def check_resume_made_progress(*, crashed_steps: int, final_steps: int) -> Criterion: """时机 A 下续跑真的接着往下跑了。 @@ -586,8 +687,15 @@ def check_no_env_error_step(read: LogRead) -> Criterion: 撞预算那两条要的是「干净地撞上限」。中途出过环境故障的话，步数与动作数的账仍然对得上， 但这次跑压到的已经不是预算这条路径了。 + + **一步都没产出时报「无法判定」**：「零步里没有 env_error」这句话恒为真，而它出现在报告上 + 的样子和「跑了十步都没出故障」一模一样。 """ entries = tagged(read, "step_completed") + if not entries: + return undetermined( + "no_env_error_step", "日志里一条步记录都没有，有没有环境故障的步无从判起" + ) bad: list[int] = [] for index, item in enumerate(entries): outcome = item.get("action_outcome") @@ -784,24 +892,39 @@ def check_steps_before_last_all_executed(read: LogRead) -> Criterion: return passed(name, f"环境坏掉之前的 {len(steps) - 1} 步动作状态全是 executed") -def check_env_broken_after_a_full_step(*, broken: bool, executions_before_break: int) -> Criterion: - """动手弄坏环境之前，至少已经完整执行过一次动作。 +def check_env_broken_after_a_full_step( + *, broken: bool, executions_before_break: int, expected_after: int +) -> Criterion: + """动手弄坏环境的时机正是声明的那一次：完整执行过 `expected_after` 次动作之后。 第一次执行之前就把容器打掉的话，验的是「环境起不来」而不是「跑到一半环境不能接着服务 了」——前者落在会话初始化上，根本走不到动作执行接缝，而这一类要压的正是那个接缝。 + + **动手时机与声明的不一致是击穿，不是无法判定。** 这一条判的是注入器自己：它对外宣称在第 + `expected_after` 次执行之后动手，整类故障的结论都建立在这句话上。真在别的时刻动手的话， + 这一类压到的是另一件事，而报告上仍然写着「环境故障那一类通过了」——那正是「判据验的不是 + 它声称要验的东西」这种坏法。判成无法判定则把这件事说成「看不清」，可它看得很清楚：注入器 + 自己记下了动手时它数到几。 + + `broken` 为假是另一回事，那一档确实什么都没发生（模型一次动作都没成功执行过），报无法判定。 """ name = "env_broken_after_a_full_step" if not broken: return undetermined( name, "这次运行结束时环境一次都没被弄坏过：动作执行没走到该动手的那一次" ) - if executions_before_break < 1: - return undetermined( + if executions_before_break != expected_after: + return breached( name, - f"弄坏环境之前只完整执行过 {executions_before_break} 次动作，" - "压到的是初始化而不是执行接缝", + f"声明的是完整执行 {expected_after} 次动作之后动手，实际在第 " + f"{executions_before_break} 次之后动手：" + + ( + "压到的是会话初始化而不是动作执行接缝" + if executions_before_break < expected_after + else "比声明的晚，这一类压到的不是它声称的那个时刻" + ), ) - return passed(name, f"完整执行过 {executions_before_break} 次动作之后才把环境弄坏") + return passed(name, f"完整执行过 {executions_before_break} 次动作之后才把环境弄坏，与声明一致") def check_all_steps_parse_failed(read: LogRead) -> Criterion: @@ -855,6 +978,91 @@ def check_cancelled_raised(*, raised: BaseException | None) -> Criterion: return passed("cancelled_error_propagated", "取消之后 await 原样抛出了 CancelledError") +def check_env_quiet_after_cancel( + *, + dispatched_before: int, + dispatched_after: int, + completed_before: int, + completed_after: int, + waited_s: float, +) -> Criterion: + """取消之后环境侧不再有动静：动作执行接缝没被再进入，也没有哪次执行又跑完了。 + + `dispatched_*` 数的是动作执行接缝被**进入**过几次（由 `TriggeringExecutor` 记）， + `completed_*` 数的是环境侧确认**跑完**过几次（`AppWorldSession.n_executions`）。前者涨了 + 说明库在取消之后还在往环境派活；后者涨了说明有一次执行在取消之后才落地。两个数分别对应 + 两种不同的坏法，所以一起判。 + + **已知看不见的那一半**：`n_executions` 是客户端侧的计数，`tools/soak/appworld.py` 在 + HTTP 响应返回之后才加一。取消把那个协程掐断了，于是「代码已经发到容器、容器仍然把它跑完 + 并改了环境状态」这种情形在这两个数上都留不下痕迹。容器那侧没有可查的执行计数接口 + （只有 `/execute`、`/task_completed`、`/evaluate`、`/close`），补上它要改环境层。这条缺口 + 写在模块 docstring 的已知缺口一节里，这里的「通过」只覆盖上面那两种坏法。 + """ + name = "env_quiet_after_cancel" + if dispatched_after > dispatched_before: + return breached( + name, + f"取消之后动作执行接缝又被进入了 {dispatched_after - dispatched_before} 次" + f"（等了 {waited_s} 秒再数的）：库在取消之后还在往环境派活", + ) + if completed_after > completed_before: + return breached( + name, + f"取消之后又有 {completed_after - completed_before} 次执行在环境侧跑完了" + f"（等了 {waited_s} 秒再数的）", + ) + if completed_after < completed_before or dispatched_after < dispatched_before: + return breached( + name, + f"取消之后计数倒退了（派活 {dispatched_before}→{dispatched_after}，" + f"跑完 {completed_before}→{completed_after}），这两个数只该单调不减", + ) + return passed( + name, + f"取消之后等了 {waited_s} 秒，派活次数仍是 {dispatched_after}、" + f"环境侧跑完次数仍是 {completed_after}", + ) + + +def check_events_file_intact(path: Path) -> Criterion: + """事件文件的每一行都是完整的一条 JSON 对象。 + + 取消可能落在事件写盘的途中。**撕裂尾行在这里算击穿，不像日志那侧算「没发生过」**：库有 + 宽限期（`RunRequest.cancel_grace_seconds`），在飞的写有机会收尾，而事件出口每条只写一次 + `handle.write`。真留下半行，说明取消穿过了一次没有被保护起来的写——那种写下次也会半途而废， + 而下游读事件流时看到的是一份解不开的文件。 + """ + name = "events_file_intact" + if not path.is_file(): + return undetermined(name, f"没有 {path.name}：这次运行一条事件都没发出过，完整性无从判起") + raw = path.read_bytes() + if not raw: + return undetermined(name, f"{path.name} 是空的，完整性无从判起") + chunks = raw.split(b"\n") + if chunks[-1].strip(): + return breached( + name, + f"{path.name} 末尾有 {len(chunks[-1])} 字节没有被换行终结：取消穿过了一次事件写入的中途", + ) + bad: list[int] = [] + total = 0 + for number, chunk in enumerate(chunks[:-1], start=1): + if not chunk.strip(): + continue + total += 1 + try: + payload = json.loads(chunk) + except (UnicodeDecodeError, json.JSONDecodeError): + bad.append(number) + continue + if not isinstance(payload, dict): + bad.append(number) + if bad: + return breached(name, f"{path.name} 的第 {bad} 行不是一个完整的 JSON 对象") + return passed(name, f"{path.name} 的 {total} 行全都是完整的 JSON 对象") + + def check_lease_returned(*, borrowed: bool, timeout_s: float, pool_size: int) -> Criterion: """容器租约被归还：取消结束之后还借得到。 @@ -1230,6 +1438,79 @@ class AlwaysInvalidParser: ) +class PromptSizeRecordingClient: + """转发模型调用，并把**每次真正发出去的那份消息**有多少字符按调用序号记下来。 + + 没有它的话，提示词那两条判据读的全是库写进步记录的 `prompt_chars`——那是库对自己行为的 + 陈述。失败场景很具体：库真的把历史静默截断了，同时仍然把 `prompt_chars` 记成一路不减的 + 自述值，两条判据照样全绿。这与「动作那一维要数环境侧的账、不数库报的步数」是同一条原则， + 这里把它落在提示词这一维上。 + + 字符数的口径逐字照 `polyloop._assembly.prompt_chars`：所有消息的所有内容块的文本长度之和。 + 口径不同的话对不上账，而对不上会被判成击穿，那是一次假故障。 + + **`parameters()` 原样转发内层的**，不加自己的键：它进参数快照，续跑时那边用的是没包过的 + 客户端，加一个键就是一次假的参数漂移。 + """ + + __slots__ = ("_inner", "chars_by_call_index") + + def __init__(self, *, inner: object) -> None: + self._inner = inner + #: 调用序号 → 那次调用真正发出去的字符数。 + self.chars_by_call_index: dict[int, int] = {} + + def parameters(self) -> Mapping[str, str]: + return self._inner.parameters() # type: ignore[attr-defined] + + async def call(self, call: object) -> ModelReply: + self.chars_by_call_index[call.call_index] = sum( # type: ignore[attr-defined] + len(block.text) + for message in call.messages # type: ignore[attr-defined] + for block in message.content + ) + return await self._inner.call(call) # type: ignore[attr-defined] + + +def check_prompt_chars_match_what_was_sent( + read: LogRead, *, observed: Mapping[int, int] +) -> Criterion: + """步记录里的 `prompt_chars` 与库那一步真正发出去的字符数逐步相等。 + + 这一条是提示词那两条判据的地基：它们都只读 `prompt_chars`，而 `prompt_chars` 是库自己写 + 下的数。库把历史截断了却照旧记一个单调递增的值，那两条都会通过，而这一条会当场对不上账。 + + `observed` 由 `PromptSizeRecordingClient` 在调用发生的那一刻记下，键是调用序号；步记录的 + 步号与调用序号是同一个数（`polyloop._recovery` 模块 docstring 那条对齐），所以直接按步号查。 + """ + name = "prompt_chars_match_what_was_sent" + steps = _step_payloads(read) + if not steps: + return undetermined(name, "日志里一条步记录都没有，对不了账") + if not observed: + return undetermined(name, "一次模型调用都没被记到，对不了账") + mismatches: list[str] = [] + compared = 0 + for step in steps: + index = step.get("step_idx") + if not isinstance(index, int) or isinstance(index, bool) or index not in observed: + continue + logged = step.get("prompt_chars") + if not isinstance(logged, int) or isinstance(logged, bool): + mismatches.append(f"第 {index} 步的 prompt_chars 不是整数") + continue + compared += 1 + if logged != observed[index]: + mismatches.append( + f"第 {index} 步记的是 {logged} 字符，实际发出去 {observed[index]} 字符" + ) + if mismatches: + return breached(name, "；".join(mismatches)) + if compared == 0: + return undetermined(name, "没有一条步记录的步号能和记下来的调用序号对上，对不了账") + return passed(name, f"{compared} 步的 prompt_chars 与真正发出去的字符数逐步相等") + + class TriggeringModelClient: """转发模型调用，并在进入第 N 次调用时通知父协程。父协程收到通知立刻取消。 @@ -1856,6 +2137,8 @@ async def _resume_and_judge( criteria.append(check_step_indices_dense(read)) criteria.append(check_intents_settled(read)) criteria.append(check_never_action_not_replayed(audit_after)) + criteria.append(check_audit_prefix_preserved(before=audit_before, after=audit_after)) + criteria.append(check_no_cross_segment_replay(before=audit_before, after=audit_after)) if timing is KillTiming.AFTER_STEP: criteria.append( check_resume_made_progress( @@ -1923,7 +2206,10 @@ async def run_context_overflow_fault( request = replace(base, budget=budget) store = JsonlRunStore(directory=runs_dir) sink = JsonlEventSink(runs_dir / f"{run_id}.events.jsonl") - definition = build_crash_definition(model_client=model_client, store=store, sink=sink) + # 提示词那几条判据全部读库自己写下的 `prompt_chars`。包一层把真正发出去的规模记下来， + # 跑完对账——不对账的话，一次静默截断加一列伪造的自述值可以让它们全绿。 + recorder = PromptSizeRecordingClient(inner=model_client) + definition = build_crash_definition(model_client=recorder, store=store, sink=sink) started = time.monotonic() result = await session.run(definition, request) @@ -1936,6 +2222,7 @@ async def run_context_overflow_fault( check_run_finished_present(read), check_stop_reason(read, "context_overflow"), check_at_least_one_step(read), + check_prompt_chars_match_what_was_sent(read, observed=recorder.chars_by_call_index), check_prompt_chars_monotonic(read), check_prompt_reached_max_prompt_chars(read), ) @@ -2005,6 +2292,12 @@ LEASE_TIMEOUT_S = 120.0 #: 等取消触发器的上限。等不到说明这次运行在触发点之前就结束了。 TRIGGER_TIMEOUT_S = 300.0 +#: 取消回来之后再等几秒，然后重数一次环境侧的账。见 `check_env_quiet_after_cancel`：立刻就数 +#: 的话两个数只是同一个瞬间的两份拷贝，什么都验不到。三秒盖得住一次已经发出去的 `/execute` +#: 在容器里跑完并返回——AppWorld 那侧单次执行的超时是一百秒，所以盖不住最慢的那一档；这一路 +#: 的动作都是简短的 API 调用，实测在秒级以内。 +POST_CANCEL_SETTLE_S = 3.0 + def _appworld_definition( *, model_client: object, store: JsonlRunStore, sink: JsonlEventSink, parser: object @@ -2046,6 +2339,7 @@ async def run_cancel_fault( notes: list[str] = [] raised: BaseException | None = None env_executions = 0 + quiet: Criterion | None = None started = time.monotonic() async with pool.session(task_id) as handle: @@ -2055,11 +2349,10 @@ async def run_cancel_fault( app_descriptions=app_descriptions, model_binding=APPWORLD_MODEL_BINDING, ) + probe: TriggeringExecutor | None = None if fault == "cancel_env": - request = replace( - request, - action_executor=TriggeringExecutor(inner=request.action_executor, trigger=trigger), # type: ignore[arg-type] - ) + probe = TriggeringExecutor(inner=request.action_executor, trigger=trigger) + request = replace(request, action_executor=probe) # type: ignore[arg-type] client: object = model_client else: # 第 1 次（从 0 数起的第二次）调用时触发：让日志里先有一步完整的记录，取消才落在 @@ -2083,6 +2376,19 @@ async def run_cancel_fault( except asyncio.CancelledError as exc: raised = exc env_executions = handle.n_executions + if probe is not None: + # 取消刚回来就数一次，等一小会儿再数一次。在飞的那次执行要是没被真正掐断，它会在 + # 这段等待里跑完并把计数推上去——立刻就数的话，两个数只是同一个瞬间的两份拷贝。 + dispatched_before, completed_before = probe.executions, handle.n_executions + await asyncio.sleep(POST_CANCEL_SETTLE_S) + quiet = check_env_quiet_after_cancel( + dispatched_before=dispatched_before, + dispatched_after=probe.executions, + completed_before=completed_before, + completed_after=handle.n_executions, + waited_s=POST_CANCEL_SETTLE_S, + ) + env_executions = handle.n_executions wall_ms = int((time.monotonic() - started) * 1000) borrowed = await _can_borrow(pool, task_id, LEASE_TIMEOUT_S) @@ -2093,7 +2399,9 @@ async def run_cancel_fault( check_log_readable(read), check_cancelled_raised(raised=raised), check_stop_reason(read, "cancelled"), + check_events_file_intact(runs_dir / f"{run_id}.events.jsonl"), check_lease_returned(borrowed=borrowed, timeout_s=LEASE_TIMEOUT_S, pool_size=1), + *((quiet,) if quiet is not None else ()), ) write_sidecars( runs_dir=runs_dir, @@ -2326,7 +2634,9 @@ async def run_env_error_fault( ), check_steps_before_last_all_executed(read), check_env_broken_after_a_full_step( - broken=breaker.broken, executions_before_break=breaker.executions_before_break + broken=breaker.broken, + executions_before_break=breaker.executions_before_break, + expected_after=ENV_ERROR_BREAK_AFTER_EXECUTIONS, ), ) write_sidecars( @@ -2703,6 +3013,7 @@ __all__ = [ "FaultInjectionError", "FaultReport", "JsonlEventSink", + "PromptSizeRecordingClient", "KillOutcome", "KillTiming", "LogRead", @@ -2710,11 +3021,14 @@ __all__ = [ "build_context_overflow_budget", "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", @@ -2722,9 +3036,11 @@ __all__ = [ "check_lease_returned", "check_log_readable", "check_never_action_not_replayed", + "check_no_cross_segment_replay", "check_no_env_error_step", "check_prompt_chars_monotonic", "check_prompt_reached_max_prompt_chars", + "check_prompt_chars_match_what_was_sent", "check_resume_made_progress", "check_run_finished_present", "check_step_count", diff --git a/tools/soak/tests/test_faults.py b/tools/soak/tests/test_faults.py index 78f2c48..fbf3c8d 100644 --- a/tools/soak/tests/test_faults.py +++ b/tools/soak/tests/test_faults.py @@ -61,16 +61,20 @@ from tools.soak.faults import ( 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, @@ -78,7 +82,9 @@ from tools.soak.faults import ( 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, @@ -1573,23 +1579,42 @@ def test_steps_before_last_all_executed_is_undetermined_with_a_single_step() -> def test_env_broken_after_a_full_step_passes() -> None: - outcome = check_env_broken_after_a_full_step(broken=True, executions_before_break=1) + 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) + 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_is_undetermined_when_it_broke_too_early() -> None: - """第一次执行之前就打死容器的话，压到的是初始化而不是动作执行接缝。""" - outcome = check_env_broken_after_a_full_step(broken=True, executions_before_break=0) - assert outcome.status is CriterionStatus.UNDETERMINED +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 + + # --------------------------------------------------------------------------- # 十八、环境故障：杀容器那段编排里的纯函数与执行器包装 # --------------------------------------------------------------------------- @@ -1727,3 +1752,276 @@ def test_parser_accepts_the_new_faults() -> None: ] ) 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