feat(stores): 落成逐行追加的日志存储，契约套件第一次真的在跑
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 登记两条：运行标识要带齐现在文件名里那五维，以及这份意图日志和它那份 逐步轨迹是两样东西不要混。
This commit is contained in:
@@ -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"]
|
||||
|
||||
Reference in New Issue
Block a user