iomgaa
aeb575e0f7
最实的一条：读取端只要一段能解析成 JSON 就收下，没检查它后面有没有换行。而短写完全可能
正好写完整个 JSON 对象、只差那个换行——那次写从来没被确认过（调用方的 await 还没返回），
按契约就是「没发生」，但它会被当成一条有效的动作意图读回来，恢复据此判成「状态未知」并可能
重放，而那个动作一定没执行过（调用方是在写意图返回之后才去执行的）。
判据改成「这一行有没有被换行终结」，不是「能不能解析」。同一个改动顺带修掉第三条：一行完整
终结的坏行（比如被外部追加的 {}）此前会被当成撕裂尾行吞掉，读成「少了一条记录但看起来完整」
的日志；现在终结过的行解不开就是损坏，直接报错。
其余三条：
- 新建日志文件不 fsync 父目录。os.fsync(fd) 刷的是文件内容，刷不到「这个目录里多了一个
文件」这条目录项；掉电后内容可能在而文件不存在，read_log 走「文件不存在」返回空日志，
驱动入口判成全新运行，一次已经花过钱的运行静默没了留痕。只在新建时刷。
- 同一运行标识上的并发写会交错：一条记录可能由不止一次 os.write 写完，而 O_APPEND 只保证
每次 write 的追加位置原子，保证不了一条逻辑行整体原子。按运行标识加锁串起来（不同运行
照样并行），跨进程那一半仍靠独占创建挡。有一条用短写逼出那个窗口的测试。
- 往返测试的 TOTAL_WRITES 是硬编码，而且漏写结束标记它发现不了（恢复会把最后一步之后那次
停止判定重演一遍，得出同样结果）。加一条把十次写的记录类型序列整个钉死的测试。
490 lines
18 KiB
Python
490 lines
18 KiB
Python
"""逐行追加那份存储实现的行为。
|
|
|
|
**行为契约本身在 `tests/contract/test_run_store.py`**，那套套件现在就接着这个实现跑。这里只写
|
|
契约套件覆盖不到的部分：文件长什么样、坏行怎么算、`fsync` 在哪几处、以及那条契约测试明说
|
|
「这一层验不了」的原子性——它点名要在这里用可注入的故障点补上。
|
|
"""
|
|
|
|
import asyncio
|
|
import json
|
|
import os
|
|
from pathlib import Path
|
|
|
|
import pytest
|
|
|
|
from polyloop import stores
|
|
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"}
|
|
|
|
|
|
async def test_a_torn_tail_that_happens_to_parse_is_still_dropped(tmp_path: Path) -> None:
|
|
"""短写正好写完整个 JSON、只差最后那个换行——这条记录照样不算数。
|
|
|
|
判据是「有没有被换行终结」，不是「能不能解析」。照后者判会漏掉一个很具体的场景：那次写
|
|
从来没有被确认过（调用方那个 await 还没返回），而它会被当成一条有效的动作意图读回来，
|
|
恢复据此判成「状态未知」并可能重放——可那个动作一定没执行过，因为调用方是在写意图返回
|
|
之后才去执行的。
|
|
"""
|
|
store = JsonlRunStore(directory=tmp_path)
|
|
await store.write_run_started(_started())
|
|
intact = json.dumps(
|
|
{
|
|
"record": "intent",
|
|
**{
|
|
"run_id": "r1",
|
|
"kind": "action",
|
|
"call_index": 0,
|
|
"result_id": "a0",
|
|
"replay_policy": "never",
|
|
},
|
|
},
|
|
ensure_ascii=False,
|
|
)
|
|
with _log_file(tmp_path).open("a", encoding="utf-8") as handle:
|
|
handle.write(intact) # 完整 JSON，但没有换行
|
|
|
|
log = await store.read_log("r1")
|
|
|
|
assert log.started == _started()
|
|
assert log.intents == ()
|
|
|
|
|
|
async def test_a_complete_bad_line_is_corruption_even_at_the_end(tmp_path: Path) -> None:
|
|
"""被换行终结的坏行是损坏，哪怕它在文件末尾。
|
|
|
|
当成撕裂尾行吞掉的话，一份被外部追加过一行垃圾的日志会读成「少了一条记录但看起来完整」，
|
|
而恢复会照它做判断。
|
|
"""
|
|
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("{}\n") # 合法 JSON、完整终结，但没有类型标签
|
|
|
|
with pytest.raises(DecodeError, match="标签"):
|
|
await store.read_log("r1")
|
|
|
|
|
|
async def test_creating_the_log_file_syncs_the_directory_entry(
|
|
tmp_path: Path, monkeypatch: pytest.MonkeyPatch
|
|
) -> None:
|
|
"""新建文件要把目录项也刷下去，往已有文件追加则不必。
|
|
|
|
`os.fsync(fd)` 刷的是文件内容，刷不到「这个目录里多了一个文件」这条目录项。掉电之后内容
|
|
可能在、而文件根本不存在——那时 read_log 走「文件不存在」返回空日志，驱动入口判成一次全新
|
|
的运行，于是一次已经花过钱的运行静默没了留痕。
|
|
|
|
这条持久性没有别的进程内可观测形态，所以只能盯着那次调用本身。
|
|
"""
|
|
synced: list[Path] = []
|
|
monkeypatch.setattr(stores, "_fsync_directory", synced.append)
|
|
store = JsonlRunStore(directory=tmp_path)
|
|
|
|
await store.write_run_started(_started())
|
|
assert synced == [tmp_path]
|
|
|
|
await store.write_intent(_intent())
|
|
assert synced == [tmp_path] # 追加不改目录项
|
|
|
|
|
|
async def test_concurrent_writes_to_one_run_do_not_interleave(tmp_path: Path) -> None:
|
|
"""同一个运行标识上的写被串起来，一条记录不会被另一条切开。
|
|
|
|
一条记录可能由不止一次 os.write 写完（os.write 允许短写），而 O_APPEND 只保证每一次
|
|
os.write 的追加位置原子。两个协程同时写同一个文件时，一次短写会让两条记录交错成一段谁也
|
|
解不开的字节。
|
|
"""
|
|
store = JsonlRunStore(directory=tmp_path)
|
|
real = store._write_all # noqa: SLF001
|
|
|
|
def _short_write(descriptor: int, payload: bytes) -> None:
|
|
"""每次只写一半，逼出「一条记录两次 write」那个窗口。"""
|
|
real(descriptor, payload[: len(payload) // 2])
|
|
real(descriptor, payload[len(payload) // 2 :])
|
|
|
|
store._write_all = _short_write # noqa: SLF001
|
|
await asyncio.gather(*(store.write_intent(_intent(call_index=index)) for index in range(6)))
|
|
|
|
log = await store.read_log("r1")
|
|
|
|
assert len(log.intents) == 6
|