diff --git a/src/polyloop/stores/__init__.py b/src/polyloop/stores/__init__.py index 673a892..a79dbb9 100644 --- a/src/polyloop/stores/__init__.py +++ b/src/polyloop/stores/__init__.py @@ -1,310 +1,19 @@ -"""存储接缝的第一个实现:一次运行一个文件,一行一条记录,逐行追加。 +"""存储接缝的两个实现:一个逐行追加进本地文件,一个只留在进程内存里。 -**这个模块公开,但不进 `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` 的字段,将来也不许加:加了的话编码出来的键会和标签撞,而撞的表现是解码时 -把一条记录读成另一种。 +**这一层公开,但不进 `polyloop/__init__.py`**,必须显式 import(`0003` 决策八第 9 条)。存储 +是必填的装配项,库不给默认值:顺手提供一个默认实现等于替所有下游选了日志落在哪儿,而那是 +每个项目自己的运维决定。调用方要么从这里挑一个,要么自己写一个。 **关系数据库那一种形态本库不提供。** 表结构、事务边界、连接管理都在项目那边,库替它写一个 -通用实现只会写出一个谁都不合用的。存储接缝的意义就是让它自己实现,而 `tests/contract/` 是 -它的准入标准。 +通用实现只会写出一个谁都不合用的。存储接缝的意义就是让它自己实现,而契约套件 +(`polyloop.testing`,见 `research-wiki/design/0014-contract-suite-distribution.md` 决策一) +是它的准入标准:那套用例只说行为、不碰形态,跑全绿就算合格。 + +**这一层也不提供「把日志转成别的格式」「按时间范围查」这类操作。** 存储接缝上多一个方法,就是 +给每一个下游实现多加一份永久要求,而这些事拿 `read_log` 读回来的日志在库外面做即可。 """ -import asyncio -import json -import os -import re -from collections.abc import Mapping -from pathlib import Path +from polyloop.stores._jsonl import RECORD_KEY, JsonlRunStore +from polyloop.stores._volatile import VolatileRunStore -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", "_locks", "_write_all") - - def __init__(self, *, directory: Path | str) -> None: - self._directory = Path(directory) - #: 每个运行标识一把锁,把同一个文件上的写串起来。 - #: - #: 一条记录可能由不止一次 `os.write` 写完(`os.write` 允许短写),而 `O_APPEND` 只保证 - #: 每一次 `os.write` 的追加位置原子,保证不了「一条逻辑行整体原子」。两个协程同时往同一 - #: 个文件写时,一次短写会让两条记录交错成一段谁也解不开的字节。锁把这件事挡在进程内; - #: 跨进程那一半靠运行开始记录的独占创建挡(见 `write_run_started`)。 - self._locks: dict[str, asyncio.Lock] = {} - #: **可注入的故障点**,见 `_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) - lock = self._locks.setdefault(record.run_id, asyncio.Lock()) - async with lock: - # 写入与 `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) - if exclusive: - _fsync_directory(path.parent) - - -def _fsync_directory(directory: Path) -> None: - """把新建文件的目录项刷下去。 - - `os.fsync(fd)` 刷的是那个文件的内容,刷不到「这个目录里多了一个文件」这条目录项。掉电之后 - 内容可能在、而文件根本不存在——那时 `read_log` 走「文件不存在」返回空日志,驱动入口据此 - 判成一次全新的运行,于是一次已经开始过、可能已经花过钱的运行静默没了留痕。 - - 只在新建文件时做:往已有文件追加不改目录项。 - """ - descriptor = os.open(directory, os.O_RDONLY) - try: - 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: - """把一份文件内容还原成日志。 - - **判据是「这一行有没有被换行终结」,不是「它能不能解析」。** 一次写入是先写整行再由调用方 - 等到它返回,所以文件末尾那段没有换行的字节对应的那次写**从来没有被确认过**——按契约它就是 - 没发生,丢掉它正是「要么都可见、要么都不可见」的落地方式。 - - 照「能不能解析」判会漏掉一个很具体的场景:短写正好写完了整个 JSON 对象、只差最后那个换行。 - 那段字节解得开,于是一条从没被确认的动作意图被当成有效记录读回来,恢复据此判成「状态未知」 - 并可能重放——而那个动作其实一定没执行过,因为调用方是在写意图返回之后才去执行的。 - - **被换行终结的行必须解得开**,解不开就是损坏,直接报错。追加写只在末尾产生撕裂;一条完整 - 终结的行读不了,说明别的东西动过这个文件,那时跳过它接着读会拼出一份少了几条记录、看起来 - 却完整的日志,而恢复会照它做判断。 - """ - started: RunStarted | None = None - intents: list[Intent] = [] - model_results: list[ModelCallResult] = [] - steps: list[StepCompleted] = [] - finished: RunFinished | None = None - - chunks = raw.split(b"\n") - # 文件以换行结尾时最后一段是空的;不以换行结尾说明最后那次写没写完。 - terminated = chunks[:-1] if chunks and chunks[-1].strip() else chunks - - for number, chunk in enumerate(terminated, start=1): - if not chunk.strip(): - # 空行不携带记录,也不是撕裂的证据。 - continue - record = _decode_line(chunk, run_id=run_id, number=number) - 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, *, run_id: str, number: int) -> object: - """解一条被换行终结的行。解不开就是损坏,直接报错。 - - 这里不再有「解不开就当撕裂尾行」那条路——撕裂由有没有换行判定,进不到这个函数。 - """ - where = f"运行 {run_id!r} 的日志第 {number} 行" - try: - payload = json.loads(chunk) - except (UnicodeDecodeError, json.JSONDecodeError) as exc: - raise DecodeError( - f"{where}读不了({type(exc).__name__})。它是被换行终结的完整一行," - "说明这个文件被别的东西动过——追加写只在末尾产生撕裂" - ) from exc - if not isinstance(payload, dict) or RECORD_KEY not in payload: - raise DecodeError(f"{where}没有 {RECORD_KEY!r} 标签,这份文件不是本库写的") - tag = payload[RECORD_KEY] - decoder = _DECODERS.get(tag) - if decoder is None: - raise DecodeError(f"{where}的记录类型 {tag!r} 认不得,这份文件不是本库写的") - return decoder(payload) - - -__all__ = ["RECORD_KEY", "JsonlRunStore"] +__all__ = ["RECORD_KEY", "JsonlRunStore", "VolatileRunStore"] diff --git a/src/polyloop/stores/_jsonl.py b/src/polyloop/stores/_jsonl.py new file mode 100644 index 0000000..4deb003 --- /dev/null +++ b/src/polyloop/stores/_jsonl.py @@ -0,0 +1,264 @@ +"""存储接缝的第一个实现:一次运行一个文件,一行一条记录,逐行追加。 + +文件布局、坏行怎么算、`fsync` 在哪几处,定在 `research-wiki/design/0011-jsonl-run-store.md`。 +它满足 `0003-public-api-shape.md` 决策四那张写入序列表与 `0005-storage-atomicity-and-record-fields.md` +决策五那条前缀持久性要求。 + +**文件的字面约定**:`<目录>/<运行标识>.jsonl`,UTF-8,每行以 `\\n` 结尾,非 ASCII 原样输出。 +这份日志的一个明确用途是让人和模型直接翻文件读,所以它不转义成 `\\uXXXX`。 + +**`record` 是这份文件布局的保留键。** 每行是「一个类型标签加那条记录的全部字段」,而 +`polyloop.serialization` 编出来的载荷只有记录类自己的字段——标签是这里加的。五个记录类现在都 +没有叫 `record` 的字段,将来也不许加:加了的话编码出来的键会和标签撞,而撞的表现是解码时把 +一条记录读成另一种。 + +**`fsync` 落在三处**:运行开始、每一条意图、运行结束。前两处是耐久屏障(意图必须在副作用之前 +就持久),第三处防「跑完了的运行看起来还能续」。模型调用结果与逐步结果不刷,靠前缀持久性兜。 +运行开始那一次还要刷父目录,见 `_fsync_directory`。 +""" + +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, encode +from polyloop.stores._records import DECODERS_BY_TAG, TAGS, Record, assemble_log +from polyloop.types import Intent, ModelCallResult, RunFinished, RunStarted, StepCompleted + +#: 每行那个类型标签的键名。见模块 docstring:它是保留键。 +RECORD_KEY = "record" + +#: 运行标识同时是文件名,所以它必须是一个安全的文件名。 +#: +#: 不转义也不哈希:那样文件名就不再等于运行标识,而按运行标识去目录里找文件是最自然的用法。 +#: 这条校验挡住的不只是可读性——运行标识是调用方给的不透明字符串,里面出现 `../` 的话, +#: 写文件会跑到目录外面去。 +_SAFE_RUN_ID = re.compile(r"[A-Za-z0-9._-]+") + + +class JsonlRunStore: + """把一次运行的日志逐行追加进 `<目录>/<运行标识>.jsonl`。 + + **一次运行一个文件,不是一个大文件加一列运行标识。** 大文件上「读回某一次运行的整份日志」 + 要扫全文,而那件事在每次开工前都会做一遍;更要命的是两次并发运行会往同一个文件追加, + 前缀持久性就从「同一文件的追加序」退化成「两条交错的序」。 + + 它满足 `polyloop.ports.RunStore`,但不显式继承那个 Protocol:结构化子类型不需要继承。 + """ + + __slots__ = ("_directory", "_locks", "_write_all") + + def __init__(self, *, directory: Path | str) -> None: + self._directory = Path(directory) + #: 每个运行标识一把锁,把同一个文件上的写串起来。 + #: + #: 一条记录可能由不止一次 `os.write` 写完(`os.write` 允许短写),而 `O_APPEND` 只保证 + #: 每一次 `os.write` 的追加位置原子,保证不了「一条逻辑行整体原子」。两个协程同时往同一 + #: 个文件写时,一次短写会让两条记录交错成一段谁也解不开的字节。锁把这件事挡在进程内; + #: 跨进程那一半靠运行开始记录的独占创建挡(见 `write_run_started`)。 + self._locks: dict[str, asyncio.Lock] = {} + #: **可注入的故障点**,见 `_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: Record, + *, + 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) + lock = self._locks.setdefault(record.run_id, asyncio.Lock()) + async with lock: + # 写入与 `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) + if exclusive: + _fsync_directory(path.parent) + + +def _fsync_directory(directory: Path) -> None: + """把新建文件的目录项刷下去。 + + `os.fsync(fd)` 刷的是那个文件的内容,刷不到「这个目录里多了一个文件」这条目录项。掉电之后 + 内容可能在、而文件根本不存在——那时 `read_log` 走「文件不存在」返回空日志,驱动入口据此 + 判成一次全新的运行,于是一次已经开始过、可能已经花过钱的运行静默没了留痕。 + + 只在新建文件时做:往已有文件追加不改目录项。 + """ + descriptor = os.open(directory, os.O_RDONLY) + try: + 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: + """把一份文件内容还原成日志。 + + **判据是「这一行有没有被换行终结」,不是「它能不能解析」。** 一次写入是先写整行再由调用方 + 等到它返回,所以文件末尾那段没有换行的字节对应的那次写**从来没有被确认过**——按契约它就是 + 没发生,丢掉它正是「要么都可见、要么都不可见」的落地方式。 + + 照「能不能解析」判会漏掉一个很具体的场景:短写正好写完了整个 JSON 对象、只差最后那个换行。 + 那段字节解得开,于是一条从没被确认的动作意图被当成有效记录读回来,恢复据此判成「状态未知」 + 并可能重放——而那个动作其实一定没执行过,因为调用方是在写意图返回之后才去执行的。 + + **被换行终结的行必须解得开**,解不开就是损坏,直接报错。追加写只在末尾产生撕裂;一条完整 + 终结的行读不了,说明别的东西动过这个文件,那时跳过它接着读会拼出一份少了几条记录、看起来 + 却完整的日志,而恢复会照它做判断。 + """ + chunks = raw.split(b"\n") + # 文件以换行结尾时最后一段是空的;不以换行结尾说明最后那次写没写完。 + terminated = chunks[:-1] if chunks and chunks[-1].strip() else chunks + + records: list[Record] = [] + for number, chunk in enumerate(terminated, start=1): + if not chunk.strip(): + # 空行不携带记录,也不是撕裂的证据。 + continue + records.append(_decode_line(chunk, run_id=run_id, number=number)) + + return assemble_log(records) + + +def _decode_line(chunk: bytes, *, run_id: str, number: int) -> Record: + """解一条被换行终结的行。解不开就是损坏,直接报错。 + + 这里不再有「解不开就当撕裂尾行」那条路——撕裂由有没有换行判定,进不到这个函数。 + """ + where = f"运行 {run_id!r} 的日志第 {number} 行" + try: + payload = json.loads(chunk) + except (UnicodeDecodeError, json.JSONDecodeError) as exc: + raise DecodeError( + f"{where}读不了({type(exc).__name__})。它是被换行终结的完整一行," + "说明这个文件被别的东西动过——追加写只在末尾产生撕裂" + ) from exc + if not isinstance(payload, dict) or RECORD_KEY not in payload: + raise DecodeError(f"{where}没有 {RECORD_KEY!r} 标签,这份文件不是本库写的") + tag = payload[RECORD_KEY] + decoder = DECODERS_BY_TAG.get(tag) + if decoder is None: + raise DecodeError(f"{where}的记录类型 {tag!r} 认不得,这份文件不是本库写的") + return decoder(payload) diff --git a/src/polyloop/stores/_records.py b/src/polyloop/stores/_records.py new file mode 100644 index 0000000..b842ead --- /dev/null +++ b/src/polyloop/stores/_records.py @@ -0,0 +1,94 @@ +"""五个记录类在存储这一层的类型标签与解码器,以及把一串记录装配成一份日志。 + +**一张三元组表,不是几张平行的表。** 逐行追加那个实现按记录类找标签、按标签找解码器;易失 +那个实现按记录类找解码器。三张平行的表之间有一个谁也不检查的一致性要求——必须覆盖同样那五个 +记录类,漏掉一处的表现是某种记录写得进去、读不回来。派生视图全部从同一张表算出来,那个要求 +就不可能被违反(`research-wiki/design/0014-contract-suite-distribution.md` 决策五)。 + +**这个模块只认识记录,不认识存储形态。** 标签取值是文件里那一行的事,但按记录类找标签这件事 +两个实现都做不到自己一份还保持一致,所以表在这里,怎么用它由各自的实现定。 +""" + +from collections.abc import Callable, Iterable, Mapping +from typing import NamedTuple + +from polyloop.ports import RunLog +from polyloop.serialization import ( + decode_intent, + decode_model_call_result, + decode_run_finished, + decode_run_started, + decode_step_completed, +) +from polyloop.types import Intent, ModelCallResult, RunFinished, RunStarted, StepCompleted + +#: 存储接缝收得下的五种记录。 +#: +#: 步记录与运行结果不在里面:它们是别的记录的字段,`polyloop.serialization` 单独给它们留了 +#: 编解码入口是为了下游脱离运行时读轨迹,不是为了让它们自己成为一条日志行。 +Record = RunStarted | Intent | ModelCallResult | StepCompleted | RunFinished + +Decoder = Callable[[Mapping[str, object]], Record] + + +class RecordType(NamedTuple): + """一个记录类在这一层的三件事:它自己、它的标签、把载荷变回它的那个函数。""" + + record_cls: type[Record] + #: 记录类名的蛇形写法(`design/0011` 决策二)。它写进文件,改一个字就读不了旧日志。 + tag: str + decode: Decoder + + +#: 顺序无意义,覆盖范围有意义:这五条就是存储这一层认得的全部记录。 +RECORD_TYPES: tuple[RecordType, ...] = ( + RecordType(RunStarted, "run_started", decode_run_started), + RecordType(Intent, "intent", decode_intent), + RecordType(ModelCallResult, "model_call_result", decode_model_call_result), + RecordType(StepCompleted, "step_completed", decode_step_completed), + RecordType(RunFinished, "run_finished", decode_run_finished), +) + +TAGS: Mapping[type[Record], str] = {entry.record_cls: entry.tag for entry in RECORD_TYPES} + +DECODERS_BY_TAG: Mapping[str, Decoder] = {entry.tag: entry.decode for entry in RECORD_TYPES} + +DECODERS_BY_TYPE: Mapping[type[Record], Decoder] = { + entry.record_cls: entry.decode for entry in RECORD_TYPES +} + + +def assemble_log(records: Iterable[Record]) -> RunLog: + """把一串按写入顺序排好的记录装配成一份日志。 + + 运行开始与运行结束各只保留最后一条。它们一次运行各写一回,同一份日志里出现第二条说明这个 + 标识被重开过,而挡这件事是写入那一侧的责任(见两个实现的 `write_run_started`)。 + + **两个实现共用这一段。** 各写一遍的话,某一处漏掉一种记录——比如只往步序列里放、忘了模型 + 调用结果——不会有任何检查发现,只表现成恢复少看见一条记录,而恢复据此判定的那一步会被重跑。 + """ + started: RunStarted | None = None + intents: list[Intent] = [] + model_results: list[ModelCallResult] = [] + steps: list[StepCompleted] = [] + finished: RunFinished | None = None + + for record in records: + 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, + ) diff --git a/src/polyloop/stores/_volatile.py b/src/polyloop/stores/_volatile.py new file mode 100644 index 0000000..1683db9 --- /dev/null +++ b/src/polyloop/stores/_volatile.py @@ -0,0 +1,176 @@ +"""存储接缝的第二个实现:日志留在进程内存里,进程一退就没了。 + +`research-wiki/design/0014-contract-suite-distribution.md` 决策五定的那个实现。 +""" + +from collections.abc import Mapping +from typing import NamedTuple + +from polyloop.ports import RunLog +from polyloop.serialization import encode +from polyloop.stores._records import DECODERS_BY_TYPE, Decoder, Record, assemble_log +from polyloop.types import Intent, ModelCallResult, RunFinished, RunStarted, StepCompleted + + +class _Stored(NamedTuple): + """桶里的一条记录:编出来的载荷,加上把它解回来要用的那个函数。 + + 解码器在写入那一刻就查好存下来,而不是读的时候按记录类现查——查这一下同时是「这一层认不认得 + 这个记录类」的判据,和逐行追加那个实现在写入时查类型标签是同一件事。留到读的时候查,一条 + 本层不认得的记录会先安静地进桶,直到整份日志读不回来。 + """ + + decode: Decoder + payload: Mapping[str, object] + + +class VolatileRunStore: + """把一次运行的日志攒在进程内存里。 + + **它不提供的是跨进程恢复,不是「读不回来」。** 同一个进程里写进去的意图照样读得回来—— + 这一层的准入标准是那套契约套件,而套件第一条要的就是这件事,一个过不了自家准入标准的实现 + 不该存在。真正没有的是「进程重启之后接着跑」:那份日志随进程一起消失,续跑时读到的是一份 + 空日志,于是那次运行会被判成从没开始过。选它就是选「我不要跨进程恢复」。 + + **运行标识在这里没有形状要求。** 逐行追加那个实现要求标识是个安全的文件名,因为标识就是 + 文件名;这里什么都不是文件名,`a/b` 与 `../x` 都收得下。拿它跑测试跑通了、换成落盘那个 + 实现才撞上校验,是这两个实现之间唯一一处不对等,认下来是因为反过来更糟:库替一个不落盘的 + 实现编一条文件名规矩,等于把某一种存储形态的约束写进接缝。 + + **并发隔离的承诺到「同一个事件循环」为止。** 每个写方法从头到尾没有一个 await 点,而 + asyncio 只在 await 处切协程,所以同一个循环上并发跑的多次运行看到的每一次写都是原子的。 + 多个线程各跑一个事件循环、共享同一个实例,不在这条承诺里:两个线程可以都判出这个运行标识 + 还没有日志,然后后写的那个把先写的整桶换掉,而全程没有任何冲突报错。逐行追加那个实现的 + 边界一样——它那把按运行标识分的 `asyncio.Lock` 本身也不是线程安全的。要跨线程用,一个 + 线程一个实例。 + + 它满足 `polyloop.ports.RunStore`,但不显式继承那个 Protocol:结构化子类型不需要继承。 + """ + + __slots__ = ("_runs",) + + def __init__(self) -> None: + #: 运行标识 → 那次运行写下的记录载荷,按写入顺序。 + #: + #: **存载荷不存记录对象**,理由和取值形态见 `_stored` 与 `read_log`。 + #: + #: **按运行标识分桶,端口不持有「当前运行」的隐式状态**(见 `polyloop.ports.RunStore`): + #: 运行标识住在每一条记录里,写哪一桶每次现看。一个有隐式当前运行的实现会在并发下把 A + #: 的意图写进 B 的日志,而那种错在单线程测试里永远不出现。 + self._runs: dict[str, list[_Stored]] = {} + + def __repr__(self) -> str: + return f"VolatileRunStore(runs={len(self._runs)})" + + def parameters(self) -> Mapping[str, str]: + """上报可复现参数:只有形态,没有任何实例状态。 + + 续跑时这份快照要和日志里存下来的那份逐键比对,对不上就报参数漂移。把「现在攒了几次 + 运行」这类实例状态写进去的话,同一个存储对象在两次比对之间会给出不同的答案,于是一次 + 装配完全没变的续跑被判成漂移。 + """ + return {"kind": "volatile"} + + # -- 写 ------------------------------------------------------------------ + + async def write_run_started(self, record: RunStarted) -> None: + """写运行开始记录。**独占**:这个标识已经有日志了就直接失败。 + + 和逐行追加那个实现的独占创建对齐,包括判据——那边失败的条件是文件已存在,而任何一次写 + 都会把文件建出来,所以这里的判据也是「这个桶在不在」,不是「有没有一条运行开始记录」。 + 只看运行开始记录的话,一个先写过意图的标识还能再开一次,而同一段代码换成落盘那个实现 + 会失败。 + + 驱动入口在开工前会先读一次日志判断这个标识有没有用过,但那是先读后写;独占把那个窗口 + 关掉。同一个标识被重开的后果很具体:交错的记录序会让恢复读到同一步的两条意图,判成 + 「日志被并发写过」,于是这次运行从此续不了——而两边的模型调用都已经花过钱了。 + + **失败抛 `FileExistsError`。** 这里没有文件,但调用方要处理的是同一件事——「这个运行 + 标识已经开过了」——而它多半接的是一个由配置决定的存储实现。换一个自造的异常类型的话, + 那段处理代码得先判自己拿到的是哪一种实现,或者干脆两种都捕获。 + """ + stored = _stored(record) + if record.run_id in self._runs: + raise FileExistsError( + f"运行标识 {record.run_id!r} 已经有日志了,不能再开一次。" + "要从这份日志接着跑用 resume" + ) + self._runs[record.run_id] = [stored] + + async def write_intent(self, record: Intent) -> None: + """写一条意图。 + + **没有耐久屏障可言**:这份日志比进程活不长,「必须落盘才能往下走」在这里没有对应物。 + 它仍然满足意图日志在进程内的那一半——意图在副作用之前就读得到了。 + """ + self._append(record) + + async def write_model_call_result(self, record: ModelCallResult) -> None: + 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: + self._append(record) + + # -- 读 ------------------------------------------------------------------ + + async def read_log(self, run_id: str) -> RunLog: + """读回整份日志。**从没写过的运行标识返回空日志,不抛异常。** + + 驱动入口靠这条判断「这个标识是不是已经有日志了」。抛异常的话那个判断就得写成捕获 + 异常,而用捕获异常做流程控制会把真正的存储故障一起吞掉——于是「存储连不上」会被读成 + 「这是一次全新的运行」,然后覆盖式地重跑一遍。 + + **每次读现解一遍,交出去的是一批新对象。** 把桶里那份直接交出去的话,读回来之后改它 + 就能倒着改掉存储里那一份——`RunStarted` 带的参数快照是个 dict,一次普通读取就足以把 + 日志改掉,而续跑的参数漂移判断读的正是这份。 + + **字段类型不对的记录在这里炸,不在写的时候炸。** `encode` 只把字段取出来装进 dict, + 不查类型,所以坏记录写得进去;逐行追加那个实现同理——`json.dumps` 收得下一个本该是 + 整数的字符串,它也要到读的时候才报 `DecodeError`。两个实现的失败时机必须一样,否则 + 同一段下游代码在易失存储上写就红、换成落盘存储要到读才红。 + """ + return assemble_log(entry.decode(entry.payload) for entry in self._runs.get(run_id, ())) + + # -- 内部 ---------------------------------------------------------------- + + def _append(self, record: Record) -> None: + """把一条记录记进它自己那一桶。 + + **不加锁。** 这几行里没有一个 await 点,而 asyncio 只在 await 处切协程,所以同一个事件 + 循环上的并发运行看到的这一段是原子的(跨线程的边界见类 docstring)。逐行追加那个实现 + 要按运行标识加锁,是因为它把写丢进了线程、一条记录可能由不止一次 `os.write` 写完; + 这里没有那个窗口,照抄一把锁只会让读代码的人以为这里有一个不存在的竞争。 + + **先编码再建桶。** 反过来的话,一条编不出来的记录会留下一个空桶,而那个标识从此开不了 + 新运行——`write_run_started` 的独占会撞上它。 + """ + stored = _stored(record) + self._runs.setdefault(record.run_id, []).append(stored) + + +def _stored(record: Record) -> _Stored: + """编成载荷再存,不存调用方手上那个对象的引用。 + + 存引用会有别名 bug:`RunStarted` 带着的参数快照是一个 `Mapping`,调用方构造完之后还能改 + 自己手里那个 dict,而读回来的快照会跟着变(`design/0014` 决策五)。存载荷把这一半和读那 + 一半一起堵上,形态也和逐行追加那个实现对齐——那边存的是文本,同样是读的时候才解。 + + **只编不解。** 解一遍再把结果丢掉能让字段类型不对的记录在写入时就炸,但逐行追加那个实现 + 在写入时不炸,于是同一条记录在两个实现上一个写得进去一个写不进去。等价的对齐点是读: + 两边都放行、两边都在 `read_log` 抛同一个 `DecodeError`。 + """ + # 本层不认得的记录类在这里抛 `KeyError`,和逐行追加那个实现查类型标签的位置对齐; + # `encode` 不认得的记录类型在下一行抛 `TypeError`。两者都发生在建桶之前,所以写失败 + # 不留下半个桶。 + decode = DECODERS_BY_TYPE[type(record)] + return _Stored(decode, encode(record)) diff --git a/tests/unit/test_stores.py b/tests/unit/test_stores.py index 73fd8a6..cfb4a9d 100644 --- a/tests/unit/test_stores.py +++ b/tests/unit/test_stores.py @@ -1,8 +1,12 @@ -"""逐行追加那份存储实现的行为。 +"""两份存储实现的行为:逐行追加进文件的那个,以及只留在进程内存里的那个。 -**行为契约本身在 `tests/contract/test_run_store.py`**,那套套件现在就接着这个实现跑。这里只写 -契约套件覆盖不到的部分:文件长什么样、坏行怎么算、`fsync` 在哪几处、以及那条契约测试明说 -「这一层验不了」的原子性——它点名要在这里用可注入的故障点补上。 +**行为契约本身在契约套件里**,那套套件不针对任何具体实现,写的是「不管你怎么实现,都必须满足 +这些行为」。这里只写套件覆盖不到的部分:文件长什么样、坏行怎么算、`fsync` 在哪几处、那条契约 +测试明说「这一层验不了」的原子性(它点名要在这里用可注入的故障点补上),以及易失那个实现独有 +的几条——它的独占判据、它靠存载荷换来的两个方向的别名免疫。 + +还有几条跨着两个实现跑:坏记录在哪一步炸、写入失败留不留痕。那种「两个实现必须一样」的事只在 +这里守得住——契约套件对每个实现分别跑,两边各自全绿并不代表它们一致。 """ import asyncio @@ -12,9 +16,9 @@ from pathlib import Path import pytest -from polyloop import stores from polyloop.serialization import DecodeError -from polyloop.stores import RECORD_KEY, JsonlRunStore +from polyloop.stores import RECORD_KEY, JsonlRunStore, VolatileRunStore +from polyloop.stores import _jsonl as jsonl_module from polyloop.types import ( ActionOutcome, ActionStatus, @@ -248,7 +252,7 @@ async def test_an_unknown_record_type_is_refused_not_skipped(tmp_path: Path) -> async def test_a_write_that_dies_halfway_leaves_nothing_readable(tmp_path: Path) -> None: """崩在一次写中途,那条记录**整条不可见**,不是半条可见。 - 这是 `tests/contract/test_run_store.py` 里那条 `xfail` 点名要在这一层补的:契约套件跑在 + 这是 `polyloop.testing.RunStoreContract` 里那条跳过点名要在这一层补的:契约套件跑在 一个进程里、面对一个已经装配好的实现,没有位置插入那次崩溃。这里靠替换掉那个内部的 「把这些字节写进去」来造它。 @@ -456,7 +460,7 @@ async def test_creating_the_log_file_syncs_the_directory_entry( 这条持久性没有别的进程内可观测形态,所以只能盯着那次调用本身。 """ synced: list[Path] = [] - monkeypatch.setattr(stores, "_fsync_directory", synced.append) + monkeypatch.setattr(jsonl_module, "_fsync_directory", synced.append) store = JsonlRunStore(directory=tmp_path) await store.write_run_started(_started()) @@ -487,3 +491,199 @@ async def test_concurrent_writes_to_one_run_do_not_interleave(tmp_path: Path) -> log = await store.read_log("r1") assert len(log.intents) == 6 + + +# --------------------------------------------------------------------------- +# 易失存储:只在进程内存里 +# --------------------------------------------------------------------------- + + +async def test_a_volatile_store_reads_back_what_it_wrote() -> None: + """写进去的读得回来,五种记录都算数。 + + 「不提供恢复」说的是跨进程那一层。进程内读不回来的话它连自家的准入标准都过不了——契约 + 套件第一条要的就是「写进去的意图读得回来」。 + """ + store = VolatileRunStore() + result = RunResult( + run_id="r1", stop_reason=StopReason.TASK_COMPLETED, final_answer="42", steps=() + ) + model_result = ModelCallResult( + run_id="r1", + result_id="m0", + reply=ModelReply(call_id="c1", content="hi", thinking=""), + failure=None, + ) + finished = RunFinished(run_id="r1", result=result) + + await store.write_run_started(_started()) + await store.write_intent(_intent()) + await store.write_model_call_result(model_result) + await store.write_step_completed(_step()) + await store.write_run_finished(finished) + + log = await store.read_log("r1") + + assert log.started == _started() + assert log.intents == (_intent(),) + assert log.model_results == (model_result,) + assert log.steps == (_step(),) + assert log.finished == finished + + +async def test_a_volatile_store_keeps_two_runs_apart() -> None: + """按运行标识分桶,两次运行互相看不见对方的记录。 + + 端口不许持有「当前运行」的隐式状态:那种实现会在并发下把 A 的意图写进 B 的日志,而这种 + 错在单线程测试里永远不出现。 + """ + store = VolatileRunStore() + + await store.write_intent(_intent("run-a", call_index=0)) + await store.write_intent(_intent("run-b", call_index=1)) + + assert (await store.read_log("run-a")).intents == (_intent("run-a", call_index=0),) + assert (await store.read_log("run-b")).intents == (_intent("run-b", call_index=1),) + + +async def test_a_volatile_store_refuses_to_start_the_same_run_twice() -> None: + """独占和逐行追加那个实现对齐:这个标识已经有日志了就直接失败。 + + 重开一个已经开过的标识,交错的记录序会让恢复读到同一步的两条意图、判成日志被并发写过, + 于是这次运行从此续不了——而两边的模型调用都已经花过钱了。 + """ + store = VolatileRunStore() + await store.write_run_started(_started()) + + with pytest.raises(FileExistsError): + await store.write_run_started(_started()) + + +async def test_a_volatile_store_refuses_to_start_a_run_that_already_has_records() -> None: + """判据是「这个标识有没有日志」,不是「有没有一条运行开始记录」。 + + 逐行追加那边任何一次写都会把文件建出来,于是独占创建会拦下这一种;两处判据不一样的话, + 同一段代码在两个实现上一个通过一个失败。 + """ + store = VolatileRunStore() + await store.write_intent(_intent()) + + with pytest.raises(FileExistsError): + await store.write_run_started(_started()) + + +async def test_an_unwritten_run_reads_back_empty_from_a_volatile_store() -> None: + """驱动入口靠这条判断「这个标识是不是已经有日志了」。 + + 抛异常的话那个判断就得写成捕获异常,而用捕获异常做流程控制会把真正的存储故障一起吞掉。 + """ + store = VolatileRunStore() + + log = await store.read_log("never-written") + + assert log.started is None + assert log.intents == () + assert log.finished is None + + +async def test_mutating_the_snapshot_after_writing_does_not_change_the_log() -> None: + """写进去之后调用方改自己手上那个映射,读回来的还是写进去时那份。 + + 参数快照是一个 `Mapping`,存对象引用就有别名 bug:续跑时拿它和当前装配比对,而它已经跟着 + 调用方后来的改动变了,于是一次真的参数漂移被判成没漂移。逐行追加那个实现因为要序列化成 + JSON 文本,天然免疫这件事;这个实现靠桶里存载荷换到同样的免疫。 + """ + store = VolatileRunStore() + snapshot = {"model": "m-1"} + await store.write_run_started(RunStarted(run_id="r1", parameter_snapshot=snapshot)) + + snapshot["model"] = "m-2" + + log = await store.read_log("r1") + assert log.started is not None + assert log.started.parameter_snapshot == {"model": "m-1"} + + +async def test_a_bad_field_type_lands_in_both_stores_and_fails_on_read_the_same_way( + tmp_path: Path, +) -> None: + """字段类型不对的记录两个实现都收得下,都要到 `read_log` 才抛同一个 `DecodeError`。 + + 落盘那个实现写入时只做 `json.dumps`,而一个本该是整数的字符串是合法 JSON——它写得进去。 + 易失那个实现要是在写入时就解一遍、当场拒绝,同一段下游代码在它上面写就红、换成落盘存储 + 要到读才红,而两个实现的准入标准是同一套契约套件、套件里没有一条覆盖得到这处分歧。 + + 这两个实现在「什么时候炸」上必须一致,所以这条测试对两个都跑,而且比对异常文本本身。 + """ + bad = Intent( + run_id="r1", + kind=IntentKind.MODEL_CALL, + call_index="zero", # type: ignore[arg-type] + result_id="m0", + replay_policy=ReplayPolicy.NEVER, + ) + jsonl = JsonlRunStore(directory=tmp_path) + volatile = VolatileRunStore() + + await jsonl.write_intent(bad) + await volatile.write_intent(bad) + + with pytest.raises(DecodeError) as jsonl_error: + await jsonl.read_log("r1") + with pytest.raises(DecodeError) as volatile_error: + await volatile.read_log("r1") + assert str(volatile_error.value) == str(jsonl_error.value) + + +async def test_a_record_class_this_layer_does_not_know_leaves_no_trace_in_either_store( + tmp_path: Path, +) -> None: + """这一层不认得的记录类两个实现都在写入时拒绝,而且什么都不留下。 + + 「不留痕迹」这一半要紧的地方在易失那边:失败的那次写要是先建了桶,这个标识就再也开不了 + 新运行,`write_run_started` 的独占会撞上那个空桶。落盘那边对应的是「不留下一个空文件」。 + """ + # 步记录是别的记录的字段,不是一条日志行——它编得出来,但存储这一层不收它。 + not_a_record = _step().step + jsonl = JsonlRunStore(directory=tmp_path) + volatile = VolatileRunStore() + + with pytest.raises(KeyError): + await jsonl.write_intent(not_a_record) # type: ignore[arg-type] + with pytest.raises(KeyError): + await volatile.write_intent(not_a_record) # type: ignore[arg-type] + + assert list(tmp_path.iterdir()) == [] + assert (await volatile.read_log("r1")).started is None + await volatile.write_run_started(_started()) + assert (await volatile.read_log("r1")).started == _started() + + +async def test_mutating_a_record_read_back_from_a_volatile_store_does_not_change_the_log() -> None: + """读回来之后改它,再读一次还是原来的值。 + + 桶里存的是载荷,每次读现解出一批新对象。把桶里那份直接交出去的话,一次普通读取就足以 + 篡改日志——`RunStarted` 带的参数快照是个 dict,改它一下,续跑的参数漂移判断读到的就是 + 改过的那份,于是一次真的漂移被判成没漂移。 + """ + store = VolatileRunStore() + await store.write_run_started(RunStarted(run_id="r1", parameter_snapshot={"model": "m-1"})) + + first = await store.read_log("r1") + assert first.started is not None + first.started.parameter_snapshot["model"] = "m-2" # type: ignore[index] + + second = await store.read_log("r1") + assert second.started is not None + assert second.started.parameter_snapshot == {"model": "m-1"} + + +def test_the_volatile_store_reports_only_its_kind() -> None: + """快照里只有形态,没有实例状态。 + + 把「现在攒了几次运行」这类状态写进去的话,同一个存储对象在两次比对之间会给出不同的答案, + 于是一次装配完全没变的续跑被判成参数漂移。 + """ + store = VolatileRunStore() + + assert store.parameters() == {"kind": "volatile"}