rotation: tests, replay-base cutoff state fix, formatting

- tests/test_rotation.py covers both framers: history split, idempotent
  reopen, no-op cases, aged-out reduction to a single snapshot, internal
  snapshot validation, extension-agnostic rotated naming, and the
  disabled default.
- Fix plan_rotation to seed the cutoff state from the replay base
  snapshot (previously asserted when the base already covered the cutoff).
- docs: snapshots carry the schema version in effect at their position,
  keeping replay/migration behavior identical to the unrotated file.

docs/rotation.md

@@ -155,16 +155,18 @@ into memory by `open_and_read`; no second disk read is needed.
 5. **Rewrite the main file in place.** On the locked fd: seek to 0, truncate
    to 0, write the new content, `fdatasync`. The new content is, in order:
    1. A **snapshot of the state at the cutoff** — the replayed state after
-      applying all records with `ts < cutoff`, stamped with the **current
-      schema version**. Its `ts` is the **ts of the last pre-cutoff record**
-      (not the rotation time), and this is exactly the timestamp used in the
-      rotated filename. This snapshot is the new replay base and carries the
-      version forward so migrations are not re-run; it is always written.
+      applying all records with `ts < cutoff`, stamped with the **schema
+      version in effect at the cutoff**. Its `ts` is the **ts of the last
+      pre-cutoff record** (not the rotation time), and this is exactly the
+      timestamp used in the rotated filename. This snapshot is the new replay
+      base and carries the version forward so migrations are not re-run; it is
+      always written.
    2. The retained change records (`ts >= cutoff`), **recreated record by
       record** — no internal snapshots are carried over, even if the original
       file had many in the retained range.
    3. A **final snapshot** of the state after the last retained record,
-      stamped with the current schema version — written **only if** there were
+      stamped with the version of the last retained record — written **only
+      if** there were
       retained change records (and, in line with the existing snapshot policy
       in `kanta/snapshot.py`, only when a meaningful number of changes
       accumulated; a handful of trailing changes need not force one). If no

kanta/kantaimpl.py

@@ -399,7 +399,9 @@ class KantaImpl(PersistenceMixin, Generic[T]):
         )
         if plan is None:
             return content
-        await asyncio.to_thread(execute_rotation, self.filename, self.file, plan, log=log)
+        await asyncio.to_thread(
+            execute_rotation, self.filename, self.file, plan, log=log
+        )
         return plan.new_content
 
     async def close(self) -> None:

kanta/rotation.py

@@ -67,9 +67,7 @@ class _Entry:
         self.end_pos = end_pos
 
 
-def _scan(
-    content: bytes, *, framer: Framer, serializer: Serializer
-) -> list[_Entry]:
+def _scan(content: bytes, *, framer: Framer, serializer: Serializer) -> list[_Entry]:
     """Decode every record in *content* with byte ranges."""
     raw = list(framer.iter_records(content, 0))
     entries: list[_Entry] = []
@@ -131,7 +129,11 @@ def plan_rotation(
         version = snap.v
         m = snap.m
 
-    state_at_cutoff: dict[str, Any] | None = None
+    # The cutoff state starts from the replay base: when the base snapshot
+    # already predates the cutoff, it may itself be the cutoff state.
+    state_at_cutoff: dict[str, Any] | None = (
+        copy.deepcopy(state) if base is not None else None
+    )
     version_at_cutoff = version
     m_at_cutoff = m
     final_version = version
@@ -173,7 +175,9 @@ def plan_rotation(
     # survived to warrant one (mirrors the regular snapshot policy).
     out = bytearray()
     leading = serializer.encode(
-        Snapshot(ts=rotated_ts, v=version_at_cutoff, state=state_at_cutoff, m=m_at_cutoff)
+        Snapshot(
+            ts=rotated_ts, v=version_at_cutoff, state=state_at_cutoff, m=m_at_cutoff
+        )
     )
     out += framer.frame_snapshot(leading, record_offset=0)
     for e in retained:

tests/test_rotation.py

@@ -0,0 +1,200 @@
+"""Tests for retention-based database rotation (docs/rotation.md)."""
+
+from datetime import UTC, datetime, timedelta
+from pathlib import Path
+
+import pytest
+
+from kanta.structs import ChangeRecord, Snapshot
+from tests.support import Data, make_kanta, read_changes
+
+pytestmark = pytest.mark.asyncio
+
+DAY = timedelta(days=1)
+T0 = datetime(2026, 1, 1, tzinfo=UTC)
+
+
+def make_clock(cell: list[datetime]):
+    def clock() -> datetime:
+        return cell[0]
+
+    return clock
+
+
+async def write_history(path: Path, format_config, days: list[int]) -> None:
+    """Write one change per day offset (relative to T0) with a fake clock."""
+    cell = [T0 + (days[0] - 1) * DAY]  # bootstrap predates all history
+    kanta = make_kanta(path, Data, format_config)
+    kanta.clock(make_clock(cell))
+    await kanta.open(log=False)
+    for day in days:
+        cell[0] = T0 + day * DAY
+        with kanta.transaction(f"day{day}", log=False) as data:
+            data.counter += 1
+        await kanta.flush()
+    await kanta.close()
+
+
+def read_all(path: Path, format_config):
+    """All records (changes and snapshots) in file order."""
+    _, serializer_cls = format_config
+    serializer = serializer_cls()
+    framer = serializer.framer_cls()
+    out = []
+    for is_snapshot, payload, _, _ in framer.iter_records(path.read_bytes(), 0):
+        out.append(
+            serializer.decode(payload, type=Snapshot if is_snapshot else ChangeRecord)
+        )
+    return out
+
+
+def rotated_files(path: Path) -> list[Path]:
+    return sorted(path.parent.glob(f"{path.stem}@*.kantadb"))
+
+
+async def test_rotation_splits_history(tmp_path, format_config):
+    path = tmp_path / "data.kantadb"
+    await write_history(path, format_config, days=[-40, -20, -5])
+
+    cell = [T0]
+    kanta = make_kanta(path, Data, format_config, retention=30 * DAY)
+    kanta.clock(make_clock(cell))
+    await kanta.open(log=False)
+    assert kanta.data.counter == 3
+    await kanta.close()
+
+    rotated = rotated_files(path)
+    assert len(rotated) == 1
+
+    # Main file: leading snapshot (ts = last dropped record), the retained
+    # changes, and no final snapshot (too few retained changes).
+    records = read_all(path, format_config)
+    assert isinstance(records[0], Snapshot)
+    assert records[0].ts == T0 - 40 * DAY
+    assert records[0].state["counter"] == 1
+    changes = [r for r in records if isinstance(r, ChangeRecord)]
+    assert [c.a for c in changes] == ["day-20", "day-5"]
+
+    # Rotated file holds exactly the dropped history, ending at the last
+    # dropped record whose ts matches the filename.
+    stamp = (T0 - 40 * DAY).strftime("%Y%m%dT%H%M%S.%fZ")
+    assert rotated[0].name == f"data@{stamp}.kantadb"
+    rrecords = read_all(rotated[0], format_config)
+    assert [r.a for r in rrecords] == ["bootstrap", "day-40"]
+
+
+async def test_rotation_reopens_cleanly_and_does_not_rerotate(tmp_path, format_config):
+    path = tmp_path / "data.kantadb"
+    await write_history(path, format_config, days=[-40, -5])
+
+    cell = [T0]
+    for expected_changes in (["day-5"], ["day-5"]):
+        kanta = make_kanta(path, Data, format_config, retention=30 * DAY)
+        kanta.clock(make_clock(cell))
+        async with kanta:
+            assert kanta.data.counter == 2
+        assert [c.a for c in read_changes(path, format_config)] == expected_changes
+
+    # Second open found a file whose history already fits the window.
+    assert len(rotated_files(path)) == 1
+
+
+async def test_rotation_noop_when_retention_covers_all(tmp_path, format_config):
+    path = tmp_path / "data.kantadb"
+    await write_history(path, format_config, days=[-5])
+    before = path.read_bytes()
+
+    cell = [T0]
+    kanta = make_kanta(path, Data, format_config, retention=30 * DAY)
+    kanta.clock(make_clock(cell))
+    async with kanta:
+        assert kanta.data.counter == 1
+
+    assert rotated_files(path) == []
+    assert path.read_bytes() == before
+
+
+async def test_rotation_aged_out_database_reduces_to_single_snapshot(
+    tmp_path, format_config
+):
+    path = tmp_path / "data.kantadb"
+    await write_history(path, format_config, days=[-40, -35])
+
+    cell = [T0]
+    kanta = make_kanta(path, Data, format_config, retention=30 * DAY)
+    kanta.clock(make_clock(cell))
+    async with kanta:
+        assert kanta.data.counter == 2
+
+    records = read_all(path, format_config)
+    assert len(records) == 1
+    assert isinstance(records[0], Snapshot)
+    assert records[0].state["counter"] == 2
+
+    # Opening again must not rotate the snapshot-only file.
+    before = path.read_bytes()
+    kanta = make_kanta(path, Data, format_config, retention=30 * DAY)
+    kanta.clock(make_clock(cell))
+    async with kanta:
+        assert kanta.data.counter == 2
+    assert path.read_bytes() == before
+    assert len(rotated_files(path)) == 1
+
+
+async def test_rotation_validates_against_internal_snapshots(tmp_path, format_config):
+    path = tmp_path / "data.kantadb"
+    cell = [T0 - 40 * DAY]
+    kanta = make_kanta(path, Data, format_config)
+    kanta.clock(make_clock(cell))
+    await kanta.open(log=False)
+    with kanta.transaction("old", log=False) as data:
+        data.counter = 1
+    await kanta.flush()
+    kanta.request_snapshot()
+    kanta._impl.maybe_snapshot()
+    cell[0] = T0 - 1 * DAY
+    with kanta.transaction("new", log=False) as data:
+        data.counter = 2
+    await kanta.flush()
+    await kanta.close()
+
+    cell[0] = T0
+    kanta = make_kanta(path, Data, format_config, retention=30 * DAY)
+    kanta.clock(make_clock(cell))
+    async with kanta:
+        assert kanta.data.counter == 2
+
+    records = read_all(path, format_config)
+    assert isinstance(records[0], Snapshot)
+    assert records[0].state["counter"] == 1
+    assert [r.a for r in records if isinstance(r, ChangeRecord)] == ["new"]
+
+
+@pytest.mark.parametrize("name", ["data", "data.db", "data.kantadb"])
+async def test_rotated_naming_normalizes_extension(tmp_path, format_config, name):
+    path = tmp_path / name
+    await write_history(path, format_config, days=[-40, -5])
+
+    cell = [T0]
+    kanta = make_kanta(path, Data, format_config, retention=30 * DAY)
+    kanta.clock(make_clock(cell))
+    async with kanta:
+        pass
+
+    stamp = (T0 - 40 * DAY).strftime("%Y%m%dT%H%M%S.%fZ")
+    assert (tmp_path / f"data@{stamp}.kantadb").exists()
+
+
+async def test_rotation_disabled_by_default(tmp_path, format_config):
+    path = tmp_path / "data.kantadb"
+    await write_history(path, format_config, days=[-40, -5])
+    before = path.read_bytes()
+
+    cell = [T0]
+    kanta = make_kanta(path, Data, format_config)
+    kanta.clock(make_clock(cell))
+    async with kanta:
+        assert kanta.data.counter == 2
+
+    assert rotated_files(path) == []
+    assert path.read_bytes() == before