fix(stores): 修 Codex 对抗审查报的五条，其中两条同一根因

最实的一条：读取端只要一段能解析成 JSON 就收下，没检查它后面有没有换行。而短写完全可能正好写完整个 JSON 对象、只差那个换行——那次写从来没被确认过（调用方的 await 还没返回），按契约就是「没发生」，但它会被当成一条有效的动作意图读回来，恢复据此判成「状态未知」并可能重放，而那个动作一定没执行过（调用方是在写意图返回之后才去执行的）。判据改成「这一行有没有被换行终结」，不是「能不能解析」。同一个改动顺带修掉第三条：一行完整终结的坏行（比如被外部追加的 {}）此前会被当成撕裂尾行吞掉，读成「少了一条记录但看起来完整」的日志；现在终结过的行解不开就是损坏，直接报错。其余三条： - 新建日志文件不 fsync 父目录。os.fsync(fd) 刷的是文件内容，刷不到「这个目录里多了一个文件」这条目录项；掉电后内容可能在而文件不存在，read_log 走「文件不存在」返回空日志，驱动入口判成全新运行，一次已经花过钱的运行静默没了留痕。只在新建时刷。 - 同一运行标识上的并发写会交错：一条记录可能由不止一次 os.write 写完，而 O_APPEND 只保证每次 write 的追加位置原子，保证不了一条逻辑行整体原子。按运行标识加锁串起来（不同运行照样并行），跨进程那一半仍靠独占创建挡。有一条用短写逼出那个窗口的测试。 - 往返测试的 TOTAL_WRITES 是硬编码，而且漏写结束标记它发现不了（恢复会把最后一步之后那次停止判定重演一遍，得出同样结果）。加一条把十次写的记录类型序列整个钉死的测试。
2026-08-10 03:52:35 -04:00
parent 7f6066701e
commit aeb575e0f7
4 changed files with 299 additions and 34 deletions
@@ -12,6 +12,7 @@ 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 (
@@ -394,3 +395,95 @@ def test_the_directory_does_not_enter_the_parameter_snapshot(tmp_path: Path) ->
    记进去只会在换一台机器、挂载点变了的时候报出一次假的漂移，而那次续跑其实完全正常。
    """
    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