From bcf8ccb8307114616e9d7f6abe00a183e8821775 Mon Sep 17 00:00:00 2001
From: Leo Vasanko <git@vasanko.com>
Date: Fri, 28 Aug 2026 00:33:06 +0000
Subject: [PATCH 1/8] Add homepage

---
 pyproject.toml | 1 +
 1 file changed, 1 insertion(+)

diff --git a/pyproject.toml b/pyproject.toml
index a492a0a..7896564 100644
--- a/pyproject.toml
+++ b/pyproject.toml
@@ -30,6 +30,7 @@ bin = [
 ]
 
 [project.urls]
+Homepage = "https://vasanko.com/coders/kanta"
 Repository = "https://git.zi.fi/LeoVasanko/kanta"
 
 [dependency-groups]
-- 
2.55.0


From 379b0de1dc35088bcad46460ee78dd8244c394e0 Mon Sep 17 00:00:00 2001
From: Leo Vasanko <git@vasanko.com>
Date: Wed, 2 Sep 2026 04:06:36 +0000
Subject: [PATCH 2/8] Replace jsondiff producer with own diff; support list
 edits in replay

compute_diff now emits a simple subset of jsondiff's marshaled syntax:
plain key assignment for adds and full-value changes (lists replaced
wholesale), $delete for removed object keys, and $replace only when a
dict replaces a non-dict. Removed keys are always $delete, even when the
object becomes empty (jsondiff emitted $replace: {}).

apply_diff (single implementation in serialization.base, used by both
replay and patch_state) additionally accepts jsondiff-produced patches:
positional $insert/$delete and per-index nested diffs on lists, plus
$ escaping of user keys and values per the marshal format.

jsondiff moves to the dev dependency group; it is used only by tests
that verify patch compatibility in both directions.
---
 kanta/diff.py                   | 101 ++++++++-------
 kanta/logging.py                |  19 ++-
 kanta/serialization/base.py     | 104 ++++++++++++---
 pyproject.toml                  |   2 +-
 tests/test_diff.py              | 216 +++++++++++++++++++++++++++++++-
 tests/test_kanta_integration.py |   5 +-
 tests/test_migrations.py        |   2 +-
 7 files changed, 376 insertions(+), 73 deletions(-)

diff --git a/kanta/diff.py b/kanta/diff.py
index 725ab70..1f4db0d 100644
--- a/kanta/diff.py
+++ b/kanta/diff.py
@@ -1,63 +1,72 @@
-"""Diff computation and replay utilities."""
+"""Diff computation and replay utilities.
 
-import jsondiff
+Diffs use the jsondiff *marshal* format: JSON-serializable dicts where
+command keys are ``$delete``, ``$insert`` and ``$replace``, and any user
+string (key or value) starting with ``$`` is escaped by prepending another
+``$`` (``$foo`` -> ``$$foo``).
+
+Our producer is deliberately simpler than jsondiff's: only object keys are
+added/removed recursively; every other change (scalars, lists, type
+changes) is a plain full-value assignment. Removed keys are always emitted
+as ``$delete``, even when the object becomes empty. The output stays a
+subset of jsondiff's marshaled syntax, so ``jsondiff.patch(...,
+marshal=True)`` can apply it, and our patcher can apply
+jsondiff-produced diffs (including positional ``$insert``/``$delete``
+list edits and per-index nested diffs).
+"""
 
 from kanta.structs import ChangeRecord
-from kanta.serialization.base import ReplayResult, replay
+from kanta.serialization.base import ReplayResult, apply_diff, marshal_value, replay
 from kanta.serialization.framing import LineFramer
 from kanta.serialization.json import JsonSerializer
 
+_UNCHANGED = object()
+
+
+def _escape_key(key: str) -> str:
+    """Escape a user key for use as a diff key (``$foo`` -> ``$$foo``)."""
+    if isinstance(key, str) and key.startswith("$"):
+        return "$" + key
+    return key
+
+
+def _diff(previous, current):
+    """Compute a raw diff, or _UNCHANGED if there is no difference."""
+    if isinstance(previous, dict) and isinstance(current, dict):
+        result = {}
+        deleted = [_escape_key(k) for k in previous if k not in current]
+        if deleted:
+            result["$delete"] = deleted
+        for key, new_value in current.items():
+            ekey = _escape_key(key)
+            if key not in previous:
+                result[ekey] = marshal_value(new_value)
+            else:
+                sub = _diff(previous[key], new_value)
+                if sub is not _UNCHANGED:
+                    result[ekey] = sub
+        return result if result else _UNCHANGED
+    if previous == current:
+        return _UNCHANGED
+    if isinstance(current, dict):
+        # A dict assigned over a non-dict is ambiguous with a nested diff;
+        # like jsondiff, use $replace for that case.
+        return {"$replace": marshal_value(current)}
+    return marshal_value(current)
+
 
 def compute_diff(previous: dict, current: dict) -> dict | None:
-    """Compute a jsondiff patch between two dicts.
+    """Compute a marshaled diff between two state dicts.
 
     Returns None if there is no difference.
     """
-    return jsondiff.diff(previous, current, marshal=True) or None
-
-
-def _apply_diff(state: dict, diff: dict) -> dict:
-    """Apply a jsondiff patch manually, handling ``$replace`` and ``$delete``.
-
-    jsondiff.patch does not handle nested ``$replace`` commands when the
-    parent key is missing from the state.  This function recursively applies
-    diffs, treating ``$replace`` as full replacement and ``$delete`` as
-    key removal.
-    """
-    if not isinstance(diff, dict):
-        return diff
-
-    result = dict(state) if isinstance(state, dict) else state
-    if not isinstance(result, dict):
-        result = {}
-
-    for key, value in diff.items():
-        if key == "$replace":
-            return value
-        elif key == "$delete":
-            if isinstance(value, list):
-                for k in value:
-                    result.pop(k, None)
-            else:
-                result.pop(value, None)
-        elif isinstance(value, dict):
-            old = result.get(key, {})
-            if not isinstance(old, dict):
-                old = {}
-            result[key] = _apply_diff(old, value)
-        else:
-            result[key] = value
-
-    return result
+    diff = _diff(previous, current)
+    return diff if diff is not _UNCHANGED else None
 
 
 def patch_state(state: dict, diff: dict) -> dict:
-    """Apply a jsondiff patch to a state dict.
-
-    The diff was produced with ``marshal=True`` (string keys like
-    ``"$replace"`` and ``"$delete"``) and decoded from JSON.
-    """
-    return _apply_diff(state, diff)
+    """Apply a marshaled diff to a state dict."""
+    return apply_diff(state, diff)
 
 
 # Backward-compatible JSONL replay using the default serializer.
diff --git a/kanta/logging.py b/kanta/logging.py
index b34fe44..2d9fff2 100644
--- a/kanta/logging.py
+++ b/kanta/logging.py
@@ -15,6 +15,7 @@ from typing import Any
 
 import msgspec
 
+from kanta.serialization.base import _apply, unmarshal
 from kanta.tty import Line, displaywidth
 
 transaction_logger = logging.getLogger("kanta.transaction")
@@ -312,6 +313,13 @@ def _collect_changes(
         changes.append(("update" if existed else "add", path, diff))
         return
 
+    old_at_path = _get_nested(previous, path)
+    if isinstance(old_at_path, list):
+        # List edits ($insert/$delete/per-index) are shown as one whole-list
+        # update; the diff is already unmarshaled at this point.
+        changes.append(("update", path, _apply(old_at_path, diff)))
+        return
+
     for key, value in diff.items():
         if key == "$delete":
             if isinstance(value, list):
@@ -340,7 +348,14 @@ def _collect_changes(
                     ("update" if old_collection is not None else "add", path, value)
                 )
         elif isinstance(key, str) and key.startswith("$"):
-            changes.append(("add", path, {key: value}))
+            # Unknown $-command or (post-unmarshal) a user key starting with
+            # "$": treat as a normal key.
+            new_path = path + [str(key)]
+            existed = _get_nested(previous, new_path) is not None
+            if existed:
+                _collect_changes(value, new_path, changes, previous)
+            else:
+                changes.append(("add", new_path, value))
         else:
             new_path = path + [str(key)]
             existed = _get_nested(previous, new_path) is not None
@@ -418,7 +433,7 @@ def format_diff(
     Returns a list of formatted lines (without newlines).
     """
     changes: list[tuple[str, list[str], Any]] = []
-    _collect_changes(diff, [], changes, previous)
+    _collect_changes(unmarshal(diff), [], changes, previous)
     if not changes:
         return []
     lines = []
diff --git a/kanta/serialization/base.py b/kanta/serialization/base.py
index 9008c3e..abb37f3 100644
--- a/kanta/serialization/base.py
+++ b/kanta/serialization/base.py
@@ -121,33 +121,97 @@ def replay(
 
 
 def _patch_state(state: dict, diff: dict) -> dict:
-    return _apply_diff(state, diff)
+    return apply_diff(state, diff)
 
 
-def _apply_diff(state: dict, diff: dict) -> dict:
+_COMMAND_KEYS = frozenset({"$delete", "$insert", "$replace"})
+
+
+def _escape(value: str) -> str:
+    """Escape a user string for the marshaled format (``$x`` -> ``$$x``)."""
+    return "$" + value if value.startswith("$") else value
+
+
+def _unescape(value: str) -> str:
+    """Reverse escaping; command strings pass through unchanged."""
+    if value in _COMMAND_KEYS:
+        return value
+    if value.startswith("$"):
+        return value[1:]
+    return value
+
+
+def marshal_value(value: Any) -> Any:
+    """Escape all ``$``-prefixed strings in a value stored in a diff."""
+    if isinstance(value, dict):
+        return {
+            _escape(k) if isinstance(k, str) else k: marshal_value(v)
+            for k, v in value.items()
+        }
+    if isinstance(value, list):
+        return [marshal_value(v) for v in value]
+    if isinstance(value, str):
+        return _escape(value)
+    return value
+
+
+def unmarshal(diff: Any) -> Any:
+    """Unescape a marshaled diff (keys, values and ``$delete`` entries)."""
+    if isinstance(diff, dict):
+        return {
+            _unescape(k) if isinstance(k, str) else k: unmarshal(v)
+            for k, v in diff.items()
+        }
+    if isinstance(diff, list):
+        return [unmarshal(v) for v in diff]
+    if isinstance(diff, str):
+        return _unescape(diff)
+    return diff
+
+
+def apply_diff(state: Any, diff: Any) -> Any:
+    """Apply a marshaled jsondiff-format diff.
+
+    Mirrors ``jsondiff.patch(..., marshal=True)``: supports ``$replace``,
+    key-based ``$delete`` on dicts, and positional ``$delete``/``$insert``
+    plus per-index nested diffs on lists.
+    """
+    return _apply(state, unmarshal(diff))
+
+
+def _apply(state: Any, diff: Any) -> Any:
     if not isinstance(diff, dict):
         return diff
+    if not diff:
+        return state
+    if "$replace" in diff:
+        return diff["$replace"]
 
-    result = dict(state) if isinstance(state, dict) else state
-    if not isinstance(result, dict):
-        result = {}
+    if isinstance(state, list):
+        result = list(state)
+        deletes = diff.get("$delete")
+        if deletes:
+            for pos in deletes:
+                result.pop(pos)
+        for pos, value in diff.get("$insert", []):
+            result.insert(pos, value)
+        for key, value in diff.items():
+            if key in ("$delete", "$insert"):
+                continue
+            pos = int(key)
+            result[pos] = _apply(result[pos], value)
+        return result
 
+    result = dict(state) if isinstance(state, dict) else {}
     for key, value in diff.items():
-        if key == "$replace":
-            return value
         if key == "$delete":
-            if isinstance(value, list):
-                for k in value:
-                    result.pop(k, None)
-            else:
-                result.pop(value, None)
+            keys = value if isinstance(value, list) else [value]
+            for k in keys:
+                result.pop(k, None)
+        elif key == "$insert":
             continue
-        if isinstance(value, dict):
-            old = result.get(key, {})
-            if not isinstance(old, dict):
-                old = {}
-            result[key] = _apply_diff(old, value)
-            continue
-        result[key] = value
-
+        elif key in result:
+            result[key] = _apply(result[key], value)
+        else:
+            result[key] = value
     return result
diff --git a/pyproject.toml b/pyproject.toml
index 7896564..ac084ea 100644
--- a/pyproject.toml
+++ b/pyproject.toml
@@ -17,7 +17,6 @@ readme = "README.md"
 requires-python = ">=3.11"
 dependencies = [
     "blake3>=1.0.8",
-    "jsondiff>=2.2.1",
     "msgspec>=0.20.0",
 ]
 
@@ -35,6 +34,7 @@ Repository = "https://git.zi.fi/LeoVasanko/kanta"
 
 [dependency-groups]
 dev = [
+    "jsondiff>=2.2.1",
     "pytest>=9.0.2",
     "pytest-asyncio>=1.3.0",
 ]
diff --git a/tests/test_diff.py b/tests/test_diff.py
index 67de6e6..f6fe410 100644
--- a/tests/test_diff.py
+++ b/tests/test_diff.py
@@ -1,8 +1,26 @@
-from kanta.diff import compute_diff
+"""Tests for our own diff producer/consumer and jsondiff compatibility.
+
+jsondiff is a dev dependency used only here, to verify that:
+
+- jsondiff.patch(..., marshal=True) can apply patches produced by
+  compute_diff (our format is a subset of jsondiff's marshaled syntax);
+- apply_diff can apply patches produced by jsondiff.diff(..., marshal=True),
+  including positional $insert/$delete list edits and per-index nested diffs.
+"""
+
+import jsondiff
+import pytest
+
+from kanta.diff import compute_diff, patch_state
+from kanta.logging import format_diff
+from kanta.serialization.base import apply_diff
+
+# --- Producer: compute_diff ------------------------------------------------
 
 
 def test_no_diff():
     assert compute_diff({"a": 1}, {"a": 1}) is None
+    assert compute_diff({}, {}) is None
 
 
 def test_simple_diff():
@@ -14,3 +32,199 @@ def test_simple_diff():
 def test_nested_diff():
     diff = compute_diff({"x": {"y": 1}}, {"x": {"y": 2}})
     assert diff == {"x": {"y": 2}}
+
+
+def test_key_added():
+    assert compute_diff({"a": 1}, {"a": 1, "b": 2}) == {"b": 2}
+
+
+def test_key_removed():
+    assert compute_diff({"a": 1, "b": 2}, {"a": 1}) == {"$delete": ["b"]}
+
+
+def test_last_key_removed_is_delete_not_replace():
+    # jsondiff's minimal-diff search emits {"$replace": {}} here; we emit
+    # what actually happened: the key was deleted.
+    assert compute_diff({"a": 1}, {}) == {"$delete": ["a"]}
+    assert compute_diff({"x": {"y": 1}}, {"x": {}}) == {"x": {"$delete": ["y"]}}
+
+
+def test_list_changes_are_full_assignment():
+    # No $insert/$delete positional edits: lists are replaced wholesale.
+    assert compute_diff({"l": [1, 2]}, {"l": [1, 2, 3]}) == {"l": [1, 2, 3]}
+    assert compute_diff({"l": [1, 2, 3]}, {"l": [1, 3]}) == {"l": [1, 3]}
+    assert compute_diff({"l": [1]}, {"l": []}) == {"l": []}
+
+
+def test_list_with_unchanged_prefix_is_full_assignment():
+    diff = compute_diff({"l": ["a", "b", "c"]}, {"l": ["a", "x", "b", "c"]})
+    assert diff == {"l": ["a", "x", "b", "c"]}
+
+
+def test_type_changes_are_full_assignment():
+    assert compute_diff({"a": {"x": 1}}, {"a": [1]}) == {"a": [1]}
+    # A dict replacing a non-dict needs $replace (a bare dict would read as
+    # a nested diff); this matches jsondiff.
+    assert compute_diff({"a": [1]}, {"a": {"x": 1}}) == {"a": {"$replace": {"x": 1}}}
+    assert compute_diff({"a": 1}, {"a": None}) == {"a": None}
+
+
+def test_new_dict_value_assigned_wholesale():
+    assert compute_diff({}, {"a": {"x": 1}}) == {"a": {"x": 1}}
+
+
+def test_dollar_keys_escaped():
+    assert compute_diff({}, {"$weird": 1}) == {"$$weird": 1}
+    assert compute_diff({"$weird": 1}, {"$weird": 2}) == {"$$weird": 2}
+    assert compute_diff({"$weird": 1}, {}) == {"$delete": ["$$weird"]}
+
+
+def test_dollar_values_escaped():
+    assert compute_diff({"s": 1}, {"s": "$y"}) == {"s": "$$y"}
+    assert compute_diff({"s": 1}, {"s": "$delete"}) == {"s": "$$delete"}
+    # Nested values in wholesale assignments are escaped too.
+    assert compute_diff({}, {"o": {"s": "$y", "l": ["$z"]}}) == {
+        "o": {"s": "$$y", "l": ["$$z"]}
+    }
+
+
+# --- Consumer: apply_diff / patch_state -------------------------------------
+
+
+def test_patch_state_delegates():
+    assert patch_state({"a": 1}, {"a": 2}) == {"a": 2}
+
+
+def test_apply_scalar_and_add():
+    assert apply_diff({"a": 1}, {"a": 2, "b": 3}) == {"a": 2, "b": 3}
+
+
+def test_apply_delete():
+    assert apply_diff({"a": 1, "b": 2}, {"$delete": ["b"]}) == {"a": 1}
+    assert apply_diff({"a": 1}, {"$delete": ["a"]}) == {}
+
+
+def test_apply_replace():
+    assert apply_diff({"a": 1, "b": 2}, {"$replace": {"c": 3}}) == {"c": 3}
+    assert apply_diff({"x": {"a": 1}}, {"x": {"$replace": [1]}}) == {"x": [1]}
+
+
+def test_apply_nested_delete():
+    diff = {"x": {"$delete": ["y"]}}
+    assert apply_diff({"x": {"y": 2, "z": 3}}, diff) == {"x": {"z": 3}}
+
+
+def test_apply_list_insert():
+    diff = {"l": {"$insert": [[1, "x"]]}}
+    assert apply_diff({"l": ["a", "b"]}, diff) == {"l": ["a", "x", "b"]}
+
+
+def test_apply_list_delete():
+    diff = {"l": {"$delete": [1]}}
+    assert apply_diff({"l": ["a", "b", "c"]}, diff) == {"l": ["a", "c"]}
+
+
+def test_apply_list_delete_multiple_positions():
+    # jsondiff emits positions in descending order for sequential pops.
+    diff = {"l": {"$delete": [4, 2, 0]}}
+    assert apply_diff({"l": [0, 1, 2, 3, 4]}, diff) == {"l": [1, 3]}
+
+
+def test_apply_list_insert_and_delete():
+    diff = {"l": {"$insert": [[0, 9], [2, 8], [4, 7]], "$delete": [2, 0]}}
+    assert apply_diff({"l": [0, 1, 2, 3]}, diff) == {"l": [9, 1, 8, 3, 7]}
+
+
+def test_apply_list_per_index_nested_diff():
+    diff = {"l": {"1": {"y": 3}}}
+    state = {"l": [{"x": 1}, {"y": 2}]}
+    assert apply_diff(state, diff) == {"l": [{"x": 1}, {"y": 3}]}
+
+
+def test_apply_escaped_keys_and_values():
+    assert apply_diff({}, {"$$weird": 1}) == {"$weird": 1}
+    assert apply_diff({"$weird": 1}, {"$delete": ["$$weird"]}) == {}
+    assert apply_diff({"s": 1}, {"s": "$$y"}) == {"s": "$y"}
+    assert apply_diff({"s": 1}, {"s": "$$delete"}) == {"s": "$delete"}
+    assert apply_diff({}, {"o": {"s": "$$y", "l": ["$$z"]}}) == {
+        "o": {"s": "$y", "l": ["$z"]}
+    }
+
+
+def test_apply_empty_diff():
+    assert apply_diff({"a": 1}, {}) == {"a": 1}
+
+
+def test_apply_diff_on_missing_state():
+    assert apply_diff({}, {"a": {"b": 1}}) == {"a": {"b": 1}}
+
+
+# --- jsondiff compatibility, both directions --------------------------------
+
+COMPAT_CASES = [
+    ("scalar change", {"a": 1}, {"a": 2}),
+    ("key add", {"a": 1}, {"a": 1, "b": 2}),
+    ("key remove", {"a": 1, "b": 2}, {"a": 1}),
+    ("last key removed", {"a": 1}, {}),
+    ("nested delete", {"a": {"x": 1, "y": 2}}, {"a": {"x": 1}}),
+    ("nested mixed", {"a": {"x": 1, "y": 2}}, {"a": {"x": 9, "z": 3}}),
+    ("list append", {"l": [1, 2]}, {"l": [1, 2, 3]}),
+    ("list insert mid", {"l": [1, 2, 3]}, {"l": [1, 9, 2, 3]}),
+    ("list remove mid", {"l": [1, 2, 3]}, {"l": [1, 3]}),
+    ("list remove many", {"l": [0, 1, 2, 3, 4]}, {"l": [1, 3]}),
+    ("list replace all", {"l": [1, 2]}, {"l": [3, 4]}),
+    ("list insert+delete", {"l": [0, 1, 2, 3]}, {"l": [9, 1, 8, 3, 7]}),
+    ("dict in list", {"l": [{"x": 1}, {"y": 2}]}, {"l": [{"x": 1}, {"y": 3}]}),
+    ("list to empty", {"l": [1]}, {"l": []}),
+    ("type change dict->list", {"a": {"x": 1}}, {"a": [1]}),
+    ("type change list->dict", {"a": [1]}, {"a": {"x": 1}}),
+    ("dollar key", {"$k": 1, "b": 1}, {"$k": 2}),
+    ("dollar value", {"s": "$x"}, {"s": "$y"}),
+    (
+        "deep nesting",
+        {"a": {"b": {"c": {"d": 1, "e": 2}}}},
+        {"a": {"b": {"c": {"d": 9}}}},
+    ),
+]
+
+
+@pytest.mark.parametrize("name,old,new", COMPAT_CASES, ids=[c[0] for c in COMPAT_CASES])
+def test_jsondiff_applies_our_patches(name, old, new):
+    diff = compute_diff(old, new)
+    assert diff is not None
+    assert jsondiff.patch(old, diff, marshal=True) == new
+
+
+@pytest.mark.parametrize("name,old,new", COMPAT_CASES, ids=[c[0] for c in COMPAT_CASES])
+def test_we_apply_jsondiff_patches(name, old, new):
+    diff = jsondiff.diff(old, new, marshal=True)
+    assert apply_diff(old, diff) == new
+
+
+@pytest.mark.parametrize("name,old,new", COMPAT_CASES, ids=[c[0] for c in COMPAT_CASES])
+def test_our_own_round_trip(name, old, new):
+    diff = compute_diff(old, new)
+    assert diff is not None
+    assert apply_diff(old, diff) == new
+
+
+def test_no_diff_means_equal_states():
+    for _name, old, new in COMPAT_CASES:
+        assert compute_diff(old, new) is not None  # cases really differ
+    assert compute_diff({"a": [1, {"b": "$x"}]}, {"a": [1, {"b": "$x"}]}) is None
+
+
+# --- Logging ----------------------------------------------------------------
+
+
+def test_format_diff_list_edit_shows_whole_list():
+    diff = jsondiff.diff({"l": [1, 2, 3]}, {"l": [1, 9, 3]}, marshal=True)
+    lines = format_diff(diff, previous={"l": [1, 2, 3]})
+    text = "\n".join(lines)
+    assert "$insert" not in text
+    assert "[1, 9, 3]" in text
+
+
+def test_format_diff_unescapes_dollar_keys():
+    lines = format_diff({"$$weird": 1}, previous={})
+    assert any("$weird" in line and "$$weird" not in line for line in lines)
diff --git a/tests/test_kanta_integration.py b/tests/test_kanta_integration.py
index 7971575..563ec39 100644
--- a/tests/test_kanta_integration.py
+++ b/tests/test_kanta_integration.py
@@ -47,7 +47,7 @@ async def test_new_file_writes_bootstrap_record_without_handlers(
     records = read_changes(path, format_config)
     assert len(records) == 1
     assert records[0].a == "bootstrap"
-    assert records[0].diff == {"$replace": {"users": {}, "counter": 0}}
+    assert records[0].diff == {"users": {}, "counter": 0}
 
 
 @pytest.mark.asyncio
@@ -63,7 +63,8 @@ async def test_new_file_persists_initial_state_for_roundtrip(tmp_path, format_co
     assert len(records) == 1
     assert records[0].a == "bootstrap"
     assert records[0].diff == {
-        "$replace": {"users": {"alice": {"name": "Alice", "age": 0}}, "counter": 5}
+        "users": {"alice": {"name": "Alice", "age": 0}},
+        "counter": 5,
     }
 
     kanta2 = make_kanta(path, Data, format_config)
diff --git a/tests/test_migrations.py b/tests/test_migrations.py
index 8cd8452..f881063 100644
--- a/tests/test_migrations.py
+++ b/tests/test_migrations.py
@@ -183,7 +183,7 @@ def test_apply_returns_change_information():
     assert result.migrations[0].name == "migrate_v1"
     assert result.migrations[0].description == "Set x"
     assert result.migrations[0].changed is True
-    assert result.migrations[0].diff == {"$replace": {"x": 1}}
+    assert result.migrations[0].diff == {"x": 1}
 
     assert result.migrations[1].name == "migrate_v2"
     assert result.migrations[1].description == "No-op"
-- 
2.55.0


From 1bf52629a34c0c129c25ff84185d253307876f0e Mon Sep 17 00:00:00 2001
From: Leo Vasanko <git@vasanko.com>
Date: Wed, 2 Sep 2026 15:11:52 +0000
Subject: [PATCH 3/8] Simplify diff producer: no $replace, no value escaping

The producer now emits only plain key assignment (full-value replacement
for scalars, lists and type changes, including dict-over-non-dict) and
$delete for removed keys. $-prefixed keys are still escaped as $$;
values are stored verbatim.

The consumer remains compatible with jsondiff-produced diffs ($replace,
positional $insert/$delete, per-index list diffs). A dict diff against
a list state is treated as jsondiff list-ops only when every key is a
command or integer position, otherwise as our wholesale replacement.
Unescaping is narrowed to the $$ prefix so verbatim single-$ values
pass through untouched.
---
 kanta/diff.py               | 36 +++++++++-----------
 kanta/serialization/base.py | 65 +++++++++++++++++++------------------
 tests/test_diff.py          | 46 +++++++++++++++++++++-----
 3 files changed, 86 insertions(+), 61 deletions(-)

diff --git a/kanta/diff.py b/kanta/diff.py
index 1f4db0d..ce7d36c 100644
--- a/kanta/diff.py
+++ b/kanta/diff.py
@@ -1,22 +1,21 @@
 """Diff computation and replay utilities.
 
-Diffs use the jsondiff *marshal* format: JSON-serializable dicts where
-command keys are ``$delete``, ``$insert`` and ``$replace``, and any user
-string (key or value) starting with ``$`` is escaped by prepending another
-``$`` (``$foo`` -> ``$$foo``).
+Diff format: JSON-serializable dicts where ``$delete`` is the only command
+our producer emits; added keys and changed values (scalars, lists, type
+changes — lists always wholesale) are plain assignments. A dict value
+assigned over a non-dict needs no ``$replace``: the consumer can see from
+the old value whether to patch (old is a dict) or replace. User keys
+starting with ``$`` are escaped by prepending another ``$``
+(``$foo`` -> ``$$foo``); values are stored verbatim.
 
-Our producer is deliberately simpler than jsondiff's: only object keys are
-added/removed recursively; every other change (scalars, lists, type
-changes) is a plain full-value assignment. Removed keys are always emitted
-as ``$delete``, even when the object becomes empty. The output stays a
-subset of jsondiff's marshaled syntax, so ``jsondiff.patch(...,
-marshal=True)`` can apply it, and our patcher can apply
-jsondiff-produced diffs (including positional ``$insert``/``$delete``
-list edits and per-index nested diffs).
+The consumer additionally stays compatible with jsondiff's marshaled
+syntax, so it can replay diffs produced by jsondiff itself: ``$replace``,
+positional ``$insert``/``$delete`` and per-index nested diffs on lists,
+and jsondiff's escaping of ``$``-prefixed values.
 """
 
 from kanta.structs import ChangeRecord
-from kanta.serialization.base import ReplayResult, apply_diff, marshal_value, replay
+from kanta.serialization.base import ReplayResult, apply_diff, replay
 from kanta.serialization.framing import LineFramer
 from kanta.serialization.json import JsonSerializer
 
@@ -38,21 +37,16 @@ def _diff(previous, current):
         if deleted:
             result["$delete"] = deleted
         for key, new_value in current.items():
-            ekey = _escape_key(key)
             if key not in previous:
-                result[ekey] = marshal_value(new_value)
+                result[_escape_key(key)] = new_value
             else:
                 sub = _diff(previous[key], new_value)
                 if sub is not _UNCHANGED:
-                    result[ekey] = sub
+                    result[_escape_key(key)] = sub
         return result if result else _UNCHANGED
     if previous == current:
         return _UNCHANGED
-    if isinstance(current, dict):
-        # A dict assigned over a non-dict is ambiguous with a nested diff;
-        # like jsondiff, use $replace for that case.
-        return {"$replace": marshal_value(current)}
-    return marshal_value(current)
+    return current
 
 
 def compute_diff(previous: dict, current: dict) -> dict | None:
diff --git a/kanta/serialization/base.py b/kanta/serialization/base.py
index abb37f3..0970e4f 100644
--- a/kanta/serialization/base.py
+++ b/kanta/serialization/base.py
@@ -124,39 +124,25 @@ def _patch_state(state: dict, diff: dict) -> dict:
     return apply_diff(state, diff)
 
 
-_COMMAND_KEYS = frozenset({"$delete", "$insert", "$replace"})
-
-
-def _escape(value: str) -> str:
-    """Escape a user string for the marshaled format (``$x`` -> ``$$x``)."""
-    return "$" + value if value.startswith("$") else value
-
-
 def _unescape(value: str) -> str:
-    """Reverse escaping; command strings pass through unchanged."""
-    if value in _COMMAND_KEYS:
-        return value
-    if value.startswith("$"):
+    """Reverse jsondiff's ``$$`` escaping; command strings pass through.
+
+    Only a ``$$`` prefix is stripped: jsondiff escapes ``$x`` to ``$$x``,
+    while single ``$`` strings occur verbatim in our own diffs (we do not
+    escape values) and must be left alone.
+    """
+    if value.startswith("$$"):
         return value[1:]
     return value
 
 
-def marshal_value(value: Any) -> Any:
-    """Escape all ``$``-prefixed strings in a value stored in a diff."""
-    if isinstance(value, dict):
-        return {
-            _escape(k) if isinstance(k, str) else k: marshal_value(v)
-            for k, v in value.items()
-        }
-    if isinstance(value, list):
-        return [marshal_value(v) for v in value]
-    if isinstance(value, str):
-        return _escape(value)
-    return value
-
-
 def unmarshal(diff: Any) -> Any:
-    """Unescape a marshaled diff (keys, values and ``$delete`` entries)."""
+    """Unescape a marshaled diff (keys, values and ``$delete`` entries).
+
+    Needed for jsondiff-produced diffs, which escape ``$``-prefixed values
+    as well as keys; our own producer escapes keys only, so unescaping
+    values is a no-op for them.
+    """
     if isinstance(diff, dict):
         return {
             _unescape(k) if isinstance(k, str) else k: unmarshal(v)
@@ -170,15 +156,28 @@ def unmarshal(diff: Any) -> Any:
 
 
 def apply_diff(state: Any, diff: Any) -> Any:
-    """Apply a marshaled jsondiff-format diff.
+    """Apply a diff.
 
-    Mirrors ``jsondiff.patch(..., marshal=True)``: supports ``$replace``,
-    key-based ``$delete`` on dicts, and positional ``$delete``/``$insert``
-    plus per-index nested diffs on lists.
+    Understands our own format (plain assignment + ``$delete``) and
+    jsondiff's marshaled syntax: ``$replace``, positional
+    ``$delete``/``$insert`` and per-index nested diffs on lists. A bare
+    dict over a non-dict old value is a wholesale replacement.
     """
     return _apply(state, unmarshal(diff))
 
 
+def _is_list_patch(diff: dict) -> bool:
+    """Whether a dict diff against a list state is a jsondiff list edit."""
+    for key in diff:
+        if key in ("$delete", "$insert"):
+            continue
+        try:
+            int(key)
+        except (ValueError, TypeError):
+            return False
+    return True
+
+
 def _apply(state: Any, diff: Any) -> Any:
     if not isinstance(diff, dict):
         return diff
@@ -188,6 +187,10 @@ def _apply(state: Any, diff: Any) -> Any:
         return diff["$replace"]
 
     if isinstance(state, list):
+        if not _is_list_patch(diff):
+            # Our own producer replaces a list with a dict (or any other
+            # type) by plain assignment — no $replace wrapper.
+            return diff
         result = list(state)
         deletes = diff.get("$delete")
         if deletes:
diff --git a/tests/test_diff.py b/tests/test_diff.py
index f6fe410..c92dbce 100644
--- a/tests/test_diff.py
+++ b/tests/test_diff.py
@@ -63,9 +63,9 @@ def test_list_with_unchanged_prefix_is_full_assignment():
 
 def test_type_changes_are_full_assignment():
     assert compute_diff({"a": {"x": 1}}, {"a": [1]}) == {"a": [1]}
-    # A dict replacing a non-dict needs $replace (a bare dict would read as
-    # a nested diff); this matches jsondiff.
-    assert compute_diff({"a": [1]}, {"a": {"x": 1}}) == {"a": {"$replace": {"x": 1}}}
+    # A dict replacing a non-dict is a plain assignment too: the consumer
+    # sees from the old value whether to patch (dict) or replace.
+    assert compute_diff({"a": [1]}, {"a": {"x": 1}}) == {"a": {"x": 1}}
     assert compute_diff({"a": 1}, {"a": None}) == {"a": None}
 
 
@@ -79,12 +79,12 @@ def test_dollar_keys_escaped():
     assert compute_diff({"$weird": 1}, {}) == {"$delete": ["$$weird"]}
 
 
-def test_dollar_values_escaped():
-    assert compute_diff({"s": 1}, {"s": "$y"}) == {"s": "$$y"}
-    assert compute_diff({"s": 1}, {"s": "$delete"}) == {"s": "$$delete"}
-    # Nested values in wholesale assignments are escaped too.
+def test_dollar_values_not_escaped():
+    # Only keys are escaped; values are stored verbatim, even "$delete".
+    assert compute_diff({"s": 1}, {"s": "$y"}) == {"s": "$y"}
+    assert compute_diff({"s": 1}, {"s": "$delete"}) == {"s": "$delete"}
     assert compute_diff({}, {"o": {"s": "$y", "l": ["$z"]}}) == {
-        "o": {"s": "$$y", "l": ["$$z"]}
+        "o": {"s": "$y", "l": ["$z"]}
     }
 
 
@@ -144,8 +144,12 @@ def test_apply_list_per_index_nested_diff():
 def test_apply_escaped_keys_and_values():
     assert apply_diff({}, {"$$weird": 1}) == {"$weird": 1}
     assert apply_diff({"$weird": 1}, {"$delete": ["$$weird"]}) == {}
+    # jsondiff escapes $-values as "$$.."; those are unescaped on apply.
     assert apply_diff({"s": 1}, {"s": "$$y"}) == {"s": "$y"}
     assert apply_diff({"s": 1}, {"s": "$$delete"}) == {"s": "$delete"}
+    # Our own producer stores values verbatim; single-$ stays as-is.
+    assert apply_diff({"s": 1}, {"s": "$y"}) == {"s": "$y"}
+    assert apply_diff({"s": 1}, {"s": "$delete"}) == {"s": "$delete"}
     assert apply_diff({}, {"o": {"s": "$$y", "l": ["$$z"]}}) == {
         "o": {"s": "$y", "l": ["$z"]}
     }
@@ -159,6 +163,19 @@ def test_apply_diff_on_missing_state():
     assert apply_diff({}, {"a": {"b": 1}}) == {"a": {"b": 1}}
 
 
+def test_apply_bare_dict_replaces_non_dict():
+    # Our own producer emits no $replace; a dict over a non-dict old value
+    # is a wholesale replacement.
+    assert apply_diff({"a": [1, 2]}, {"a": {"x": 1}}) == {"a": {"x": 1}}
+    assert apply_diff({"a": 5}, {"a": {"x": 1}}) == {"a": {"x": 1}}
+    assert apply_diff({"a": None}, {"a": {"x": 1}}) == {"a": {"x": 1}}
+
+
+def test_apply_list_patch_still_works_on_lists():
+    # jsondiff-style per-index diff keeps list-op semantics on list state.
+    assert apply_diff({"l": [1, 2]}, {"l": {"1": 9}}) == {"l": [1, 9]}
+
+
 # --- jsondiff compatibility, both directions --------------------------------
 
 COMPAT_CASES = [
@@ -188,7 +205,18 @@ COMPAT_CASES = [
 ]
 
 
-@pytest.mark.parametrize("name,old,new", COMPAT_CASES, ids=[c[0] for c in COMPAT_CASES])
+# jsondiff.patch cannot apply our patches for "type change list->dict"
+# (we emit a bare dict where jsondiff needs $replace) and "dollar value"
+# (we do not escape "$"-prefixed values; jsondiff.patch would strip the
+# "$"), so those cases are excluded from this direction.
+JSONDIFF_APPLIES_CASES = [
+    c for c in COMPAT_CASES if c[0] not in {"type change list->dict", "dollar value"}
+]
+
+
+@pytest.mark.parametrize(
+    "name,old,new", JSONDIFF_APPLIES_CASES, ids=[c[0] for c in JSONDIFF_APPLIES_CASES]
+)
 def test_jsondiff_applies_our_patches(name, old, new):
     diff = compute_diff(old, new)
     assert diff is not None
-- 
2.55.0


From 1e43f29eeca0ec49166933a5226d7436b9ac6d75 Mon Sep 17 00:00:00 2001
From: Leo Vasanko <git@vasanko.com>
Date: Wed, 2 Sep 2026 16:14:54 +0000
Subject: [PATCH 4/8] docs: add database rotation design

---
 docs/rotation.md | 261 +++++++++++++++++++++++++++++++++++++++++++++++
 1 file changed, 261 insertions(+)
 create mode 100644 docs/rotation.md

diff --git a/docs/rotation.md b/docs/rotation.md
new file mode 100644
index 0000000..d7756ab
--- /dev/null
+++ b/docs/rotation.md
@@ -0,0 +1,261 @@
+# Database Rotation
+
+Goal: bound the on-disk history of a kanta database to a configurable retention
+window (e.g. the last 30 days) by *rotating* the database file: the old content
+is copied to a timestamped sibling file and the main file is truncated and
+rewritten in place with only the retained history plus fresh snapshots. Normal
+operation stays append-only under the exclusive lock; rotation is the only
+operation that rewrites the file.
+
+## Current facts the design must respect
+
+- The writer holds an exclusive `flock` on the file from `open()` until
+  `close()` (`kanta/filelock.py`). No other process can safely touch the file
+  while a writer has it open.
+- Records are append-only frames. Each `ChangeRecord` carries `ts` (record time)
+  and `m` (modification time); snapshots carry `ts`, `v` (schema version) and
+  `state` (`kanta/structs.py`).
+- Replay reads the whole file, then starts from the **last snapshot**
+  (`framer.scan_last_snapshot`, `serialization/base.py:replay`). Anything before
+  the last snapshot is already logically dead.
+- Snapshot state is validated in tooling: replayed state must equal snapshot
+  state (`kanta/replaylog.py`). A snapshot is therefore a consistency
+  checkpoint, not just an accelerator.
+- `BinFramer` checksums are **offset-keyed** (checksum includes the absolute
+  `record_offset`). A binary frame copied to a different byte offset is
+  corrupted. `LineFramer` (JSONL) has no checksums.
+- There is **no fsync/fdatasync** anywhere; durability currently relies on the
+  OS page cache. Rotation must not make this worse, and should fix it for the
+  rotation path at minimum.
+- Migrations run on open, after replay, against the snapshot/replay version.
+  A snapshot records the version it was written at, so "db already migrated"
+  survives in the snapshot even if the migrations produced no change records.
+
+## Rotated file naming
+
+The history that aged out is preserved at:
+
+```
+{stem}@{ISO-8601 timestamp}.kantadb
+```
+
+- `{stem}` is the original filename with its extension stripped
+  (`Path(filename).stem`).
+- The timestamp is the **ts of the last record dropped by the rotation** (see
+  step 4 — the leading snapshot of the rewritten main file carries the same
+  ts), not the current time. The name tells you exactly which point in history
+  the rotated file ends at. Use a filesystem-safe rendering (e.g.
+  `2026-09-02T15-24-57` — no `:` characters, which are awkward on some
+  filesystems).
+- The rotated name always ends in `.kantadb`, regardless of the original
+  extension. Users may name their databases with no extension, `.kantadb`, or
+  anything else (`.db`, …). Since the rotated name is derived from the *stem*,
+  all of these work uniformly: `data` → `data@2026-09-02T15-24-57.kantadb`,
+  `data.kantadb` → `data@….kantadb`, `data.db` → `data@….kantadb`.
+- Rotated files live in the same directory.
+- Collision: if a rotated file with the same name already exists (rotation
+  rerun over identical history — should be prevented by the eligibility check
+  below, but be defensive), append a disambiguating suffix rather than
+  overwriting.
+
+## Why in-place rewrite (and not rename-and-recreate)
+
+An earlier draft renamed the locked file away and created a fresh file at the
+main path. That opens a race: between the rename and the creation of the new
+file, a second instance can open the (now missing) main path with `O_CREAT`,
+acquire its own lock on the fresh inode, and bootstrap an empty database. The
+rotating instance then cannot lock the path it needs, and two divergent
+databases exist. `flock` is attached to the open file description (inode), not
+the path — renaming never blocks a newcomer.
+
+Instead, rotation **never renames or unlinks the main file and never releases
+its lock**:
+
+- Unix: `ftruncate(fd, 0)` on the open, locked fd is unaffected by the flock
+  and does not affect it. Subsequent writes use `lseek(fd, 0, SEEK_END)` +
+  `os.write` (`filelock.py:226`), which work identically after a truncate, so
+  append-mode operation continues unchanged.
+- Windows: this is also the *more* portable option — the DB is opened with
+  `FILE_SHARE_READ` only (`filelock.py:240`), so renaming the locked file would
+  fail outright on Windows. In-place rewrite only needs `SetFilePointer(0)` +
+  `SetEndOfFile` on a handle we own.
+- The main path therefore exists and remains locked throughout; a second
+  instance opening it at any moment gets either the old content or the new,
+  never a missing or half-created file, and never its own lock.
+
+The only new capability `LockedFile` needs is a `replace_content(data)` method
+(seek 0, truncate, write, fsync) implemented per platform.
+
+## When to rotate: at open time, not at runtime
+
+Rotation happens **inside `Kanta.open()`, after acquiring the lock, before
+replay**, gated by a retention option (see Configuration). Rationale:
+
+- The lock is already held and no background flush loop is running yet, so the
+  file is quiescent — no in-flight `pending_changes`, no concurrent snapshots.
+- Runtime rotation would have to fence the background writer, drain the queue,
+  and prove no record lands in the file after the cutoff was computed. That
+  is a second synchronization protocol for a rare operation; not worth it.
+- Open-time rotation also means rotation never races with `request_snapshot()`
+  or migration snapshot writes, which all happen under the same open() sequence.
+
+Consequence: a database that is never reopened never rotates. Document this;
+for long-running services, rotation takes effect on the next restart.
+
+## Rotation algorithm (under the exclusive lock)
+
+Let `cutoff = now - retention`. Steps 1–3 operate on the bytes already read
+into memory by `open_and_read`; no second disk read is needed.
+
+1. **Check eligibility.** Skip rotation when there is nothing to do:
+   - The file contains **no change records older than `cutoff`** — the
+     retention window already covers all history.
+   - The file contains **no change records at all** (snapshot-only file).
+     Opening a long-untouched database may legitimately rotate it down to a
+     single snapshot (that *is* the intended purge), but once a file has been
+     reduced to just a snapshot, rotating it again would be a pure no-op
+     rewrite. Treat "no change records" as "already fully rotated" and skip.
+
+2. **Find the replay base.** Replay normally starts at the most recent
+   snapshot, but that snapshot's `ts` is likely newer than `cutoff` — replaying
+   from it would silently drop history we intend to keep. Instead, scan
+   **backwards from the end of file**, collecting snapshots newest-first, and
+   pick the oldest snapshot `S` whose `ts <= cutoff` (i.e. walk back past
+   snapshots until one covers the required range, or until start of file). If
+   no such snapshot exists, `S` is "start of file" and the retained range is
+   replayed from the empty initial state.
+   - For `LineFramer` this is a reverse scan for `\nSNAPSHOT ` lines.
+   - For `BinFramer` frames are forward-scannable only; keep the forward scan
+     but record every snapshot position, then pick from the collected list.
+
+3. **Replay and validate.** Replay from `S` (or start of file) forward to end
+   of file, keeping every record with `ts >= cutoff`. At **every** snapshot
+   encountered after `S`, validate that the replayed state equals the snapshot
+   state; a mismatch means the history is corrupt or the chosen base is wrong —
+   abort rotation (leave the original file untouched) and surface the error.
+   The last snapshot in the file must always validate; if even that fails,
+   rotation must not proceed.
+   - Records with `ts < cutoff` are applied to the replay (they are needed to
+     reach the cutoff state) but not retained in the output.
+   - Remember `cutoff_end`: the byte offset in the original content just after
+     the last record with `ts < cutoff` (frame-boundary aligned). The rotated
+     file will be truncated to this length in step 6.
+
+4. **Copy the original aside.** `shutil.copy2(main_path, rotated_path)` —
+   no lock needed on the copy, and no temporary name: the content is written
+   directly to its final `{stem}@{ts}.kantadb` name. `copy2` preserves
+   metadata and, on filesystems with copy-on-write (btrfs, XFS with reflinks,
+   APFS, …), performs a cheap reflink copy instead of duplicating data; it is
+   also generally faster than re-writing the same bytes from memory. The
+   original bytes remain readable from the locked fd if the copy fails, so a
+   failure here simply aborts rotation.
+
+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.
+   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
+      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
+      records survived the cutoff, the new file consists of the single leading
+      snapshot and nothing else — this is the steady state for databases whose
+      history has fully aged out, and the eligibility check in step 1 prevents
+      re-rotating such files.
+
+6. **Trim the rotated copy.** Truncate `{stem}@{ts}.kantadb` to `cutoff_end`
+   bytes, so it contains **only the dropped history** and does not duplicate
+   the records retained in the main file. The cut is at a frame boundary, so
+   the rotated file remains a valid, replayable database on its own (it is a
+   prefix of a valid log). This truncation happens only after step 5's fsync,
+   so until then the rotated file still holds the complete original content as
+   a crash-recovery anchor.
+
+7. **Continue normal open.** Replay/migrations proceed on the same locked fd.
+   Because the leading snapshot carries the current version, migrations run
+   exactly as they would have against the old content.
+
+Failure rule: any error before step 5 leaves the main file byte-identical
+(only an extra copy exists). A crash during step 5 may leave the main file
+torn, but the rotated copy still holds the complete original content
+(truncated only after the main file is durable) — recovery is copying it back.
+After step 6 the split is complete and both files are consistent.
+
+## Verbatim copy or rewrite?
+
+**Rewrite (re-frame), not verbatim copy**, for all records written to the main
+file:
+
+- `BinFramer` checksums include `record_offset`, so a verbatim byte copy to a
+  new offset is unreadable. Binary records must be re-framed at their new
+  offsets regardless.
+- Rewriting also normalizes encoding drift and lets us drop the redundant
+  intermediate snapshots the original file accumulated: none of them are
+  carried over — the new file contains only the leading cutoff snapshot, the
+  recreated change records, and (conditionally) the final snapshot.
+
+The rotated copy is the one place where verbatim bytes are used — a raw
+`copy2` plus a frame-aligned tail truncation — which is safe precisely because
+it preserves original offsets (the truncated prefix keeps every frame at its
+original `record_offset`, so binary checksums stay valid).
+
+## Configuration
+
+Add keyword options to `Kanta(...)` (`kanta/kanta.py`), surfaced through
+`open()`:
+
+- `retention: timedelta | None = None` — history window to keep. `None`
+  (default) disables rotation entirely; current behavior is unchanged.
+- `rotate_keep: int = 3` (optional, later) — how many rotated backups to
+  retain; older ones are pruned at rotation time.
+
+Rotation uses `impl.now()` so the `@Kanta.clock` test clock controls it, same
+as record timestamps.
+
+## Integrity checklist
+
+- Rotation runs under the exclusive lock, before the background writer starts.
+- The main path is never renamed, unlinked, or unlocked during rotation; no
+  bootstrap race with a second instance is possible.
+- Replay base is chosen by walking snapshots backwards until the retained range
+  is covered; replay is validated against every snapshot in range.
+- A leading cutoff snapshot (ts = last pre-cutoff record, current schema
+  version) is always written; a final snapshot is written only when warranted
+  by retained changes.
+- The full original content sits at `{stem}@{ts}.kantadb` before the main file
+  is touched, and is only trimmed to the dropped-history prefix after the
+  rewritten main file is `fdatasync`ed.
+- Rotated files are never deleted by the rotation itself.
+- Any validation failure aborts rotation with the original file intact.
+- Files with no change records (already reduced to a snapshot) are never
+  re-rotated.
+
+## Testing notes
+
+- Use the test clock (`tests/test_clock.py`) to age records past the cutoff.
+- Cover both framers: JSONL rotation and BinFramer rotation (assert the
+  rewritten binary file passes checksum validation and replays identically,
+  and that the truncated rotated prefix still passes checksum validation).
+- Assert state equality before/after rotation, version continuity (no
+  re-migration), correct behavior when no snapshot precedes the cutoff, when
+  the newest snapshot is already older than the cutoff, and when retention
+  covers everything (no-op).
+- Naming: databases named `x`, `x.kantadb`, and `x.db` all rotate to
+  `x@{ts}.kantadb`; the timestamp equals the last dropped record's ts and the
+  leading snapshot's ts.
+- Assert the rotated file ends exactly at the last dropped record's frame
+  boundary (no overlap with the retained history in the main file).
+- No-change files: a snapshot-only database opened with retention set is left
+  untouched (no copy, no rewrite).
+- Aged-out database: all history older than the cutoff → new file contains
+  exactly one snapshot; opening it again performs no rotation.
+- Concurrency: while one instance rotates, a second instance opening the main
+  path must fail with the normal "already locked" error at every stage.
-- 
2.55.0


From 010b690b4774b0954f8b99d0fd592c3d4e0eb378 Mon Sep 17 00:00:00 2001
From: Leo Vasanko <git@vasanko.com>
Date: Wed, 2 Sep 2026 16:21:11 +0000
Subject: [PATCH 5/8] filelock: add replace_content for in-place locked rewrite

Rotation rewrites the database file while holding the exclusive lock;
flock follows the open file description across ftruncate, and on Windows
in-place rewrite avoids share-mode rename restrictions. Also update
rotation doc timestamp format to ISO basic with microseconds.
---
 docs/rotation.md       | 10 +++++---
 kanta/filelock.py      | 50 ++++++++++++++++++++++++++++++++++++++
 tests/test_filelock.py | 55 ++++++++++++++++++++++++++++++++++++++++++
 3 files changed, 111 insertions(+), 4 deletions(-)
 create mode 100644 tests/test_filelock.py

diff --git a/docs/rotation.md b/docs/rotation.md
index d7756ab..7a0405f 100644
--- a/docs/rotation.md
+++ b/docs/rotation.md
@@ -44,13 +44,15 @@ The history that aged out is preserved at:
 - The timestamp is the **ts of the last record dropped by the rotation** (see
   step 4 — the leading snapshot of the rewritten main file carries the same
   ts), not the current time. The name tells you exactly which point in history
-  the rotated file ends at. Use a filesystem-safe rendering (e.g.
-  `2026-09-02T15-24-57` — no `:` characters, which are awkward on some
-  filesystems).
+  the rotated file ends at. Rendered in ISO 8601 basic format with the same
+  microsecond precision as the record's ``ts`` in the database (e.g.
+  `20260902T143000.123456Z`), so the filename matches precisely the ``ts`` of
+  the final line of the rotated file and of the snapshot at the start of the
+  new file.
 - The rotated name always ends in `.kantadb`, regardless of the original
   extension. Users may name their databases with no extension, `.kantadb`, or
   anything else (`.db`, …). Since the rotated name is derived from the *stem*,
-  all of these work uniformly: `data` → `data@2026-09-02T15-24-57.kantadb`,
+  all of these work uniformly: `data` → `data@20260902T143000.123456Z.kantadb`,
   `data.kantadb` → `data@….kantadb`, `data.db` → `data@….kantadb`.
 - Rotated files live in the same directory.
 - Collision: if a rotated file with the same name already exists (rotation
diff --git a/kanta/filelock.py b/kanta/filelock.py
index 369eea7..3166538 100644
--- a/kanta/filelock.py
+++ b/kanta/filelock.py
@@ -83,6 +83,10 @@ if sys.platform == "win32":
     ]
     _kernel32.CloseHandle.restype = wintypes.BOOL
     _kernel32.CloseHandle.argtypes = [wintypes.HANDLE]
+    _kernel32.SetEndOfFile.restype = wintypes.BOOL
+    _kernel32.SetEndOfFile.argtypes = [wintypes.HANDLE]
+    _kernel32.FlushFileBuffers.restype = wintypes.BOOL
+    _kernel32.FlushFileBuffers.argtypes = [wintypes.HANDLE]
 
     def _is_invalid_handle(handle) -> bool:
         return ctypes.c_void_p(handle).value == ctypes.c_void_p(-1).value
@@ -177,6 +181,22 @@ class LockedFile:
         os.lseek(self._fd, current, os.SEEK_SET)
         return end
 
+    def replace_content(self, data: bytes) -> None:
+        """Atomically-ish rewrite the file's content in place, lock retained.
+
+        Seeks to the start, truncates, writes *data* and fsyncs, all on the
+        already-locked descriptor. The path is never unlinked or renamed, so
+        no other process can observe a missing file or acquire its own lock.
+        Used by database rotation.
+        """
+        if self._fd is None:
+            raise RuntimeError("LockedFile.replace_content() called on a closed file")
+
+        if sys.platform == "win32":
+            self._replace_content_win32(data)
+        else:
+            self._replace_content_unix(data)
+
     def close(self) -> None:
         """Release the lock and close the file."""
         if self._fd is None:
@@ -227,6 +247,15 @@ class LockedFile:
         os.lseek(self._fd, 0, os.SEEK_END)
         os.write(self._fd, data)
 
+    def _replace_content_unix(self, data: bytes) -> None:
+        os.lseek(self._fd, 0, os.SEEK_SET)
+        os.ftruncate(self._fd, 0)
+        view = memoryview(data)
+        while view:
+            written = os.write(self._fd, view)
+            view = view[written:]
+        os.fdatasync(self._fd)
+
     # -- Windows -------------------------------------------------------------
 
     def _open_win32(self, path: Path, create: bool, readonly: bool) -> None:
@@ -288,3 +317,24 @@ class LockedFile:
         )
         if not ok:
             raise OSError(f"WriteFile failed: Windows error {ctypes.get_last_error()}")
+
+    def _replace_content_win32(self, data: bytes) -> None:
+        _kernel32.SetFilePointer(self._fd, 0, None, _FILE_BEGIN)
+        written = wintypes.DWORD()
+        ok = _kernel32.WriteFile(
+            self._fd,
+            data,
+            len(data),
+            ctypes.byref(written),
+            None,
+        )
+        if not ok:
+            raise OSError(f"WriteFile failed: Windows error {ctypes.get_last_error()}")
+        if not _kernel32.SetEndOfFile(self._fd):
+            raise OSError(
+                f"SetEndOfFile failed: Windows error {ctypes.get_last_error()}"
+            )
+        if not _kernel32.FlushFileBuffers(self._fd):
+            raise OSError(
+                f"FlushFileBuffers failed: Windows error {ctypes.get_last_error()}"
+            )
diff --git a/tests/test_filelock.py b/tests/test_filelock.py
new file mode 100644
index 0000000..ed8e435
--- /dev/null
+++ b/tests/test_filelock.py
@@ -0,0 +1,55 @@
+"""Tests for LockedFile low-level behaviors."""
+
+from kanta.exceptions import FileLockError
+from kanta.filelock import LockedFile
+
+
+def test_replace_content_rewrites_in_place(tmp_path):
+    path = tmp_path / "data.kantadb"
+    path.write_bytes(b"original content here")
+
+    f = LockedFile()
+    f.open(path)
+    try:
+        f.replace_content(b"new")
+        assert f.size() == 3
+        f.write(b"!")
+    finally:
+        f.close()
+
+    assert path.read_bytes() == b"new!"
+
+
+def test_replace_content_keeps_lock(tmp_path):
+    path = tmp_path / "data.kantadb"
+    path.write_bytes(b"abc")
+
+    f = LockedFile()
+    f.open(path)
+    try:
+        f.replace_content(b"xyz")
+        other = LockedFile()
+        try:
+            other.open(path)
+            raise AssertionError("second open should fail while lock is held")
+        except FileLockError:
+            pass
+    finally:
+        f.close()
+
+
+def test_replace_content_grow_and_shrink(tmp_path):
+    path = tmp_path / "data.kantadb"
+    path.write_bytes(b"x" * 100)
+
+    f = LockedFile()
+    f.open(path)
+    try:
+        f.replace_content(b"")
+        assert f.size() == 0
+        f.replace_content(b"y" * 200)
+        assert f.size() == 200
+    finally:
+        f.close()
+
+    assert path.read_bytes() == b"y" * 200
-- 
2.55.0


From 5a5f8b011c039a4f504a772f48d8756ef40b8b90 Mon Sep 17 00:00:00 2001
From: Leo Vasanko <git@vasanko.com>
Date: Wed, 2 Sep 2026 16:24:26 +0000
Subject: [PATCH 6/8] rotation: implement retention-based history rotation

New kanta.rotation module plans rotation from in-memory bytes: walks
snapshots backwards to a replay base predating the cutoff, validates
replayed state against every snapshot in range, and rebuilds the file as
leading cutoff snapshot + re-framed retained changes (+ final snapshot
when enough changes survived). KantaImpl.open executes the plan under
the exclusive lock before replay: copy2 aside to {stem}@{ts}.kantadb,
in-place locked rewrite with fdatasync, then trim the rotated copy to
the dropped-history prefix. Enabled via Kanta(retention=timedelta).
---
 kanta/kanta.py     |  10 ++-
 kanta/kantaimpl.py |  29 +++++-
 kanta/rotation.py  | 220 +++++++++++++++++++++++++++++++++++++++++++++
 kanta/snapshot.py  |   5 ++
 4 files changed, 262 insertions(+), 2 deletions(-)
 create mode 100644 kanta/rotation.py

diff --git a/kanta/kanta.py b/kanta/kanta.py
index 7e01d91..30feac0 100644
--- a/kanta/kanta.py
+++ b/kanta/kanta.py
@@ -2,7 +2,7 @@
 
 from __future__ import annotations
 import logging
-from datetime import datetime
+from datetime import datetime, timedelta
 from pathlib import Path
 from types import ModuleType, SimpleNamespace
 from typing import Any, Generic, TypeVar
@@ -53,6 +53,7 @@ class Kanta(Generic[T]):
         migrations: ModuleType | str | None = None,
         serializer: Serializer | None = None,
         flush_interval: float = 0.1,
+        retention: timedelta | None = None,
     ):
         """Initialize a Kanta persistence instance.
 
@@ -63,6 +64,12 @@ class Kanta(Generic[T]):
             migrations: Optional migrations module object or import path.
             flush_interval: Background flush interval in seconds.
             serializer: Optional serializer implementation.
+            retention: Optional history retention window. When set, opening the
+                database rotates it: history older than ``now - retention`` is
+                moved to a ``{stem}@{timestamp}.kantadb`` sibling file and the
+                main file is rewritten with a fresh snapshot plus the retained
+                records (see ``docs/rotation.md``). ``None`` (default) disables
+                rotation.
 
         Raises:
             ImportError: If ``migrations`` is a string path that cannot be imported.
@@ -78,6 +85,7 @@ class Kanta(Generic[T]):
             type=data_type,
             migrations=migrations,
             flush_interval=flush_interval,
+            retention=retention,
             kanta=self,
         )
 
diff --git a/kanta/kantaimpl.py b/kanta/kantaimpl.py
index 38b4d6e..220ef3d 100644
--- a/kanta/kantaimpl.py
+++ b/kanta/kantaimpl.py
@@ -6,7 +6,7 @@ import asyncio
 import copy
 import importlib
 import logging
-from datetime import UTC, datetime
+from datetime import UTC, datetime, timedelta
 from types import SimpleNamespace
 from typing import Any, Generic, TypeVar
 
@@ -21,6 +21,7 @@ from kanta.logging import (
 )
 from kanta.migrations import MigrationReport, Migrations
 from kanta.persistence import PersistenceMixin
+from kanta.rotation import execute_rotation, plan_rotation
 from kanta.serialization import restore_data_in_place, struct_to_dict
 from kanta.serialization.base import replay
 
@@ -42,6 +43,7 @@ class KantaImpl(PersistenceMixin, Generic[T]):
         self.data: T = kwargs.pop("data")
         self._kanta = kwargs.pop("kanta", None)
         migrations = kwargs.pop("migrations", None)
+        self.retention: timedelta | None = kwargs.pop("retention", None)
         self.ctx = SimpleNamespace()
         super().__init__(**kwargs)
         self.migrations: Migrations | None = None
@@ -181,6 +183,9 @@ class KantaImpl(PersistenceMixin, Generic[T]):
         # From this point the file is open and must be closed via close().
         self.opened = True
 
+        if content and self.retention is not None and not readonly:
+            content = await self._maybe_rotate(content, log)
+
         if content:
             try:
                 rr = replay(
@@ -375,6 +380,28 @@ class KantaImpl(PersistenceMixin, Generic[T]):
         if not self.readonly:
             self.background_task = asyncio.create_task(self._background_loop())
 
+    async def _maybe_rotate(self, content: bytes, log: bool | logging.Logger) -> bytes:
+        """Rotate history older than the retention window (see docs/rotation.md).
+
+        Runs while the file is locked and quiescent, before replay. Returns
+        the (possibly replaced) content to replay. Rotation failures abort the
+        open with the original file intact.
+        """
+        cutoff = self.now() - self.retention
+        plan = await asyncio.to_thread(
+            plan_rotation,
+            content,
+            framer=self.framer,
+            serializer=self.serializer,
+            cutoff=cutoff,
+            now=self.now(),
+            min_diffs=self.snapshot.min_diffs,
+        )
+        if plan is None:
+            return content
+        await asyncio.to_thread(execute_rotation, self.filename, self.file, plan, log=log)
+        return plan.new_content
+
     async def close(self) -> None:
         """Stop the background task, flush pending changes, and release the file lock."""
         if not self.opened:
diff --git a/kanta/rotation.py b/kanta/rotation.py
new file mode 100644
index 0000000..410e535
--- /dev/null
+++ b/kanta/rotation.py
@@ -0,0 +1,220 @@
+"""Database rotation: bound on-disk history to a retention window.
+
+See docs/rotation.md for the design. All planning happens on the in-memory
+bytes of the database file; the caller (KantaImpl.open) performs the actual
+copy-aside, in-place rewrite and rotated-file trimming under the file lock.
+"""
+
+from __future__ import annotations
+
+import copy
+import logging
+import shutil
+from dataclasses import dataclass
+from datetime import datetime
+from pathlib import Path
+from typing import Any
+
+from kanta.exceptions import DatabaseError
+from kanta.structs import ChangeRecord, Snapshot
+from kanta.serialization.base import Serializer, apply_diff
+from kanta.serialization.framing import Framer
+
+_logger = logging.getLogger(__name__)
+
+
+@dataclass
+class RotationPlan:
+    """Everything needed to execute a rotation on disk."""
+
+    new_content: bytes
+    cutoff_end: int  # byte length of the dropped-history prefix
+    rotated_ts: datetime  # ts of the last dropped change record
+    retained_changes: int
+
+
+def rotated_path_for(path: Path, ts: datetime) -> Path:
+    """Sibling path for the rotated history: ``{stem}@{ISO-basic-ts}.kantadb``.
+
+    The timestamp uses ISO 8601 basic format with microsecond precision so it
+    matches the record ``ts`` values stored in the database. On collision an
+    incrementing suffix is inserted before the extension.
+    """
+    stamp = ts.strftime("%Y%m%dT%H%M%S.%fZ")
+    candidate = path.with_name(f"{path.stem}@{stamp}.kantadb")
+    n = 1
+    while candidate.exists():
+        candidate = path.with_name(f"{path.stem}@{stamp}.{n}.kantadb")
+        n += 1
+    return candidate
+
+
+class _Entry:
+    """One parsed record frame with its byte range."""
+
+    __slots__ = ("is_snapshot", "record", "byte_pos", "end_pos")
+
+    def __init__(
+        self,
+        is_snapshot: bool,
+        record: ChangeRecord | Snapshot,
+        byte_pos: int,
+        end_pos: int,
+    ) -> None:
+        self.is_snapshot = is_snapshot
+        self.record = record
+        self.byte_pos = byte_pos
+        self.end_pos = end_pos
+
+
+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] = []
+    for i, (is_snapshot, payload, _line, byte_pos) in enumerate(raw):
+        end_pos = raw[i + 1][3] if i + 1 < len(raw) else len(content)
+        record = serializer.decode(
+            payload, type=Snapshot if is_snapshot else ChangeRecord
+        )
+        entries.append(_Entry(is_snapshot, record, byte_pos, end_pos))
+    return entries
+
+
+def plan_rotation(
+    content: bytes,
+    *,
+    framer: Framer,
+    serializer: Serializer,
+    cutoff: datetime,
+    now: datetime,
+    min_diffs: int,
+) -> RotationPlan | None:
+    """Plan a rotation of *content*, or return None when there is nothing to do.
+
+    Raises:
+        DatabaseError: If replay from the chosen base snapshot does not match
+            a snapshot found inside the file (corrupt history). Rotation must
+            be aborted and the original file left untouched.
+    """
+    if not content:
+        return None
+    entries = _scan(content, framer=framer, serializer=serializer)
+    changes = [e for e in entries if not e.is_snapshot]
+    if not changes:
+        return None  # snapshot-only file: already fully rotated
+
+    dropped = [e for e in changes if e.record.ts < cutoff]
+    if not dropped:
+        return None  # retention window covers all history
+
+    retained = [e for e in changes if e.record.ts >= cutoff]
+    rotated_ts = dropped[-1].record.ts
+    cutoff_end = retained[0].byte_pos if retained else len(content)
+
+    # Replay base: walk snapshots newest-first and take the first (newest)
+    # one predating the cutoff; fall back to start of file.
+    base: _Entry | None = None
+    for e in reversed([e for e in entries if e.is_snapshot]):
+        if e.record.ts <= cutoff:
+            base = e
+            break
+
+    state: dict[str, Any] = {}
+    version = 0
+    m: datetime | None = None
+    if base is not None:
+        snap = base.record
+        assert isinstance(snap, Snapshot)
+        state = dict(snap.state)
+        version = snap.v
+        m = snap.m
+
+    state_at_cutoff: dict[str, Any] | None = None
+    version_at_cutoff = version
+    m_at_cutoff = m
+    final_version = version
+    final_m = m
+
+    for e in entries:
+        if base is not None and e.byte_pos <= base.byte_pos:
+            continue
+        if e.is_snapshot:
+            snap = e.record
+            assert isinstance(snap, Snapshot)
+            if snap.state != state:
+                raise DatabaseError(
+                    "rotation aborted: replayed state does not match snapshot "
+                    f"at byte {e.byte_pos}",
+                    action="rotate",
+                )
+            if snap.m is not None:
+                m = snap.m
+            continue
+        change = e.record
+        assert isinstance(change, ChangeRecord)
+        state = apply_diff(state, change.diff)
+        version = change.v
+        if change.m is not None:
+            m = change.m
+        if change.ts < cutoff:
+            state_at_cutoff = copy.deepcopy(state)
+            version_at_cutoff = version
+            m_at_cutoff = m
+        final_version = version
+        final_m = m
+
+    # There is at least one dropped change, so the cutoff state is known.
+    assert state_at_cutoff is not None
+
+    # Build the new content: leading cutoff snapshot, retained changes
+    # re-framed at fresh offsets, and a final snapshot when enough changes
+    # 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)
+    )
+    out += framer.frame_snapshot(leading, record_offset=0)
+    for e in retained:
+        payload = serializer.encode(e.record)
+        out += framer.frame_change(payload, record_offset=len(out))
+    if len(retained) >= min_diffs:
+        closing = serializer.encode(
+            Snapshot(ts=now, v=final_version, state=state, m=final_m)
+        )
+        out += framer.frame_snapshot(closing, record_offset=len(out))
+
+    return RotationPlan(
+        new_content=bytes(out),
+        cutoff_end=cutoff_end,
+        rotated_ts=rotated_ts,
+        retained_changes=len(retained),
+    )
+
+
+def execute_rotation(
+    path: Path, file, plan: RotationPlan, *, log: bool | logging.Logger = True
+) -> Path:
+    """Execute a planned rotation on disk. Caller must hold the lock on *file*.
+
+    1. Copy the original content aside to ``{stem}@{ts}.kantadb``.
+    2. Rewrite the locked file in place with the new content and fsync.
+    3. Trim the rotated copy to the dropped-history prefix.
+
+    Returns the rotated file path.
+    """
+    rotated = rotated_path_for(path, plan.rotated_ts)
+    shutil.copy2(path, rotated)
+    file.replace_content(plan.new_content)
+    with open(rotated, "r+b") as f:
+        f.truncate(plan.cutoff_end)
+    if log:
+        _logger.info(
+            "rotated %s: kept %d change record(s), history before %s moved to %s",
+            path,
+            plan.retained_changes,
+            plan.rotated_ts.isoformat(),
+            rotated,
+        )
+    return rotated
diff --git a/kanta/snapshot.py b/kanta/snapshot.py
index e97357b..dd92ebe 100644
--- a/kanta/snapshot.py
+++ b/kanta/snapshot.py
@@ -35,6 +35,11 @@ class SnapshotState:
         """Force snapshot write on next check."""
         self._force_pending = True
 
+    @property
+    def min_diffs(self) -> int:
+        """Minimum accumulated changes before a snapshot may be written."""
+        return self._min_diffs
+
     def record_changes(self, count: int) -> None:
         self.changes += count
 
-- 
2.55.0


From b8285e91709d4ff461c13ac9500820945ec14c35 Mon Sep 17 00:00:00 2001
From: Leo Vasanko <git@vasanko.com>
Date: Wed, 2 Sep 2026 16:30:00 +0000
Subject: [PATCH 7/8] 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       |  14 +--
 kanta/kantaimpl.py     |   4 +-
 kanta/rotation.py      |  14 +--
 tests/test_rotation.py | 200 +++++++++++++++++++++++++++++++++++++++++
 4 files changed, 220 insertions(+), 12 deletions(-)
 create mode 100644 tests/test_rotation.py

diff --git a/docs/rotation.md b/docs/rotation.md
index 7a0405f..8d9f6a6 100644
--- a/docs/rotation.md
+++ b/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
diff --git a/kanta/kantaimpl.py b/kanta/kantaimpl.py
index 220ef3d..952c3c1 100644
--- a/kanta/kantaimpl.py
+++ b/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:
diff --git a/kanta/rotation.py b/kanta/rotation.py
index 410e535..58f97c9 100644
--- a/kanta/rotation.py
+++ b/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:
diff --git a/tests/test_rotation.py b/tests/test_rotation.py
new file mode 100644
index 0000000..08d38a0
--- /dev/null
+++ b/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
-- 
2.55.0


From 3074de2950e09b707cb680c4af633c3a9cfe1c16 Mon Sep 17 00:00:00 2001
From: Leo Vasanko <git@vasanko.com>
Date: Wed, 2 Sep 2026 16:40:13 +0000
Subject: [PATCH 8/8] rotation: second-precision rotated filenames, int-days
 retention

- Rotated filename timestamp is now ISO basic at second precision
  (20260902T143000Z); the exact cutoff ts remains inside the files.
- Kanta(retention=N) accepts a plain int as number of days.
---
 docs/rotation.md       | 18 ++++++++++--------
 kanta/kanta.py         | 15 ++++++++-------
 kanta/kantaimpl.py     |  5 ++++-
 kanta/rotation.py      |  9 +++++----
 tests/test_rotation.py | 18 ++++++++++++++++--
 5 files changed, 43 insertions(+), 22 deletions(-)

diff --git a/docs/rotation.md b/docs/rotation.md
index 8d9f6a6..b0292c3 100644
--- a/docs/rotation.md
+++ b/docs/rotation.md
@@ -44,15 +44,16 @@ The history that aged out is preserved at:
 - The timestamp is the **ts of the last record dropped by the rotation** (see
   step 4 — the leading snapshot of the rewritten main file carries the same
   ts), not the current time. The name tells you exactly which point in history
-  the rotated file ends at. Rendered in ISO 8601 basic format with the same
-  microsecond precision as the record's ``ts`` in the database (e.g.
-  `20260902T143000.123456Z`), so the filename matches precisely the ``ts`` of
-  the final line of the rotated file and of the snapshot at the start of the
-  new file.
+  the rotated file ends at. Rendered in ISO 8601 basic format at second
+  precision (e.g. `20260902T143000Z`). The exact microsecond timestamp of the
+  cutoff remains available inside the file (it is the ``ts`` of the final
+  line of the rotated file and of the snapshot at the start of the new file);
+  a second rotation within the same second cannot occur because rotation
+  requires history to have aged past the cutoff.
 - The rotated name always ends in `.kantadb`, regardless of the original
   extension. Users may name their databases with no extension, `.kantadb`, or
   anything else (`.db`, …). Since the rotated name is derived from the *stem*,
-  all of these work uniformly: `data` → `data@20260902T143000.123456Z.kantadb`,
+  all of these work uniformly: `data` → `data@20260902T143000Z.kantadb`,
   `data.kantadb` → `data@….kantadb`, `data.db` → `data@….kantadb`.
 - Rotated files live in the same directory.
 - Collision: if a rotated file with the same name already exists (rotation
@@ -216,8 +217,9 @@ original `record_offset`, so binary checksums stay valid).
 Add keyword options to `Kanta(...)` (`kanta/kanta.py`), surfaced through
 `open()`:
 
-- `retention: timedelta | None = None` — history window to keep. `None`
-  (default) disables rotation entirely; current behavior is unchanged.
+- `retention: timedelta | int | None = None` — history window to keep; a plain
+  `int` is interpreted as a number of days. `None` (default) disables rotation
+  entirely; current behavior is unchanged.
 - `rotate_keep: int = 3` (optional, later) — how many rotated backups to
   retain; older ones are pruned at rotation time.
 
diff --git a/kanta/kanta.py b/kanta/kanta.py
index 30feac0..5641a67 100644
--- a/kanta/kanta.py
+++ b/kanta/kanta.py
@@ -53,7 +53,7 @@ class Kanta(Generic[T]):
         migrations: ModuleType | str | None = None,
         serializer: Serializer | None = None,
         flush_interval: float = 0.1,
-        retention: timedelta | None = None,
+        retention: timedelta | int | None = None,
     ):
         """Initialize a Kanta persistence instance.
 
@@ -64,12 +64,13 @@ class Kanta(Generic[T]):
             migrations: Optional migrations module object or import path.
             flush_interval: Background flush interval in seconds.
             serializer: Optional serializer implementation.
-            retention: Optional history retention window. When set, opening the
-                database rotates it: history older than ``now - retention`` is
-                moved to a ``{stem}@{timestamp}.kantadb`` sibling file and the
-                main file is rewritten with a fresh snapshot plus the retained
-                records (see ``docs/rotation.md``). ``None`` (default) disables
-                rotation.
+            retention: Optional history retention window, either a
+                :class:`~datetime.timedelta` or a plain number of days. When
+                set, opening the database rotates it: history older than
+                ``now - retention`` is moved to a ``{stem}@{timestamp}.kantadb``
+                sibling file and the main file is rewritten with a fresh
+                snapshot plus the retained records (see ``docs/rotation.md``).
+                ``None`` (default) disables rotation.
 
         Raises:
             ImportError: If ``migrations`` is a string path that cannot be imported.
diff --git a/kanta/kantaimpl.py b/kanta/kantaimpl.py
index 952c3c1..caa02f7 100644
--- a/kanta/kantaimpl.py
+++ b/kanta/kantaimpl.py
@@ -43,7 +43,10 @@ class KantaImpl(PersistenceMixin, Generic[T]):
         self.data: T = kwargs.pop("data")
         self._kanta = kwargs.pop("kanta", None)
         migrations = kwargs.pop("migrations", None)
-        self.retention: timedelta | None = kwargs.pop("retention", None)
+        retention = kwargs.pop("retention", None)
+        if isinstance(retention, int) and not isinstance(retention, bool):
+            retention = timedelta(days=retention)
+        self.retention: timedelta | None = retention
         self.ctx = SimpleNamespace()
         super().__init__(**kwargs)
         self.migrations: Migrations | None = None
diff --git a/kanta/rotation.py b/kanta/rotation.py
index 58f97c9..41d124a 100644
--- a/kanta/rotation.py
+++ b/kanta/rotation.py
@@ -36,11 +36,12 @@ class RotationPlan:
 def rotated_path_for(path: Path, ts: datetime) -> Path:
     """Sibling path for the rotated history: ``{stem}@{ISO-basic-ts}.kantadb``.
 
-    The timestamp uses ISO 8601 basic format with microsecond precision so it
-    matches the record ``ts`` values stored in the database. On collision an
-    incrementing suffix is inserted before the extension.
+    The timestamp uses ISO 8601 basic format at second precision (e.g.
+    ``20260902T143000Z``); the exact microsecond timestamp remains available
+    inside the file if ever needed. On collision an incrementing suffix is
+    inserted before the extension.
     """
-    stamp = ts.strftime("%Y%m%dT%H%M%S.%fZ")
+    stamp = ts.strftime("%Y%m%dT%H%M%SZ")
     candidate = path.with_name(f"{path.stem}@{stamp}.kantadb")
     n = 1
     while candidate.exists():
diff --git a/tests/test_rotation.py b/tests/test_rotation.py
index 08d38a0..204fd3e 100644
--- a/tests/test_rotation.py
+++ b/tests/test_rotation.py
@@ -77,7 +77,7 @@ async def test_rotation_splits_history(tmp_path, format_config):
 
     # 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")
+    stamp = (T0 - 40 * DAY).strftime("%Y%m%dT%H%M%SZ")
     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"]
@@ -181,10 +181,24 @@ async def test_rotated_naming_normalizes_extension(tmp_path, format_config, name
     async with kanta:
         pass
 
-    stamp = (T0 - 40 * DAY).strftime("%Y%m%dT%H%M%S.%fZ")
+    stamp = (T0 - 40 * DAY).strftime("%Y%m%dT%H%M%SZ")
     assert (tmp_path / f"data@{stamp}.kantadb").exists()
 
 
+async def test_retention_accepts_int_days(tmp_path, format_config):
+    path = tmp_path / "data.kantadb"
+    await write_history(path, format_config, days=[-40, -5])
+
+    cell = [T0]
+    kanta = make_kanta(path, Data, format_config, retention=30)
+    kanta.clock(make_clock(cell))
+    async with kanta:
+        assert kanta.data.counter == 2
+
+    assert len(rotated_files(path)) == 1
+    assert [c.a for c in read_changes(path, format_config)] == ["day-5"]
+
+
 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])
-- 
2.55.0