Stricter versioning: always store migrate version record, file must be within versions included in migrations. Log only for migrations that made changes. Allow deleting older migration functions when no longer required.

README.md

@@ -51,6 +51,80 @@ asyncio.run(main())
 3. Let Kanta flush queued changes to disk in the background.
 4. Use snapshots and replay for fast startup and full history.
 
+## Bootstrap and Open Modes
+
+When `open()` creates a brand-new database, it always writes a single bootstrap
+change record from the initial data object you passed to `Kanta(...)`. The
+simplest bootstrap is therefore the object itself — no extra code is required.
+
+Bootstrap handlers are optional. Use them only when you need to modify the
+initial state at creation time, for example to seed defaults or perform
+expensive/external setup that should happen exactly once:
+
+```python
+kanta = Kanta("data.kantadb", Data())
+
+@kanta.bootstrap(action="seed", user="system")
+def seed_defaults(data) -> None:
+    data.users["admin"] = User(name="Admin")
+
+await kanta.open()
+```
+
+You can also use `@kanta.bootstrap` with no arguments and async handlers:
+
+```python
+@kanta.bootstrap
+async def bootstrap_async(data) -> None:
+    data.counter = 1
+```
+
+Whether or not handlers are registered, exactly one bootstrap change record is
+written when a new database is created. The record contains the initial object,
+or the state after all bootstrap handlers have run. When handlers are present:
+- they run in registration order,
+- bootstrap metadata (`action`, `user`, `mtime`) is taken from the last
+  registration.
+
+If any bootstrap handler raises, Kanta closes and removes the database file,
+then re-raises the error.
+
+`open()` also supports strict open mode:
+
+```python
+await kanta.open(create=False)
+```
+
+With `create=False`, open fails if the database file does not exist or is
+empty.
+
+Read-only mode opens an existing database without locking it or starting the
+background flush task. This is useful for readers that must not block the
+writer or modify the file:
+
+```python
+await kanta.open(readonly=True)
+```
+
+In read-only mode, records are replayed and migrations are applied in memory,
+but transactions and explicit flushes are rejected and the file is never
+created if missing.
+
+## Fatal Error Handlers
+
+Fatal background write errors can be observed with a decorator:
+
+```python
+import os
+import signal
+
+@kanta.fatal_error
+async def on_fatal(err):
+    os.kill(os.getpid(), signal.SIGTERM)  # Die
+```
+
+Multiple fatal handlers are supported and run in registration order.
+
 ## Migrations
 
 Adding or removing a field and other such simple operations are automatic, but when the time comes to really change your data model, implement a `migrate_v1` function that converts your old data to the new form. This works on plain built-in dict and other types, to avoid needing to preserve old versions of your structs.

__init__.py

@@ -1,16 +0,0 @@
-"""Project-root shim package for local development layout.
-
-This forwards imports to the inner `kanta/` package directory so
-`from kanta import ...` works when running tests from the workspace root.
-"""
-
-import importlib
-from pathlib import Path
-
-_inner_pkg = Path(__file__).with_name("kanta")
-if str(_inner_pkg) not in __path__:
-    __path__.append(str(_inner_pkg))
-
-_pkg = importlib.import_module(".kanta", __name__)
-__all__ = list(getattr(_pkg, "__all__", ()))
-globals().update({name: getattr(_pkg, name) for name in __all__})

docs/database.md

@@ -82,15 +82,30 @@ history.
 ## Transaction Semantics
 
 - `kanta.transaction(action=...)` captures a pre-transaction snapshot dict.
+- By default a transaction updates the modification time `m` to the current UTC
+  time.
+- `mtime=True|False|datetime` controls the modification time `m`:
+  - `True` (default) sets `m` to the current UTC time.
+  - `False` omits `m`, leaving the previous modification time in effect.
+  - A `datetime` sets `m` to that explicit value.
+- System operations such as `migrate:msgspec` use `mtime=False` so they are not
+  considered modifications and do not advance `m`.
 - On success:
   - compute diff between previous builtins and current builtins,
-  - queue a `ChangeRecord` if non-empty.
+  - queue a `ChangeRecord` if non-empty,
+  - update `kanta.mtime` when the change carries an `m` value.
 - On exception:
   - restore in-memory data from snapshot,
   - re-raise the exception.
 
 Nested transactions are rejected.
 
+## Modification Time
+
+`kanta.mtime` exposes the last modification time carried forward from change
+records. It is updated by normal transactions and preserved across snapshots and
+reloads, while system operations such as migrations leave it unchanged.
+
 ## Flush and Lifecycle
 
 - Writes are queued in memory.
@@ -99,6 +114,89 @@ Nested transactions are rejected.
 - `kanta.close()` performs final flush and releases file resources.
 - `async with Kanta(...)` guarantees open/close lifecycle management.
 
+### Open Modes
+
+- `await kanta.open()` (default) creates the database file if missing.
+- `await kanta.open(create=False)` fails when the file is missing or empty.
+- `await kanta.open(readonly=True)` opens an existing database read-only.
+  - The file is opened without acquiring a lock and without a background flush
+    task.
+  - Existing records are replayed and migrations are still applied in memory.
+  - Transactions and explicit flushes are rejected.
+  - The file is never created if missing.
+
+### Callbacks
+
+All callbacks are registered via decorators and receive arguments by their
+annotation types.  Parameters without a supported annotation are only allowed
+when they have a default value.
+
+#### Bootstrap Callbacks
+
+- When `open()` creates a new database, it always writes a single bootstrap
+  `ChangeRecord`.
+- The simplest bootstrap is the initial data object passed to `Kanta(...)`;
+  bootstrap callbacks are optional and only needed when you want to modify or
+  enrich that object at creation time.
+- Register callbacks via:
+  - `@kanta.bootstrap`
+  - `@kanta.bootstrap(action=..., user=..., mtime=...)`
+- Bootstrap callbacks may be sync or async.  The live root data object is
+  injected by annotating a parameter with the struct type passed to `Kanta`,
+  and the `Kanta` instance itself can be injected by annotating a parameter
+  with `Kanta`.
+- Multiple bootstrap callbacks are supported:
+  - callbacks execute in registration order,
+  - exactly one bootstrap `ChangeRecord` is queued,
+  - bootstrap metadata (`action`, `user`, `mtime`) is taken from the last
+    callback registration.
+- If no bootstrap callbacks are registered, the bootstrap record still uses
+  `action="bootstrap"` and contains the initial data object.
+- If any bootstrap callback raises, Kanta closes and removes the database file,
+  then re-raises the exception.
+
+#### Fatal Error Handlers
+
+- Fatal background persistence errors can be handled with `@kanta.fatal_error`.
+- Handlers may be sync or async.  The `DatabaseError` is injected by annotating
+  a parameter with `DatabaseError`; `Kanta` may also be injected.
+- Multiple handlers are supported and invoked in registration order.  A failing
+  handler is logged and does not prevent subsequent handlers from running.
+
+#### Transaction Log Formatting
+
+- Logfmt callbacks prettify identifiers in the change log and are registered with
+  `@kanta.logfmt`.
+- A logfmt callback is called for every value Kanta renders: diff values, path
+  components, and the transaction `user`.  It receives the value as its first
+  parameter and optionally a `path: str` parameter with the dot-notation path
+  to the value.  The special path `"$user"` is used when rendering the
+  transaction actor, replacing the old `user_display` parameter.
+- The callback returns `str | None`: a string replaces the default rendering,
+  while `None` means "fall through to the next formatter".
+- State dicts can be injected via `DictPre` (`Annotated[dict, "pre"]`)
+  and `DictPost` (`Annotated[dict, "post"]`); the `Kanta` instance can also be
+  injected.
+- Alternatively, a logfmt callback can be a class inheriting from `LogFmt`; the
+  framework instantiates it with the state dicts and calls its
+  `resolve(value, path) -> str | None` method.
+- Multiple logfmt callbacks are stacked in registration order; the first
+  callback to return a non-`None` result wins.  If none handle a value, Kanta
+  falls back to its default formatting.
+
+The decorator accepts an optional ``path`` so the callback only runs for
+values at that exact path:
+
+```python
+@kanta.logfmt(path="$user")
+def resolve_user(value: str, current: DictPost) -> str | None:
+    return current.get("users", {}).get(value, {}).get("name")
+
+@kanta.logfmt(path="users.uuid-1")
+def resolve_user_key(value: str) -> str | None:
+    return names_by_id.get(value)
+```
+
 ## Migrations
 
 - Migration source is configured on `Kanta(...)` via `migrations=`.

kanta/__init__.py

@@ -1,26 +1,5 @@
-from .diff import compute_diff
-from .diff import replay_jsonl as replay
-from .exceptions import DatabaseError, DataIntegrityError, FileLockError, ReplayError
-from .filelock import LockedFile
 from .kanta import Kanta
-from .logging import configure_logging, format_diff, log_change
-from .serialization import JsonSerializer, MsgPackSerializer
-from .structs import ChangeRecord, Snapshot
 
 __all__ = [
-    "ChangeRecord",
-    "compute_diff",
-    "configure_logging",
-    "DataIntegrityError",
-    "DatabaseError",
-    "FileLockError",
-    "format_diff",
-    "JsonSerializer",
     "Kanta",
-    "LockedFile",
-    "log_change",
-    "MsgPackSerializer",
-    "ReplayError",
-    "replay",
-    "Snapshot",
 ]

kanta/callbacks.py

@@ -0,0 +1,530 @@
+"""Unified decorator-based callback registry for Kanta.
+
+Callbacks are registered once and invoked with arguments filled by their
+annotation types.  Unknown arguments are only permitted when they have a
+default value.
+
+Log formatters are a special case: they are called per value being rendered
+and receive the value plus an optional ``path`` string.  They return
+``str | None``; ``None`` means "fall through to the next formatter".
+"""
+
+from __future__ import annotations
+
+import inspect
+import types
+from collections.abc import Callable
+from dataclasses import dataclass
+from typing import Annotated, Any, Union, get_args, get_origin
+
+from kanta.exceptions import DatabaseError
+
+DictPre = Annotated[dict, "pre"]
+DictPost = Annotated[dict, "post"]
+
+
+class LogFmt:
+    """Base class for stateful logfmt callbacks.
+
+    Subclasses only need to override :meth:`resolve`.  The framework injects
+    ``previous_state`` and ``current_state`` through ``__init__``.
+    """
+
+    def __init__(
+        self,
+        previous: DictPre | None = None,
+        current: DictPost | None = None,
+    ) -> None:
+        self.previous_state = previous
+        self.current_state = current
+
+    def __call__(self, value: Any, path: str) -> str | None:
+        return self.resolve(value, path)
+
+    def resolve(self, value: Any, path: str) -> str | None:
+        """Resolve *value* into a display string.
+
+        The default implementation returns ``None`` so other formatters are
+        tried.
+        """
+        return None
+
+
+@dataclass
+class InjectionContext:
+    """Runtime values available for injection into callbacks."""
+
+    kanta: Any | None = None
+    data: Any | None = None
+    error: DatabaseError | None = None
+    previous_state: dict | None = None
+    current_state: dict | None = None
+
+
+@dataclass
+class _CallbackRegistration:
+    callback: Callable[..., Any]
+    params: list[tuple[str, type]]
+    is_async: bool = False
+
+
+@dataclass
+class _LogFmtFunctionSpec:
+    callback: Callable[..., Any]
+    value_type: type | Any
+    has_path: bool
+    inject_params: list[tuple[str, type]]
+    path: str | None = None
+
+
+@dataclass
+class _LogFmtClassSpec:
+    cls: type[LogFmt]
+    inject_params: list[tuple[str, type]]
+    path: str | None = None
+
+
+class CallbackRegistry:
+    """Stores and invokes callbacks, resolving arguments by annotation."""
+
+    def __init__(
+        self,
+        *,
+        kanta_class: type | None = None,
+        data_type: type | None = None,
+    ) -> None:
+        self._kanta_class = kanta_class
+        self._data_type = data_type
+        self._callbacks: dict[str, list[_CallbackRegistration]] = {
+            "bootstrap": [],
+            "fatal_error": [],
+        }
+        self._logfmt_callbacks: list[_LogFmtFunctionSpec | _LogFmtClassSpec] = []
+
+    def register(
+        self,
+        kind: str,
+        callback: Callable[..., Any],
+        *,
+        path: str | None = None,
+    ) -> Callable[..., Any]:
+        """Register *callback* for *kind* after validating its signature."""
+        if kind == "logfmt":
+            if inspect.isclass(callback):
+                self._logfmt_callbacks.append(
+                    self._validate_logfmt_class(callback, path=path)
+                )
+            else:
+                self._logfmt_callbacks.append(
+                    self._validate_logfmt_function(callback, path=path)
+                )
+            return callback
+
+        if kind not in self._callbacks:
+            raise ValueError(f"unknown callback kind: {kind}")
+
+        if inspect.isclass(callback):
+            raise TypeError(f"{kind} callbacks must be functions, not classes")
+        if not callable(callback):
+            raise TypeError(f"{kind} callback must be callable")
+
+        params = self._validate_function(callback, kind)
+        is_async = inspect.iscoroutinefunction(callback)
+
+        self._callbacks[kind].append(
+            _CallbackRegistration(
+                callback=callback,
+                params=params,
+                is_async=is_async,
+            )
+        )
+        return callback
+
+    async def invoke(
+        self,
+        kind: str,
+        ctx: InjectionContext,
+        *,
+        on_error: Callable[[Exception, Callable[..., Any]], bool | None] | None = None,
+    ) -> list[Any]:
+        """Invoke all callbacks of *kind* with arguments from *ctx*.
+
+        If *on_error* is provided it is called for each exception and may return
+        ``False`` to stop invoking further callbacks.  When *on_error* is not
+        provided the first exception is raised immediately.
+        """
+        results: list[Any] = []
+        for reg in self._callbacks[kind]:
+            try:
+                kwargs = self._build_kwargs(reg.params, ctx)
+                result = reg.callback(**kwargs)
+                if inspect.isawaitable(result):
+                    result = await result
+                results.append(result)
+            except Exception as exc:
+                if on_error is None:
+                    raise
+                if on_error(exc, reg.callback) is False:
+                    break
+        return results
+
+    def has(self, kind: str) -> bool:
+        """Return True if any callback of *kind* is registered."""
+        if kind == "logfmt":
+            return bool(self._logfmt_callbacks)
+        return bool(self._callbacks[kind])
+
+    def build_logfmt(self, ctx: InjectionContext) -> Callable[[Any, str], str | None]:
+        """Build a chained formatter from registered logfmt callbacks."""
+        formatters: list[tuple[Callable[[Any, str], str | None], str | None]] = []
+        for spec in self._logfmt_callbacks:
+            if isinstance(spec, _LogFmtClassSpec):
+                kwargs = self._build_kwargs(spec.inject_params, ctx)
+                instance: Callable[[Any, str], str | None] = spec.cls(**kwargs)
+                formatters.append((instance, spec.path))
+            else:
+                kwargs = self._build_kwargs(spec.inject_params, ctx)
+
+                def make_formatter(
+                    callback: Callable[..., Any] = spec.callback,
+                    value_type: type | Any = spec.value_type,
+                    has_path: bool = spec.has_path,
+                    state_kwargs: dict[str, Any] = kwargs,
+                ) -> Callable[[Any, str], str | None]:
+                    def formatter(value: Any, path: str) -> str | None:
+                        if value_type is str and not isinstance(value, str):
+                            return None
+                        call_kwargs = dict(state_kwargs)
+                        if has_path:
+                            call_kwargs["path"] = path
+                        return callback(value, **call_kwargs)
+
+                    return formatter
+
+                formatters.append((make_formatter(), spec.path))
+
+        def format_value(value: Any, path: str) -> str | None:
+            for fn, pattern in formatters:
+                if pattern is not None and path != pattern:
+                    continue
+                resolved = fn(value, path)
+                if resolved is not None:
+                    return resolved
+            return None
+
+        return format_value
+
+    def _validate_function(
+        self,
+        callback: Callable[..., Any],
+        kind: str,
+    ) -> list[tuple[str, type]]:
+        sig = inspect.signature(callback)
+        params: list[tuple[str, type]] = []
+        for name, param in sig.parameters.items():
+            if param.kind in (param.VAR_POSITIONAL, param.VAR_KEYWORD):
+                raise TypeError(
+                    f"{kind} callback {callback.__name__} must not use "
+                    f"*args or **kwargs"
+                )
+
+            if param.annotation is inspect.Parameter.empty:
+                if param.default is inspect.Parameter.empty:
+                    raise TypeError(
+                        f"{kind} callback {callback.__name__} has parameter "
+                        f"'{name}' without an annotation or default value"
+                    )
+                continue
+
+            ann = self._resolve_raw_annotation(param.annotation, callback)
+            if not self._is_allowed(kind, ann):
+                if param.default is inspect.Parameter.empty:
+                    raise TypeError(
+                        f"{kind} callback {callback.__name__} has parameter "
+                        f"'{name}' with unsupported annotation {ann!r}. "
+                        f"Allowed: {self._allowed_message(kind)}"
+                    )
+                continue
+
+            params.append((name, ann))
+
+        return params
+
+    def _validate_logfmt_function(
+        self,
+        callback: Callable[..., Any],
+        *,
+        path: str | None = None,
+    ) -> _LogFmtFunctionSpec:
+        sig = inspect.signature(callback)
+        if inspect.iscoroutinefunction(callback):
+            raise TypeError("logfmt callbacks must not be async")
+
+        params = list(sig.parameters.items())
+        if not params:
+            raise TypeError(
+                f"logfmt callback {callback.__name__} must accept a value parameter"
+            )
+
+        value_name, value_param = params[0]
+        if value_param.kind in (value_param.VAR_POSITIONAL, value_param.VAR_KEYWORD):
+            raise TypeError(
+                f"logfmt callback {callback.__name__} must not use *args or **kwargs"
+            )
+        if value_param.annotation is inspect.Parameter.empty:
+            raise TypeError(
+                f"logfmt callback {callback.__name__} value parameter "
+                f"'{value_name}' must be annotated as str or Any"
+            )
+        value_ann = self._resolve_raw_annotation(value_param.annotation, callback)
+        value_bare = self._unwrap_optional(value_ann)
+        if value_bare is str:
+            value_type = str
+        elif value_bare is Any:
+            value_type = Any
+        else:
+            raise TypeError(
+                f"logfmt callback {callback.__name__} value parameter "
+                f"'{value_name}' must be annotated as str or Any, got {value_ann!r}"
+            )
+
+        has_path = False
+        inject_params: list[tuple[str, type]] = []
+        for name, param in params[1:]:
+            if param.kind in (param.VAR_POSITIONAL, param.VAR_KEYWORD):
+                raise TypeError(
+                    f"logfmt callback {callback.__name__} must not use "
+                    f"*args or **kwargs"
+                )
+            if param.annotation is inspect.Parameter.empty:
+                if param.default is inspect.Parameter.empty:
+                    raise TypeError(
+                        f"logfmt callback {callback.__name__} has parameter "
+                        f"'{name}' without an annotation or default value"
+                    )
+                continue
+
+            ann = self._resolve_raw_annotation(param.annotation, callback)
+            if name == "path" and self._unwrap_optional(ann) is str:
+                has_path = True
+                continue
+            if self._is_allowed("logfmt", ann):
+                inject_params.append((name, ann))
+                continue
+            if param.default is inspect.Parameter.empty:
+                raise TypeError(
+                    f"logfmt callback {callback.__name__} has parameter "
+                    f"'{name}' with unsupported annotation {ann!r}. "
+                    f"Allowed: str path, {self._allowed_message('logfmt')}"
+                )
+
+        if sig.return_annotation is not inspect.Signature.empty:
+            return_ann = self._resolve_raw_annotation(sig.return_annotation, callback)
+            if not self._is_optional_str(return_ann):
+                raise TypeError(
+                    f"logfmt callback {callback.__name__} must return str | None, "
+                    f"got {return_ann!r}"
+                )
+
+        return _LogFmtFunctionSpec(
+            callback=callback,
+            value_type=value_type,
+            has_path=has_path,
+            inject_params=inject_params,
+            path=path,
+        )
+
+    def _validate_logfmt_class(
+        self,
+        cls: type[LogFmt],
+        *,
+        path: str | None = None,
+    ) -> _LogFmtClassSpec:
+        if not issubclass(cls, LogFmt):
+            raise TypeError("logfmt classes must inherit from kanta.callbacks.LogFmt")
+        if inspect.iscoroutinefunction(cls.__init__):
+            raise TypeError("logfmt class __init__ must not be async")
+
+        sig = inspect.signature(cls.__init__)
+        inject_params: list[tuple[str, type]] = []
+        first = True
+        for name, param in sig.parameters.items():
+            if first and name == "self":
+                first = False
+                continue
+            first = False
+
+            if param.kind in (param.VAR_POSITIONAL, param.VAR_KEYWORD):
+                raise TypeError(
+                    f"logfmt class {cls.__name__}.__init__ must not use "
+                    f"*args or **kwargs"
+                )
+            if param.annotation is inspect.Parameter.empty:
+                if param.default is inspect.Parameter.empty:
+                    raise TypeError(
+                        f"logfmt class {cls.__name__}.__init__ has parameter "
+                        f"'{name}' without an annotation or default value"
+                    )
+                continue
+
+            ann = self._resolve_raw_annotation(param.annotation, cls.__init__)
+            if self._is_allowed("logfmt", ann):
+                inject_params.append((name, ann))
+                continue
+            if param.default is inspect.Parameter.empty:
+                raise TypeError(
+                    f"logfmt class {cls.__name__}.__init__ has parameter "
+                    f"'{name}' with unsupported annotation {ann!r}. "
+                    f"Allowed: {self._allowed_message('logfmt')}"
+                )
+
+        resolve = getattr(cls, "resolve", None)
+        if resolve is None:
+            raise TypeError(f"logfmt class {cls.__name__} must define a resolve method")
+        resolve_sig = inspect.signature(resolve)
+        resolve_params = list(resolve_sig.parameters.items())
+        if not resolve_params or resolve_params[0][0] != "self":
+            raise TypeError(
+                f"logfmt class {cls.__name__}.resolve must have 'self' as first parameter"
+            )
+        if len(resolve_params) < 2:
+            raise TypeError(
+                f"logfmt class {cls.__name__}.resolve must accept a value parameter"
+            )
+
+        value_name, value_param = resolve_params[1]
+        value_ann = self._resolve_raw_annotation(value_param.annotation, resolve)
+        value_bare = self._unwrap_optional(value_ann)
+        if value_bare not in (inspect.Parameter.empty, str, Any):
+            raise TypeError(
+                f"logfmt class {cls.__name__}.resolve value parameter "
+                f"'{value_name}' must be annotated as str or Any, got {value_ann!r}"
+            )
+
+        path_found = False
+        for name, param in resolve_params[2:]:
+            path_ann = self._resolve_raw_annotation(param.annotation, resolve)
+            path_bare = self._unwrap_optional(path_ann)
+            if name == "path" and path_bare in (inspect.Parameter.empty, str):
+                path_found = True
+                break
+        if not path_found:
+            raise TypeError(
+                f"logfmt class {cls.__name__}.resolve must accept a 'path: str' parameter"
+            )
+
+        if resolve_sig.return_annotation is not inspect.Signature.empty:
+            return_ann = self._resolve_raw_annotation(
+                resolve_sig.return_annotation, resolve
+            )
+            if not self._is_optional_str(return_ann):
+                raise TypeError(
+                    f"logfmt class {cls.__name__}.resolve must return str | None, "
+                    f"got {return_ann!r}"
+                )
+
+        return _LogFmtClassSpec(cls=cls, inject_params=inject_params, path=path)
+
+    def _build_kwargs(
+        self,
+        params: list[tuple[str, type]],
+        ctx: InjectionContext,
+    ) -> dict[str, Any]:
+        kwargs: dict[str, Any] = {}
+        for name, ann in params:
+            value = self._resolve_annotation(ann, ctx)
+            if value is _UNRESOLVED:
+                raise RuntimeError(f"no value available for annotation {ann!r}")
+            kwargs[name] = value
+        return kwargs
+
+    def _is_allowed(self, kind: str, ann: Any) -> bool:
+        bare = self._unwrap_optional(ann)
+        if self._matches_state_annotation(bare, "pre"):
+            return kind == "logfmt"
+        if self._matches_state_annotation(bare, "post"):
+            return kind == "logfmt"
+        if bare is DatabaseError:
+            return kind == "fatal_error"
+        if self._data_type is not None and bare is self._data_type:
+            return kind == "bootstrap"
+        if self._kanta_class is not None and bare is self._kanta_class:
+            return kind in {"bootstrap", "fatal_error", "logfmt"}
+        return False
+
+    def _allowed_message(self, kind: str) -> str:
+        parts: list[str] = []
+        if kind == "bootstrap":
+            if self._data_type is not None:
+                parts.append(self._data_type.__name__)
+        if kind in {"bootstrap", "fatal_error", "logfmt"}:
+            if self._kanta_class is not None:
+                parts.append(self._kanta_class.__name__)
+        if kind == "fatal_error":
+            parts.append("DatabaseError")
+        if kind == "logfmt":
+            parts.append("Annotated[dict, 'pre']")
+            parts.append("Annotated[dict, 'post']")
+        return ", ".join(parts) if parts else "none"
+
+    def _resolve_annotation(self, ann: Any, ctx: InjectionContext) -> Any:
+        bare = self._unwrap_optional(ann)
+        if self._matches_state_annotation(bare, "pre"):
+            return ctx.previous_state
+        if self._matches_state_annotation(bare, "post"):
+            return ctx.current_state
+        if bare is DatabaseError:
+            return ctx.error
+        if self._data_type is not None and bare is self._data_type:
+            return ctx.data
+        if self._kanta_class is not None and bare is self._kanta_class:
+            return ctx.kanta
+        return _UNRESOLVED
+
+    def _resolve_raw_annotation(
+        self,
+        raw_ann: Any,
+        callback: Callable[..., Any],
+    ) -> Any:
+        if isinstance(raw_ann, str):
+            try:
+                return eval(raw_ann, callback.__globals__)
+            except Exception as exc:
+                raise TypeError(
+                    f"could not resolve annotation {raw_ann!r} for "
+                    f"{callback.__name__}: {exc}"
+                ) from exc
+        return raw_ann
+
+    @staticmethod
+    def _matches_state_annotation(ann: Any, marker: str) -> bool:
+        origin = get_origin(ann)
+        if origin is not Annotated:
+            return False
+        args = get_args(ann)
+        if not args:
+            return False
+        return args[0] is dict and marker in args[1:]
+
+    @staticmethod
+    def _unwrap_optional(ann: Any) -> Any:
+        origin = get_origin(ann)
+        if origin not in (Union, types.UnionType):
+            return ann
+        args = [arg for arg in get_args(ann) if arg is not type(None)]
+        return args[0] if len(args) == 1 else ann
+
+    @staticmethod
+    def _is_optional_str(ann: Any) -> bool:
+        origin = get_origin(ann)
+        if origin not in (Union, types.UnionType):
+            return ann is str
+        args = get_args(ann)
+        return type(None) in args and any(arg is str for arg in args)
+
+
+class _Unresolved:
+    pass
+
+
+_UNRESOLVED = _Unresolved()

kanta/diff.py

@@ -2,7 +2,7 @@
 
 import jsondiff
 
-from kanta.kanta.structs import ChangeRecord
+from kanta.structs import ChangeRecord
 from kanta.serialization.base import ReplayResult, replay
 from kanta.serialization.framing import LineFramer
 from kanta.serialization.json import JsonSerializer

kanta/filelock.py

@@ -34,6 +34,7 @@ if sys.platform == "win32":
     _GENERIC_READ = 0x80000000
     _GENERIC_WRITE = 0x40000000
     _FILE_SHARE_READ = 0x00000001
+    _FILE_SHARE_WRITE = 0x00000002
     _OPEN_EXISTING = 3
     _OPEN_ALWAYS = 4
     _FILE_ATTRIBUTE_NORMAL = 0x80
@@ -91,12 +92,13 @@ else:
 
 
 class LockedFile:
-    """A file opened with an exclusive write lock.
+    """A file opened for read+write with an optional exclusive lock.
 
     Usage::
 
         f = LockedFile()
         f.open(path)                    # open + lock (read+write)
+        f.open(path, readonly=True)     # open read-only without locking
         content = f.read()              # read entire content
         f.write(data)                   # append data (seeks to end first)
         f.close()                       # release lock + close fd
@@ -108,12 +110,13 @@ class LockedFile:
     def __init__(self) -> None:
         self._fd: int | None = None  # Unix fd or Windows HANDLE
 
-    def open(self, path: Path, *, create: bool = False) -> None:
-        """Open *path* for read+write with an exclusive lock.
+    def open(self, path: Path, *, create: bool = False, readonly: bool = False) -> None:
+        """Open *path* and optionally acquire an exclusive lock.
 
         Args:
             path:   File to open and lock.
             create: If True, create the file if it doesn't exist (bootstrap).
+            readonly: If True, open read-only without acquiring a lock.
 
         Raises:
             FileLockError: If the file is locked by another process or not found.
@@ -122,16 +125,18 @@ class LockedFile:
             return  # Already open (idempotent)
 
         if sys.platform == "win32":
-            self._open_win32(path, create)
+            self._open_win32(path, create, readonly)
         else:
-            self._open_unix(path, create)
+            self._open_unix(path, create, readonly)
 
-    def open_and_read(self, path: Path, create: bool = False) -> bytes:
-        """Open *path* with exclusive lock and read all content.
+    def open_and_read(
+        self, path: Path, create: bool = False, readonly: bool = False
+    ) -> bytes:
+        """Open *path* and read all content.
 
         Combined operation for efficient use with asyncio.to_thread().
         """
-        self.open(path, create=create)
+        self.open(path, create=create, readonly=readonly)
         return self.read()
 
     def read(self) -> bytes:
@@ -188,12 +193,16 @@ class LockedFile:
 
     # -- Unix ----------------------------------------------------------------
 
-    def _open_unix(self, path: Path, create: bool) -> None:
+    def _open_unix(self, path: Path, create: bool, readonly: bool) -> None:
+        if readonly:
+            flags = os.O_RDONLY
+        else:
             flags = os.O_RDWR | (os.O_CREAT if create else 0)
         try:
             fd = os.open(path, flags, 0o666)
         except FileNotFoundError:
             _fatal(f"Database file not found: {path.resolve()}", db_path=path)
+        if not readonly:
             try:
                 fcntl.flock(fd, fcntl.LOCK_EX | fcntl.LOCK_NB)
             except OSError:
@@ -220,12 +229,19 @@ class LockedFile:
 
     # -- Windows -------------------------------------------------------------
 
-    def _open_win32(self, path: Path, create: bool) -> None:
+    def _open_win32(self, path: Path, create: bool, readonly: bool) -> None:
+        if readonly:
+            disposition = _OPEN_EXISTING
+            access = _GENERIC_READ
+            share = _FILE_SHARE_READ | _FILE_SHARE_WRITE
+        else:
             disposition = _OPEN_ALWAYS if create else _OPEN_EXISTING
+            access = _GENERIC_READ | _GENERIC_WRITE
+            share = _FILE_SHARE_READ
         handle = _kernel32.CreateFileW(
             str(path),
-            _GENERIC_READ | _GENERIC_WRITE,
-            _FILE_SHARE_READ,
+            access,
+            share,
             None,
             disposition,
             _FILE_ATTRIBUTE_NORMAL,

kanta/kanta.py

@@ -1,14 +1,12 @@
-"""JSONL persistence layer with background flush task."""
+"""Kanta DB main public API"""
 
 from __future__ import annotations
-
-from collections.abc import Callable
+from datetime import datetime
 from pathlib import Path
-from types import ModuleType
-from typing import Any, Generic, TypeVar
+from types import ModuleType, SimpleNamespace
+from typing import Generic, TypeVar
 
-from kanta.exceptions import DatabaseError
-from kanta.kanta.kantaimpl import KantaImpl
+from kanta.kantaimpl import KantaImpl
 from kanta.serialization import JsonSerializer, Serializer
 from kanta.transaction import transaction as _transaction
 
@@ -52,9 +50,7 @@ class Kanta(Generic[T]):
         *,
         type: type[T] | None = None,
         migrations: ModuleType | str | None = None,
-        migration_ctx: Any | None = None,
         serializer: Serializer | None = None,
-        fatal_error: Callable[[DatabaseError], None] | None = None,
         flush_interval: float = 0.1,
     ):
         """Initialize a Kanta persistence instance.
@@ -64,11 +60,8 @@ class Kanta(Generic[T]):
             data: Caller-owned root msgspec.Struct state instance.
             type: Optional explicit root type. Defaults to ``type(data)``.
             migrations: Optional migrations module object or import path.
-            migration_ctx: Optional context object passed to migration functions.
             flush_interval: Background flush interval in seconds.
             serializer: Optional serializer implementation.
-            fatal_error: Optional callback invoked immediately when the
-                background writer encounters a DatabaseError.
 
         Raises:
             ImportError: If ``migrations`` is a string path that cannot be imported.
@@ -79,13 +72,12 @@ class Kanta(Generic[T]):
 
         self._impl = KantaImpl(
             serializer=active_serializer,
-            fatal_error=fatal_error,
             filename=filename,
             data=data,
             type=data_type,
             migrations=migrations,
-            migration_ctx=migration_ctx,
             flush_interval=flush_interval,
+            kanta=self,
         )
 
     @property
@@ -132,19 +124,46 @@ class Kanta(Generic[T]):
         """
         return self._impl.filename
 
-    async def open(self) -> None:
+    @property
+    def ctx(self) -> SimpleNamespace:
+        """User-writable context namespace.
+
+        Migration functions receive the ``Kanta`` instance and can read or
+        mutate ``kanta.ctx`` during migrations.
+        """
+        return self._impl.ctx
+
+    @property
+    def mtime(self) -> datetime | None:
+        """Last modification time carried forward from change records.
+
+        Returns:
+            The latest ``m`` value, or ``None`` if no modification time has
+            been set yet. System operations such as migrations do not update
+            this value.
+        """
+        return self._impl.mtime
+
+    async def open(self, *, create: bool = True, readonly: bool = False) -> None:
         """Open the database file and start background persistence.
 
         This loads existing records, applies configured migrations, and starts
         the background flush task.
 
+        Args:
+            create: Whether to create the database file when missing.
+                If False, opening fails when the file does not exist or is empty.
+            readonly: If True, open the database read-only. No lock is acquired,
+                no background flush task is started, and transactions are
+                rejected. The file is not created if missing.
+
         Calling ``open`` more than once on the same instance is not allowed.
 
         Raises:
             kanta.exceptions.DatabaseError: If replay or decoding fails.
             kanta.exceptions.DataIntegrityError: If the instance is already open.
         """
-        await self._impl.open()
+        await self._impl.open(create=create, readonly=readonly)
 
     async def __aenter__(self) -> Kanta[T]:
         """Enter async context manager and open the database.
@@ -177,21 +196,90 @@ class Kanta(Generic[T]):
         """Stop background task, flush pending changes, and close file lock."""
         await self._impl.close()
 
+    def bootstrap(
+        self,
+        fn=None,
+        *,
+        action: str = "bootstrap",
+        user: str | None = None,
+        mtime: bool | datetime = True,
+    ):
+        """Register a bootstrap callback executed during :meth:`open`.
+
+        Can be used as ``@kanta.bootstrap`` or ``@kanta.bootstrap(...)``.
+        The callback receives the live ``data`` object and may be sync or async.
+        """
+
+        def _register(callback):
+            self._impl.add_bootstrap(
+                callback=callback,
+                action=action,
+                user=user,
+                mtime=mtime,
+            )
+            return callback
+
+        if fn is None:
+            return _register
+        return _register(fn)
+
+    def fatal_error(self, fn=None):
+        """Register fatal error handler callback.
+
+        Can be used as ``@kanta.fatal_error``.
+        The callback receives a :class:`kanta.exceptions.DatabaseError` and may
+        be sync or async.
+        """
+
+        def _register(callback):
+            self._impl.add_fatal_error(callback)
+            return callback
+
+        if fn is None:
+            return _register
+        return _register(fn)
+
+    def logfmt(self, fn=None, *, path: str | None = None):
+        """Register a transaction logfmt callback.
+
+        Can be used as ``@kanta.logfmt`` or ``@kanta.logfmt(path=...)``.
+        The callback is called for each value being rendered and receives the
+        value plus an optional ``path: str`` parameter.  It must return
+        ``str | None`` (or inherit from :class:`kanta.callbacks.LogFmt`).
+
+        When ``path`` is given, the callback is only invoked for values whose
+        dot-notation path matches the pattern (full match, shell-style wildcards
+        such as ``*`` are supported).
+        """
+
+        def _register(callback):
+            self._impl.add_logfmt(callback, path=path)
+            return callback
+
+        if fn is None:
+            return _register
+        return _register(fn)
+
     def transaction(
         self,
         action: str,
         *,
         user: str | None = None,
-        user_display: str | None = None,
-        resolver: Any = None,
+        mtime: bool | datetime = True,
     ):
         """Create a transactional mutation context manager.
 
         Args:
             action: Action label stored in the change record.
-            user: Optional user identifier stored in metadata.
-            user_display: Optional display name used for logging/resolution.
-            resolver: Optional callable for resolving identifiers in logs.
+            user: Optional user identifier stored in metadata and rendered in
+                the log header.  Register a ``@kanta.logfmt`` callback to format
+                the user value; the path ``"$user"`` is passed for this case.
+            mtime: Controls the modification time ``m``. ``True`` (default)
+                sets ``m`` to the current UTC time. ``False`` omits ``m`` so the
+                previous modification time remains in effect; this is used for
+                system operations that are not considered modifications. A
+                :class:`~datetime.datetime` value sets ``m`` to that explicit
+                time.
 
         Returns:
             A context manager yielding the live state object for mutation.
@@ -202,5 +290,8 @@ class Kanta(Generic[T]):
             rolled back.
         """
         return _transaction(
-            self._impl, action, user=user, user_display=user_display, resolver=resolver
+            self._impl,
+            action,
+            user=user,
+            mtime=mtime,
         )

kanta/kantaimpl.py

@@ -7,10 +7,12 @@ import copy
 import importlib
 import logging
 from datetime import UTC, datetime
+from types import SimpleNamespace
 from typing import Any, Generic, TypeVar
 
+from kanta.callbacks import CallbackRegistry, InjectionContext
 from kanta.exceptions import DatabaseError, DataIntegrityError, ReplayError
-from kanta.migrate import MigrationRegistry
+from kanta.migrations import Migrations
 from kanta.persistence import PersistenceMixin
 from kanta.serialization import restore_data_in_place, struct_to_dict
 from kanta.serialization.base import replay
@@ -26,28 +28,54 @@ class KantaImpl(PersistenceMixin, Generic[T]):
     def __init__(self, **kwargs: Any):
         self.data_type = kwargs.pop("type")
         self.data: T = kwargs.pop("data")
-        self.migrations = kwargs.pop("migrations", None)
-        self.migration_ctx = kwargs.pop("migration_ctx", None)
+        self._kanta = kwargs.pop("kanta", None)
+        migrations = kwargs.pop("migrations", None)
+        self.ctx = SimpleNamespace()
         super().__init__(**kwargs)
-        self.migration_registry: MigrationRegistry | None = None
-        if self.migrations is not None:
+        self.migrations: Migrations | None = None
+        if migrations is not None:
             module = (
-                importlib.import_module(self.migrations)
-                if isinstance(self.migrations, str)
-                else self.migrations
+                importlib.import_module(migrations)
+                if isinstance(migrations, str)
+                else migrations
             )
-            self.migration_registry = MigrationRegistry.from_module(module)
+            self.migrations = Migrations.from_module(module)
 
         self.in_transaction = False
         self.transaction_snapshot: dict[str, Any] | None = None
         self.opened = False
+        self.readonly = False
+        self.bootstrap_action = "bootstrap"
+        self.bootstrap_user: str | None = None
+        self.bootstrap_mtime: bool | datetime = True
 
-        self.statedict = struct_to_dict(self.data, serializer=self.serializer)
-        self.version = (
-            self.migration_registry.dbver if self.migration_registry is not None else 0
+        self.callback_registry = CallbackRegistry(
+            kanta_class=type(self._kanta) if self._kanta is not None else None,
+            data_type=self.data_type,
         )
 
-    async def open(self) -> None:
+        self.statedict = struct_to_dict(self.data, serializer=self.serializer)
+        self.version = self.migrations.dbver if self.migrations is not None else 0
+
+    def add_bootstrap(
+        self,
+        *,
+        callback,
+        action: str,
+        user: str | None,
+        mtime: bool | datetime,
+    ) -> None:
+        """Add bootstrap callback and update bootstrap metadata."""
+        self.callback_registry.register("bootstrap", callback)
+        self.bootstrap_action = action
+        self.bootstrap_user = user
+        self.bootstrap_mtime = mtime
+
+    def add_logfmt(self, callback, *, path: str | None = None) -> None:
+        """Register one transaction logfmt callback."""
+        self.callback_registry.register("logfmt", callback, path=path)
+
+    async def open(self, *, create: bool = True, readonly: bool = False) -> None:
         """Open the database: load from disk, apply migrations, start background task."""
         if self.opened:
             raise DataIntegrityError(
@@ -56,10 +84,30 @@ class KantaImpl(PersistenceMixin, Generic[T]):
                 action="open",
             )
 
+        self.readonly = readonly
+        existed_before_open = self.filename.exists()
+
+        # Read-only mode never creates the file.
+        open_create = create and not readonly
+
         content = await asyncio.to_thread(
             self.file.open_and_read,
             self.filename,
-            create=True,
+            create=open_create,
+            readonly=readonly,
+        )
+
+        if not create and (not existed_before_open or not content):
+            self.file.close()
+            reason = (
+                "database file did not exist"
+                if not existed_before_open
+                else "database file is empty"
+            )
+            raise DataIntegrityError(
+                f"Cannot open database: {reason}",
+                db_path=self.filename,
+                action="open",
             )
 
         if content:
@@ -91,10 +139,11 @@ class KantaImpl(PersistenceMixin, Generic[T]):
                     cause_type=type(e).__name__,
                 ) from e
 
-            if self.migration_registry is not None:
-                rr.version = self.migration_registry.apply(
-                    rr.state, rr.version, self.migration_ctx
-                )
+            migrations_ran = False
+            if self.migrations is not None:
+                previous_version = rr.version
+                rr.version = self.migrations.apply(rr.state, rr.version, self._kanta)
+                migrations_ran = rr.version != previous_version
 
             self.statedict = copy.deepcopy(rr.state)
             self.data = restore_data_in_place(
@@ -104,16 +153,59 @@ class KantaImpl(PersistenceMixin, Generic[T]):
                 serializer=self.serializer,
             )
             self.version = rr.version
+            self.mtime = rr.m
             normalized = struct_to_dict(self.data, serializer=self.serializer)
-            self.queue_change("migrate:msgspec", normalized)
+            if self.readonly:
+                self.statedict = copy.deepcopy(normalized)
+            else:
+                if migrations_ran:
+                    self.queue_change(
+                        f"migrate:v{self.version}",
+                        self.statedict,
+                        mtime=False,
+                        force=True,
+                    )
+                self.queue_change("migrate:msgspec", normalized, mtime=False)
             self.snapshot.ts = (
                 datetime.fromtimestamp(rr.last_snapshot_mtime, UTC)
                 if rr.last_snapshot_mtime is not None
                 else None
             )
+        elif self.readonly:
+            self.file.close()
+            raise DataIntegrityError(
+                "Cannot open empty database in read-only mode",
+                db_path=self.filename,
+                action="open",
+            )
+        else:
+            try:
+                if self.callback_registry.has("bootstrap"):
+                    await self.callback_registry.invoke(
+                        "bootstrap",
+                        InjectionContext(data=self.data, kanta=self._kanta),
+                    )
+
+                self.statedict = {}
+                current = struct_to_dict(self.data, serializer=self.serializer)
+                self.queue_change(
+                    self.bootstrap_action,
+                    current,
+                    user=self.bootstrap_user,
+                    mtime=self.bootstrap_mtime,
+                    force=True,
+                )
+            except Exception:
+                self.file.close()
+                try:
+                    await asyncio.to_thread(self.filename.unlink, missing_ok=True)
+                except FileNotFoundError:
+                    pass
+                raise
 
         self.opened = True
 
+        if not self.readonly:
             self.background_task = asyncio.create_task(self._background_loop())
 
     async def close(self) -> None:
@@ -131,6 +223,7 @@ class KantaImpl(PersistenceMixin, Generic[T]):
 
         # Always run a final flush in case the background task never reached
         # its cancellation handler.
+        if not self.readonly:
             await self.flush()
 
         self.file.close()

kanta/logging.py

@@ -31,11 +31,30 @@ _ADD = "\033[0;32m"  # Green for additions
 _ACTION = "\033[1;34m"  # Bold blue for action name
 _USER = "\033[0;34m"  # Blue for user display
 
+# Metadata path used when formatting the transaction actor.
+_USER_PATH = "$user"
+
+
+def _join_path(path: str, key: str) -> str:
+    """Append *key* to a dot-notation *path*."""
+    if not path:
+        return key
+    return f"{path}.{key}"
+
 
 def _format_value(
-    value: Any, max_len: int = 60, resolver: Callable[[str], str] | None = None
+    value: Any,
+    path: str,
+    *,
+    max_len: int = 60,
+    logfmt: Callable[[Any, str], str | None] | None = None,
 ) -> str:
     """Format a value for display, truncating if needed."""
+    if logfmt is not None:
+        resolved = logfmt(value, path)
+        if resolved is not None:
+            return resolved
+
     if value is None:
         return "null"
     if isinstance(value, bool):
@@ -44,10 +63,6 @@ def _format_value(
         return str(value)
     if isinstance(value, str):
         value = _UNSAFE_CHARS.sub("", value)
-        if resolver is not None:
-            resolved = resolver(value)
-            if resolved != value:
-                return resolved
         if len(value) > max_len:
             return value[: max_len - 3] + "..."
         return value
@@ -57,17 +72,21 @@ def _format_value(
         all_true = all(v is True for v in value.values())
         parts = []
         for k, v in value.items():
-            key_display = resolver(k) if resolver is not None else k
+            key_path = _join_path(path, str(k))
+            key_display = _format_value(k, key_path, max_len=30, logfmt=logfmt)
             if all_true:
                 parts.append(key_display)
             else:
-                val_display = _format_value(v, max_len=30, resolver=resolver)
+                val_display = _format_value(v, key_path, max_len=30, logfmt=logfmt)
                 parts.append(f"{key_display}: {val_display}")
         return "{" + ", ".join(parts) + "}"
     if isinstance(value, list):
         if not value:
             return "[]"
-        parts = [_format_value(v, max_len=30, resolver=resolver) for v in value]
+        parts = []
+        for i, v in enumerate(value):
+            item_path = _join_path(path, str(i))
+            parts.append(_format_value(v, item_path, max_len=30, logfmt=logfmt))
         return "[" + ", ".join(parts) + "]"
     text = str(value)
     if len(text) > max_len:
@@ -75,16 +94,35 @@ def _format_value(
     return text
 
 
-def _format_path(path: list[str], resolver: Callable[[str], str] | None = None) -> str:
-    """Format a path as dot notation with prefix in dark grey, final in default."""
+def _format_path_components(
+    path: list[str], logfmt: Callable[[Any, str], str | None] | None
+) -> list[str]:
+    """Return path components after applying formatters."""
     if not path:
+        return []
+    result = []
+    for i, component in enumerate(path):
+        prefix_path = ".".join(path[: i + 1])
+        display = component
+        if logfmt is not None:
+            resolved = logfmt(component, prefix_path)
+            if resolved is not None:
+                display = resolved
+        result.append(display)
+    return result
+
+
+def _format_path(
+    path: list[str], logfmt: Callable[[Any, str], str | None] | None
+) -> str:
+    """Format a path as dot notation with prefix in dark grey, final in default."""
+    components = _format_path_components(path, logfmt)
+    if not components:
         return ""
-    if resolver is not None:
-        path = [resolver(p) for p in path]
-    if len(path) == 1:
-        return f"{_PATH_FINAL}{path[0]}{_RESET}"
-    prefix = ".".join(path[:-1])
-    final = path[-1]
+    if len(components) == 1:
+        return f"{_PATH_FINAL}{components[0]}{_RESET}"
+    prefix = ".".join(components[:-1])
+    final = components[-1]
     return f"{_PATH_PREFIX}{prefix}.{_RESET}{_PATH_FINAL}{final}{_RESET}"
 
 
@@ -158,74 +196,56 @@ def _format_change_lines(
     change_type: str,
     path: list[str],
     value: Any,
-    resolver: Callable[[str], str] | None = None,
+    logfmt: Callable[[Any, str], str | None] | None = None,
 ) -> list[str]:
     """Format a single change as one or more lines."""
-
-    def fmt_value(v: Any, child_path: list[str]) -> str:
-        return _format_value(v, resolver=resolver)
-
-    formatted_path = list(path)
-    if resolver is not None:
-        formatted_path = [resolver(p) for p in formatted_path]
+    path_str = _format_path(path, logfmt=logfmt)
 
     if change_type == "delete":
-        if len(formatted_path) == 1:
-            return [f"  {_DELETE}{formatted_path[0]} ✗{_RESET}"]
-        prefix = ".".join(formatted_path[:-1])
-        final = formatted_path[-1]
+        components = _format_path_components(path, logfmt)
+        if len(components) == 1:
+            return [f"  {_DELETE}{components[0]} ✗{_RESET}"]
+        prefix = ".".join(components[:-1])
+        final = components[-1]
         return [f"  {_PATH_PREFIX}{prefix}.{_RESET}{_DELETE}{final} ✗{_RESET}"]
 
     if change_type == "add":
         if isinstance(value, dict) and value:
-            lines = []
-            if len(formatted_path) == 1:
-                lines.append(f"  {_ADD}{formatted_path[0]}{_RESET} {_SEP}={_RESET}")
-            else:
-                prefix = ".".join(formatted_path[:-1])
-                final = formatted_path[-1]
-                lines.append(
-                    f"  {_PATH_PREFIX}{prefix}.{_RESET}{_ADD}{final}{_RESET} {_SEP}={_RESET}"
-                )
+            lines = [f"  {path_str} {_SEP}={_RESET}"]
             formatted_items = []
+            base_path = ".".join(path)
             for k, v in value.items():
-                k_display = resolver(k) if resolver is not None else k
-                v_str = fmt_value(v, path + [k])
-                formatted_items.append((k_display, v_str))
+                key_path = _join_path(base_path, str(k))
+                key_display = _format_value(k, key_path, max_len=30, logfmt=logfmt)
+                v_str = _format_value(v, key_path, max_len=30, logfmt=logfmt)
+                formatted_items.append((key_display, v_str))
             max_key_len = max(len(k) for k, _ in formatted_items)
             field_width = max(max_key_len, 12)
             for k_display, v_str in formatted_items:
                 padding = " " * (field_width - len(k_display))
                 lines.append(f"    {k_display}{_SEP}:{_RESET}{padding} {v_str}")
             return lines
-        else:
-            value_str = fmt_value(value, path)
-            if len(formatted_path) == 1:
-                return [
-                    f"  {_ADD}{formatted_path[0]}{_RESET} {_SEP}={_RESET} {value_str}"
-                ]
-            prefix = ".".join(formatted_path[:-1])
-            final = formatted_path[-1]
-            return [
-                f"  {_PATH_PREFIX}{prefix}.{_RESET}{_ADD}{final}{_RESET} {_SEP}={_RESET} {value_str}"
-            ]
+        value_str = _format_value(value, ".".join(path), logfmt=logfmt)
+        return [f"  {path_str} {_SEP}={_RESET} {value_str}"]
 
-    value_str = fmt_value(value, path)
-    path_str = _format_path(path, resolver=resolver)
+    value_str = _format_value(value, ".".join(path), logfmt=logfmt)
     return [f"  {path_str} {_SEP}={_RESET} {value_str}"]
 
 
 def format_diff(
     diff: dict,
     previous: dict | None = None,
-    resolver: Callable[[str], str] | None = None,
+    logfmt: Callable[[Any, str], str | None] | None = None,
 ) -> list[str]:
     """Format a JSON diff as human-readable lines.
 
     Args:
         diff: The JSON diff dict.
         previous: The previous state dict (for determining add vs update).
-        resolver: Optional callable to resolve path components (e.g. UUID→name).
+        logfmt: Optional formatter callable ``(value, path) -> str | None``.
+            ``path`` is a dot-notation string; ``"$user"`` is used for the
+            transaction actor.  If the callable returns ``None``, default
+            formatting is used.
 
     Returns a list of formatted lines (without newlines).
     """
@@ -235,15 +255,15 @@ def format_diff(
         return []
     lines = []
     for change_type, path, value in changes:
-        lines.extend(_format_change_lines(change_type, path, value, resolver))
+        lines.extend(_format_change_lines(change_type, path, value, logfmt))
     return lines
 
 
-def format_action_header(action: str, user_display: str | None = None) -> str:
+def format_action_header(action: str, user: str | None = None) -> str:
     """Format the action header line."""
     action_str = f"{_ACTION}{action}{_RESET}"
-    if user_display:
-        user_str = f"{_USER}{user_display}{_RESET}"
+    if user:
+        user_str = f"{_USER}{user}{_RESET}"
         return f"{action_str} by {user_str}"
     return action_str
 
@@ -251,21 +271,21 @@ def format_action_header(action: str, user_display: str | None = None) -> str:
 def log_change(
     action: str,
     diff: dict,
-    user_display: str | None = None,
+    user: str | None = None,
     previous: dict | None = None,
-    resolver: Callable[[str], str] | None = None,
+    logfmt: Callable[[Any, str], str | None] | None = None,
 ) -> None:
     """Log a database change with pretty-printed diff.
 
     Args:
         action: The action name (e.g., "login", "admin:delete_user").
         diff: The JSON diff dict.
-        user_display: Optional display name of the user who performed the action.
+        user: Optional already-formatted user name to show in the header.
         previous: The previous state dict (for determining add vs update).
-        resolver: Optional callable to resolve path components (e.g. UUID→name).
+        logfmt: Optional formatter callable ``(value, path) -> str | None``.
     """
-    header = format_action_header(action, user_display)
-    diff_lines = format_diff(diff, previous, resolver)
+    header = format_action_header(action, user)
+    diff_lines = format_diff(diff, previous, logfmt)
 
     if not diff_lines:
         logger.info(header)

kanta/migrate.py

@@ -1,117 +0,0 @@
-"""Database schema migration framework.
-
-Migrations are numbered functions discovered automatically via a decorator
-or by prefix. Each runs exactly once based on the current version.
-"""
-
-from __future__ import annotations
-
-import importlib
-import logging
-from types import ModuleType
-from typing import Any
-
-import msgspec
-
-_logger = logging.getLogger(__name__)
-
-
-class MigrationCtx(msgspec.Struct, omit_defaults=True):
-    """Context passed to each migration function.
-
-    Subclass or replace this with your own context type.
-    """
-
-    pass
-
-
-class MigrationRegistry:
-    """Registry of schema migration functions.
-
-    Usage::
-
-        registry = MigrationRegistry()
-
-        @registry.register
-        def migrate_v1(d: dict, ctx: MigrationCtx) -> None:
-            d.setdefault("version", 1)
-
-        new_version = registry.apply(state, current_version=0)
-
-    Or load from a module::
-
-        registry = MigrationRegistry.from_module("myapp.migrations")
-        new_version = registry.apply(state, current_version=0)
-    """
-
-    def __init__(self) -> None:
-        self._migrations: dict[int, Any] = {}
-
-    @staticmethod
-    def _migration_version(fn: Any) -> int:
-        name = getattr(fn, "__name__", "")
-        if not name.startswith("migrate_v"):
-            raise ValueError(f"Invalid migration function name: {name!r}")
-        suffix = name.removeprefix("migrate_v")
-        if not suffix.isdigit() or int(suffix) <= 0:
-            raise ValueError(f"Invalid migration version in function name: {name!r}")
-        return int(suffix)
-
-    def register(self, fn):
-        """Decorator to register a migration function."""
-        version = self._migration_version(fn)
-        self._migrations[version] = fn
-        return fn
-
-    @classmethod
-    def from_module(cls, module: str | ModuleType) -> MigrationRegistry:
-        """Create a registry by scanning a module for ``migrate_vN`` functions.
-
-        Args:
-            module: A module name (string) or an imported module object.
-        """
-        reg = cls()
-        if isinstance(module, str):
-            mod = importlib.import_module(module)
-        else:
-            mod = module
-
-        for name in dir(mod):
-            if name.startswith("migrate_v"):
-                fn = getattr(mod, name)
-                if callable(fn):
-                    version = reg._migration_version(fn)
-                    reg._migrations[version] = fn
-        return reg
-
-    @property
-    def dbver(self) -> int:
-        """Current schema version (= highest discovered migration, or 0)."""
-        return max(self._migrations.keys(), default=0)
-
-    def apply(
-        self,
-        data_dict: dict[str, Any],
-        current_version: int,
-        ctx: MigrationCtx | None = None,
-        *,
-        silent: bool = False,
-    ) -> int:
-        """Apply pending migrations to *data_dict* in place.
-
-        Returns the new version after all migrations.
-        """
-        while current_version < self.dbver:
-            next_version = current_version + 1
-            fn = self._migrations.get(next_version)
-            if fn is None:
-                raise ValueError(
-                    f"Missing migration step migrate_v{next_version} "
-                    f"(highest discovered is v{self.dbver})"
-                )
-            fn(data_dict, ctx or MigrationCtx())
-            current_version = next_version
-            if not silent:
-                desc = (fn.__doc__ or fn.__name__).split("\n")[0].rstrip(".")
-                _logger.info("Applied migration %s: %s", fn.__name__, desc)
-        return current_version

kanta/migrations.py

@@ -0,0 +1,154 @@
+"""Database schema migration framework.
+
+Migrations are numbered functions discovered automatically via a decorator
+or by prefix. Each runs exactly once based on the current version.
+"""
+
+from __future__ import annotations
+
+import copy
+import importlib
+import inspect
+import logging
+from types import ModuleType
+from typing import Any
+
+from kanta.exceptions import DatabaseError
+
+_logger = logging.getLogger(__name__)
+
+# Cache registries by imported module object so that many Kanta instances using
+# the same migrations module do not re-scan it each time.
+_module_registry_cache: dict[ModuleType, Migrations] = {}
+
+
+class Migrations:
+    """Registry of schema migration functions.
+
+    Usage::
+
+        migrations = Migrations()
+
+        @migrations.register
+        def migrate_v1(d: dict, kanta) -> None:
+            d.setdefault("version", 1)
+            kanta.ctx.note = "migrated"
+
+        @migrations.register
+        def migrate_v2(d: dict) -> None:
+            d.setdefault("version", 2)
+
+        new_version = migrations.apply(state, current_version=0, kanta=kanta)
+
+    Or load from a module::
+
+        migrations = Migrations.from_module("myapp.migrations")
+        new_version = migrations.apply(state, current_version=0, kanta=kanta)
+    """
+
+    def __init__(self) -> None:
+        self._migrations: dict[int, Any] = {}
+
+    @staticmethod
+    def _migration_version(fn: Any) -> int:
+        name = getattr(fn, "__name__", "")
+        if not name.startswith("migrate_v"):
+            raise ValueError(f"Invalid migration function name: {name!r}")
+        suffix = name.removeprefix("migrate_v")
+        if not suffix.isdigit() or int(suffix) <= 0:
+            raise ValueError(f"Invalid migration version in function name: {name!r}")
+        return int(suffix)
+
+    def register(self, fn):
+        """Decorator to register a migration function."""
+        version = self._migration_version(fn)
+        self._migrations[version] = fn
+        return fn
+
+    @classmethod
+    def from_module(cls, module: str | ModuleType) -> Migrations:
+        """Create or retrieve a cached registry by scanning a module.
+
+        Args:
+            module: A module name (string) or an imported module object.
+        """
+        if isinstance(module, str):
+            mod = importlib.import_module(module)
+        else:
+            mod = module
+
+        try:
+            return _module_registry_cache[mod]
+        except KeyError:
+            pass
+
+        reg = cls()
+        for name in dir(mod):
+            if name.startswith("migrate_v"):
+                fn = getattr(mod, name)
+                if callable(fn):
+                    version = reg._migration_version(fn)
+                    reg._migrations[version] = fn
+        _module_registry_cache[mod] = reg
+        return reg
+
+    @property
+    def dbver(self) -> int:
+        """Current schema version (= highest discovered migration, or 0)."""
+        return max(self._migrations.keys(), default=0)
+
+    @property
+    def minver(self) -> int:
+        """Minimum supported current version (first migration minus 1, or 0)."""
+        return min(self._migrations.keys(), default=1) - 1
+
+    @staticmethod
+    def _call_migration(fn: Any, data_dict: dict[str, Any], kanta: Any) -> None:
+        """Call *fn* with the data dict and, if accepted, the Kanta instance."""
+        try:
+            inspect.signature(fn).bind(data_dict, kanta)
+        except TypeError:
+            fn(data_dict)
+        else:
+            fn(data_dict, kanta)
+
+    def apply(
+        self,
+        data_dict: dict[str, Any],
+        current_version: int,
+        kanta: Any,
+        *,
+        silent: bool = False,
+    ) -> int:
+        """Apply pending migrations to *data_dict* in place.
+
+        Missing intermediate migration steps are silently skipped.
+
+        Raises:
+            DatabaseError: If the database version is newer than the highest
+                supported version or older than the minimum supported version.
+
+        Returns the new version after all migrations.
+        """
+        if current_version > self.dbver:
+            raise DatabaseError(
+                f"Database version v{current_version} is newer than the "
+                f"highest supported version v{self.dbver}"
+            )
+        if current_version < self.minver:
+            raise DatabaseError(
+                f"Database version v{current_version} is older than the "
+                f"minimum supported version v{self.minver}"
+            )
+
+        for version in sorted(self._migrations.keys()):
+            if version <= current_version:
+                continue
+            fn = self._migrations[version]
+            before = copy.deepcopy(data_dict) if not silent else None
+            self._call_migration(fn, data_dict, kanta)
+            current_version = version
+            if not silent and before != data_dict:
+                desc = (fn.__doc__ or fn.__name__).split("\n")[0].rstrip(".")
+                _logger.info("Applied migration %s: %s", fn.__name__, desc)
+        return current_version

kanta/persistence.py

@@ -5,17 +5,16 @@ from __future__ import annotations
 import asyncio
 import copy
 import logging
-import threading
 from collections import deque
-from collections.abc import Callable
-from datetime import datetime
+from datetime import UTC, datetime
 from pathlib import Path
 from typing import Any
 
+from kanta.callbacks import CallbackRegistry, InjectionContext
 from kanta.diff import compute_diff
 from kanta.exceptions import DatabaseError, DataIntegrityError
 from kanta.filelock import LockedFile
-from kanta.kanta.structs import ChangeRecord
+from kanta.structs import ChangeRecord
 from kanta.serialization import JsonSerializer, Serializer
 from kanta.serialization.framing import Framer
 from kanta.snapshot import SnapshotState
@@ -31,41 +30,47 @@ class PersistenceMixin:
     flush_failed: bool
     statedict: dict[str, Any]
     pending_changes: deque[ChangeRecord]
-    pending_lock: threading.Lock
     snapshot: SnapshotState
     serializer: Serializer
     framer: Framer
     background_task: asyncio.Task | None
-    fatal_error: Callable[[DatabaseError], None] | None
+    callback_registry: CallbackRegistry
     background_error: DatabaseError | None
     flush_interval: float
     version: int
     opened: bool
+    readonly: bool
+    mtime: datetime | None
 
     def __init__(self, **kwargs: Any) -> None:
         """Initialize persistence-owned state used by mixin methods."""
         filename = kwargs.pop("filename")
         flush_interval = kwargs.pop("flush_interval", 0.1)
         serializer = kwargs.pop("serializer", None)
-        fatal_error = kwargs.pop("fatal_error", None)
         super().__init__(**kwargs)
         self.filename = Path(filename)
         self.file = LockedFile()
         self.flush_failed = False
         self.statedict = {}
         self.pending_changes = deque()
-        self.pending_lock = threading.Lock()
         self.serializer = serializer or JsonSerializer()
         self.framer = self.serializer.framer_cls()
         self.snapshot = SnapshotState(serializer=self.serializer, framer=self.framer)
         self.background_task = None
-        self.fatal_error = fatal_error
+        self.callback_registry = CallbackRegistry()
         self.background_error = None
         self.flush_interval = flush_interval
         self.version = 0
+        self.mtime: datetime | None = None
+
+    def add_fatal_error(self, callback) -> None:
+        """Register one fatal error callback in call order."""
+        self.callback_registry.register("fatal_error", callback)
 
     async def _background_loop(self) -> None:
         """Background task that periodically flushes changes to disk."""
+        if self.readonly:
+            return
         while True:
             try:
                 await asyncio.sleep(self.flush_interval)
@@ -77,42 +82,82 @@ class PersistenceMixin:
                 break
             except DatabaseError as e:
                 self.background_error = e
-                if self.fatal_error is not None:
-                    try:
-                        self.fatal_error(e)
-                    except Exception as callback_error:
+
+                def _log_callback_error(callback_error, callback):
                     _logger.exception(
-                            "Background error callback failed: %s", callback_error
+                        "Background error callback %r failed: %s",
+                        callback,
+                        callback_error,
+                    )
+
+                await self.callback_registry.invoke(
+                    "fatal_error",
+                    InjectionContext(error=e, kanta=self._kanta),
+                    on_error=_log_callback_error,
                 )
                 _logger.error("Background flush loop stopped: %s", e)
                 break
 
     def maybe_snapshot(self) -> None:
         """Evaluate and possibly write a snapshot from current state."""
-        self.snapshot.maybe_write(self.file, self.version, self.statedict)
+        self.snapshot.maybe_write(self.file, self.version, self.statedict, m=self.mtime)
 
     def queue_change(
         self,
         action: str,
         current: dict,
+        *,
         user: str | None = None,
-        m: datetime | None = None,
-    ) -> None:
-        """Queue a change record internally (thread-safe)."""
+        mtime: bool | datetime = True,
+        force: bool = False,
+    ) -> ChangeRecord | None:
+        """Queue a change record internally (thread-safe).
+
+        Args:
+            action: Action label stored in the change record.
+            current: New serialized state after the change.
+            user: Optional actor identifier.
+            mtime: Controls the modification timestamp. ``True`` (default)
+                sets ``m`` to the current UTC time. ``False`` omits ``m`` so the
+                previous modification time remains in effect; this is used for
+                system operations that are not considered modifications. A
+                :class:`~datetime.datetime` value sets ``m`` to that explicit time.
+            force: If ``True``, queue the record even when the diff is empty.
+
+        Returns:
+            The queued :class:`ChangeRecord`, or ``None`` if the diff was empty
+            and *force* is ``False``.
+        """
+        now = datetime.now(UTC)
+
+        if mtime is True:
+            m = now
+        elif mtime is False:
+            m = None
+        elif isinstance(mtime, datetime):
+            m = mtime
+        else:
+            raise TypeError("mtime must be True, False, or a datetime")
+
         diff = compute_diff(self.statedict, current)
         if not diff:
-            return
-        with self.pending_lock:
-            self.pending_changes.append(
-                ChangeRecord(
+            if not force:
+                return None
+            diff = {}
+
+        record = ChangeRecord(
+            ts=now,
             a=action,
             v=self.version,
             u=user,
             m=m,
             diff=diff,
         )
-            )
+        self.pending_changes.append(record)
         self.statedict = copy.deepcopy(current)
+        if m is not None:
+            self.mtime = m
+        return record
 
     def flush_sync(self) -> None:
         """Synchronously flush all pending changes to disk."""
@@ -123,10 +168,16 @@ class PersistenceMixin:
                 action="flush_sync",
             )
 
+        if self.readonly:
+            raise DataIntegrityError(
+                "Cannot flush in read-only mode",
+                db_path=self.filename,
+                action="flush_sync",
+            )
+
         if self.flush_failed:
             return
 
-        with self.pending_lock:
         if not self.pending_changes:
             return
         changes_to_write = list(self.pending_changes)
@@ -146,13 +197,11 @@ class PersistenceMixin:
                 records.append(framed)
                 running_size += len(framed)
             if not records:
-                with self.pending_lock:
                 self.pending_changes.clear()
                 return
 
             self.file.write(b"".join(records))
             self.snapshot.record_changes(len(records))
-            with self.pending_lock:
             for _ in changes_to_write:
                 self.pending_changes.popleft()
         except OSError as e:
@@ -173,10 +222,16 @@ class PersistenceMixin:
                 action="flush",
             )
 
+        if self.readonly:
+            raise DataIntegrityError(
+                "Cannot flush in read-only mode",
+                db_path=self.filename,
+                action="flush",
+            )
+
         if self.flush_failed:
             return
 
-        with self.pending_lock:
         if not self.pending_changes:
             return
         changes_to_write = list(self.pending_changes)
@@ -196,13 +251,11 @@ class PersistenceMixin:
                 records.append(framed)
                 running_size += len(framed)
             if not records:
-                with self.pending_lock:
                 self.pending_changes.clear()
                 return
 
             await asyncio.to_thread(self.file.write, b"".join(records))
             self.snapshot.record_changes(len(records))
-            with self.pending_lock:
             for _ in changes_to_write:
                 self.pending_changes.popleft()
         except OSError as e:

kanta/serialization/base.py

@@ -9,7 +9,7 @@ from typing import Any, Protocol, TypeVar
 import msgspec
 
 from kanta.exceptions import ReplayError
-from kanta.kanta.structs import ChangeRecord, Snapshot
+from kanta.structs import ChangeRecord, Snapshot
 from kanta.serialization.framing import Framer
 
 T = TypeVar("T")
@@ -23,14 +23,12 @@ class ReplayResult:
         state: dict[str, Any],
         version: int = 0,
         has_migration: bool = False,
-        last_patch_mtime: float | None = None,
         last_snapshot_mtime: float | None = None,
         m: datetime | None = None,
     ):
         self.state = state
         self.version = version
         self.has_migration = has_migration
-        self.last_patch_mtime = last_patch_mtime
         self.last_snapshot_mtime = last_snapshot_mtime
         self.m = m
 
@@ -63,7 +61,6 @@ def replay(
     last_snapshot_mtime: float | None = None
     m: datetime | None = None
     has_migration = False
-    last_patch_mtime: float | None = None
 
     if snap_payload is not None:
         try:
@@ -112,14 +109,12 @@ def replay(
             has_migration = True
         if change.m is not None:
             m = change.m
-        last_patch_mtime = change.ts.timestamp()
         version = change.v
         state = _patch_state(state, change.diff)
     return ReplayResult(
         state=state,
         version=version,
         has_migration=has_migration,
-        last_patch_mtime=last_patch_mtime,
         last_snapshot_mtime=last_snapshot_mtime,
         m=m,
     )

kanta/serialization/json.py

@@ -6,7 +6,7 @@ from typing import Any, TypeVar
 
 import msgspec
 
-from kanta.serialization.framing import LineFramer
+from kanta.serialization.framing import Framer, LineFramer
 
 T = TypeVar("T")
 
@@ -14,7 +14,7 @@ T = TypeVar("T")
 class JsonSerializer:
     """Line-based JSON serializer."""
 
-    framer_cls = LineFramer
+    framer_cls: type[Framer] = LineFramer
 
     def encode(self, obj: Any) -> bytes:
         return msgspec.json.encode(obj)

kanta/serialization/msgpack.py

@@ -6,7 +6,7 @@ from typing import Any, TypeVar
 
 import msgspec
 
-from kanta.serialization.framing import BinFramer
+from kanta.serialization.framing import BinFramer, Framer
 
 T = TypeVar("T")
 
@@ -14,7 +14,7 @@ T = TypeVar("T")
 class MsgPackSerializer:
     """Binary serializer using MessagePack format."""
 
-    framer_cls = BinFramer
+    framer_cls: type[Framer] = BinFramer
 
     def encode(self, obj: Any) -> bytes:
         return msgspec.msgpack.encode(obj)

kanta/snapshot.py

@@ -5,7 +5,7 @@ from __future__ import annotations
 import logging
 from datetime import UTC, datetime
 
-from kanta.kanta.structs import Snapshot
+from kanta.structs import Snapshot
 from kanta.serialization import JsonSerializer, Serializer
 from kanta.serialization.framing import Framer, LineFramer
 
@@ -37,7 +37,9 @@ class SnapshotState:
     def record_changes(self, count: int) -> None:
         self.changes += count
 
-    def maybe_write(self, file, version: int, state: dict) -> None:
+    def maybe_write(
+        self, file, version: int, state: dict, m: datetime | None = None
+    ) -> None:
         """Write snapshot when thresholds/time policy allows it."""
         if self.changes < self._min_diffs:
             return
@@ -51,14 +53,16 @@ class SnapshotState:
         if not file.is_open:
             return
         try:
-            self._write(file, version, state, now)
+            self._write(file, version, state, now, m=m)
             self._force_pending = False
         except Exception as exc:
             _logger.error("snapshot: failed to write snapshot: %r", exc)
 
-    def _write(self, file, version: int, state: dict, now: datetime) -> None:
+    def _write(
+        self, file, version: int, state: dict, now: datetime, m: datetime | None = None
+    ) -> None:
         """Write a snapshot and update internal state."""
-        payload = self._serializer.encode(Snapshot(ts=now, v=version, state=state))
+        payload = self._serializer.encode(Snapshot(ts=now, v=version, state=state, m=m))
         record_offset = file.size() if hasattr(file, "size") else 0
         file.write(self._framer.frame_snapshot(payload, record_offset=record_offset))
         self.changes = 0

kanta/transaction.py

@@ -4,11 +4,12 @@ from __future__ import annotations
 
 import logging
 from contextlib import contextmanager
-from typing import Any
+from datetime import datetime
 
 from kanta.diff import compute_diff
 from kanta.exceptions import DataIntegrityError
-from kanta.logging import log_change
+from kanta.callbacks import InjectionContext
+from kanta.logging import _USER_PATH, log_change
 from kanta.serialization import restore_data_in_place, struct_to_dict
 
 _logger = logging.getLogger(__name__)
@@ -20,10 +21,16 @@ def transaction(
     action: str,
     *,
     user: str | None = None,
-    user_display: str | None = None,
-    resolver: Any = None,
+    mtime: bool | datetime = True,
 ):
     """Wrap writes in a transaction and yield the live db object."""
+    if impl.readonly:
+        raise DataIntegrityError(
+            "Cannot start transaction in read-only mode",
+            db_path=impl.filename,
+            action=action,
+        )
+
     if impl.in_transaction:
         raise RuntimeError(
             "Nested or simultaneous transactions are not supported "
@@ -58,9 +65,22 @@ def transaction(
         new_dict = struct_to_dict(impl.data, serializer=impl.serializer)
         diff = compute_diff(impl.statedict, new_dict)
         if diff:
-            impl.queue_change(action, new_dict, user=user)
-            log_change(action, diff, user_display, impl.statedict, resolver)
-            impl.statedict = new_dict
+            previous = impl.statedict
+            record = impl.queue_change(action, new_dict, user=user, mtime=mtime)
+            if record is not None:
+                logfmt = impl.callback_registry.build_logfmt(
+                    InjectionContext(
+                        previous_state=previous,
+                        current_state=new_dict,
+                        kanta=impl._kanta,
+                    )
+                )
+                formatted_user = user
+                if user is not None and logfmt is not None:
+                    resolved = logfmt(user, _USER_PATH)
+                    if resolved is not None:
+                        formatted_user = resolved
+                log_change(action, record.diff, formatted_user, previous, logfmt)
     except Exception:
         _logger.warning("Transaction '%s' failed, rolling back changes", action)
         if impl.transaction_snapshot is not None:

tests/conftest.py

@@ -1,6 +1,6 @@
 import pytest
 
-from kanta import JsonSerializer, MsgPackSerializer
+from kanta.serialization import JsonSerializer, MsgPackSerializer
 
 
 @pytest.fixture(

tests/support.py

@@ -6,7 +6,8 @@ from uuid import UUID
 
 import msgspec
 
-from kanta import ChangeRecord, Kanta
+from kanta.kanta import Kanta
+from kanta.structs import ChangeRecord
 
 
 class User(msgspec.Struct):
@@ -69,6 +70,18 @@ def change_actions(path: Path, format_config) -> list[str]:
     return actions
 
 
+def read_changes(path: Path, format_config) -> list[ChangeRecord]:
+    _, serializer_cls = format_config
+    serializer = serializer_cls()
+    framer = serializer.framer_cls()
+    records: list[ChangeRecord] = []
+    for is_snapshot, payload, _, _ in framer.iter_records(path.read_bytes(), 0):
+        if is_snapshot:
+            continue
+        records.append(serializer.decode(payload, type=ChangeRecord))
+    return records
+
+
 def make_migrations_module(name: str, fn_name: str, fn):
     mod = ModuleType(name)
     mod.__dict__[fn_name] = fn

tests/test_callbacks.py

@@ -0,0 +1,349 @@
+from typing import Any, Optional, Union
+
+import pytest
+
+from kanta import Kanta
+from kanta.callbacks import DictPost, DictPre, LogFmt
+from kanta.exceptions import DatabaseError
+
+from .support import Data, User, make_kanta
+
+
+def test_bootstrap_rejects_unannotated_param(tmp_path, format_config):
+    kanta = make_kanta(tmp_path / "test.db", Data, format_config)
+
+    with pytest.raises(TypeError, match="without an annotation or default"):
+
+        @kanta.bootstrap
+        def seed(data):
+            data.counter = 1
+
+
+def test_bootstrap_accepts_unknown_with_default(tmp_path, format_config):
+    kanta = make_kanta(tmp_path / "test.db", Data, format_config)
+
+    @kanta.bootstrap
+    def seed(data: Data, extra: int = 0) -> None:
+        data.counter = extra + 1
+
+    # Should register without error.
+
+
+def test_bootstrap_rejects_unknown_annotation(tmp_path, format_config):
+    kanta = make_kanta(tmp_path / "test.db", Data, format_config)
+
+    with pytest.raises(TypeError, match="unsupported annotation"):
+
+        @kanta.bootstrap
+        def seed(data: int):
+            pass
+
+
+def test_logfmt_requires_value_annotation(tmp_path, format_config):
+    kanta = make_kanta(tmp_path / "test.db", Data, format_config)
+
+    with pytest.raises(TypeError, match="value parameter.*must be annotated"):
+
+        @kanta.logfmt
+        def resolve_names(previous: DictPre, current: DictPost) -> str | None:
+            return None
+
+
+def test_logfmt_allows_missing_return_annotation(tmp_path, format_config):
+    kanta = make_kanta(tmp_path / "test.db", Data, format_config)
+
+    @kanta.logfmt
+    def resolve_names(value: str, current: DictPost):
+        return None
+
+
+def test_logfmt_class_allows_missing_return_annotation(tmp_path, format_config):
+    kanta = make_kanta(tmp_path / "test.db", Data, format_config)
+
+    @kanta.logfmt
+    class UserLogFmt(LogFmt):
+        def resolve(self, value: str, path: str):
+            return None
+
+
+# fmt: off
+def test_logfmt_accepts_optional_return_typing_forms(tmp_path, format_config):
+    kanta = make_kanta(tmp_path / "test.db", Data, format_config)
+
+    @kanta.logfmt
+    def resolve_optional(value: str) -> Optional[str]:  # noqa: UP007
+        return value
+
+    @kanta.logfmt
+    def resolve_union(value: str) -> Union[str, None]:  # noqa: UP007
+        return value
+
+    @kanta.logfmt
+    def resolve_pipe(value: "str") -> "str | None":
+        return value
+
+
+def test_logfmt_class_accepts_optional_return_typing_forms(tmp_path, format_config):
+    kanta = make_kanta(tmp_path / "test.db", Data, format_config)
+
+    @kanta.logfmt
+    class OptionalStyle(LogFmt):
+        def resolve(self, value: str, path: str) -> Optional[str]:  # noqa: UP007
+            return value
+
+    @kanta.logfmt
+    class UnionStyle(LogFmt):
+        def resolve(self, value: str, path: str) -> Union[str, None]:  # noqa: UP007
+            return value
+
+    @kanta.logfmt
+    class StringStyle(LogFmt):
+        def resolve(self, value: "str", path: "str") -> "str | None":
+            return value
+# fmt: on
+
+
+def test_logfmt_rejects_async_callback(tmp_path, format_config):
+    kanta = make_kanta(tmp_path / "test.db", Data, format_config)
+
+    with pytest.raises(TypeError, match="must not be async"):
+
+        @kanta.logfmt
+        async def resolve_names(value: str, current: DictPost) -> str | None:
+            return None
+
+
+@pytest.mark.asyncio
+async def test_bootstrap_injects_data_by_type(tmp_path, format_config):
+    path = tmp_path / "test.db"
+    kanta = make_kanta(path, Data, format_config)
+
+    @kanta.bootstrap
+    def seed(data: Data) -> None:
+        data.counter = 7
+
+    await kanta.open()
+    assert kanta.data.counter == 7
+    await kanta.close()
+
+
+@pytest.mark.asyncio
+async def test_bootstrap_injects_kanta(tmp_path, format_config):
+    path = tmp_path / "test.db"
+    kanta = make_kanta(path, Data, format_config)
+    seen: list[Kanta] = []
+
+    @kanta.bootstrap
+    def seed(data: Data, kanta_ref: Kanta) -> None:
+        seen.append(kanta_ref)
+        data.counter = 8
+
+    await kanta.open()
+    assert seen == [kanta]
+    assert kanta.data.counter == 8
+    await kanta.close()
+
+
+@pytest.mark.asyncio
+async def test_logfmt_injects_states(tmp_path, format_config, caplog):
+    import logging
+
+    caplog.set_level(logging.INFO, logger="kanta.changes")
+    path = tmp_path / "test.db"
+    kanta = make_kanta(path, Data, format_config)
+
+    @kanta.logfmt
+    def resolve_users(value: str, current: DictPost) -> str | None:
+        return current.get("users", {}).get(value, {}).get("name")
+
+    await kanta.open()
+
+    with kanta.transaction(action="create_user") as data:
+        data.users["uuid-1"] = User(name="Alice")
+
+    await kanta.close()
+
+    assert "Alice" in caplog.text
+
+
+@pytest.mark.asyncio
+async def test_logfmt_class_injection(tmp_path, format_config, caplog):
+    import logging
+
+    caplog.set_level(logging.INFO, logger="kanta.changes")
+    path = tmp_path / "test.db"
+    kanta = make_kanta(path, Data, format_config)
+
+    @kanta.logfmt
+    class UserLogFmt(LogFmt):
+        def resolve(self, value: str, path: str) -> str | None:
+            return self.current_state.get("users", {}).get(value, {}).get("name")
+
+    await kanta.open()
+
+    with kanta.transaction(action="create_user") as data:
+        data.users["uuid-2"] = User(name="Bob")
+
+    await kanta.close()
+
+    assert "Bob" in caplog.text
+
+
+@pytest.mark.asyncio
+async def test_multiple_logfmt_chain(tmp_path, format_config, caplog):
+    import logging
+
+    caplog.set_level(logging.INFO, logger="kanta.changes")
+    path = tmp_path / "test.db"
+    kanta = make_kanta(path, Data, format_config)
+
+    @kanta.logfmt
+    def resolve_a(value: str) -> str | None:
+        return "A" if value == "a" else None
+
+    @kanta.logfmt
+    def resolve_b(value: str) -> str | None:
+        return "B" if value == "b" else None
+
+    await kanta.open()
+
+    with kanta.transaction(action="create_user") as data:
+        data.users["a"] = User(name="first")
+        data.users["b"] = User(name="second")
+
+    await kanta.close()
+
+    assert "A" in caplog.text
+    assert "B" in caplog.text
+
+
+@pytest.mark.asyncio
+async def test_logfmt_path_context(tmp_path, format_config, caplog):
+    import logging
+
+    caplog.set_level(logging.INFO, logger="kanta.changes")
+    path = tmp_path / "test.db"
+    kanta = make_kanta(path, Data, format_config)
+
+    @kanta.logfmt(path="users.uuid-1")
+    def resolve_user_key(value: str) -> str | None:
+        if value == "uuid-1":
+            return "user-alice"
+        return None
+
+    await kanta.open()
+
+    with kanta.transaction(action="create_user") as data:
+        data.users["uuid-1"] = User(name="Alice")
+
+    await kanta.close()
+
+    assert "user-alice" in caplog.text
+
+
+@pytest.mark.asyncio
+async def test_logfmt_decorator_path_filters_calls(tmp_path, format_config, caplog):
+    import logging
+
+    caplog.set_level(logging.INFO, logger="kanta.changes")
+    path = tmp_path / "test.db"
+    kanta = make_kanta(path, Data, format_config)
+
+    @kanta.logfmt(path="counter")
+    def fmt_counter(value: Any) -> str | None:
+        if value == 1:
+            return "one"
+        return None
+
+    await kanta.open()
+
+    with kanta.transaction(action="create_user") as data:
+        data.users["uuid-1"] = User(name="Alice")
+        data.counter = 1
+
+    await kanta.close()
+
+    assert "one" in caplog.text
+    assert "uuid-1" in caplog.text
+
+
+@pytest.mark.asyncio
+async def test_logfmt_user_path_replaces_user_display(tmp_path, format_config, caplog):
+    import logging
+
+    caplog.set_level(logging.INFO, logger="kanta.changes")
+    path = tmp_path / "test.db"
+    kanta = make_kanta(path, Data, format_config)
+
+    @kanta.logfmt(path="$user")
+    def resolve_user(value: str, current: DictPost) -> str | None:
+        return current.get("users", {}).get(value, {}).get("name")
+
+    await kanta.open()
+
+    with kanta.transaction(action="create_user", user="uuid-1") as data:
+        data.users["uuid-1"] = User(name="Alice")
+
+    await kanta.close()
+
+    assert "by Alice" in caplog.text
+
+
+@pytest.mark.asyncio
+async def test_logfmt_non_string_value(tmp_path, format_config, caplog):
+    import logging
+
+    caplog.set_level(logging.INFO, logger="kanta.changes")
+    path = tmp_path / "test.db"
+    kanta = make_kanta(path, Data, format_config)
+
+    @kanta.logfmt
+    def fmt_count(value: Any, path: str) -> str | None:
+        if path == "counter" and value == 1:
+            return "one"
+        return None
+
+    await kanta.open()
+
+    with kanta.transaction(action="inc") as data:
+        data.counter = 1
+
+    await kanta.close()
+
+    assert "one" in caplog.text
+
+
+@pytest.mark.asyncio
+async def test_fatal_error_injects_kanta_and_error(
+    tmp_path, format_config, monkeypatch
+):
+    import asyncio
+
+    path = tmp_path / "test.db"
+    errors: list[DatabaseError] = []
+    kantas: list[Kanta] = []
+    signaled = asyncio.Event()
+
+    kanta = make_kanta(path, Data, format_config, flush_interval=0.01)
+
+    @kanta.fatal_error
+    def on_fatal(error: DatabaseError, kanta_ref: Kanta) -> None:
+        errors.append(error)
+        kantas.append(kanta_ref)
+        signaled.set()
+
+    await kanta.open()
+
+    with kanta.transaction(action="inc") as data:
+        data.counter = 1
+
+    def fail_write(_data: bytes) -> None:
+        raise OSError("simulated background write failure")
+
+    monkeypatch.setattr(kanta._impl.file, "write", fail_write)
+
+    await asyncio.wait_for(signaled.wait(), timeout=1.0)
+    assert errors
+    assert kantas == [kanta]
+
+    await kanta.close()

tests/test_diff.py

@@ -1,4 +1,4 @@
-from kanta import compute_diff
+from kanta.diff import compute_diff
 
 
 def test_no_diff():

tests/test_format_diff.py

@@ -1,4 +1,4 @@
-from kanta import format_diff
+from kanta.logging import format_diff
 
 
 def test_add():
@@ -16,10 +16,34 @@ def test_delete():
     assert any("old_key" in line for line in lines)
 
 
-def test_resolver():
+def test_logfmt():
     lines = format_diff(
         {"users": {"uuid-1": {"name": "Alice"}}},
         previous={},
-        resolver=lambda x: "Alice" if x == "uuid-1" else x,
+        logfmt=lambda value, path: "Alice" if value == "uuid-1" else None,
     )
     assert any("Alice" in line for line in lines)
+
+
+def test_logfmt_uses_path_context():
+    lines = format_diff(
+        {
+            "users": {"uuid-1": {"name": "Alice"}},
+            "groups": {"uuid-1": {"name": "Admins"}},
+        },
+        previous={},
+        logfmt=lambda value, path: (
+            "User Alice" if path.startswith("users.") and value == "uuid-1" else None
+        ),
+    )
+    assert any("User Alice" in line for line in lines)
+    assert any("uuid-1" in line for line in lines)
+
+
+def test_logfmt_formats_non_string_value():
+    lines = format_diff(
+        {"count": 42},
+        previous={},
+        logfmt=lambda value, path: "forty-two" if value == 42 else None,
+    )
+    assert any("forty-two" in line for line in lines)

tests/test_kanta_integration.py

@@ -17,6 +17,8 @@ from .support import (
     change_actions,
     fixed_change,
     make_kanta,
+    make_migrations_module,
+    read_changes,
     seed_single_change,
 )
 
@@ -30,6 +32,44 @@ async def test_load_empty(tmp_path, format_config):
     await kanta.close()
 
 
+@pytest.mark.asyncio
+async def test_new_file_writes_bootstrap_record_without_handlers(
+    tmp_path, format_config
+):
+    path = tmp_path / "test.db"
+    kanta = make_kanta(path, Data, format_config)
+    await kanta.open()
+    await kanta.close()
+
+    records = read_changes(path, format_config)
+    assert len(records) == 1
+    assert records[0].a == "bootstrap"
+    assert records[0].diff == {"$replace": {"users": {}, "counter": 0}}
+
+
+@pytest.mark.asyncio
+async def test_new_file_persists_initial_state_for_roundtrip(tmp_path, format_config):
+    path = tmp_path / "test.db"
+    kanta = make_kanta(
+        path, Data(counter=5, users={"alice": User(name="Alice")}), format_config
+    )
+    await kanta.open()
+    await kanta.close()
+
+    records = read_changes(path, format_config)
+    assert len(records) == 1
+    assert records[0].a == "bootstrap"
+    assert records[0].diff == {
+        "$replace": {"users": {"alice": {"name": "Alice", "age": 0}}, "counter": 5}
+    }
+
+    kanta2 = make_kanta(path, Data, format_config)
+    await kanta2.open()
+    assert kanta2.data.counter == 5
+    assert kanta2.data.users["alice"].name == "Alice"
+    await kanta2.close()
+
+
 @pytest.mark.asyncio
 async def test_open_overwrites_caller_owned_root_data(tmp_path, format_config):
     path = tmp_path / "test.db"
@@ -100,6 +140,202 @@ async def test_bootstrap_creates_file(tmp_path, format_config):
     assert path.exists()
 
 
+@pytest.mark.asyncio
+async def test_bootstrap_decorator_with_args(tmp_path, format_config):
+    path = tmp_path / "test.db"
+    kanta = make_kanta(path, Data, format_config)
+
+    @kanta.bootstrap(action="seed_init", user="system")
+    def seed(data: Data):
+        data.counter = 3
+
+    await kanta.open()
+    await kanta.close()
+
+    assert change_actions(path, format_config) == ["seed_init"]
+
+
+@pytest.mark.asyncio
+async def test_bootstrap_decorator_without_args(tmp_path, format_config):
+    path = tmp_path / "test.db"
+    kanta = make_kanta(path, Data, format_config)
+
+    @kanta.bootstrap
+    def seed(data: Data):
+        data.counter = 4
+
+    await kanta.open()
+    await kanta.close()
+
+    assert change_actions(path, format_config) == ["bootstrap"]
+
+
+@pytest.mark.asyncio
+async def test_bootstrap_decorator_async(tmp_path, format_config):
+    path = tmp_path / "test.db"
+    kanta = make_kanta(path, Data, format_config)
+
+    @kanta.bootstrap(action="async_seed")
+    async def seed(data: Data):
+        await asyncio.sleep(0)
+        data.counter = 5
+
+    await kanta.open()
+    await kanta.close()
+
+    assert change_actions(path, format_config) == ["async_seed"]
+
+
+@pytest.mark.asyncio
+async def test_bootstrap_decorator_multiple_handlers_in_order(tmp_path, format_config):
+    path = tmp_path / "test.db"
+    kanta = make_kanta(path, Data, format_config)
+
+    @kanta.bootstrap(action="boot_1")
+    def seed_one(data: Data):
+        data.counter = 1
+
+    @kanta.bootstrap(action="boot_2")
+    async def seed_two(data: Data):
+        await asyncio.sleep(0)
+        data.counter = 2
+
+    await kanta.open()
+    await kanta.close()
+
+    assert change_actions(path, format_config) == ["boot_2"]
+
+
+@pytest.mark.asyncio
+async def test_bootstrap_failure_removes_database_file(tmp_path, format_config):
+    path = tmp_path / "test.db"
+    kanta = make_kanta(path, Data, format_config)
+
+    @kanta.bootstrap(action="boot_fail")
+    def seed_fail(data: Data):
+        data.counter = 10
+        raise RuntimeError("bootstrap failed")
+
+    with pytest.raises(RuntimeError, match="bootstrap failed"):
+        await kanta.open()
+
+    assert not path.exists()
+
+
+@pytest.mark.asyncio
+async def test_bootstrap_async_failure_removes_database_file(tmp_path, format_config):
+    path = tmp_path / "test.db"
+    kanta = make_kanta(path, Data, format_config)
+
+    @kanta.bootstrap(action="boot_fail_async")
+    async def seed_fail(data: Data):
+        await asyncio.sleep(0)
+        data.counter = 10
+        raise RuntimeError("bootstrap async failed")
+
+    with pytest.raises(RuntimeError, match="bootstrap async failed"):
+        await kanta.open()
+
+    assert not path.exists()
+
+
+@pytest.mark.asyncio
+async def test_open_create_false_missing_file_fails(tmp_path, format_config):
+    path = tmp_path / "missing.db"
+    kanta = make_kanta(path, Data, format_config)
+
+    with pytest.raises(FileLockError):
+        await kanta.open(create=False)
+
+
+@pytest.mark.asyncio
+async def test_open_create_false_empty_file_fails(tmp_path, format_config):
+    path = tmp_path / "empty.db"
+    path.touch()
+    kanta = make_kanta(path, Data, format_config)
+
+    with pytest.raises(DataIntegrityError, match="empty"):
+        await kanta.open(create=False)
+
+
+@pytest.mark.asyncio
+async def test_background_write_failure_notifies_decorator_callback(
+    tmp_path, format_config, monkeypatch
+):
+    path = tmp_path / "test.db"
+    errors: list[DatabaseError] = []
+    signaled = asyncio.Event()
+
+    kanta = make_kanta(
+        path,
+        Data,
+        format_config,
+        flush_interval=0.01,
+    )
+
+    @kanta.fatal_error
+    async def on_fatal_error(err: DatabaseError) -> None:
+        errors.append(err)
+        signaled.set()
+
+    await kanta.open()
+
+    with kanta.transaction(action="inc") as data:
+        data.counter = 1
+
+    def fail_write(_data: bytes) -> None:
+        raise OSError("simulated background write failure")
+
+    monkeypatch.setattr(kanta._impl.file, "write", fail_write)
+
+    await asyncio.wait_for(signaled.wait(), timeout=1.0)
+    assert errors
+    assert "Failed to flush database" in str(errors[0])
+
+    await kanta.close()
+
+
+@pytest.mark.asyncio
+async def test_background_write_failure_notifies_multiple_callbacks_in_order(
+    tmp_path, format_config, monkeypatch
+):
+    path = tmp_path / "test.db"
+    calls: list[str] = []
+    signaled = asyncio.Event()
+
+    kanta = make_kanta(
+        path,
+        Data,
+        format_config,
+        flush_interval=0.01,
+    )
+
+    @kanta.fatal_error
+    def on_fatal_error_sync(err: DatabaseError) -> None:
+        calls.append("sync")
+
+    @kanta.fatal_error
+    async def on_fatal_error_async(err: DatabaseError) -> None:
+        await asyncio.sleep(0)
+        calls.append("async")
+        signaled.set()
+
+    await kanta.open()
+
+    with kanta.transaction(action="inc") as data:
+        data.counter = 1
+
+    def fail_write(_data: bytes) -> None:
+        raise OSError("simulated background write failure")
+
+    monkeypatch.setattr(kanta._impl.file, "write", fail_write)
+
+    await asyncio.wait_for(signaled.wait(), timeout=1.0)
+    assert calls == ["sync", "async"]
+
+    await kanta.close()
+
+
 @pytest.mark.asyncio
 async def test_snapshot(tmp_path, format_config):
     path = tmp_path / "test.db"
@@ -208,7 +444,7 @@ async def test_migrations_from_module(tmp_path, format_config):
 
     mod = type(sys)("test_migrations")
 
-    def migrate_v1(d, ctx):
+    def migrate_v1(d, kanta):
         d["version"] = 1
 
     mod.__dict__["migrate_v1"] = migrate_v1
@@ -238,6 +474,43 @@ async def test_msgspec_normalization_logs_migration(tmp_path, format_config):
     assert "migrate:msgspec" in change_actions(path, format_config)
 
 
+@pytest.mark.asyncio
+async def test_empty_migration_is_recorded_and_not_reapplied(tmp_path, format_config):
+    path = tmp_path / "test.db"
+    seed_single_change(
+        path, fixed_change("init", {"counter": 0, "users": {}}), format_config
+    )
+
+    def migrate_v1(d, kanta):
+        """No-op migration that only bumps the schema version."""
+        pass
+
+    mod = make_migrations_module("empty_migration_mod", "migrate_v1", migrate_v1)
+
+    try:
+        kanta = make_kanta(path, Data, format_config, migrations=mod)
+        await kanta.open()
+        assert kanta.version == 1
+        await kanta.flush()
+        await kanta.close()
+
+        records = read_changes(path, format_config)
+        migration_records = [r for r in records if r.a.startswith("migrate")]
+        assert len(migration_records) == 1
+        assert migration_records[0].v == 1
+        assert migration_records[0].diff == {}
+
+        kanta2 = make_kanta(path, Data, format_config, migrations=mod)
+        await kanta2.open()
+        assert kanta2.version == 1
+        await kanta2.close()
+
+        records2 = read_changes(path, format_config)
+        assert len([r for r in records2 if r.a.startswith("migrate")]) == 1
+    finally:
+        sys.modules.pop("empty_migration_mod", None)
+
+
 @pytest.mark.asyncio
 async def test_open_locked_file_raises_filelock_error(tmp_path, format_config):
     path = tmp_path / "test.db"
@@ -281,17 +554,18 @@ async def test_background_write_failure_notifies_callback(
     errors: list[DatabaseError] = []
     signaled = asyncio.Event()
 
-    def on_fatal_error(err: DatabaseError) -> None:
-        errors.append(err)
-        signaled.set()
-
     kanta = make_kanta(
         path,
         Data,
         format_config,
         flush_interval=0.01,
-        fatal_error=on_fatal_error,
     )
+
+    @kanta.fatal_error
+    def on_fatal_error(err: DatabaseError) -> None:
+        errors.append(err)
+        signaled.set()
+
     await kanta.open()
 
     with kanta.transaction(action="inc") as data:
@@ -317,7 +591,7 @@ async def test_migrations_from_module_path(tmp_path, format_config):
     module_name = "test_migrations_path"
     mod = type(sys)(module_name)
 
-    def migrate_v1(d, ctx):
+    def migrate_v1(d, kanta):
         d["counter"] = 2
 
     mod.__dict__["migrate_v1"] = migrate_v1

tests/test_logging.py

@@ -1,6 +1,6 @@
 import logging
 
-from kanta import configure_logging, log_change
+from kanta.logging import configure_logging, log_change
 from kanta.logging import logger
 
 

tests/test_migrations.py

@@ -1,53 +1,204 @@
-from types import ModuleType
+import logging
+from types import ModuleType, SimpleNamespace
 
-from kanta.migrate import MigrationRegistry
+import pytest
+
+from kanta.exceptions import DatabaseError
+from kanta.migrations import Migrations
+
+
+class _DummyKanta:
+    def __init__(self):
+        self.ctx = SimpleNamespace()
 
 
 def test_register_and_apply():
-    reg = MigrationRegistry()
+    reg = Migrations()
+    kanta = _DummyKanta()
 
     @reg.register
-    def migrate_v1(d, ctx):
+    def migrate_v1(d, kanta):
         d["version"] = 1
 
     @reg.register
-    def migrate_v2(d, ctx):
+    def migrate_v2(d, kanta):
         d["version"] = 2
 
     state = {}
-    new_ver = reg.apply(state, current_version=0, silent=True)
+    new_ver = reg.apply(state, current_version=0, kanta=kanta, silent=True)
     assert new_ver == 2
     assert state["version"] == 2
 
 
 def test_no_migrations_needed():
-    reg = MigrationRegistry()
+    reg = Migrations()
+    kanta = _DummyKanta()
 
     @reg.register
-    def migrate_v1(d, ctx):
+    def migrate_v1(d, kanta):
         d["x"] = 1
 
     state = {"x": 1}
-    new_ver = reg.apply(state, current_version=1, silent=True)
+    new_ver = reg.apply(state, current_version=1, kanta=kanta, silent=True)
     assert new_ver == 1
 
 
 def test_from_module():
     mod = ModuleType("fake_migrations")
+    kanta = _DummyKanta()
 
-    def migrate_v1(d, ctx):
+    def migrate_v1(d, kanta):
         d["v"] = 1
 
-    def migrate_v2(d, ctx):
+    def migrate_v2(d, kanta):
         d["v"] = 2
 
     mod.__dict__["migrate_v1"] = migrate_v1
     mod.__dict__["migrate_v2"] = migrate_v2
 
-    reg = MigrationRegistry.from_module(mod)
+    reg = Migrations.from_module(mod)
     assert reg.dbver == 2
 
     state = {}
-    new_ver = reg.apply(state, current_version=0, silent=True)
+    new_ver = reg.apply(state, current_version=0, kanta=kanta, silent=True)
     assert new_ver == 2
     assert state["v"] == 2
+
+
+def test_migrations_can_use_kanta_ctx():
+    reg = Migrations()
+    kanta = _DummyKanta()
+
+    @reg.register
+    def migrate_v1(d, kanta):
+        kanta.ctx.source = "migration"
+        d["source"] = kanta.ctx.source
+
+    state = {}
+    new_ver = reg.apply(state, current_version=0, kanta=kanta, silent=True)
+    assert new_ver == 1
+    assert state["source"] == "migration"
+    assert kanta.ctx.source == "migration"
+
+
+def test_migration_can_omit_kanta_argument():
+    reg = Migrations()
+    kanta = _DummyKanta()
+
+    @reg.register
+    def migrate_v1(d):
+        d["x"] = 1
+
+    state = {}
+    new_ver = reg.apply(state, current_version=0, kanta=kanta, silent=True)
+    assert new_ver == 1
+    assert state["x"] == 1
+
+
+def test_version_too_new():
+    reg = Migrations()
+    kanta = _DummyKanta()
+
+    @reg.register
+    def migrate_v1(d):
+        d["x"] = 1
+
+    with pytest.raises(
+        DatabaseError,
+        match="Database version v2 is newer than the highest supported version v1",
+    ):
+        reg.apply({}, current_version=2, kanta=kanta, silent=True)
+
+
+def test_version_too_old():
+    reg = Migrations()
+    kanta = _DummyKanta()
+
+    @reg.register
+    def migrate_v3(d):
+        d["x"] = 3
+
+    with pytest.raises(
+        DatabaseError,
+        match="Database version v1 is older than the minimum supported version v2",
+    ):
+        reg.apply({}, current_version=1, kanta=kanta, silent=True)
+
+
+def test_missing_middle_migration_is_skipped():
+    reg = Migrations()
+    kanta = _DummyKanta()
+
+    @reg.register
+    def migrate_v1(d):
+        d["x"] = 1
+
+    @reg.register
+    def migrate_v3(d):
+        d["y"] = 3
+
+    state = {"x": 1}
+    new_ver = reg.apply(state, current_version=1, kanta=kanta, silent=True)
+    assert new_ver == 3
+    assert state["x"] == 1
+    assert state["y"] == 3
+
+
+def test_old_migrations_deleted_current_supported():
+    reg = Migrations()
+    kanta = _DummyKanta()
+
+    @reg.register
+    def migrate_v3(d):
+        d["x"] = 3
+
+    state = {"x": 2}
+    new_ver = reg.apply(state, current_version=2, kanta=kanta, silent=True)
+    assert new_ver == 3
+    assert state["x"] == 3
+
+
+def test_migration_log_only_when_changed(caplog):
+    reg = Migrations()
+    kanta = _DummyKanta()
+
+    @reg.register
+    def migrate_v1(d):
+        """Set x."""
+        d["x"] = 1
+
+    @reg.register
+    def migrate_v2(d):
+        """No-op."""
+        pass
+
+    @reg.register
+    def migrate_v3(d):
+        """Set y."""
+        d["y"] = 3
+
+    with caplog.at_level(logging.INFO, logger="kanta.migrations"):
+        reg.apply({}, current_version=0, kanta=kanta)
+
+    messages = [r.message for r in caplog.records if r.levelno == logging.INFO]
+    assert len(messages) == 2
+    assert "migrate_v1" in messages[0]
+    assert "Set x" in messages[0]
+    assert "migrate_v3" in messages[1]
+    assert "Set y" in messages[1]
+
+
+def test_no_op_migration_produces_no_log(caplog):
+    reg = Migrations()
+    kanta = _DummyKanta()
+
+    @reg.register
+    def migrate_v1(d):
+        """No-op."""
+        pass
+
+    with caplog.at_level(logging.INFO, logger="kanta.migrations"):
+        reg.apply({}, current_version=0, kanta=kanta)
+
+    info_messages = [r for r in caplog.records if r.levelno == logging.INFO]
+    assert not info_messages

tests/test_mtime.py

@@ -0,0 +1,144 @@
+"""Tests for mtime handling and the public ``kanta.mtime`` property."""
+
+from datetime import UTC, datetime
+
+import pytest
+
+from kanta.structs import ChangeRecord
+
+from .support import Data, make_kanta, seed_single_change
+
+
+def _read_last_change(path, format_config):
+    name, serializer_cls = format_config
+    serializer = serializer_cls()
+    framer = serializer.framer_cls()
+    last = None
+    for is_snapshot, payload, _, _ in framer.iter_records(path.read_bytes(), 0):
+        if is_snapshot:
+            continue
+        last = serializer.decode(payload, type=ChangeRecord)
+    assert last is not None
+    return last
+
+
+@pytest.mark.asyncio
+async def test_default_transaction_updates_mtime(tmp_path, format_config):
+    path = tmp_path / "test.db"
+    kanta = make_kanta(path, Data, format_config)
+    await kanta.open()
+
+    before = datetime.now(UTC)
+    with kanta.transaction(action="inc") as data:
+        data.counter = 1
+    await kanta.flush()
+    await kanta.close()
+
+    rec = _read_last_change(path, format_config)
+    assert rec.ts == rec.m
+    assert before <= rec.m <= datetime.now(UTC)
+    assert kanta.mtime == rec.m
+
+
+@pytest.mark.asyncio
+async def test_transaction_custom_mtime(tmp_path, format_config):
+    path = tmp_path / "test.db"
+    kanta = make_kanta(path, Data, format_config)
+    await kanta.open()
+
+    custom_m = datetime(2026, 1, 1, 8, 0, tzinfo=UTC)
+    with kanta.transaction(action="inc", mtime=custom_m) as data:
+        data.counter = 1
+    await kanta.flush()
+    await kanta.close()
+
+    rec = _read_last_change(path, format_config)
+    assert rec.m == custom_m
+    assert kanta.mtime == custom_m
+
+
+@pytest.mark.asyncio
+async def test_transaction_mtime_false_preserves_mtime(tmp_path, format_config):
+    path = tmp_path / "test.db"
+    kanta = make_kanta(path, Data, format_config)
+    await kanta.open()
+
+    first_m = datetime(2026, 1, 1, 10, 0, tzinfo=UTC)
+    with kanta.transaction(action="first", mtime=first_m) as data:
+        data.counter = 1
+
+    with kanta.transaction(action="second", mtime=False) as data:
+        data.counter = 2
+
+    await kanta.flush()
+    await kanta.close()
+
+    records = []
+    name, serializer_cls = format_config
+    serializer = serializer_cls()
+    framer = serializer.framer_cls()
+    for is_snapshot, payload, _, _ in framer.iter_records(path.read_bytes(), 0):
+        if is_snapshot:
+            continue
+        records.append(serializer.decode(payload, type=ChangeRecord))
+
+    assert records[0].a == "bootstrap"
+    assert records[1].m == first_m
+    assert records[2].m is None
+    assert kanta.mtime == first_m
+
+
+@pytest.mark.asyncio
+async def test_migration_does_not_update_mtime(tmp_path, format_config):
+    path = tmp_path / "test.db"
+    seed_m = datetime(2025, 12, 31, 23, 0, tzinfo=UTC)
+    seed_single_change(
+        path,
+        ChangeRecord(
+            ts=seed_m,
+            m=seed_m,
+            a="seed",
+            v=0,
+            diff={"counter": 0},
+        ),
+        format_config,
+    )
+
+    kanta = make_kanta(path, Data, format_config)
+    await kanta.open()
+
+    assert kanta.mtime == seed_m
+
+    new_m = datetime(2026, 1, 5, 10, 0, tzinfo=UTC)
+    with kanta.transaction(action="inc", mtime=new_m) as data:
+        data.counter = 5
+    await kanta.flush()
+
+    assert kanta.mtime == new_m
+    await kanta.close()
+
+
+@pytest.mark.asyncio
+async def test_rollback_does_not_update_mtime(tmp_path, format_config):
+    path = tmp_path / "test.db"
+    kanta = make_kanta(path, Data, format_config)
+    await kanta.open()
+
+    seed_m = datetime(2026, 1, 1, 10, 0, tzinfo=UTC)
+    with kanta.transaction(action="seed", mtime=seed_m) as data:
+        data.counter = 1
+
+    before = kanta.mtime
+
+    try:
+        with kanta.transaction(
+            action="boom", mtime=datetime(2099, 1, 1, tzinfo=UTC)
+        ) as data:
+            data.counter = 99
+            raise RuntimeError("fail")
+    except RuntimeError:
+        pass
+
+    assert kanta.data.counter == 1
+    assert kanta.mtime == before
+    await kanta.close()

tests/test_readonly.py

@@ -0,0 +1,153 @@
+"""Tests for Kanta read-only mode."""
+
+import pytest
+
+from kanta.exceptions import DataIntegrityError, FileLockError
+from kanta.serialization import struct_to_dict
+
+from .support import (
+    Data,
+    EvolvableDataV2,
+    fixed_change,
+    make_kanta,
+    make_migrations_module,
+    seed_single_change,
+)
+
+
+@pytest.mark.asyncio
+async def test_readonly_opens_existing_database(tmp_path, format_config):
+    path = tmp_path / "test.db"
+    seed_single_change(path, fixed_change("seed", {"counter": 7}), format_config)
+
+    kanta = make_kanta(path, Data, format_config)
+    await kanta.open(readonly=True)
+
+    assert isinstance(kanta.data, Data)
+    assert kanta.data.counter == 7
+    assert kanta._impl.readonly is True
+    assert kanta._impl.background_task is None
+
+    await kanta.close()
+
+
+@pytest.mark.asyncio
+async def test_readonly_missing_file_fails(tmp_path, format_config):
+    path = tmp_path / "missing.db"
+    kanta = make_kanta(path, Data, format_config)
+
+    with pytest.raises(FileLockError):
+        await kanta.open(readonly=True)
+
+    assert not path.exists()
+
+
+@pytest.mark.asyncio
+async def test_readonly_empty_file_fails(tmp_path, format_config):
+    path = tmp_path / "empty.db"
+    path.touch()
+    kanta = make_kanta(path, Data, format_config)
+
+    with pytest.raises(DataIntegrityError, match="empty"):
+        await kanta.open(readonly=True)
+
+
+@pytest.mark.asyncio
+async def test_readonly_transaction_fails(tmp_path, format_config):
+    path = tmp_path / "test.db"
+    seed_single_change(path, fixed_change("seed", {"counter": 1}), format_config)
+
+    kanta = make_kanta(path, Data, format_config)
+    await kanta.open(readonly=True)
+
+    with pytest.raises(DataIntegrityError, match="read-only"):
+        with kanta.transaction(action="inc") as data:
+            data.counter = 2
+
+    # In-memory state must remain unchanged.
+    assert kanta.data.counter == 1
+    await kanta.close()
+
+
+@pytest.mark.asyncio
+async def test_readonly_flush_fails(tmp_path, format_config):
+    path = tmp_path / "test.db"
+    seed_single_change(path, fixed_change("seed", {"counter": 1}), format_config)
+
+    kanta = make_kanta(path, Data, format_config)
+    await kanta.open(readonly=True)
+
+    with pytest.raises(DataIntegrityError, match="read-only"):
+        await kanta.flush()
+
+    await kanta.close()
+
+
+@pytest.mark.asyncio
+async def test_readonly_create_true_does_not_create_file(tmp_path, format_config):
+    path = tmp_path / "test.db"
+    kanta = make_kanta(path, Data, format_config)
+
+    with pytest.raises(FileLockError):
+        await kanta.open(create=True, readonly=True)
+
+    assert not path.exists()
+
+
+@pytest.mark.asyncio
+async def test_readonly_does_not_persist_changes(tmp_path, format_config):
+    path = tmp_path / "test.db"
+    seed_single_change(path, fixed_change("seed", {"counter": 1}), format_config)
+    original_content = path.read_bytes()
+
+    kanta = make_kanta(path, Data, format_config)
+    await kanta.open(readonly=True)
+    await kanta.close()
+
+    assert path.read_bytes() == original_content
+
+
+@pytest.mark.asyncio
+async def test_readonly_runs_migrations(tmp_path, format_config):
+    path = tmp_path / "test.db"
+    seed_single_change(
+        path,
+        fixed_change("seed", {"counter": 1}, version=0),
+        format_config,
+    )
+
+    def migrate_v1(data, kanta):
+        data.setdefault("enabled", True)
+
+    migrations = make_migrations_module("readonly_migrations", "migrate_v1", migrate_v1)
+
+    kanta = make_kanta(path, EvolvableDataV2, format_config, migrations=migrations)
+    await kanta.open(readonly=True)
+
+    assert kanta.data.counter == 1
+    # Migration ran in memory even though no change was persisted.
+    assert struct_to_dict(kanta.data, serializer=kanta._impl.serializer) == {
+        "counter": 1,
+        "enabled": True,
+    }
+    assert not kanta._impl.pending_changes
+
+    await kanta.close()
+
+
+@pytest.mark.asyncio
+async def test_readwrite_and_readonly_can_open_together(tmp_path, format_config):
+    path = tmp_path / "test.db"
+    seed_single_change(path, fixed_change("seed", {"counter": 1}), format_config)
+
+    rw = make_kanta(path, Data, format_config)
+    await rw.open()
+
+    ro = make_kanta(path, Data, format_config)
+    await ro.open(readonly=True)
+
+    assert rw.data.counter == 1
+    assert ro.data.counter == 1
+
+    await ro.close()
+    await rw.close()

tests/test_replay.py

@@ -1,6 +1,7 @@
 from datetime import UTC, datetime
 
-from kanta import ChangeRecord, Snapshot, replay
+from kanta.diff import replay_jsonl as replay
+from kanta.structs import ChangeRecord, Snapshot
 from kanta.serialization.framing import LineFramer