From 6af289d28305b27dae7065fc1401b5863dfac620 Mon Sep 17 00:00:00 2001 From: iomgaa Date: Mon, 10 Aug 2026 03:38:31 -0400 Subject: [PATCH] =?UTF-8?q?feat(stores):=20=E8=90=BD=E6=88=90=E9=80=90?= =?UTF-8?q?=E8=A1=8C=E8=BF=BD=E5=8A=A0=E7=9A=84=E6=97=A5=E5=BF=97=E5=AD=98?= =?UTF-8?q?=E5=82=A8=EF=BC=8C=E5=A5=91=E7=BA=A6=E5=A5=97=E4=BB=B6=E7=AC=AC?= =?UTF-8?q?=E4=B8=80=E6=AC=A1=E7=9C=9F=E7=9A=84=E5=9C=A8=E8=B7=91?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit design 0011（待确认）定了六条：一次运行一个文件且文件名就是运行标识（不转义不哈希，按标识 去目录里找文件是最自然的用法；标识必须是安全文件名，否则 ../ 会把文件写到目录外面）；一行 一条记录加一个 record 类型标签（serialization 编出来的载荷没有元信息键，标签是存储这层加的， record 从此是保留键）；第一条解不开的行就是日志结尾、它后面还有内容就是损坏；fsync 只在 运行开始、两条意图、运行结束四处（其余两处靠前缀持久性兜）；写入走 to_thread；运行开始记录 用 O_EXCL 兜住跨进程撞车。 契约套件里那条 test_step_without_an_action_is_still_recorded 转成真断言——它标着 xfail 的 理由是「StepCompleted.result_id 在 0006 里是必填字符串」，而 0006 决策七早就把它改成可为空 并加了不变量。xfail 8→7，跳过 24→14。 原子写「一起不可见」那一半按契约套件的点名在这一层补上了：给实现留一个可注入的故障点 （一个可替换的「把这些字节写进去」），测试把它换成写一半就抛异常，断言那条记录整条不可见。 前缀持久性仍然验不了（掉电才看得出来），继续登记为已知缺口。 调研三条实据写进了 0011：两个下游 fsync 全仓零处（一个的 SQLite 还开着 synchronous=NORMAL）， 所以这条比它们都严、代价是每步两次 fsync；一个下游的轨迹检查器同样是「碰到第一条坏行就放弃 整个文件」；另一个下游踩过「文件名少一维导致两个阶段静默互相覆盖」，O_EXCL 把那类静默覆盖 变成显式失败。这几条我自己逐条核过——那份调研的 subagent 承认它编过一句「我抽查过了」。 migrations/dissect.md 登记两条：运行标识要带齐现在文件名里那五维，以及这份意图日志和它那份 逐步轨迹是两样东西不要混。 --- research-wiki/design/0011-jsonl-run-store.md | 162 ++++++++ research-wiki/migrations/dissect.md | 10 + src/polyloop/stores/__init__.py | 278 ++++++++++++- tests/contract/conftest.py | 203 ++++++++-- tests/contract/test_run_store.py | 48 +-- tests/unit/test_stores.py | 396 +++++++++++++++++++ 6 files changed, 1037 insertions(+), 60 deletions(-) create mode 100644 research-wiki/design/0011-jsonl-run-store.md create mode 100644 tests/unit/test_stores.py diff --git a/research-wiki/design/0011-jsonl-run-store.md b/research-wiki/design/0011-jsonl-run-store.md new file mode 100644 index 0000000..a4b18a8 --- /dev/null +++ b/research-wiki/design/0011-jsonl-run-store.md @@ -0,0 +1,162 @@ +# Design 0011 · 逐行追加的日志存储 + +**日期** 2026-08-10 · **状态** 待确认 + +**落实** `0003-public-api-shape.md` 决策四与 `0005-storage-atomicity-and-record-fields.md` +决策二、五。那两份定了存储接缝有哪六个方法、写入粒度是什么、哪两次写是耐久屏障、以及前缀 +持久性这条要求；本文定**第一个真实实现**怎么满足它们。 + +**触及** `../../src/polyloop/stores/`，以及 `../../tests/contract/` 那套套件——它现在 24 条 +全跳过，因为没有实现可接。 + +## 这份实现为什么值得一份 design doc + +它写下去的东西**会被下游直接读**。某个下游的分析方式是让模型去翻文件，那份日志离开这个库 +也得读得懂；文件名、行的形状、坏行怎么算，一旦有人的分析代码依赖上就改不动了。 + +它还是**唯一一处能验原子写与前缀持久性的地方**。契约套件明写这两条它验不了（要在写入中途 +杀进程，而套件跑在一个进程里），并且点名「落地时要在 `stores` 的 unit 测试里用可注入的故障点 +覆盖」。那个故障点的形状是本文要定的。 + +## 决策一：一次运行一个文件，文件名就是运行标识 + +`<目录>/.jsonl`。 + +**不做成一个大文件加运行标识列。** 一个大文件上，「读回某一次运行的整份日志」要扫全文，而 +`read_log` 在每次 `run` 开工前都会被调用一次；更要命的是两次并发运行会往同一个文件追加， +前缀持久性从「同一文件的追加序」退化成「两条交错的序」，`0005` 决策五那条论证就不成立了。 + +**运行标识必须是一个安全的文件名，不合就报错，不做转义。** 判据是 `[A-Za-z0-9._-]+` 且不以 +点开头。转义（百分号编码、哈希）能接受任意标识，但那样文件名就不再等于运行标识，而下游按 +运行标识去目录里找文件是最自然的用法——`0005` 那句「那份文件离开数据库也得能读懂」说的正是 +这种用法。报错的代价是调用方要约束自己的标识格式，那是一行校验；转义的代价是从此有两套标识 +互相翻译。 + +**这条校验挡住的不只是可读性。** 运行标识是调用方给的不透明字符串，里面出现 `../` 或者绝对 +路径的话，写文件会跑到目录外面去。 + +**迁移注意：运行标识要带齐能唯一定位这次运行的每一维。** 某个下游今天的轨迹文件名是 +`r{轮次}__{阶段}__{题目}__s{种子}__a{尝试}.jsonl` 五维拼出来的，而且它的注释记了一个踩过的 +坑——阶段那一维原先漏了，导致同一次 run 先后两个阶段的轨迹静默互相覆盖。它迁过来时那五维要 +拼进运行标识，否则同样的覆盖会以「日志里有别人的记录」的形式重演。这条登记进 +`../migrations/dissect.md`。 + +## 决策二：一行一条记录，行首加一个类型标签 + +```json +{"record": "intent", "run_id": "r1", "kind": "model_call", ...} +``` + +`polyloop.serialization` 编出来的载荷**只有记录类自己的字段，没有任何元信息键**——那是刻意 +的，为的是一条步记录的载荷和迁移前那份逐行轨迹同形。哪一行是哪种记录由存储自己解决，所以 +标签是这一层加的。 + +**键名 `record` 从此是保留键。** 五个记录类现在都没有叫这个名字的字段，将来也不许加——加了 +的话，编码出来的字典会和标签撞，而撞的表现是解码时把一条记录读成另一种。这条约束写在 +`stores` 的模块 docstring 与 `serialization` 的编码说明里。 + +**取值就是记录类名的蛇形写法**（`run_started` / `intent` / `model_call_result` / +`step_completed` / `run_finished`），不另起一套短名。短名省的那几个字节抵不上「查一个名字要 +先查一张对照表」的成本。 + +## 决策三：撕裂的尾行丢掉，中间的坏行是损坏 + +按行扫，**第一条解不开的行就是日志的结尾**；如果它后面还有解得开的行，那不是撕裂而是损坏， +直接报错。 + +**为什么尾行可以丢。** 进程被杀在一次 `write` 中途，文件末尾会留下半行。那半行对应的那次 +写入从来没有被确认过——调用方还没等到那个 `await` 返回，所以按契约它就是「没发生」。丢掉它 +正是「要么都可见、要么都不可见」的落地方式。 + +**为什么中间的坏行不能丢。** 追加写只在末尾产生撕裂；中间出现读不了的字节意味着别的东西 +（写坏、外部改动、两个进程交错写）动过这个文件。这时候跳过那一行接着读，会拼出一份少了几条 +记录但看起来完整的日志，而恢复会照它做判断。`0002` 决策二那条「读到说不通的状态就失败，不 +修复也不带着它继续」在这里同样适用。 + +**「碰到坏行就放弃这个文件」和某个下游今天的做法一致**：它的轨迹检查器把整个文件的读取包在 +一个 `except (json.JSONDecodeError, UnicodeDecodeError)` 里，撞到第一条坏行就把整个文件降级成 +一条违规，而不是跳过坏行接着读；它还有一条专门造「轨迹文件被截断成半行」的测试。这条不是我们 +新发明的谨慎。 + +**空行跳过，不算坏行。** 它不携带任何记录，也不是撕裂的证据。 + +## 决策四：`fsync` 在三处，其余三处不做 + +| 写什么 | `fsync` | 为什么 | +|---|---|---| +| 运行开始 | 是 | 它是整份日志的头，丢了就读不出这次运行按哪份配置跑 | +| 两条意图 | 是 | `0003` 决策四定的耐久屏障：必须落盘才能发出调用 / 执行动作 | +| 模型调用结果 | 否 | 后面紧跟的不是副作用 | +| 动作结果与步记录 | 否 | 同上 | +| 运行结束 | 是 | 丢了的话这次运行看起来还能续，而它已经跑完了 | + +**不做 `fsync` 的那两次靠前缀持久性兜。** 同一个文件的追加写，后一次 `fsync` 会把它之前的 +全部内容一起刷下去，所以「模型调用结果还没落盘，动作意图（屏障）落了盘」这个状态在这份实现 +上不可能出现——`0005` 决策五要求的正是这个，而这份实现是天然满足的那一类。 + +**每次写都是「打开、追加、按需 `fsync`、关闭」，不长期持有文件句柄。** 持有句柄要为每个运行 +标识维护一份状态，而那份状态在并发下就是共享可变状态；打开的成本相对于一次 `fsync` 可以忽略， +而一次 `fsync` 相对于一次模型调用又可以忽略。 + +**这条比两个下游今天的做法都严，代价要认下。** 它们的仓库里 `fsync` 一处都没有：一个的轨迹 +文件是整体写完再关，它的 SQLite 开着 `synchronous = NORMAL`（WAL 下不对每次提交刷盘）；另一个 +的「原子写」是临时文件加 `rename`，关文件之前不 `fsync`、`rename` 之后也不 `fsync` 父目录， +所以它只保证不出现半截文件，不保证掉电后内容还在。严这一档的直接代价是每一步多两次 `fsync` +——一次几万步的批跑要多花几秒到几十秒，相对于同一批里几万次模型调用可以忽略。 + +**不提供「关掉 `fsync`」的开关。** 关掉之后耐久屏障就不成立了，而恢复的全部正确性建立在它 +上面；一个能把正确性关掉的开关，迟早会有人为了跑得快一点打开它，然后在半年后的一次崩溃里 +发现日志对不上。真要更快，该换一种存储形态，而不是把这一种的保证削掉。 + +## 决策五：阻塞 I/O 丢进线程 + +存储接缝的六个方法都是协程，而文件写入与 `fsync` 是阻塞调用。`fsync` 在忙盘上可以到几十 +毫秒甚至更久，直接在事件循环里做会把同一个循环上所有并发运行一起卡住。所以每次写入走 +`asyncio.to_thread`。 + +**两个下游今天都是直接在事件循环里写文件的**（都没有 `to_thread` / `run_in_executor` / +`aiofiles`），其中一个的循环上还挂着一个心跳协程——写文件一慢，心跳跟着晚。它们现在没被这件事 +咬到，是因为写得少：一个是整次跑完写一个文件，另一个根本不落盘。本库是每步四次写，量级不同。 + +**取消能穿过去。** `to_thread` 那一下被取消时，协程立刻抛出取消，而那个线程会把手上这次写 +做完——写完的东西留在文件里，没写完的是尾行，按决策三丢掉。两种结果都不会让日志进入说不通 +的状态。 + +## 决策六：运行开始记录用独占创建兜住跨进程撞车 + +`write_run_started` 用 `O_CREAT | O_EXCL` 打开文件，文件已存在就直接失败。 + +`run` 在开工前会先 `read_log` 判断这个标识有没有日志，但那是**先读后写**，两个进程同时读到 +空、同时开始写的窗口它挡不住。独占创建把这个窗口关掉，代价是一个标志位。 + +两个进程同时跑同一个运行标识的后果很具体：两条交错的记录序会让恢复读到同一步的两条意图， +按 `_recovery` 的判定那是「日志被并发写过」，于是这次运行从此续不了——而两边的模型调用都 +已经花过钱了。 + +**这一档有真实先例。** 某个下游靠文件名的唯一性避免撞车、不加任何锁，而它的注释记了一次 +踩坑：文件名少了一维，同一次 run 的两个阶段写进了同一个文件、后者静默覆盖前者。独占创建把 +这类「静默覆盖」变成一次显式失败——**发生了就报错，比发生了没人知道好**。 + +## 怎么验那两条没有机器兜底的承诺 + +契约套件验不了原子性的「一起不可见」那一半和前缀持久性，它点名要在这一层补。做法是**给这份 +实现一个只在测试里用的故障注入点**：一个可替换的「把这些字节写进去」的内部函数，测试把它换成 +「写一半就抛异常」。 + +**故障点是内部的，不进公共 API。** 契约套件那条 `xfail` 说「给端口加一个『故意在这里失败』 +的钩子能验，但那个钩子会变成公共 API 的一部分」——那说的是给**接缝**加钩子。给一个具体实现 +的内部留一个可替换点不一样：它不出现在任何接缝签名上，换一个存储实现就没有它。 + +**前缀持久性仍然验不了**，理由和契约套件那条一样：「已经持久」是掉电之后才看得出来的性质。 +这份实现靠形态满足它（同一文件的追加写），只能靠评审看，不靠测试。这一条继续登记为已知缺口。 + +## 留给后续的 + +**这套契约测试怎么交给下游跑。** 套件现在住在 `tests/contract/`，随仓库走，不进发布包。下游 +要拿它验自己的存储实现，得能 import 到它。做成一个 pytest 插件、做成一个可安装的子包、还是让 +下游把仓库作为测试依赖装进去——三条路各有代价，而现在还没有一个下游真的试过跑它。等 ⑥ 迁移 +验收撞上这件事再定，不提前挑一条。 + +**关系数据库那一种形态本文不做。** 另一个下游要写库，但它的表结构、事务边界、连接管理都在它 +自己那边，库替它写一个通用实现只会写出一个谁都不合用的。存储接缝的意义就是让它自己实现，而 +契约套件是它的准入标准。 diff --git a/research-wiki/migrations/dissect.md b/research-wiki/migrations/dissect.md index 94256a8..3d63008 100644 --- a/research-wiki/migrations/dissect.md +++ b/research-wiki/migrations/dissect.md @@ -164,6 +164,16 @@ dissect 有一档实验要在同一时刻用不同的注入内容跑同一批题 ## 缺口登记 +**运行标识要带齐现在文件名里那五维，否则会重演一次静默覆盖。** 现在的轨迹文件名是 +`r{轮次}__{阶段}__{题目}__s{种子}__a{尝试}.jsonl`，`_trajectory_path` 的 docstring 记着一个 +踩过的坑：阶段那一维原先漏了，导致同一次 run 先跑 train 再跑 gate 时后者静默覆盖前者。 +PolyLoop 的日志按运行标识分文件（`../design/0011-jsonl-run-store.md` 决策一），所以那五维要 +拼进运行标识；少一维的表现不再是覆盖，而是「日志里有别人的记录」，恢复会判成日志损坏。 + +**这条日志和 `Rollout.to_jsonl` 那份轨迹是两样东西，不要混。** 前者是意图日志，记的是「准备 +做什么、做完了没有」，为的是崩了能续；后者是产物，记的是逐步轨迹，给反思模型读。前者由库写， +后者迁移后由 dissect 自己从 `RunResult.steps` 重组（见下一条）。 + **轨迹文件的格式是反思模型的唯一输入界面，迁移后要由 dissect 自己重组。** `Rollout.to_jsonl` 现在把项目 metadata 头和逐步轨迹写在同一个文件里，第一行是头、后面每行 一步。PolyLoop 只返回运行标识和步序列，所以 dissect 要自己把两者拼回那个格式，而且格式 diff --git a/src/polyloop/stores/__init__.py b/src/polyloop/stores/__init__.py index 6cf8167..bee7ba5 100644 --- a/src/polyloop/stores/__init__.py +++ b/src/polyloop/stores/__init__.py @@ -1,5 +1,277 @@ -"""库自带的存储实现。**须由使用者显式 import。** +"""存储接缝的第一个实现：一次运行一个文件，一行一条记录，逐行追加。 -不进 `polyloop/__init__.py`，也不许被 `session` import：`session` 一旦 import 了某个 -存储实现，那个实现就成了隐式默认，而不传存储的人不会知道自己这次运行没有恢复能力。 +**这个模块公开，但不进 `polyloop/__init__.py`**，必须显式 import（`0003` 决策八第 9 条）。 + +它满足 `research-wiki/design/0003-public-api-shape.md` 决策四那张写入序列表与 +`0005-storage-atomicity-and-record-fields.md` 决策五那条前缀持久性要求；文件布局、坏行怎么算、 +`fsync` 在哪几处，定在 `research-wiki/design/0011-jsonl-run-store.md`。 + +**`record` 是这一层的保留键。** 每行是「一个类型标签加那条记录的全部字段」，而 +`polyloop.serialization` 编出来的载荷只有记录类自己的字段——标签是这里加的。五个记录类现在 +都没有叫 `record` 的字段，将来也不许加：加了的话编码出来的键会和标签撞，而撞的表现是解码时 +把一条记录读成另一种。 + +**关系数据库那一种形态本库不提供。** 表结构、事务边界、连接管理都在项目那边，库替它写一个 +通用实现只会写出一个谁都不合用的。存储接缝的意义就是让它自己实现，而 `tests/contract/` 是 +它的准入标准。 """ + +import asyncio +import json +import os +import re +from collections.abc import Mapping +from pathlib import Path + +from polyloop.ports import RunLog +from polyloop.serialization import ( + DecodeError, + decode_intent, + decode_model_call_result, + decode_run_finished, + decode_run_started, + decode_step_completed, + encode, +) +from polyloop.types import Intent, ModelCallResult, RunFinished, RunStarted, StepCompleted + +#: 每行那个类型标签的键名。见模块 docstring：它是保留键。 +RECORD_KEY = "record" + +#: 运行标识同时是文件名，所以它必须是一个安全的文件名。 +#: +#: 不转义也不哈希：那样文件名就不再等于运行标识，而按运行标识去目录里找文件是最自然的用法。 +#: 这条校验挡住的不只是可读性——运行标识是调用方给的不透明字符串，里面出现 `../` 的话， +#: 写文件会跑到目录外面去。 +_SAFE_RUN_ID = re.compile(r"[A-Za-z0-9._-]+") + +_TAGS: Mapping[type, str] = { + RunStarted: "run_started", + Intent: "intent", + ModelCallResult: "model_call_result", + StepCompleted: "step_completed", + RunFinished: "run_finished", +} + +_DECODERS = { + "run_started": decode_run_started, + "intent": decode_intent, + "model_call_result": decode_model_call_result, + "step_completed": decode_step_completed, + "run_finished": decode_run_finished, +} + + +class JsonlRunStore: + """把一次运行的日志逐行追加进 `<目录>/<运行标识>.jsonl`。 + + **一次运行一个文件，不是一个大文件加一列运行标识。** 大文件上「读回某一次运行的整份日志」 + 要扫全文，而那件事在每次开工前都会做一遍；更要命的是两次并发运行会往同一个文件追加， + 前缀持久性就从「同一文件的追加序」退化成「两条交错的序」。 + + 它满足 `polyloop.ports.RunStore`，但不显式继承那个 Protocol：结构化子类型不需要继承。 + """ + + __slots__ = ("_directory", "_write_all") + + def __init__(self, *, directory: Path | str) -> None: + self._directory = Path(directory) + #: **可注入的故障点**，见 `_write_all_bytes` 的 docstring。做成实例属性而不是方法， + #: 是因为 `__slots__` 让方法替换不掉，而替换它正是那条测试唯一的做法。 + self._write_all = _write_all_bytes + + def __repr__(self) -> str: + return f"JsonlRunStore(directory={str(self._directory)!r})" + + def parameters(self) -> Mapping[str, str]: + """上报可复现参数。 + + **目录不进快照。** 它是这份日志本身所在的地方——目录要是不一样，根本读不到这份日志、 + 也就走不到比对那一步。把它记进去只会在换一台机器、挂载点变了的时候报出一次假的漂移， + 而那次续跑其实完全正常。 + """ + return {"kind": "jsonl"} + + # -- 写 ------------------------------------------------------------------ + + async def write_run_started(self, record: RunStarted) -> None: + """写运行开始记录。**独占创建**：文件已存在就直接失败。 + + 驱动入口在开工前会先读一次日志判断这个标识有没有用过，但那是先读后写，两个进程同时 + 读到空、同时开始写的窗口它挡不住。独占创建把那个窗口关掉，代价是一个标志位。 + + 两个进程同时跑同一个运行标识的后果很具体：交错的记录序会让恢复读到同一步的两条意图， + 判成「日志被并发写过」，于是这次运行从此续不了——而两边的模型调用都已经花过钱了。 + """ + await self._append(record, fsync=True, exclusive=True) + + async def write_intent(self, record: Intent) -> None: + """写一条意图。**耐久屏障**：必须落盘才能往下走。 + + 意图必须在副作用之前就持久，这是意图日志的全部意义。 + """ + await self._append(record, fsync=True) + + async def write_model_call_result(self, record: ModelCallResult) -> None: + """不做 `fsync`：后面紧跟的不是副作用。 + + 它靠前缀持久性兜——同一个文件的追加写，下一次 `fsync`（那必定是一条意图，或者运行 + 结束）会把它一起刷下去。所以「结果还没落盘、动作意图落了盘」这个状态在这份实现上 + 不可能出现，而那正是 `0005` 决策五要防的。 + """ + await self._append(record) + + async def write_step_completed(self, record: StepCompleted) -> None: + """动作结果与步记录一次原子落地。 + + 它们本来就是同一个记录类的两个字段，所以「一次原子写」在这份实现上就是**一行**： + 一行要么完整地在文件里，要么是被丢掉的撕裂尾行，没有中间态。 + """ + await self._append(record) + + async def write_run_finished(self, record: RunFinished) -> None: + """写结束标记并 `fsync`。丢了的话这次运行看起来还能续，而它已经跑完了。""" + await self._append(record, fsync=True) + + # -- 读 ------------------------------------------------------------------ + + async def read_log(self, run_id: str) -> RunLog: + """读回整份日志。文件不存在时返回空日志，不抛异常。 + + 驱动入口靠这条判断「这个标识是不是已经有日志了」。抛异常的话那个判断就得写成捕获 + 异常，而用捕获异常做流程控制会把真正的存储故障一起吞掉——于是「磁盘挂了」会被读成 + 「这是一次全新的运行」，然后覆盖式地重跑一遍。 + """ + path = self._path(run_id) + if not path.exists(): + return RunLog() + raw = await asyncio.to_thread(path.read_bytes) + return _parse(raw, run_id) + + # -- 内部 ---------------------------------------------------------------- + + def _path(self, run_id: str) -> Path: + if not _SAFE_RUN_ID.fullmatch(run_id) or run_id.startswith("."): + raise ValueError( + f"运行标识 {run_id!r} 不能直接当文件名。这份实现要求它只含字母、数字、点、" + "下划线与连字符，且不以点开头——它同时是文件名，而按标识去目录里找文件是最" + "自然的用法" + ) + return self._directory / f"{run_id}.jsonl" + + async def _append( + self, + record: RunStarted | Intent | ModelCallResult | StepCompleted | RunFinished, + *, + fsync: bool = False, + exclusive: bool = False, + ) -> None: + tag = _TAGS[type(record)] + line = json.dumps({RECORD_KEY: tag, **encode(record)}, ensure_ascii=False) + "\n" + path = self._path(record.run_id) + # 写入与 `fsync` 都是阻塞调用，而 `fsync` 在忙盘上可以到几十毫秒。直接在事件循环里做 + # 会把同一个循环上所有并发运行一起卡住。 + await asyncio.to_thread( + self._write_line, path, line.encode("utf-8"), fsync=fsync, exclusive=exclusive + ) + + def _write_line(self, path: Path, payload: bytes, *, fsync: bool, exclusive: bool) -> None: + """打开、追加、按需 `fsync`、关闭。 + + **不长期持有文件句柄。** 持有要为每个运行标识维护一份状态，而那份状态在并发下就是共享 + 可变状态；打开的成本相对一次 `fsync` 可以忽略，一次 `fsync` 相对一次模型调用又可以忽略。 + """ + path.parent.mkdir(parents=True, exist_ok=True) + flags = os.O_WRONLY | os.O_APPEND | os.O_CREAT + if exclusive: + flags |= os.O_EXCL + descriptor = os.open(path, flags, 0o644) + try: + self._write_all(descriptor, payload) + if fsync: + os.fsync(descriptor) + finally: + os.close(descriptor) + + +def _write_all_bytes(descriptor: int, payload: bytes) -> None: + """把这些字节全部写进去。 + + **这是那个可注入的故障点。** 契约套件验不了原子写的「一起不可见」那一半（要在写入中途 + 杀进程，而它跑在一个进程里），`0011` 定的做法是在这一层留一个可替换的内部函数，单元测试 + 把它换成「写一半就抛异常」。它不出现在任何接缝签名上，换一个存储实现就没有它。 + + 循环是因为 `os.write` 允许短写。短写留下的半行正是撕裂尾行，读那边会丢掉它。 + """ + written = 0 + while written < len(payload): + written += os.write(descriptor, payload[written:]) + + +def _parse(raw: bytes, run_id: str) -> RunLog: + """把一份文件内容还原成日志。 + + **第一条解不开的行就是日志的结尾**，它后面还有内容就不是撕裂而是损坏。追加写只在末尾产生 + 撕裂；中间出现读不了的字节意味着别的东西动过这个文件，那时跳过那一行接着读会拼出一份少了 + 几条记录、看起来却完整的日志，而恢复会照它做判断。 + """ + started: RunStarted | None = None + intents: list[Intent] = [] + model_results: list[ModelCallResult] = [] + steps: list[StepCompleted] = [] + finished: RunFinished | None = None + torn_at: int | None = None + + for number, chunk in enumerate(raw.split(b"\n"), start=1): + if not chunk.strip(): + # 空行不携带记录，也不是撕裂的证据。 + continue + if torn_at is not None: + raise DecodeError( + f"运行 {run_id!r} 的日志第 {torn_at} 行读不了，而第 {number} 行还有内容。" + "追加写只在末尾产生撕裂，中间读不了说明这个文件被别的东西动过" + ) + record = _decode_line(chunk) + if record is None: + torn_at = number + continue + if isinstance(record, RunStarted): + started = record + elif isinstance(record, Intent): + intents.append(record) + elif isinstance(record, ModelCallResult): + model_results.append(record) + elif isinstance(record, StepCompleted): + steps.append(record) + else: + finished = record + + return RunLog( + started=started, + intents=tuple(intents), + model_results=tuple(model_results), + steps=tuple(steps), + finished=finished, + ) + + +def _decode_line(chunk: bytes) -> object | None: + """解一行。解不开返回 `None`——调用方据此判断它是不是撕裂的尾行。 + + **认不得的类型标签不算撕裂，直接报错。** 一行完整的 JSON 带着一个我们不认识的标签，说明 + 这份日志是别的版本或者别的东西写的，不是被杀在写一半。 + """ + try: + payload = json.loads(chunk) + except (UnicodeDecodeError, json.JSONDecodeError): + return None + if not isinstance(payload, dict) or RECORD_KEY not in payload: + return None + tag = payload[RECORD_KEY] + decoder = _DECODERS.get(tag) + if decoder is None: + raise DecodeError(f"日志里出现认不得的记录类型 {tag!r}，这份文件不是本库写的") + return decoder(payload) + + +__all__ = ["RECORD_KEY", "JsonlRunStore"] diff --git a/tests/contract/conftest.py b/tests/contract/conftest.py index 669b965..7357235 100644 --- a/tests/contract/conftest.py +++ b/tests/contract/conftest.py @@ -6,83 +6,218 @@ **接法**：下游在自己的 `conftest.py` 里覆盖同名 fixture，返回自己的实现。 -**现在这套测试全部跳过**，因为 `polyloop` 下还没有任何公共类型与 Protocol -（`research-wiki/design/0006-public-names-and-signatures.md` 还没过人类门）。跳过的理由写在 -每个 fixture 里，读跳过原因就能知道缺的是哪一块。 - **为什么在实现之前就写它**：写一条契约测试要求把每一次调用逐字写出来——方法叫什么、参数 填什么、返回值怎么取。散文里读着通顺的地方，落到这一步就会露出来。前三轮文档评审抓不到的 洞，几乎全是这么冒出来的。 """ +from collections.abc import Mapping + import pytest -#: 公共类型与 Protocol 落地之前，套件里的每一条都缺同一样东西。 -_NOT_YET = ( - "polyloop 的公共类型与 Protocol 还没落地（design/0006 待确认）。" - "这条契约要断言的行为已经写在测试的 docstring 里，落地之后去掉这个跳过即可。" +from polyloop.ports import Action, Event, ToolCall +from polyloop.stores import JsonlRunStore +from polyloop.types import ( + ActionOutcome, + ActionStatus, + Intent, + IntentKind, + ModelCallResult, + ModelReply, + ReplayPolicy, + RunFinished, + RunResult, + RunStarted, + StepCompleted, + StepRecord, + StopReason, +) + +#: 还没有默认实现的那几个接缝，套件里对应的测试全部跳过。 +_NO_IMPLEMENTATION = ( + "库不带这个接缝的默认实现——带了就等于替某一家定了它的协议。" + "下游在自己的 conftest.py 里覆盖这个 fixture，把自己的实现接进来跑。" ) +class _Records: + """构造各类记录的工厂。 + + 它不是被测对象，是让测试正文读得懂的一层薄封装：`records.model_call_intent(...)` 比直接 + 写一长串构造参数更能看出这条测试在断言什么。工厂随套件一起走，因为记录类的字段是库的 + 公共承诺，下游不该为了跑契约测试去手写构造。 + """ + + #: 让测试写 `records.action_status.EXECUTED` 而不必自己 import 那个枚举。 + action_status = ActionStatus + + def reply(self, *, call_id: str | None = "call-1", content: str = "hi") -> ModelReply: + return ModelReply(call_id=call_id, content=content, thinking="") + + def model_call_intent(self, *, run_id: str, call_index: int, result_id: str) -> Intent: + return Intent( + run_id=run_id, + kind=IntentKind.MODEL_CALL, + call_index=call_index, + result_id=result_id, + replay_policy=ReplayPolicy.NEVER, + ) + + def action_intent( + self, + *, + run_id: str, + call_index: int, + result_id: str, + replay_policy: ReplayPolicy = ReplayPolicy.NEVER, + ) -> Intent: + return Intent( + run_id=run_id, + kind=IntentKind.ACTION, + call_index=call_index, + result_id=result_id, + replay_policy=replay_policy, + ) + + def model_call_result( + self, + *, + run_id: str, + result_id: str, + reply: ModelReply | None = None, + failure: str | None = None, + ) -> ModelCallResult: + """`reply` 与 `failure` 恰好一个有值，两个都不给时默认造一条成功的。""" + if reply is None and failure is None: + reply = self.reply() + return ModelCallResult(run_id=run_id, result_id=result_id, reply=reply, failure=failure) + + def outcome( + self, + *, + status: ActionStatus = ActionStatus.EXECUTED, + observation: str = "环境的输出", + env_reported_completion: bool = False, + ) -> ActionOutcome: + return ActionOutcome( + status=status, + observation=observation, + observation_is_synthetic=False, + env_reported_completion=env_reported_completion, + observation_truncated_chars=0, + ) + + def step(self, *, step_idx: int = 0, parse_ok: bool = True) -> StepRecord: + return StepRecord( + step_idx=step_idx, + raw_output="模型说的话", + content_chars=5, + thinking_chars=0, + action="做点事" if parse_ok else None, + parse_ok=parse_ok, + parse_error=None if parse_ok else "解释不出动作", + observation="环境的输出", + observation_is_synthetic=False, + observation_truncated_chars=0, + prompt_chars=100, + call_id="call-1", + step_wall_ms=12, + ) + + def step_completed( + self, + *, + run_id: str, + result_id: str | None, + action_outcome: ActionOutcome | None, + step: StepRecord, + ) -> StepCompleted: + return StepCompleted( + run_id=run_id, result_id=result_id, action_outcome=action_outcome, step=step + ) + + def run_started(self, *, run_id: str, parameter_snapshot: Mapping[str, str]) -> RunStarted: + return RunStarted(run_id=run_id, parameter_snapshot=dict(parameter_snapshot)) + + def result( + self, *, run_id: str, stop_reason: StopReason = StopReason.TASK_COMPLETED + ) -> RunResult: + return RunResult( + run_id=run_id, stop_reason=stop_reason, final_answer="42", steps=(self.step(),) + ) + + def run_finished(self, *, run_id: str, result: RunResult) -> RunFinished: + return RunFinished(run_id=run_id, result=result) + + def action(self, *, text: str = "做点事", tool_name: str | None = None) -> Action: + return Action( + text=text, + tool_call=None if tool_name is None else ToolCall(name=tool_name, arguments={}), + ) + + def event(self) -> Event: + return Event() + + @pytest.fixture -def store(): +def records() -> _Records: + return _Records() + + +@pytest.fixture +def store(tmp_path) -> JsonlRunStore: """被测的存储接缝实现。 - 下游覆盖这个 fixture，返回自己的实例。每次调用应当返回一个**空的**存储——套件里的每条 - 测试都假设自己面对一份干净的日志，共用状态会让测试之间的顺序变成隐式依赖。 + 默认接的是库自带的那个逐行追加实现。下游覆盖这个 fixture，返回自己的实例。 + + 每次调用返回一个**空的**存储：套件里每条测试都假设自己面对一份干净的日志，共用状态会让 + 测试之间的顺序变成隐式依赖。`tmp_path` 每条测试一个新目录，这一条自动成立。 """ - pytest.skip(_NOT_YET) - - -@pytest.fixture -def records(): - """构造各类记录的辅助工厂。 - - 它不是被测对象，是让测试正文读得懂的一层薄封装：`records.model_call_intent(...)` 比 - 直接写一长串构造参数更能看出这条测试在断言什么。工厂本身由库提供，因为记录类的字段 - 是库的公共承诺，下游不该为了跑契约测试去手写构造。 - """ - pytest.skip(_NOT_YET) + return JsonlRunStore(directory=tmp_path) @pytest.fixture def samples(): """被测解释器认得的几段模型输出，由实现方提供。 - **套件不许自己写死输入。** 库不带默认解释器实现，也就不认识任何一家的动作语言：拿 - dissect 的 Python 代码围栏去喂 GovDoc 的 JSON 解析器，后者正确地返回「无效决策」， - 而写死输入的套件会把这个正确行为判成失败。 + **套件不许自己写死输入。** 库不带默认解释器实现，也就不认识任何一家的动作语言：拿一家的 + 代码围栏去喂另一家的 JSON 解析器，后者正确地返回「无效决策」，而写死输入的套件会把这个 + 正确行为判成失败。 实现方要提供两段：`yields_an_action`（一段能被解释成动作的模型回复）与 `yields_invalid` （一段解释不出动作的）。这不是给套件开后门——「我这套语言里什么算合法动作」本来就只有 实现方答得出，套件断言的是**拿到之后的形状**，不是输入长什么样。 """ - pytest.skip(_NOT_YET) + pytest.skip(_NO_IMPLEMENTATION) @pytest.fixture def action_executor(): - """被测的动作执行接缝实现。""" - pytest.skip(_NOT_YET) + """被测的动作执行接缝实现。 + + 库自带一个由工具注册表派生的分发器（`polyloop.tools.ToolRegistry.executor`），但它只覆盖 + 「工具调用」那一种动作语言；把它接在这里会让套件只验得了那一种，所以默认仍然留空。 + """ + pytest.skip(_NO_IMPLEMENTATION) @pytest.fixture def decision_parser(): """被测的决策解释接缝实现。""" - pytest.skip(_NOT_YET) + pytest.skip(_NO_IMPLEMENTATION) @pytest.fixture def model_client(): """被测的模型调用接缝实现。 - 注意这一层的契约测试**不打真实网关**——那是 e2e 的事。这里断言的是返回结构体的形状 - 与失败时的表达方式，用一个受控的替身就能验。 + 注意这一层的契约测试**不打真实网关**——那是 e2e 的事。这里断言的是返回结构体的形状与 + 失败时的表达方式，用一个受控的替身就能验。 """ - pytest.skip(_NOT_YET) + pytest.skip(_NO_IMPLEMENTATION) @pytest.fixture def event_sink(): """被测的事件出口实现。""" - pytest.skip(_NOT_YET) + pytest.skip(_NO_IMPLEMENTATION) diff --git a/tests/contract/test_run_store.py b/tests/contract/test_run_store.py index d54ed5b..0287e2c 100644 --- a/tests/contract/test_run_store.py +++ b/tests/contract/test_run_store.py @@ -6,18 +6,15 @@ **行为的理由不在这里。** 崩溃恢复为什么这么设计见 `design/0002`，写入粒度与前缀持久性见 `design/0005`。这里只断言结果。 -## 写这份文件时撞出来的、`design/0006` 还答不上的问题 +## 标成 `xfail` 的那两条 -每一条都在下面对应的测试里标成 `xfail`，摘要里每次都看得见，但不把套件拖红——**一个永远 -红的套件会训练所有人忽略红**。它们也都不带 fixture，否则会被「实现还没有」那个跳过挡住， -于是「答不上来」就伪装成了「还没轮到」。 +它们是**已知没有机器兜底的承诺**，不是还没写的测试。标成会失败的测试而不是写一句注释，是为了 +让它们在每次跑套件时都被看见；`strict=True` 是配套的：哪天真的验得了、测试过了，它会以 XPASS +报错，逼人回来把标记连同说明一起删掉。它们不带 fixture，否则会被「实现还没有」那个跳过挡住， +于是「验不了」就伪装成了「还没轮到」。 -`strict=True` 是配套的：哪天这个洞被补上、测试真的能过了，它会以 XPASS 报错，逼人回来把 -这个标记连同这段说明一起删掉。 - -1. 没有动作的那些步，`StepCompleted.result_id` 填什么。 -2. 「重放」是把动作再执行一次，还是把上次的结果填回去。 -3. 原子性与前缀持久性能不能写成契约测试。 +剩下那条曾经答不上的——没有动作的步 `StepCompleted.result_id` 填什么——已经由 `design/0006` +决策七答掉（可为空，且为空当且仅当动作结果也为空），对应的测试已经改写成真断言。 """ import pytest @@ -149,22 +146,27 @@ async def test_action_result_and_step_land_together(store, records): assert log.steps[0].action_outcome is not None -@pytest.mark.xfail(reason="design/0006 答不上，见 docstring", strict=True) -def test_step_without_an_action_is_still_recorded(): - """没有动作的步照样留痕：解析失败、模型调用失败、环境故障三种都算一步。 +async def test_step_without_an_action_is_still_recorded(store, records): + """没有动作的步照样留痕：解析失败、模型调用失败、最终回答三种都算一步。 - dissect 的预算对等要求它们计入步数——它们确实消耗了一次模型调用。丢掉那一步还会丢掉 - 模型在出故障时说了什么，而那正是排查「环境坏了还是模型写了危险代码」最需要的。 + 预算对等要求它们计入步数——它们确实消耗了一次模型调用。丢掉那一步还会丢掉模型在出故障时 + 说了什么，而那正是排查「环境坏了还是模型写了危险代码」最需要的。 - **这条现在写不出来。** 这一步没有写过动作意图，所以没有预分配的结果 ID，而 - `StepCompleted.result_id` 在 `design/0006` 里是必填的字符串。照那个形状写，恢复会读到 - 一条对不上任何意图的记录，按 `design/0002` 四态表最后一行判为「日志损坏，拒绝续跑」 - ——而这本该是一次恢复成 `llm_error` 正常终止的运行。 - - **它不带 fixture，所以不会被「实现还没有」那个跳过挡住。** 挡住了它就看起来像「还没 - 轮到」，而它是「答不上来」，两者要分得开。 + 这条曾经写不出来：那时 `StepCompleted.result_id` 是必填字符串，而这一步没写过动作意图、 + 没有预分配的 ID，随便编一个会让恢复读到一条对不上任何意图的记录，按四态表最后一行判成 + 日志损坏。`design/0006` 决策七把它改成可为空，并要求**它为空当且仅当动作结果也为空**， + 这个洞才补上。存储要能原样存下这个形状。 """ - pytest.fail("StepCompleted.result_id 对没有动作意图的步没有定义") + step = records.step_completed( + run_id="r1", result_id=None, action_outcome=None, step=records.step(step_idx=0) + ) + await store.write_step_completed(step) + + log = await store.read_log("r1") + + assert log.steps == (step,) + assert log.steps[0].result_id is None + assert log.steps[0].action_outcome is None # -------------------------------------------------------------------------- diff --git a/tests/unit/test_stores.py b/tests/unit/test_stores.py new file mode 100644 index 0000000..d42b3e4 --- /dev/null +++ b/tests/unit/test_stores.py @@ -0,0 +1,396 @@ +"""逐行追加那份存储实现的行为。 + +**行为契约本身在 `tests/contract/test_run_store.py`**，那套套件现在就接着这个实现跑。这里只写 +契约套件覆盖不到的部分：文件长什么样、坏行怎么算、`fsync` 在哪几处、以及那条契约测试明说 +「这一层验不了」的原子性——它点名要在这里用可注入的故障点补上。 +""" + +import asyncio +import json +import os +from pathlib import Path + +import pytest + +from polyloop.serialization import DecodeError +from polyloop.stores import RECORD_KEY, JsonlRunStore +from polyloop.types import ( + ActionOutcome, + ActionStatus, + Intent, + IntentKind, + ModelCallResult, + ModelReply, + ReplayPolicy, + RunFinished, + RunResult, + RunStarted, + StepCompleted, + StepRecord, + StopReason, +) + +pytestmark = pytest.mark.unit + + +def _started(run_id: str = "r1") -> RunStarted: + return RunStarted(run_id=run_id, parameter_snapshot={"model": "m-1"}) + + +def _intent(run_id: str = "r1", *, call_index: int = 0) -> Intent: + return Intent( + run_id=run_id, + kind=IntentKind.MODEL_CALL, + call_index=call_index, + result_id=f"m{call_index}", + replay_policy=ReplayPolicy.NEVER, + ) + + +def _step(run_id: str = "r1", *, step_idx: int = 0) -> StepCompleted: + return StepCompleted( + run_id=run_id, + result_id=f"a{step_idx}", + action_outcome=ActionOutcome( + status=ActionStatus.EXECUTED, + observation="输出", + observation_is_synthetic=False, + env_reported_completion=False, + observation_truncated_chars=0, + ), + step=StepRecord( + step_idx=step_idx, + raw_output="说的话", + content_chars=3, + thinking_chars=0, + action="做点事", + parse_ok=True, + parse_error=None, + observation="输出", + observation_is_synthetic=False, + observation_truncated_chars=0, + prompt_chars=10, + call_id="c1", + step_wall_ms=1, + ), + ) + + +def _log_file(store_dir: Path, run_id: str = "r1") -> Path: + return store_dir / f"{run_id}.jsonl" + + +# --------------------------------------------------------------------------- +# 文件长什么样 +# --------------------------------------------------------------------------- + + +async def test_one_run_one_file_named_after_the_run_id(tmp_path: Path) -> None: + """按运行标识去目录里找文件是最自然的用法，所以文件名就是标识。""" + store = JsonlRunStore(directory=tmp_path) + + await store.write_run_started(_started("my-run.1")) + + assert (tmp_path / "my-run.1.jsonl").exists() + + +async def test_each_record_is_one_line_tagged_with_its_type(tmp_path: Path) -> None: + """一行一条记录，行首带一个类型标签。 + + `serialization` 编出来的载荷只有记录类自己的字段——标签是存储这一层加的，因为「哪一行是 + 哪种记录」本来就是文件布局的事。 + """ + store = JsonlRunStore(directory=tmp_path) + + await store.write_run_started(_started()) + await store.write_intent(_intent()) + await store.write_step_completed(_step()) + + lines = _log_file(tmp_path).read_text(encoding="utf-8").splitlines() + assert [json.loads(line)[RECORD_KEY] for line in lines] == [ + "run_started", + "intent", + "step_completed", + ] + + +async def test_a_record_never_contains_a_raw_newline(tmp_path: Path) -> None: + """一条记录占一行，靠的是 JSON 把换行转义掉。 + + 转义要是失效了，一条带换行的观察会被拆成两行，其中一行必然解不开——而读那边会把它当成 + 撕裂的尾行丢掉，于是一条完整写下去的记录静默消失了。 + """ + store = JsonlRunStore(directory=tmp_path) + record = _step() + multiline = StepCompleted( + run_id=record.run_id, + result_id=record.result_id, + action_outcome=record.action_outcome, + step=StepRecord( + **{ + **{ + field: getattr(record.step, field) + for field in ( + "step_idx", + "raw_output", + "content_chars", + "thinking_chars", + "action", + "parse_ok", + "parse_error", + "observation_is_synthetic", + "observation_truncated_chars", + "prompt_chars", + "call_id", + "step_wall_ms", + ) + }, + "observation": "第一行\n第二行\n第三行", + } + ), + ) + + await store.write_step_completed(multiline) + + assert len(_log_file(tmp_path).read_text(encoding="utf-8").splitlines()) == 1 + (readback,) = (await store.read_log("r1")).steps + assert readback.step.observation == "第一行\n第二行\n第三行" + + +async def test_two_runs_go_to_two_files(tmp_path: Path) -> None: + """两次并发运行写同一个文件的话，前缀持久性就从「同一文件的追加序」退化成两条交错的序。""" + store = JsonlRunStore(directory=tmp_path) + + await store.write_intent(_intent("run-a")) + await store.write_intent(_intent("run-b")) + + assert (await store.read_log("run-a")).intents == (_intent("run-a"),) + assert (await store.read_log("run-b")).intents == (_intent("run-b"),) + + +@pytest.mark.parametrize("run_id", ["../escape", "a/b", ".hidden", "", "有中文", "a b"]) +async def test_a_run_id_that_is_not_a_safe_file_name_is_refused( + tmp_path: Path, run_id: str +) -> None: + """运行标识是调用方给的不透明字符串，里面出现 `../` 的话写文件会跑到目录外面去。 + + 不转义也不哈希：那样文件名就不再等于运行标识，而按标识去目录里找文件是最自然的用法。 + """ + store = JsonlRunStore(directory=tmp_path) + + with pytest.raises(ValueError, match="文件名"): + await store.read_log(run_id) + + +# --------------------------------------------------------------------------- +# 坏行 +# --------------------------------------------------------------------------- + + +async def test_a_torn_last_line_is_dropped(tmp_path: Path) -> None: + """进程被杀在一次写中途，末尾留下半行。那次写从来没被确认过，所以它就是「没发生」。""" + store = JsonlRunStore(directory=tmp_path) + await store.write_run_started(_started()) + await store.write_intent(_intent()) + with _log_file(tmp_path).open("a", encoding="utf-8") as handle: + handle.write('{"record": "step_completed", "run_id": "r1", "resu') + + log = await store.read_log("r1") + + assert log.started == _started() + assert log.intents == (_intent(),) + assert log.steps == () + + +async def test_a_bad_line_in_the_middle_is_corruption_not_a_torn_tail(tmp_path: Path) -> None: + """追加写只在末尾产生撕裂。中间读不了说明别的东西动过这个文件。 + + 跳过那一行接着读会拼出一份少了几条记录、看起来却完整的日志，而恢复会照它做判断。 + """ + store = JsonlRunStore(directory=tmp_path) + await store.write_run_started(_started()) + await store.write_intent(_intent()) + path = _log_file(tmp_path) + lines = path.read_text(encoding="utf-8").splitlines() + path.write_text(f"{lines[0]}\n半行不完整\n{lines[1]}\n", encoding="utf-8") + + with pytest.raises(DecodeError, match="被别的东西动过"): + await store.read_log("r1") + + +async def test_a_blank_line_is_not_a_torn_tail(tmp_path: Path) -> None: + """空行不携带记录，也不是撕裂的证据。""" + store = JsonlRunStore(directory=tmp_path) + await store.write_run_started(_started()) + path = _log_file(tmp_path) + path.write_text(path.read_text(encoding="utf-8") + "\n\n", encoding="utf-8") + + assert (await store.read_log("r1")).started == _started() + + +async def test_an_unknown_record_type_is_refused_not_skipped(tmp_path: Path) -> None: + """一行完整的 JSON 带着认不得的标签，说明这份文件不是本库写的，不是被杀在写一半。""" + store = JsonlRunStore(directory=tmp_path) + await store.write_run_started(_started()) + with _log_file(tmp_path).open("a", encoding="utf-8") as handle: + handle.write('{"record": "something_else", "run_id": "r1"}\n') + + with pytest.raises(DecodeError, match="认不得"): + await store.read_log("r1") + + +# --------------------------------------------------------------------------- +# 原子写：契约套件明说它验不了的那一半 +# --------------------------------------------------------------------------- + + +async def test_a_write_that_dies_halfway_leaves_nothing_readable(tmp_path: Path) -> None: + """崩在一次写中途，那条记录**整条不可见**，不是半条可见。 + + 这是 `tests/contract/test_run_store.py` 里那条 `xfail` 点名要在这一层补的：契约套件跑在 + 一个进程里、面对一个已经装配好的实现，没有位置插入那次崩溃。这里靠替换掉那个内部的 + 「把这些字节写进去」来造它。 + + 不成立的话，恢复会读到一条残缺的步记录——而那一步的历史文本就永远丢了。 + """ + store = JsonlRunStore(directory=tmp_path) + await store.write_run_started(_started()) + + def _die_halfway(descriptor: int, payload: bytes) -> None: + os.write(descriptor, payload[: len(payload) // 2]) + raise OSError("磁盘满了") + + store._write_all = _die_halfway # noqa: SLF001 — 那个可注入的故障点 + with pytest.raises(OSError, match="磁盘满了"): + await store.write_step_completed(_step()) + + log = await store.read_log("r1") + + assert log.started == _started() + assert log.steps == () + + +async def test_the_records_written_before_a_dead_write_survive(tmp_path: Path) -> None: + """崩在第 k 次写中途，前 k-1 次照样读得回来。 + + 这是前缀持久性在「读得回来」这一层的影子。真正的前缀持久性是掉电之后的性质，这一层验 + 不了——它靠形态满足（同一个文件的追加写），只能靠评审看。 + """ + store = JsonlRunStore(directory=tmp_path) + await store.write_run_started(_started()) + await store.write_intent(_intent(call_index=0)) + await store.write_step_completed(_step(step_idx=0)) + + def _die(descriptor: int, payload: bytes) -> None: + raise OSError("盘掉了") + + store._write_all = _die # noqa: SLF001 — 那个可注入的故障点 + with pytest.raises(OSError): + await store.write_intent(_intent(call_index=1)) + + log = await store.read_log("r1") + + assert log.intents == (_intent(call_index=0),) + assert len(log.steps) == 1 + + +# --------------------------------------------------------------------------- +# 并发与耐久 +# --------------------------------------------------------------------------- + + +async def test_starting_the_same_run_twice_fails_instead_of_interleaving(tmp_path: Path) -> None: + """独占创建把「先读后写」那个窗口关掉。 + + 两个进程同时跑同一个运行标识，交错的记录序会让恢复读到同一步的两条意图、判成日志被并发 + 写过，于是这次运行从此续不了——而两边的模型调用都已经花过钱了。 + """ + store = JsonlRunStore(directory=tmp_path) + await store.write_run_started(_started()) + + with pytest.raises(FileExistsError): + await store.write_run_started(_started()) + + +async def test_writes_do_not_block_the_event_loop(tmp_path: Path) -> None: + """写入与 `fsync` 走线程，不压在事件循环上。 + + `fsync` 在忙盘上可以到几十毫秒，直接在循环里做会把同一个循环上所有并发运行一起卡住。 + """ + store = JsonlRunStore(directory=tmp_path) + ticks = 0 + + async def _tick() -> None: + nonlocal ticks + while True: + ticks += 1 + await asyncio.sleep(0) + + ticker = asyncio.ensure_future(_tick()) + await store.write_run_started(_started()) + ticker.cancel() + + assert ticks > 0 + + +async def test_an_unwritten_run_reads_back_empty(tmp_path: Path) -> None: + """驱动入口靠这条判断「这个标识是不是已经有日志了」。""" + store = JsonlRunStore(directory=tmp_path) + + log = await store.read_log("never-written") + + assert log.started is None + assert log.intents == () + assert log.finished is None + + +async def test_the_directory_is_created_on_first_write(tmp_path: Path) -> None: + """构造廉价：目录在第一次真的要写的时候才建，不在构造时。""" + nested = tmp_path / "a" / "b" + store = JsonlRunStore(directory=nested) + assert not nested.exists() + + await store.write_run_started(_started()) + + assert nested.exists() + + +async def test_a_whole_log_round_trips(tmp_path: Path) -> None: + """五种记录写进去、读回来逐字段相等。""" + store = JsonlRunStore(directory=tmp_path) + result = RunResult( + run_id="r1", stop_reason=StopReason.TASK_COMPLETED, final_answer="42", steps=() + ) + records = [ + _started(), + _intent(), + ModelCallResult( + run_id="r1", + result_id="m0", + reply=ModelReply(call_id="c1", content="hi", thinking=""), + failure=None, + ), + _step(), + RunFinished(run_id="r1", result=result), + ] + await store.write_run_started(records[0]) # type: ignore[arg-type] + await store.write_intent(records[1]) # type: ignore[arg-type] + await store.write_model_call_result(records[2]) # type: ignore[arg-type] + await store.write_step_completed(records[3]) # type: ignore[arg-type] + await store.write_run_finished(records[4]) # type: ignore[arg-type] + + log = await store.read_log("r1") + + assert log.started == records[0] + assert log.intents == (records[1],) + assert log.model_results == (records[2],) + assert log.steps == (records[3],) + assert log.finished == records[4] + + +def test_the_directory_does_not_enter_the_parameter_snapshot(tmp_path: Path) -> None: + """目录不进快照：目录不一样就根本读不到这份日志，也走不到比对那一步。 + + 记进去只会在换一台机器、挂载点变了的时候报出一次假的漂移，而那次续跑其实完全正常。 + """ + assert JsonlRunStore(directory=tmp_path).parameters() == {"kind": "jsonl"}