Add experimental kanta CLI for inspecting databases.

- Log as a single event
- Logging config has debug parameter to lower level to DEBUG, showing migration diffs

README.md

@@ -53,10 +53,13 @@ asyncio.run(main())
 
 ## Bootstrap and Open Modes
 
-Kanta supports open-time bootstrap callbacks for initializing a brand-new
-database before `open()` returns.
+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.
 
-Register bootstrap handlers with a decorator:
+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())
@@ -76,9 +79,10 @@ async def bootstrap_async(data) -> None:
     data.counter = 1
 ```
 
-When multiple bootstrap handlers are registered:
+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,
-- exactly one bootstrap change record is queued,
 - bootstrap metadata (`action`, `user`, `mtime`) is taken from the last
   registration.
 
@@ -94,6 +98,18 @@ 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:

demo/.gitignore

@@ -0,0 +1 @@
+demo.kantadb

demo/demo.py

@@ -0,0 +1,110 @@
+#!/usr/bin/env -S uv run
+import asyncio
+import sys
+from datetime import UTC, datetime, timedelta
+from pathlib import Path
+
+import msgspec
+
+from kanta import Kanta
+from kanta.callbacks import DictPre
+from kanta.logging import configure_logging
+
+
+filename = Path(__file__).with_name("demo.kantadb")
+
+# For demonstration purposes, we use "original v0" and "modified v1" in this same script
+# Normally your app would only have the latest supported data model
+
+
+class Data(msgspec.Struct):  # type: ignore - intentionally redefined later
+    users: dict[str, dict] = {}
+    counter: int = 0
+
+
+kanta_v0 = Kanta(filename, Data())
+
+
+@kanta_v0.bootstrap
+def bootstrap(data: Data) -> None:
+    """Create the initial admin user."""
+    data.users["userid001"] = {"name": "Alice", "role": "admin"}
+
+
+# Redefinition to simulate new version
+class Data(msgspec.Struct):
+    users: dict[str, dict] = {}
+    total: int = 0  # Replaces old counter field
+    lang: str = "en"  # New field
+
+
+def migrate_v1(d: dict) -> None:
+    """Rename counter to total"""
+    d["total"] = d["counter"]
+
+
+kanta_v1 = Kanta(filename, Data(), migrations=sys.modules[__name__])
+
+
+@kanta_v1.logfmt
+def resolve_user(value: str, path: str, previous: DictPre) -> str | None:
+    """Resolve user ids to names from the database state itself."""
+    if path != "$user" and not path.startswith("users."):
+        return None
+    return previous.get("users", {}).get(value, {}).get("name")
+
+
+async def main() -> None:
+    filename.unlink(missing_ok=True)
+
+    print("Database creation with v0 schema and basic transactions:\n")
+    # Open and close automatically; you can also `await kanta.open()` instead
+    async with kanta_v0 as kanta:
+        with kanta.transaction(action="create", user="userid001") as data:
+            data.users["userid002"] = {"name": "Bob", "role": "user"}
+
+        with kanta.transaction(action="update", user="userid001") as data:
+            data.users["userid002"]["role"] = "editor"
+            data.counter = 1
+
+        # Display-only extra string, appended after the action.
+        with kanta.transaction(
+            action="export", user="userid002", extra="extra info"
+        ) as data:
+            data.counter = 2
+
+    print("\nA new data model, migrations and logfmt pretty names:\n")
+    async with kanta_v1 as kanta:
+        with kanta.transaction(
+            action="update", user="userid002", extra=filename.name
+        ) as data:
+            data.total += 1
+
+        try:
+            with kanta.transaction(action="reset", user="userid001") as data:
+                data.total = 99
+                raise ValueError("simulated failure")
+        except ValueError:
+            print(
+                f"\nReset rolled back: {data.total=} (we can always read data without tx)\n"
+            )
+
+        with kanta.transaction(action="delete", user="userid002") as data:
+            del data.users["userid001"]
+
+
+# Fake clock for deterministic timestamps
+_now = datetime(2027, 1, 1, tzinfo=UTC)
+
+
+@kanta_v0.clock
+@kanta_v1.clock
+def fake_clock() -> datetime:
+    global _now
+    _now += timedelta(hours=1)
+    return _now
+
+
+if __name__ == "__main__":
+    configure_logging(debug=True)
+    asyncio.run(main())

docs/database.md

@@ -76,8 +76,9 @@ history.
 - In-memory data is defined by an application `msgspec.Struct` type.
 - Kanta round-trips through plain builtins for persistence and diffing.
 - Dict keys are serialized as strings (`str_keys=True`) for stable JSON form.
-- Normalization changes introduced by struct decode/encode are logged as
-  `migrate:msgspec` when they produce a diff.
+- Normalization changes introduced by struct decode/encode are logged together
+  with migrations as `migrate:vN`, or as `migrate:msgspec` when no migration
+  ran but normalization still produces a diff.
 
 ## Transaction Semantics
 
@@ -118,6 +119,12 @@ reloads, while system operations such as migrations leave it unchanged.
 
 - `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
 
@@ -127,7 +134,11 @@ when they have a default value.
 
 #### Bootstrap Callbacks
 
-- Bootstrap callbacks run during `open()` when the database is empty.
+- 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=...)`
@@ -140,6 +151,8 @@ when they have a default value.
   - 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.
 
@@ -151,6 +164,17 @@ when they have a default value.
 - Multiple handlers are supported and invoked in registration order.  A failing
   handler is logged and does not prevent subsequent handlers from running.
 
+#### Clock
+
+- `@kanta.clock` registers a callback `() -> datetime` that replaces the
+  default UTC clock.  Its value is used for all record timestamps (`ts`, and
+  `m` when `mtime` is `True`) and for snapshot timestamps.
+- The clock is only read when a timestamp is actually produced; no-op
+  transactions and skipped snapshot checks do not read it.
+- Register before `open()` so that bootstrap and migration records use the
+  custom clock as well.  This is mainly useful for tests and reproducible
+  demos.
+
 #### Transaction Log Formatting
 
 - Logfmt callbacks prettify identifiers in the change log and are registered with
@@ -185,6 +209,91 @@ def resolve_user_key(value: str) -> str | None:
     return names_by_id.get(value)
 ```
 
+#### Transaction Log Headers
+
+- By default a transaction is logged with an `action by user` header followed
+  by the diff lines.  Added paths are colored green, deleted paths red.
+- `kanta.transaction(..., extra=...)` accepts a display-only value that is
+  shown after the action in the header.  Anything other than `None` is
+  printed str-converted (colored by Kanta), unless a custom logemit handler
+  does something else with it; it is never persisted in the `ChangeRecord`.
+- `kanta.transaction(..., logdiff=False)` skips building and printing the diff
+  body and logs only the header, which is useful for large or noisy
+  changesets.  Diff output can also be disabled globally with
+  `configure_logging(diff=False)`; diff lines are emitted on the
+  `kanta.transaction.diff` child logger so applications can route or silence
+  them separately from the headers.
+
+#### Log Emitters
+
+- Every change-related message Kanta emits (transaction/bootstrap/migration
+  changes, file created/opened lines, migration summaries, aborted
+  transactions) is described by a `kanta.logging.LogEvent` and dispatched
+  through
+  `kanta.logging.emit_event`.  Kanta's own output goes through the same
+  mechanism: when no `logemit` callback handles an event,
+  `kanta.logging.default_emit` renders it with the built-in formatting.
+- A `LogEvent` carries the event `kind` (`"change"`, `"created"`,
+  `"opened"`, `"migrated"`, `"aborted"`), the preferred `logger` and `level`,
+  the
+  `kanta` instance, and all relevant state: `action`, `user`, `extra`,
+  `error` (for aborted transactions), `diff`, `previous`/`current` state
+  dicts, the built `logfmt` chain, and version info for migration events.
+  Application-specific context (e.g. a connection id) can be stored in
+  `kanta.ctx` — a user-writable namespace — and read back in callbacks as
+  `event.kanta.ctx`, which also covers creation/bootstrap events.
+- The built-in formatting is assembled from standard blocks that custom
+  emitters can reuse as-is or replace piecemeal:
+  - `event.header` — a lazy property producing the default one-line header
+    for any kind: `<action>[ <extra>][ by <user>]` for changes,
+    `<action>[ <extra>][ by <user>] transaction aborted: <error>` for aborts,
+    and the `🛢️ <file> created|opened|migrated ...` summaries.  It is
+    settable: assign
+    `event.header = ...` and return truthy to restyle the header while
+    keeping the default diff routing.
+  - `event.diff_lines` — a lazy property producing the pretty diff body for
+    change events (built only if accessed).
+  - `default_emit` itself is just `header` plus the `diff_lines` routing.
+- `@kanta.logemit` registers a callback receiving the event.  The callback
+  decides what is logged and where: it may log one or more messages on
+  `event.logger`, log somewhere else, or nothing at all.  A falsy return
+  value marks the event handled and stops the chain; a truthy return value
+  passes the event — possibly modified — to the next registered callback.
+  When all callbacks pass, `default_emit` renders the event; a callback may
+  also call `default_emit(event)` itself to delegate events it does not
+  customize.  Operational diagnostics (integrity errors, background flush
+  failures) do not go through this mechanism.
+- Logging never breaks functionality: a crashing `logemit` callback is
+  reported with `logger.exception` and the event falls back to the built-in
+  formatting; if the built-in formatting itself fails, the error is reported
+  and swallowed.  The same applies to `logfmt` callbacks (a failing one is
+  treated as a fall-through) and `logmigr` callbacks.
+
+```python
+@kanta.logemit
+def emit(ev: LogEvent):
+    if ev.kind != "change":
+        return default_emit(ev)  # delegate, no chaining needed
+    # Restyle the header; default_emit keeps routing the diff body.
+    ev.header = str(Line().user(ev.user or "-", width=20)(" ").action(ev.action))
+    return True
+```
+
+#### Terminal Formatting Helpers
+
+- `kanta.tty` provides the building blocks used by Kanta's own rendering:
+  - `colors`: the mutable color palette.  Colors are bare SGR parameter
+    strings (e.g. `"1;34"`, `"38;5;226"`) without escape framing.  Attributes
+    are read at render time, so assignments (`colors.action = "36"`) and
+    additions (`colors.session = "38;5;226"`) take effect immediately.
+  - `Line`: builds a terminal string part by part.  Calling it appends
+    content (`str`-converted); `.<colorname>` arms a palette color for the
+    next call only, and the reset is folded into a single escape sequence
+    with whatever color comes next.  `width=`/`align=` pad by display width;
+    `str(line)` finishes the line and restores default colors.
+  - `strip_ansi`, `displaywidth` (wide chars and emoji count correctly) and
+    `pad` for working with pre-colored strings.
+
 ## Migrations
 
 - Migration source is configured on `Kanta(...)` via `migrations=`.

kanta/callbacks.py

@@ -7,20 +7,29 @@ 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".
+
+Log emitters (``logemit``) are another special case: plain callables that
+receive a :class:`kanta.logging.LogEvent` and are dispatched by
+:func:`kanta.logging.emit_event`.
 """
 
 from __future__ import annotations
 
 import inspect
+import logging
+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
+from kanta.migrations import MigrationResult
 
 DictPre = Annotated[dict, "pre"]
 DictPost = Annotated[dict, "post"]
 
+_logger = logging.getLogger(__name__)
+
 
 class LogFmt:
     """Base class for stateful logfmt callbacks.
@@ -58,6 +67,7 @@ class InjectionContext:
     error: DatabaseError | None = None
     previous_state: dict | None = None
     current_state: dict | None = None
+    migration_result: MigrationResult | None = None
 
 
 @dataclass
@@ -97,8 +107,10 @@ class CallbackRegistry:
         self._callbacks: dict[str, list[_CallbackRegistration]] = {
             "bootstrap": [],
             "fatal_error": [],
+            "logmigr": [],
         }
         self._logfmt_callbacks: list[_LogFmtFunctionSpec | _LogFmtClassSpec] = []
+        self._logemit_callbacks: list[Callable[..., Any]] = []
 
     def register(
         self,
@@ -119,6 +131,14 @@ class CallbackRegistry:
                 )
             return callback
 
+        if kind == "logemit":
+            if inspect.isclass(callback) or not callable(callback):
+                raise TypeError("logemit callbacks must be functions")
+            if inspect.iscoroutinefunction(callback):
+                raise TypeError("logemit callbacks must not be async")
+            self._logemit_callbacks.append(callback)
+            return callback
+
         if kind not in self._callbacks:
             raise ValueError(f"unknown callback kind: {kind}")
 
@@ -171,8 +191,15 @@ class CallbackRegistry:
         """Return True if any callback of *kind* is registered."""
         if kind == "logfmt":
             return bool(self._logfmt_callbacks)
+        if kind == "logemit":
+            return bool(self._logemit_callbacks)
         return bool(self._callbacks[kind])
 
+    @property
+    def logemit_handlers(self) -> list[Callable[..., Any]]:
+        """Registered logemit callbacks in registration order."""
+        return self._logemit_callbacks
+
     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]] = []
@@ -206,7 +233,13 @@ class CallbackRegistry:
             for fn, pattern in formatters:
                 if pattern is not None and path != pattern:
                     continue
-                resolved = fn(value, path)
+                try:
+                    resolved = fn(value, path)
+                except Exception:
+                    # Formatting must never break functionality; a failing
+                    # callback is reported and treated as a fall-through.
+                    _logger.exception("logfmt callback %r failed", fn)
+                    continue
                 if resolved is not None:
                     return resolved
             return None
@@ -317,17 +350,13 @@ class CallbackRegistry:
                     f"Allowed: str path, {self._allowed_message('logfmt')}"
                 )
 
-        if sig.return_annotation is inspect.Signature.empty:
-            raise TypeError(
-                f"logfmt callback {callback.__name__} must annotate its "
-                f"return type as str | None"
-            )
-        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}"
-            )
+        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,
@@ -416,19 +445,15 @@ class CallbackRegistry:
                 f"logfmt class {cls.__name__}.resolve must accept a 'path: str' parameter"
             )
 
-        if resolve_sig.return_annotation is inspect.Signature.empty:
-            raise TypeError(
-                f"logfmt class {cls.__name__}.resolve must annotate its "
-                f"return type as str | None"
-            )
-        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}"
+        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)
 
@@ -453,10 +478,17 @@ class CallbackRegistry:
             return kind == "logfmt"
         if bare is DatabaseError:
             return kind == "fatal_error"
+        if bare is MigrationResult:
+            return kind == "logmigr"
         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 kind in {
+                "bootstrap",
+                "fatal_error",
+                "logfmt",
+                "logmigr",
+            }
         return False
 
     def _allowed_message(self, kind: str) -> str:
@@ -464,11 +496,13 @@ class CallbackRegistry:
         if kind == "bootstrap":
             if self._data_type is not None:
                 parts.append(self._data_type.__name__)
-        if kind in {"bootstrap", "fatal_error", "logfmt"}:
+        if kind in {"bootstrap", "fatal_error", "logfmt", "logmigr"}:
             if self._kanta_class is not None:
                 parts.append(self._kanta_class.__name__)
         if kind == "fatal_error":
             parts.append("DatabaseError")
+        if kind == "logmigr":
+            parts.append("MigrationResult")
         if kind == "logfmt":
             parts.append("Annotated[dict, 'pre']")
             parts.append("Annotated[dict, 'post']")
@@ -482,6 +516,8 @@ class CallbackRegistry:
             return ctx.current_state
         if bare is DatabaseError:
             return ctx.error
+        if bare is MigrationResult:
+            return ctx.migration_result
         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:
@@ -516,7 +552,7 @@ class CallbackRegistry:
     @staticmethod
     def _unwrap_optional(ann: Any) -> Any:
         origin = get_origin(ann)
-        if origin is not Union:
+        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
@@ -524,7 +560,7 @@ class CallbackRegistry:
     @staticmethod
     def _is_optional_str(ann: Any) -> bool:
         origin = get_origin(ann)
-        if origin is not Union:
+        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)

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,15 +92,16 @@ 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)
-        content = f.read()    # read entire content
-        f.write(data)         # append data (seeks to end first)
-        f.close()             # release lock + close fd
+        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
 
     Unix:    fcntl.flock (advisory) — read-only callers that don't flock are unaffected.
     Windows: CreateFileW with FILE_SHARE_READ — OS blocks other writers.
@@ -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,20 +193,24 @@ class LockedFile:
 
     # -- Unix ----------------------------------------------------------------
 
-    def _open_unix(self, path: Path, create: bool) -> None:
-        flags = os.O_RDWR | (os.O_CREAT if create else 0)
+    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)
-        try:
-            fcntl.flock(fd, fcntl.LOCK_EX | fcntl.LOCK_NB)
-        except OSError:
-            os.close(fd)
-            _fatal(
-                f"{path.resolve()}: database already locked by another instance",
-                db_path=path,
-            )
+        if not readonly:
+            try:
+                fcntl.flock(fd, fcntl.LOCK_EX | fcntl.LOCK_NB)
+            except OSError:
+                os.close(fd)
+                _fatal(
+                    f"{path.resolve()}: database already locked by another instance",
+                    db_path=path,
+                )
         self._fd = fd
 
     def _read_unix(self) -> bytes:
@@ -220,12 +229,19 @@ class LockedFile:
 
     # -- Windows -------------------------------------------------------------
 
-    def _open_win32(self, path: Path, create: bool) -> None:
-        disposition = _OPEN_ALWAYS if create else _OPEN_EXISTING
+    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,9 +1,10 @@
 """Kanta DB main public API"""
 
 from __future__ import annotations
+import logging
 from datetime import datetime
 from pathlib import Path
-from types import ModuleType
+from types import ModuleType, SimpleNamespace
 from typing import Any, Generic, TypeVar
 
 from kanta.kantaimpl import KantaImpl
@@ -50,7 +51,6 @@ class Kanta(Generic[T]):
         *,
         type: type[T] | None = None,
         migrations: ModuleType | str | None = None,
-        migration_ctx: Any | None = None,
         serializer: Serializer | None = None,
         flush_interval: float = 0.1,
     ):
@@ -61,7 +61,6 @@ 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.
 
@@ -78,7 +77,6 @@ class Kanta(Generic[T]):
             data=data,
             type=data_type,
             migrations=migrations,
-            migration_ctx=migration_ctx,
             flush_interval=flush_interval,
             kanta=self,
         )
@@ -127,6 +125,18 @@ class Kanta(Generic[T]):
         """
         return self._impl.filename
 
+    @property
+    def ctx(self) -> SimpleNamespace:
+        """User-writable context namespace.
+
+        Migration functions receive the ``Kanta`` instance and can read or
+        mutate ``kanta.ctx`` during migrations.  Applications can also store
+        arbitrary data here (e.g. a connection id); since
+        :class:`kanta.logging.LogEvent` carries the Kanta instance, logemit
+        callbacks can read it as ``event.kanta.ctx``.
+        """
+        return self._impl.ctx
+
     @property
     def mtime(self) -> datetime | None:
         """Last modification time carried forward from change records.
@@ -138,7 +148,13 @@ class Kanta(Generic[T]):
         """
         return self._impl.mtime
 
-    async def open(self, *, create: bool = True) -> None:
+    async def open(
+        self,
+        *,
+        create: bool = True,
+        readonly: bool = False,
+        log: bool | logging.Logger = True,
+    ) -> None:
         """Open the database file and start background persistence.
 
         This loads existing records, applies configured migrations, and starts
@@ -147,6 +163,16 @@ class Kanta(Generic[T]):
         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.
+            log: Controls bootstrap and migration logging. ``True`` (default)
+                uses the ``kanta.bootstrap`` logger for bootstrap records and
+                the ``kanta.migration`` logger for migration output. ``False``
+                suppresses the default bootstrap and migration logs. A
+                :class:`~logging.Logger` instance writes default output to that
+                logger instead. Custom ``@kanta.logmigr`` callbacks run
+                regardless of this setting.
 
         Calling ``open`` more than once on the same instance is not allowed.
 
@@ -154,7 +180,7 @@ class Kanta(Generic[T]):
             kanta.exceptions.DatabaseError: If replay or decoding fails.
             kanta.exceptions.DataIntegrityError: If the instance is already open.
         """
-        await self._impl.open(create=create)
+        await self._impl.open(create=create, readonly=readonly, log=log)
 
     async def __aenter__(self) -> Kanta[T]:
         """Enter async context manager and open the database.
@@ -230,6 +256,45 @@ class Kanta(Generic[T]):
             return _register
         return _register(fn)
 
+    def clock(self, fn=None):
+        """Register a clock callback replacing the default UTC clock.
+
+        Can be used as ``@kanta.clock``.  The callback takes no arguments and
+        must return a :class:`~datetime.datetime`; its value is used for all
+        record timestamps (``ts``, and ``m`` when ``mtime`` is ``True``) and
+        snapshot timestamps.  The clock is only read when a timestamp is
+        actually produced, so read-count-dependent clocks (e.g. advancing on
+        every read) stay deterministic.  Register before :meth:`open` so that
+        bootstrap and migration records use the custom clock as well.  This is
+        mainly useful for tests and reproducible demos.
+        """
+
+        def _register(callback):
+            self._impl.add_clock(callback)
+            return callback
+
+        if fn is None:
+            return _register
+        return _register(fn)
+
+    def logmigr(self, fn=None):
+        """Register a migration logging callback.
+
+        Can be used as ``@kanta.logmigr``.
+        The callback receives a :class:`kanta.migrations.MigrationResult` and
+        may be sync or async. If registered, it replaces the default migration
+        logger output; the application is responsible for emitting any log
+        messages.
+        """
+
+        def _register(callback):
+            self._impl.add_logmigr(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.
 
@@ -251,12 +316,39 @@ class Kanta(Generic[T]):
             return _register
         return _register(fn)
 
+    def logemit(self, fn=None):
+        """Register a log emitter callback.
+
+        Can be used as ``@kanta.logemit``.  The callback receives a single
+        :class:`kanta.logging.LogEvent` describing the event, including the
+        preferred logger and level, and decides what (if anything) is logged
+        and where.
+
+        A falsy return value marks the event as handled and stops the chain.
+        A truthy return value passes the event — possibly modified — to the
+        next registered callback; when all callbacks pass, Kanta renders the
+        event with its built-in formatting
+        (:func:`kanta.logging.default_emit`), which a callback may also call
+        itself to delegate events it does not care about.
+        """
+
+        def _register(callback):
+            self._impl.add_logemit(callback)
+            return callback
+
+        if fn is None:
+            return _register
+        return _register(fn)
+
     def transaction(
         self,
         action: str,
         *,
         user: str | None = None,
+        extra: Any = None,
         mtime: bool | datetime = True,
+        log: bool | logging.Logger = True,
+        logdiff: bool = True,
     ):
         """Create a transactional mutation context manager.
 
@@ -265,12 +357,26 @@ class Kanta(Generic[T]):
             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.
+            extra: Optional display-only value shown after the action in the
+                log header.  Anything other than ``None`` is printed
+                str-converted (colored by Kanta), unless a custom
+                ``@kanta.logemit`` handler does something else with it.  It is
+                never persisted in the change record.
             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.
+            log: Controls transaction logging. ``True`` (default) uses the
+                ``kanta.transaction`` logger. ``False`` suppresses the
+                transaction log. A :class:`~logging.Logger` instance writes
+                output to that logger instead.
+            logdiff: Whether to build and print the diff body. ``False``
+                skips diff formatting entirely and logs only the header,
+                which is useful for large or noisy changesets. Diff output
+                can also be disabled globally with
+                ``configure_logging(diff=False)``.
 
         Returns:
             A context manager yielding the live state object for mutation.
@@ -284,5 +390,8 @@ class Kanta(Generic[T]):
             self._impl,
             action,
             user=user,
+            extra=extra,
             mtime=mtime,
+            log=log,
+            logdiff=logdiff,
         )

kanta/kantaimpl.py

@@ -7,11 +7,19 @@ 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.logging import (
+    _USER_PATH,
+    LogEvent,
+    bootstrap_logger,
+    emit_event,
+    migration_logger,
+)
+from kanta.migrations import MigrationResult, Migrations
 from kanta.persistence import PersistenceMixin
 from kanta.serialization import restore_data_in_place, struct_to_dict
 from kanta.serialization.base import replay
@@ -21,28 +29,34 @@ _logger = logging.getLogger(__name__)
 T = TypeVar("T")
 
 
+def _log_callback_error(callback_error, callback):
+    """Report a failing logging callback and continue with the next one."""
+    _logger.exception("Log callback %r failed: %s", callback, callback_error)
+
+
 class KantaImpl(PersistenceMixin, Generic[T]):
     """Internal state and logic for Kanta."""
 
     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
@@ -53,9 +67,7 @@ class KantaImpl(PersistenceMixin, Generic[T]):
         )
 
         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.version = self.migrations.dbver if self.migrations is not None else 0
 
     def add_bootstrap(
         self,
@@ -75,7 +87,64 @@ class KantaImpl(PersistenceMixin, Generic[T]):
         """Register one transaction logfmt callback."""
         self.callback_registry.register("logfmt", callback, path=path)
 
-    async def open(self, *, create: bool = True) -> None:
+    def add_logmigr(self, callback) -> None:
+        """Register one migration logging callback."""
+        self.callback_registry.register("logmigr", callback)
+
+    def add_logemit(self, callback) -> None:
+        """Register one log emitter callback."""
+        self.callback_registry.register("logemit", callback)
+
+    async def _handle_migration_log(
+        self,
+        migration_result: MigrationResult,
+        previous_version: int,
+        log: bool | logging.Logger,
+    ) -> None:
+        """Route migration logging to callback or default logger."""
+        assert isinstance(migration_result, MigrationResult)
+
+        if self.callback_registry.has("logmigr"):
+            await self.callback_registry.invoke(
+                "logmigr",
+                InjectionContext(
+                    kanta=self._kanta,
+                    migration_result=migration_result,
+                ),
+                on_error=_log_callback_error,
+            )
+            return
+
+        if log is False:
+            return
+
+        migration_log = log if isinstance(log, logging.Logger) else migration_logger
+
+        changed = [m for m in migration_result.migrations if m.changed]
+        if not changed:
+            return
+
+        descriptions = [f"{m.name} ({m.description})" for m in changed]
+        emit_event(
+            LogEvent(
+                kind="migrated",
+                logger=migration_log,
+                kanta=self._kanta,
+                filename=str(self.filename),
+                from_version=previous_version,
+                to_version=migration_result.version,
+                migrations=descriptions,
+            ),
+            self.callback_registry.logemit_handlers,
+        )
+
+    async def open(
+        self,
+        *,
+        create: bool = True,
+        readonly: bool = False,
+        log: bool | logging.Logger = True,
+    ) -> None:
         """Open the database: load from disk, apply migrations, start background task."""
         if self.opened:
             raise DataIntegrityError(
@@ -84,12 +153,17 @@ 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=create,
+            create=open_create,
+            readonly=readonly,
         )
 
         if not create and (not existed_before_open or not content):
@@ -105,6 +179,9 @@ class KantaImpl(PersistenceMixin, Generic[T]):
                 action="open",
             )
 
+        # From this point the file is open and must be closed via close().
+        self.opened = True
+
         if content:
             try:
                 rr = replay(
@@ -134,12 +211,29 @@ 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
+            migration_result = None
+            state_before_migrations = None
+            previous_version = rr.version
+            if self.migrations is not None:
+                state_before_migrations = copy.deepcopy(rr.state)
+                migration_result = self.migrations.apply(
+                    rr.state, rr.version, self._kanta
                 )
+                rr.version = migration_result.version
 
-            self.statedict = copy.deepcopy(rr.state)
+            migrations_ran = rr.version != previous_version
+
+            self.snapshot.ts = (
+                datetime.fromtimestamp(rr.last_snapshot_mtime, UTC)
+                if rr.last_snapshot_mtime is not None
+                else None
+            )
+
+            self.statedict = copy.deepcopy(
+                state_before_migrations
+                if state_before_migrations is not None
+                else rr.state
+            )
             self.data = restore_data_in_place(
                 self.data,
                 rr.state,
@@ -148,28 +242,132 @@ class KantaImpl(PersistenceMixin, Generic[T]):
             )
             self.version = rr.version
             self.mtime = rr.m
-            normalized = struct_to_dict(self.data, serializer=self.serializer)
-            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.callback_registry.has("bootstrap"):
-            try:
-                await self.callback_registry.invoke(
-                    "bootstrap",
-                    InjectionContext(data=self.data, kanta=self._kanta),
+            if log is not False and not migrations_ran:
+                logger = log if isinstance(log, logging.Logger) else bootstrap_logger
+                emit_event(
+                    LogEvent(
+                        kind="opened",
+                        logger=logger,
+                        level=logging.DEBUG,
+                        kanta=self._kanta,
+                        filename=str(self.filename.resolve()),
+                    ),
+                    self.callback_registry.logemit_handlers,
                 )
+            normalized = struct_to_dict(self.data, serializer=self.serializer)
+            if self.readonly:
+                self.statedict = copy.deepcopy(normalized)
+            else:
+                # One record per open: migration changes and normalization are
+                # grouped into migrate:vN, or migrate:msgspec when only the
+                # serialization drifted.
+                previous = self.statedict
+                action = (
+                    f"migrate:v{self.version}" if migrations_ran else "migrate:msgspec"
+                )
+                record = self.queue_change(action, normalized, mtime=False)
+                # The migration summary introduces the diff, so log it first.
+                if migrations_ran and migration_result is not None:
+                    await self._handle_migration_log(
+                        migration_result, previous_version, log
+                    )
+                if (
+                    record is not None
+                    and log is not False
+                    and not (migrations_ran and self.callback_registry.has("logmigr"))
+                ):
+                    logger = (
+                        log if isinstance(log, logging.Logger) else migration_logger
+                    )
+                    emit_event(
+                        LogEvent(
+                            kind="change",
+                            logger=logger,
+                            level=logging.DEBUG,
+                            kanta=self._kanta,
+                            action=action,
+                            diff=record.diff,
+                            previous=previous,
+                        ),
+                        self.callback_registry.logemit_handlers,
+                    )
+                if migrations_ran or record is not None:
+                    self.snapshot.request_force()
+                    await self.flush()
+                    self.snapshot.maybe_write(
+                        self.file,
+                        self.version,
+                        self.statedict,
+                        m=self.mtime,
+                        now=self.now,
+                    )
+        elif self.readonly:
+            self.opened = False
+            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(
+                record = self.queue_change(
                     self.bootstrap_action,
                     current,
                     user=self.bootstrap_user,
                     mtime=self.bootstrap_mtime,
+                    force=True,
                 )
+
+                if record is not None and log is not False:
+                    logger = (
+                        log if isinstance(log, logging.Logger) else bootstrap_logger
+                    )
+                    emit_event(
+                        LogEvent(
+                            kind="created",
+                            logger=logger,
+                            kanta=self._kanta,
+                            filename=str(self.filename.resolve()),
+                        ),
+                        self.callback_registry.logemit_handlers,
+                    )
+                    logfmt = self.callback_registry.build_logfmt(
+                        InjectionContext(
+                            previous_state={},
+                            current_state=current,
+                            kanta=self._kanta,
+                        )
+                    )
+                    formatted_user = self.bootstrap_user
+                    if formatted_user is not None and logfmt is not None:
+                        resolved = logfmt(formatted_user, _USER_PATH)
+                        if resolved is not None:
+                            formatted_user = resolved
+                    emit_event(
+                        LogEvent(
+                            kind="change",
+                            logger=logger,
+                            kanta=self._kanta,
+                            action=self.bootstrap_action,
+                            user=formatted_user,
+                            diff=record.diff,
+                            previous={},
+                            current=current,
+                            logfmt=logfmt,
+                        ),
+                        self.callback_registry.logemit_handlers,
+                    )
             except Exception:
+                self.opened = False
                 self.file.close()
                 try:
                     await asyncio.to_thread(self.filename.unlink, missing_ok=True)
@@ -177,9 +375,8 @@ class KantaImpl(PersistenceMixin, Generic[T]):
                     pass
                 raise
 
-        self.opened = True
-
-        self.background_task = asyncio.create_task(self._background_loop())
+        if not self.readonly:
+            self.background_task = asyncio.create_task(self._background_loop())
 
     async def close(self) -> None:
         """Stop the background task, flush pending changes, and release the file lock."""
@@ -196,7 +393,8 @@ class KantaImpl(PersistenceMixin, Generic[T]):
 
         # Always run a final flush in case the background task never reached
         # its cancellation handler.
-        await self.flush()
+        if not self.readonly:
+            await self.flush()
 
         self.file.close()
         self.opened = False

kanta/logging.py

@@ -1,16 +1,27 @@
 """Database change logging with pretty-printed diffs.
 
-Provides a logger for JSONL database changes that formats diffs
-in a human-readable path.notation style with color coding.
+All change-related output is described by a :class:`LogEvent` and dispatched
+through :func:`emit_event`, which runs any registered ``logemit`` callbacks
+and falls back to :func:`default_emit` for the built-in formatting.  Diff
+output is formatted in a human-readable path notation style with color
+coding; see :mod:`kanta.tty` for the color palette and line builder.
 """
 
 import logging
 import re
 import sys
-from collections.abc import Callable
+from collections.abc import Callable, Iterable
 from typing import Any
 
-logger = logging.getLogger("kanta.changes")
+import msgspec
+
+from kanta.tty import Line, displaywidth
+
+transaction_logger = logging.getLogger("kanta.transaction")
+bootstrap_logger = logging.getLogger("kanta.bootstrap")
+migration_logger = logging.getLogger("kanta.migration")
+
+_logger = logging.getLogger(__name__)
 
 # Pattern to match control characters and bidirectional overrides
 _UNSAFE_CHARS = re.compile(
@@ -21,20 +32,157 @@ _UNSAFE_CHARS = re.compile(
     r"]"
 )
 
-# ANSI color codes
-_RESET = "\033[0m"
-_SEP = "\033[38;5;242m"  # Dark grey for separators
-_PATH_PREFIX = "\033[38;5;242m"  # Dark grey for path prefix
-_PATH_FINAL = "\033[38;5;250m"  # Default for final element
-_DELETE = "\033[1;31m"  # Red for deletions
-_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"
 
 
+class LogEvent(msgspec.Struct, kw_only=True):
+    """All state describing one loggable event, passed to logemit callbacks.
+
+    ``kind`` is ``"change"`` (transaction, bootstrap, or migration diff),
+    ``"created"`` (database file created), ``"opened"`` (database file
+    opened), ``"migrated"`` (migration summary), or ``"aborted"``
+    (transaction rolled back).  ``logger`` and ``level`` are Kanta's
+    preferred destination; a callback may use them, log elsewhere, or not
+    log at all.
+
+    The event is mutable: a callback may modify it before returning a truthy
+    value to pass it on, affecting later callbacks and the built-in fallback.
+    """
+
+    kind: str
+    logger: logging.Logger
+    level: int = logging.INFO
+    kanta: Any = None
+    action: str | None = None
+    user: str | None = None
+    extra: Any = None
+    error: BaseException | None = None
+    diff: dict = msgspec.field(default_factory=dict)
+    previous: dict | None = None
+    current: dict | None = None
+    logfmt: Callable[[Any, str], str | None] | None = None
+    show_diff: bool = True
+    filename: str | None = None
+    from_version: int | None = None
+    to_version: int | None = None
+    migrations: list[str] = msgspec.field(default_factory=list)
+    _header: str | None = None
+    _diff_lines: list[str] | None = None
+
+    @property
+    def header(self) -> str:
+        """The default one-line header for this event, built on first access.
+
+        Covers every event kind: ``"<action>[ <extra>][ by <user>]"`` for
+        changes, ``"<action>[ <extra>][ by <user>] transaction aborted:
+        <error>"`` for aborts, and the ``🛢️ <filename> <verb>`` file
+        summaries (created / opened / migrated).
+        """
+        if self._header is None:
+            self._header = self._build_header()
+        return self._header
+
+    @header.setter
+    def header(self, value: str) -> None:
+        """Override the header, keeping the default diff routing.
+
+        A logemit callback can restyle the header and return a truthy value:
+        :func:`default_emit` then logs this header instead of building one.
+        """
+        self._header = value
+
+    def _build_header(self) -> str:
+        if self.kind == "created":
+            return f"🛢️ {self.filename} created"
+        if self.kind == "opened":
+            return f"🛢️ {self.filename} opened"
+        if self.kind == "migrated":
+            migrations = ", ".join(self.migrations)
+            return (
+                f"🛢️ {self.filename} migrated "
+                f"v{self.from_version} -> v{self.to_version}: {migrations}"
+            )
+        if self.kind == "change":
+            return format_action_header(self.action or "", self.user, self.extra)
+        line = Line().action(self.action or "")
+        if self.extra:
+            line(" ").target(self.extra)
+        if self.user:
+            line(" by ").user(self.user)
+        line(f" transaction aborted: {self.error}")
+        return str(line)
+
+    @property
+    def diff_lines(self) -> list[str]:
+        """Pretty-printed diff lines, built on first access and cached."""
+        if self._diff_lines is None:
+            self._diff_lines = format_diff(self.diff, self.previous, self.logfmt)
+        return self._diff_lines
+
+
+def emit_event(
+    ev: LogEvent,
+    handlers: Iterable[Callable[[LogEvent], Any]] = (),
+    *,
+    fallback: Callable[[LogEvent], None] | None = None,
+) -> None:
+    """Dispatch *ev* through registered logemit handlers.
+
+    Each handler receives the event and may log it (or not) as it sees fit.
+    A falsy return value stops the chain: the event is considered handled.
+    A truthy return value passes the event — possibly modified — to the next
+    handler.  When all handlers pass, the *fallback* renders the event;
+    the default fallback is :func:`default_emit` with the built-in formatting.
+
+    Logging must never break functionality: a crashing handler is reported
+    and the chain falls back to the fallback rendering, and a failure in
+    the fallback itself is reported and swallowed.
+    """
+    render = fallback if fallback is not None else default_emit
+    try:
+        for handler in handlers:
+            try:
+                proceed = handler(ev)
+            except Exception:
+                _logger.exception("logemit callback failed, using default formatting")
+                break
+            if not proceed:
+                return
+        render(ev)
+    except Exception:
+        _logger.exception("failed to emit %s log event", ev.kind)
+
+
+def default_emit(ev: LogEvent) -> None:
+    """Emit *ev* with Kanta's built-in formatting.
+
+    Logs :attr:`LogEvent.header`; for change events the
+    :attr:`LogEvent.diff_lines` body follows on the ``<logger>.diff`` child
+    logger so it can be silenced or routed separately from the headers.
+    This is what runs when no logemit callback handles the event; custom
+    callbacks may call it to delegate events they do not care about.
+    """
+    if ev.kind != "change":
+        ev.logger.log(ev.level, ev.header)
+        return
+
+    diff_logger = logging.getLogger(f"{ev.logger.name}.diff")
+    lines = ev.diff_lines if ev.show_diff and diff_logger.isEnabledFor(ev.level) else []
+
+    if not lines:
+        ev.logger.log(ev.level, ev.header)
+        return
+
+    if len(lines) == 1:
+        diff_logger.log(ev.level, f"{ev.header}{lines[0]}")
+        return
+
+    ev.logger.log(ev.level, ev.header)
+    for line in lines:
+        diff_logger.log(ev.level, line)
+
+
 def _join_path(path: str, key: str) -> str:
     """Append *key* to a dot-notation *path*."""
     if not path:
@@ -42,6 +190,11 @@ def _join_path(path: str, key: str) -> str:
     return f"{path}.{key}"
 
 
+def _dim_ellipsis() -> str:
+    """Return the truncation ellipsis in the palette's ellipsis color."""
+    return str(Line().ellipsis("…"))
+
+
 def _format_value(
     value: Any,
     path: str,
@@ -64,7 +217,7 @@ def _format_value(
     if isinstance(value, str):
         value = _UNSAFE_CHARS.sub("", value)
         if len(value) > max_len:
-            return value[: max_len - 3] + "..."
+            return value[: max_len - 1] + _dim_ellipsis()
         return value
     if isinstance(value, dict):
         if not value:
@@ -90,7 +243,7 @@ def _format_value(
         return "[" + ", ".join(parts) + "]"
     text = str(value)
     if len(text) > max_len:
-        text = text[: max_len - 3] + "..."
+        text = text[: max_len - 1] + _dim_ellipsis()
     return text
 
 
@@ -113,17 +266,22 @@ def _format_path_components(
 
 
 def _format_path(
-    path: list[str], logfmt: Callable[[Any, str], str | None] | None
+    path: list[str],
+    logfmt: Callable[[Any, str], str | None] | None,
+    final_color: str = "path_final",
 ) -> str:
-    """Format a path as dot notation with prefix in dark grey, final in default."""
+    """Format a path as dot notation with prefix in dark grey, final colored.
+
+    *final_color* names a color in the :data:`kanta.tty.colors` palette.
+    """
     components = _format_path_components(path, logfmt)
     if not components:
         return ""
-    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}"
+    line = Line()
+    if len(components) > 1:
+        line.path_prefix(".".join(components[:-1]) + ".")
+    getattr(line, final_color)(components[-1])
+    return str(line)
 
 
 def _get_nested(data: dict | None, path: list[str]) -> Any:
@@ -199,37 +357,47 @@ def _format_change_lines(
     logfmt: Callable[[Any, str], str | None] | None = None,
 ) -> list[str]:
     """Format a single change as one or more lines."""
-    path_str = _format_path(path, logfmt=logfmt)
-
     if change_type == "delete":
         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}"]
+        line = Line()("  ")
+        if len(components) > 1:
+            line.path_prefix(".".join(components[:-1]) + ".")
+        line.delete(components[-1], " ✗")
+        return [str(line)]
 
     if change_type == "add":
+        path_str = _format_path(path, logfmt, final_color="add")
         if isinstance(value, dict) and value:
-            lines = [f"  {path_str} {_SEP}={_RESET}"]
-            formatted_items = []
+            lines = [str(Line()("  ", path_str, " ").sep("="))]
             base_path = ".".join(path)
-            for k, v in value.items():
+            keys = []
+            for k in value:
                 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)
+                keys.append((k, _format_value(k, key_path, max_len=30, logfmt=logfmt)))
+            field_width = max(displaywidth(kd) for _, kd in keys)
+            field_width = max(field_width, 12)
+            # Each item line is "    {key:{field_width}}: {value}"; budget the
+            # value so the whole line fits in 80 columns.
+            value_width = max(80 - 4 - field_width - 2, 20)
+            formatted_items = []
+            for (k, key_display), v in zip(keys, value.values()):
+                key_path = _join_path(base_path, str(k))
+                v_str = _format_value(v, key_path, max_len=value_width, 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
+            return lines + [
+                str(
+                    Line()("    ", k).sep(":")(
+                        " " * (field_width - displaywidth(k)), " ", v
+                    )
+                )
+                for k, v in formatted_items
+            ]
         value_str = _format_value(value, ".".join(path), logfmt=logfmt)
-        return [f"  {path_str} {_SEP}={_RESET} {value_str}"]
+        return [str(Line()("  ", path_str, " ").sep("=")(" ", value_str))]
 
     value_str = _format_value(value, ".".join(path), logfmt=logfmt)
-    return [f"  {path_str} {_SEP}={_RESET} {value_str}"]
+    path_str = _format_path(path, logfmt=logfmt)
+    return [str(Line()("  ", path_str, " ").sep("=")(" ", value_str))]
 
 
 def format_diff(
@@ -259,13 +427,18 @@ def format_diff(
     return lines
 
 
-def format_action_header(action: str, user: str | None = None) -> str:
-    """Format the action header line."""
-    action_str = f"{_ACTION}{action}{_RESET}"
-    if user:
-        user_str = f"{_USER}{user}{_RESET}"
-        return f"{action_str} by {user_str}"
-    return action_str
+def format_action_header(
+    action: str,
+    user: str | None = None,
+    extra: Any = None,
+) -> str:
+    """Format the default action header line."""
+    line = Line().action(action)
+    if extra is not None and (extra := f"{extra}"):
+        line(" ").target(extra)
+    if user is not None and (user := f"{user}"):
+        line(" by ").user(user)
+    return str(line)
 
 
 def log_change(
@@ -273,37 +446,100 @@ def log_change(
     diff: dict,
     user: str | None = None,
     previous: dict | None = None,
+    extra: Any = None,
     logfmt: Callable[[Any, str], str | None] | None = None,
+    *,
+    logger: logging.Logger = transaction_logger,
+    level: int = logging.INFO,
+    log_diff: bool = True,
 ) -> None:
-    """Log a database change with pretty-printed diff.
+    """Log a database change with the built-in formatting.
+
+    Compatibility wrapper around :func:`emit_event` with no handlers; Kanta
+    itself builds a :class:`LogEvent` and dispatches it with the registered
+    logemit callbacks.
 
     Args:
         action: The action name (e.g., "login", "admin:delete_user").
         diff: The JSON diff dict.
         user: Optional already-formatted user name to show in the header.
         previous: The previous state dict (for determining add vs update).
+        extra: Optional display-only value shown after the action in the
+            header.  Anything other than ``None`` is printed str-converted
+            (colored by Kanta), unless a custom logemit handler does
+            something else with it.
         logfmt: Optional formatter callable ``(value, path) -> str | None``.
+        logger: Logger to write to. Defaults to the ``kanta.transaction`` logger.
+        level: Log level to use. Defaults to ``logging.INFO``.
+        log_diff: Whether to build and emit the diff lines.  ``False`` skips
+            diff formatting entirely and only the header is logged.
     """
-    header = format_action_header(action, user)
-    diff_lines = format_diff(diff, previous, logfmt)
+    emit_event(
+        LogEvent(
+            kind="change",
+            logger=logger,
+            level=level,
+            action=action,
+            user=user,
+            extra=extra,
+            diff=diff,
+            previous=previous,
+            logfmt=logfmt,
+            show_diff=log_diff,
+        )
+    )
 
-    if not diff_lines:
-        logger.info(header)
+
+def configure_logging(
+    *,
+    skiproot: bool = True,
+    bootstrap: bool = True,
+    migration: bool = True,
+    transaction: bool = True,
+    diff: bool = True,
+    debug: bool = False,
+) -> None:
+    """Configure Kanta's default logging output.
+
+    Args:
+        skiproot: If ``True`` (default), attach a no-prefix stderr handler to
+            the ``kanta`` logger and set ``kanta.propagate = False`` so Kanta
+            output is rendered directly without propagating to the root logger.
+            If ``False``, the child logger enable flags are still applied, but
+            no handler is added and ``kanta`` propagation is left untouched so
+            the application's root logger handles Kanta output.
+        bootstrap: Whether bootstrap logs are enabled.
+        migration: Whether migration logs are enabled.
+        transaction: Whether transaction logs are enabled.
+        diff: Whether transaction diff lines are enabled.  When ``False``,
+            only transaction headers are printed and diff formatting is
+            skipped.  Per transaction this is controlled by the ``logdiff``
+            argument of :meth:`Kanta.transaction`.
+        debug: Whether to set the ``kanta`` logger level to ``DEBUG`` instead
+            of ``INFO``.  This reveals debug-level output such as migration
+            diffs, which are hidden by default.
+
+    This helper is not called automatically; applications that want Kanta's
+    default output can call it, but most applications will configure logging
+    themselves.
+    """
+    logging.getLogger("kanta.transaction.diff").disabled = not diff
+
+    for name, enabled in (
+        ("kanta.bootstrap", bootstrap),
+        ("kanta.migration", migration),
+        ("kanta.transaction", transaction),
+    ):
+        logging.getLogger(name).propagate = enabled
+
+    if not skiproot:
         return
 
-    if len(diff_lines) == 1:
-        logger.info(f"{header}{diff_lines[0]}")
-    else:
-        logger.info(header)
-        for line in diff_lines:
-            logger.info(line)
+    target = logging.getLogger("kanta")
+    target.propagate = False
 
-
-def configure_logging() -> None:
-    """Configure the database logger to output to stderr without prefix."""
-    if not logger.handlers:
+    if not target.handlers:
         handler = logging.StreamHandler(sys.stderr)
         handler.setFormatter(logging.Formatter("%(message)s"))
-        logger.addHandler(handler)
-    logger.setLevel(logging.INFO)
-    logger.propagate = False
+        target.addHandler(handler)
+    target.setLevel(logging.DEBUG if debug else logging.INFO)

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,184 @@
+"""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
+from dataclasses import dataclass
+from types import ModuleType
+from typing import Any
+
+from kanta.diff import compute_diff
+from kanta.exceptions import DatabaseError
+
+# 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] = {}
+
+
+@dataclass
+class MigrationInfo:
+    """Information about a single migration that ran."""
+
+    name: str
+    description: str
+    version: int
+    changed: bool
+    diff: dict | None = None
+    before: dict | None = None
+
+
+@dataclass
+class MigrationResult:
+    """Result of applying migrations."""
+
+    version: int
+    migrations: list[MigrationInfo]
+
+
+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)
+
+        result = migrations.apply(state, current_version=0, kanta=kanta)
+        new_version = result.version
+
+    Or load from a module::
+
+        migrations = Migrations.from_module("myapp.migrations")
+        result = 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,
+    ) -> MigrationResult:
+        """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 a :class:`MigrationResult` describing the new version and every
+        migration that ran.
+        """
+        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}"
+            )
+
+        migrations: list[MigrationInfo] = []
+        for version in sorted(self._migrations.keys()):
+            if version <= current_version:
+                continue
+            fn = self._migrations[version]
+            before = copy.deepcopy(data_dict)
+            self._call_migration(fn, data_dict, kanta)
+            current_version = version
+            changed = before != data_dict
+            diff = compute_diff(before, data_dict) if changed else None
+            desc = (fn.__doc__ or f"v{version}").split("\n")[0].rstrip(".")
+            migrations.append(
+                MigrationInfo(
+                    name=fn.__name__,
+                    description=desc,
+                    version=version,
+                    changed=changed,
+                    diff=diff,
+                    before=before,
+                )
+            )
+        return MigrationResult(version=current_version, migrations=migrations)

kanta/persistence.py

@@ -4,8 +4,10 @@ from __future__ import annotations
 
 import asyncio
 import copy
+import inspect
 import logging
 from collections import deque
+from collections.abc import Callable
 from datetime import UTC, datetime
 from pathlib import Path
 from typing import Any
@@ -39,7 +41,9 @@ class PersistenceMixin:
     flush_interval: float
     version: int
     opened: bool
+    readonly: bool
     mtime: datetime | None
+    clock: Callable[[], datetime] | None
 
     def __init__(self, **kwargs: Any) -> None:
         """Initialize persistence-owned state used by mixin methods."""
@@ -61,6 +65,31 @@ class PersistenceMixin:
         self.flush_interval = flush_interval
         self.version = 0
         self.mtime: datetime | None = None
+        self.clock: Callable[[], datetime] | None = None
+
+    def add_clock(self, callback) -> None:
+        """Register a clock callback ``() -> datetime`` replacing the UTC clock."""
+        if not callable(callback):
+            raise TypeError("clock callback must be callable")
+        for param in inspect.signature(callback).parameters.values():
+            if param.default is inspect.Parameter.empty and param.kind in (
+                param.POSITIONAL_ONLY,
+                param.POSITIONAL_OR_KEYWORD,
+                param.KEYWORD_ONLY,
+            ):
+                raise TypeError("clock callback must not require arguments")
+        self.clock = callback
+
+    def now(self) -> datetime:
+        """Current time from the registered clock (default: UTC now)."""
+        if self.clock is None:
+            return datetime.now(UTC)
+        ts = self.clock()
+        if not isinstance(ts, datetime):
+            raise TypeError(
+                f"clock callback must return a datetime, got {type(ts).__name__}"
+            )
+        return ts
 
     def add_fatal_error(self, callback) -> None:
         """Register one fatal error callback in call order."""
@@ -68,6 +97,8 @@ class PersistenceMixin:
 
     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)
@@ -97,7 +128,9 @@ class PersistenceMixin:
 
     def maybe_snapshot(self) -> None:
         """Evaluate and possibly write a snapshot from current state."""
-        self.snapshot.maybe_write(self.file, self.version, self.statedict, m=self.mtime)
+        self.snapshot.maybe_write(
+            self.file, self.version, self.statedict, m=self.mtime, now=self.now
+        )
 
     def queue_change(
         self,
@@ -106,6 +139,7 @@ class PersistenceMixin:
         *,
         user: str | None = None,
         mtime: bool | datetime = True,
+        force: bool = False,
     ) -> ChangeRecord | None:
         """Queue a change record internally (thread-safe).
 
@@ -118,11 +152,20 @@ class PersistenceMixin:
                 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.
+            The queued :class:`ChangeRecord`, or ``None`` if the diff was empty
+            and *force* is ``False``.
         """
-        now = datetime.now(UTC)
+        diff = compute_diff(self.statedict, current)
+        if not diff:
+            if not force:
+                return None
+            diff = {}
+
+        # The clock is only read when a record is actually queued.
+        now = self.now()
 
         if mtime is True:
             m = now
@@ -133,10 +176,6 @@ class PersistenceMixin:
         else:
             raise TypeError("mtime must be True, False, or a datetime")
 
-        diff = compute_diff(self.statedict, current)
-        if not diff:
-            return None
-
         record = ChangeRecord(
             ts=now,
             a=action,
@@ -160,6 +199,13 @@ 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
 
@@ -207,6 +253,13 @@ 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
 

kanta/snapshot.py

@@ -3,6 +3,7 @@
 from __future__ import annotations
 
 import logging
+from collections.abc import Callable
 from datetime import UTC, datetime
 
 from kanta.structs import Snapshot
@@ -38,22 +39,29 @@ class SnapshotState:
         self.changes += count
 
     def maybe_write(
-        self, file, version: int, state: dict, m: datetime | None = None
+        self,
+        file,
+        version: int,
+        state: dict,
+        m: datetime | None = None,
+        now: Callable[[], datetime] | None = None,
     ) -> None:
         """Write snapshot when thresholds/time policy allows it."""
-        if self.changes < self._min_diffs:
-            return
         force = self._force_pending
-        now = datetime.now(UTC)
-        if not force and now.weekday() != 6:  # 6 = Sunday
-            return
-        sunday_midnight = now.replace(hour=0, minute=0, second=0, microsecond=0)
-        if not force and self.ts is not None and self.ts >= sunday_midnight:
+        if not force and self.changes < self._min_diffs:
             return
+        # The clock is only read when a snapshot may actually be written.
+        ts = now() if now is not None else datetime.now(UTC)
+        if not force:
+            if ts.weekday() != 6:  # 6 = Sunday
+                return
+            sunday_midnight = ts.replace(hour=0, minute=0, second=0, microsecond=0)
+            if self.ts is not None and self.ts >= sunday_midnight:
+                return
         if not file.is_open:
             return
         try:
-            self._write(file, version, state, now, m=m)
+            self._write(file, version, state, ts, m=m)
             self._force_pending = False
         except Exception as exc:
             _logger.error("snapshot: failed to write snapshot: %r", exc)

kanta/structs.py

@@ -23,7 +23,7 @@ class ChangeRecord(msgspec.Struct, omit_defaults=True, kw_only=True):
     v: int = 0
     u: str | None = None
     m: datetime | None = None
-    diff: dict
+    diff: dict = {}
 
 
 class Snapshot(msgspec.Struct, omit_defaults=True):

kanta/transaction.py

@@ -5,25 +5,55 @@ from __future__ import annotations
 import logging
 from contextlib import contextmanager
 from datetime import datetime
+from typing import Any
 
 from kanta.diff import compute_diff
 from kanta.exceptions import DataIntegrityError
 from kanta.callbacks import InjectionContext
-from kanta.logging import _USER_PATH, log_change
+from kanta.logging import _USER_PATH, LogEvent, emit_event, transaction_logger
 from kanta.serialization import restore_data_in_place, struct_to_dict
 
 _logger = logging.getLogger(__name__)
 
 
+def _build_logfmt(impl, previous: dict, current: dict):
+    """Build the logfmt chain for a state transition."""
+    return impl.callback_registry.build_logfmt(
+        InjectionContext(
+            previous_state=previous,
+            current_state=current,
+            kanta=impl._kanta,
+        )
+    )
+
+
+def _resolve_user(logfmt, user: str | None) -> str | None:
+    """Resolve *user* for display via the logfmt chain (raw as fallback)."""
+    if user is None:
+        return None
+    resolved = logfmt(user, _USER_PATH)
+    return resolved if resolved is not None else user
+
+
 @contextmanager
 def transaction(
     impl,
     action: str,
     *,
     user: str | None = None,
+    extra: Any = None,
     mtime: bool | datetime = True,
+    log: bool | logging.Logger = True,
+    logdiff: bool = 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 "
@@ -61,21 +91,45 @@ def transaction(
             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,
+                if log is not False:
+                    logfmt = _build_logfmt(impl, previous, new_dict)
+                    logger = (
+                        log if isinstance(log, logging.Logger) else transaction_logger
                     )
-                )
-                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)
+                    emit_event(
+                        LogEvent(
+                            kind="change",
+                            logger=logger,
+                            kanta=impl._kanta,
+                            action=action,
+                            user=_resolve_user(logfmt, user),
+                            extra=extra,
+                            diff=record.diff,
+                            previous=previous,
+                            current=new_dict,
+                            logfmt=logfmt,
+                            show_diff=logdiff,
+                        ),
+                        impl.callback_registry.logemit_handlers,
+                    )
+    except Exception as exc:
+        resolved_user = None
+        if user is not None:
+            logfmt = _build_logfmt(impl, impl.statedict, impl.statedict)
+            resolved_user = _resolve_user(logfmt, user)
+        emit_event(
+            LogEvent(
+                kind="aborted",
+                logger=transaction_logger,
+                level=logging.WARNING,
+                kanta=impl._kanta,
+                action=action,
+                user=resolved_user,
+                extra=extra,
+                error=exc,
+            ),
+            impl.callback_registry.logemit_handlers,
+        )
         if impl.transaction_snapshot is not None:
             impl.data = restore_data_in_place(
                 impl.data,

kanta/tty.py

@@ -0,0 +1,182 @@
+"""Terminal string building: ANSI colors, display widths, and a line builder.
+
+Colors are stored as bare SGR parameter strings (e.g. ``"1;34"``) without
+the ``\\x1b[`` prefix and ``m`` suffix.  The :class:`Line` builder understands
+how SGR parameters stack: ``0`` clears everything, other parameters apply
+sequentially and the last one of each class wins.  This lets it emit minimal
+escape sequences, folding a needed reset into the same sequence as the next
+color instead of emitting a separate one.
+"""
+
+from __future__ import annotations
+
+import re
+import unicodedata
+from typing import Any
+
+ESC = "\x1b["
+
+# Matches a full ANSI escape sequence (color codes, cursor movement, ...).
+ANSI_RE = re.compile(r"\x1b\[[0-9;:]*[A-Za-z]")
+
+
+def strip_ansi(text: str) -> str:
+    """Remove ANSI escape sequences from *text*."""
+    return ANSI_RE.sub("", text)
+
+
+def displaywidth(text: str) -> int:
+    """Return the terminal column width of *text*, ignoring ANSI sequences.
+
+    Wide characters (CJK, most emoji) count as two columns; combining and
+    zero-width characters count as zero.
+    """
+    return sum(
+        2
+        if unicodedata.east_asian_width(c) in "WF"
+        else 0
+        if unicodedata.category(c) in ("Mn", "Me", "Cf")
+        else 1
+        for c in strip_ansi(text)
+    )
+
+
+def pad(text: str, width: int, align: str = "left") -> str:
+    """Pad *text* to *width* columns by display width.
+
+    *align* is ``"left"`` (padding after), ``"right"`` (padding before), or
+    ``"center"``.  Text already at or above *width* is returned unchanged.
+    """
+    missing = width - displaywidth(text)
+    if missing <= 0:
+        return text
+    if align == "right":
+        return " " * missing + text
+    if align == "center":
+        left = missing // 2
+        return " " * left + text + " " * (missing - left)
+    return text + " " * missing
+
+
+class Colors:
+    """Kanta's log color palette: bare SGR parameter strings.
+
+    Attributes are looked up when a line is rendered, so assignments such as
+    ``colors.action = "36"`` or additions like ``colors.session = "38;5;226"``
+    take effect immediately, no matter how the object was imported.  Added
+    colors become available on :class:`Line` under the same name.
+    """
+
+    action = "1;34"  # Bold blue for the action name
+    user = "34"  # Blue for the user display
+    target = "38;5;250"  # White for the extra/target display
+    sep = "38;5;242"  # Dark grey for separators
+    path_prefix = "38;5;242"  # Dark grey for the leading part of a dotted path
+    path_final = "38;5;250"  # White for the final path element
+    add = "32"  # Green for additions
+    delete = "1;31"  # Bold red for deletions
+    ellipsis = "38;5;242"  # Dark grey for the truncation ellipsis
+
+
+colors = Colors()
+
+# SGR attribute classes that carry no class siblings (each clears/sets itself).
+_ATTR_CLASSES = frozenset({"1", "2", "3", "4", "7", "9"})
+
+
+def _parse_sgr(spec: str) -> dict[str, str]:
+    """Parse a bare SGR parameter string into a ``{class: group}`` state.
+
+    Applies the stacking rules: ``0`` clears everything, other parameters
+    apply sequentially and the last one of each class wins.
+    """
+    state: dict[str, str] = {}
+    tokens = spec.split(";")
+    i = 0
+    while i < len(tokens):
+        token = tokens[i]
+        if token == "0":
+            state.clear()
+        elif token in ("38", "48"):
+            cls = "fg" if token == "38" else "bg"
+            if i + 1 < len(tokens) and tokens[i + 1] == "5":
+                state[cls] = ";".join(tokens[i : i + 3])
+                i += 3
+                continue
+            if i + 1 < len(tokens) and tokens[i + 1] == "2":
+                state[cls] = ";".join(tokens[i : i + 4])
+                i += 4
+                continue
+            state[cls] = token
+        elif token.isdigit() and (30 <= int(token) <= 37 or 90 <= int(token) <= 97):
+            state["fg"] = token
+        elif token.isdigit() and (40 <= int(token) <= 47 or 100 <= int(token) <= 107):
+            state["bg"] = token
+        elif token in _ATTR_CLASSES:
+            state[token] = token
+        else:
+            state[f"other:{token}"] = token
+        i += 1
+    return state
+
+
+def _sgr_transition(current: dict[str, str], new: dict[str, str]) -> str:
+    """Return the minimal escape sequence moving from *current* to *new*."""
+    if current == new:
+        return ""
+    if not new:
+        return f"{ESC}0m" if current else ""
+    if not current:
+        return f"{ESC}{';'.join(new.values())}m"
+    if current.keys() - new.keys():
+        # Some attribute must be cleared; fold the reset into one sequence.
+        return f"{ESC}0;{';'.join(new.values())}m"
+    changed = [group for cls, group in new.items() if current.get(cls) != group]
+    return f"{ESC}{';'.join(changed)}m" if changed else ""
+
+
+class Line:
+    """Build a terminal string part by part with colors, width and alignment.
+
+    Calling the builder appends content (arguments are converted to ``str``).
+    Attribute access with a color name arms that palette color for the next
+    call; the color is reset automatically when that call ends, so a color
+    always applies to exactly one call::
+
+        str(Line().user("Alice")(" by "))  # "Alice" blue, " by " plain
+
+    ``width`` and ``align`` keyword arguments pad the content of a call by
+    display width.  ``str(line)`` finishes the line, restoring default
+    colors if any are active.
+    """
+
+    def __init__(self, palette: Colors | None = None) -> None:
+        self._palette = palette if palette is not None else colors
+        self._parts: list[str] = []
+        self._active: dict[str, str] = {}
+        self._pending: dict[str, str] = {}
+
+    def __getattr__(self, name: str) -> Line:
+        if name.startswith("_"):
+            raise AttributeError(name)
+        spec = getattr(self._palette, name, None)
+        if spec is None:
+            raise AttributeError(f"unknown color: {name!r}")
+        self._pending = _parse_sgr(spec)
+        return self
+
+    def __call__(self, *args: Any, width: int = 0, align: str = "left") -> Line:
+        text = "".join(str(arg) for arg in args)
+        if width:
+            text = pad(text, width, align)
+        if self._pending != self._active:
+            self._parts.append(_sgr_transition(self._active, self._pending))
+            self._active = self._pending
+        self._parts.append(text)
+        self._pending = {}
+        return self
+
+    def __str__(self) -> str:
+        if self._active:
+            return "".join(self._parts) + f"{ESC}0m"
+        return "".join(self._parts)

main.py

@@ -1,6 +0,0 @@
-def main():
-    print("Hello from kanta!")
-
-
-if __name__ == "__main__":
-    main()

pyproject.toml

@@ -21,6 +21,9 @@ dependencies = [
     "msgspec>=0.20.0",
 ]
 
+[project.scripts]
+kanta = "kanta.__main__:main"
+
 [project.optional-dependencies]
 bin = [
     "blake3>=1.0.8",

tests/support.py

@@ -7,7 +7,7 @@ from uuid import UUID
 import msgspec
 
 from kanta.kanta import Kanta
-from kanta.structs import ChangeRecord
+from kanta.structs import ChangeRecord, Snapshot
 
 
 class User(msgspec.Struct):
@@ -70,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
@@ -77,6 +89,17 @@ def make_migrations_module(name: str, fn_name: str, fn):
     return mod
 
 
+def read_last_snapshot(path: Path, format_config) -> Snapshot | None:
+    _, serializer_cls = format_config
+    serializer = serializer_cls()
+    framer = serializer.framer_cls()
+    data = path.read_bytes()
+    payload, _, _ = framer.scan_last_snapshot(data)
+    if payload is None:
+        return None
+    return serializer.decode(payload, type=Snapshot)
+
+
 def fixed_change(action: str, diff: dict, *, version: int = 0) -> ChangeRecord:
     return ChangeRecord(
         ts=datetime(2026, 1, 1, tzinfo=UTC), a=action, v=version, diff=diff

tests/test_callbacks.py

@@ -1,4 +1,4 @@
-from typing import Any
+from typing import Any, Optional, Union
 
 import pytest
 
@@ -49,16 +49,60 @@ def test_logfmt_requires_value_annotation(tmp_path, format_config):
             return None
 
 
-def test_logfmt_requires_return_annotation(tmp_path, format_config):
+def test_logfmt_allows_missing_return_annotation(tmp_path, format_config):
     kanta = make_kanta(tmp_path / "test.db", Data, format_config)
 
-    with pytest.raises(TypeError, match="must annotate its return"):
+    @kanta.logfmt
+    def resolve_names(value: str, current: DictPost):
+        return None
 
-        @kanta.logfmt
-        def resolve_names(value: str, current: DictPost):
+
+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)
 
@@ -104,7 +148,7 @@ async def test_bootstrap_injects_kanta(tmp_path, format_config):
 async def test_logfmt_injects_states(tmp_path, format_config, caplog):
     import logging
 
-    caplog.set_level(logging.INFO, logger="kanta.changes")
+    caplog.set_level(logging.INFO, logger="kanta.transaction")
     path = tmp_path / "test.db"
     kanta = make_kanta(path, Data, format_config)
 
@@ -126,13 +170,15 @@ async def test_logfmt_injects_states(tmp_path, format_config, caplog):
 async def test_logfmt_class_injection(tmp_path, format_config, caplog):
     import logging
 
-    caplog.set_level(logging.INFO, logger="kanta.changes")
+    caplog.set_level(logging.INFO, logger="kanta.transaction")
     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:
+            if not isinstance(value, str):
+                return None
             return self.current_state.get("users", {}).get(value, {}).get("name")
 
     await kanta.open()
@@ -149,7 +195,7 @@ async def test_logfmt_class_injection(tmp_path, format_config, caplog):
 async def test_multiple_logfmt_chain(tmp_path, format_config, caplog):
     import logging
 
-    caplog.set_level(logging.INFO, logger="kanta.changes")
+    caplog.set_level(logging.INFO, logger="kanta.transaction")
     path = tmp_path / "test.db"
     kanta = make_kanta(path, Data, format_config)
 
@@ -177,7 +223,7 @@ async def test_multiple_logfmt_chain(tmp_path, format_config, caplog):
 async def test_logfmt_path_context(tmp_path, format_config, caplog):
     import logging
 
-    caplog.set_level(logging.INFO, logger="kanta.changes")
+    caplog.set_level(logging.INFO, logger="kanta.transaction")
     path = tmp_path / "test.db"
     kanta = make_kanta(path, Data, format_config)
 
@@ -201,7 +247,7 @@ async def test_logfmt_path_context(tmp_path, format_config, caplog):
 async def test_logfmt_decorator_path_filters_calls(tmp_path, format_config, caplog):
     import logging
 
-    caplog.set_level(logging.INFO, logger="kanta.changes")
+    caplog.set_level(logging.INFO, logger="kanta.transaction")
     path = tmp_path / "test.db"
     kanta = make_kanta(path, Data, format_config)
 
@@ -227,7 +273,7 @@ async def test_logfmt_decorator_path_filters_calls(tmp_path, format_config, capl
 async def test_logfmt_user_path_replaces_user_display(tmp_path, format_config, caplog):
     import logging
 
-    caplog.set_level(logging.INFO, logger="kanta.changes")
+    caplog.set_level(logging.INFO, logger="kanta.transaction")
     path = tmp_path / "test.db"
     kanta = make_kanta(path, Data, format_config)
 
@@ -249,7 +295,7 @@ async def test_logfmt_user_path_replaces_user_display(tmp_path, format_config, c
 async def test_logfmt_non_string_value(tmp_path, format_config, caplog):
     import logging
 
-    caplog.set_level(logging.INFO, logger="kanta.changes")
+    caplog.set_level(logging.INFO, logger="kanta.transaction")
     path = tmp_path / "test.db"
     kanta = make_kanta(path, Data, format_config)
 

tests/test_clock.py

@@ -0,0 +1,135 @@
+from datetime import UTC, datetime, timedelta
+
+import pytest
+
+from .support import (
+    Data,
+    make_kanta,
+    make_migrations_module,
+    read_changes,
+    read_last_snapshot,
+)
+
+T0 = datetime(2026, 1, 1, 12, 0, tzinfo=UTC)
+
+
+def test_clock_rejects_non_callable(tmp_path, format_config):
+    kanta = make_kanta(tmp_path / "test.db", Data, format_config)
+
+    with pytest.raises(TypeError, match="must be callable"):
+        kanta.clock(42)
+
+
+def test_clock_rejects_required_argument(tmp_path, format_config):
+    kanta = make_kanta(tmp_path / "test.db", Data, format_config)
+
+    with pytest.raises(TypeError, match="must not require arguments"):
+
+        @kanta.clock
+        def fake_now(tz) -> datetime:
+            return T0
+
+
+@pytest.mark.asyncio
+async def test_clock_rejects_non_datetime_result(tmp_path, format_config):
+    kanta = make_kanta(tmp_path / "test.db", Data, format_config)
+
+    @kanta.clock
+    def fake_now() -> datetime:
+        return "noon"
+
+    with pytest.raises(TypeError, match="must return a datetime"):
+        await kanta.open(log=False)
+
+
+@pytest.mark.asyncio
+async def test_clock_controls_record_timestamps(tmp_path, format_config):
+    path = tmp_path / "test.db"
+    kanta = make_kanta(path, Data, format_config)
+    current = T0
+
+    @kanta.clock
+    def fake_now() -> datetime:
+        return current
+
+    await kanta.open(log=False)
+    current = T0 + timedelta(hours=1)
+    with kanta.transaction(action="update") as data:
+        data.counter = 1
+    current = T0 + timedelta(hours=2)
+    with kanta.transaction(action="repair", mtime=False) as data:
+        data.counter = 2
+    await kanta.close()
+
+    bootstrap, update, repair = read_changes(path, format_config)
+    assert bootstrap.ts == T0
+    assert bootstrap.m == T0
+    assert update.ts == T0 + timedelta(hours=1)
+    assert update.m == T0 + timedelta(hours=1)
+    # System operation: stamped by the clock, but m is not updated.
+    assert repair.ts == T0 + timedelta(hours=2)
+    assert repair.m is None
+    assert kanta.mtime == T0 + timedelta(hours=1)
+
+
+@pytest.mark.asyncio
+async def test_clock_not_read_without_changes(tmp_path, format_config):
+    kanta = make_kanta(tmp_path / "test.db", Data, format_config)
+    reads = 0
+
+    @kanta.clock
+    def fake_now() -> datetime:
+        nonlocal reads
+        reads += 1
+        return T0
+
+    await kanta.open(log=False)  # bootstrap record: one read
+    reads = 0
+
+    with kanta.transaction(action="noop"):
+        pass  # no changes, no record, no clock read
+    await kanta.close()  # no snapshot written, no clock read
+
+    assert reads == 0
+
+
+@pytest.mark.asyncio
+async def test_clock_controls_migration_and_snapshot_timestamps(
+    tmp_path, format_config
+):
+    path = tmp_path / "test.db"
+    kanta = make_kanta(path, Data, format_config)
+
+    @kanta.clock
+    def fake_now() -> datetime:
+        return T0
+
+    await kanta.open(log=False)
+    await kanta.close()
+
+    def migrate_v1(d):
+        """Bump counter"""
+        d["counter"] = 1
+
+    migrations = make_migrations_module("clock_migrations", "migrate_v1", migrate_v1)
+    t1 = T0 + timedelta(days=1)
+    kanta2 = make_kanta(path, Data, format_config, migrations=migrations)
+
+    @kanta2.clock
+    def fake_now2() -> datetime:
+        return t1
+
+    await kanta2.open(log=False)
+    await kanta2.close()
+
+    migrate_records = [
+        r for r in read_changes(path, format_config) if r.a.startswith("migrate:")
+    ]
+    assert migrate_records
+    assert all(r.ts == t1 for r in migrate_records)
+
+    snapshot = read_last_snapshot(path, format_config)
+    assert snapshot is not None
+    assert snapshot.ts == t1
+    # mtime is carried forward from the last real modification.
+    assert snapshot.m == T0

tests/test_format_diff.py

@@ -1,4 +1,8 @@
 from kanta.logging import format_diff
+from kanta.tty import ESC, colors
+
+_ADD = f"{ESC}{colors.add}m"
+_DELETE = f"{ESC}{colors.delete}m"
 
 
 def test_add():
@@ -6,6 +10,28 @@ def test_add():
     assert any("name" in line for line in lines)
 
 
+def test_add_path_is_green():
+    lines = format_diff({"name": "Alice"}, previous={})
+    assert any(_ADD in line for line in lines)
+
+
+def test_nested_add_path_final_element_is_green():
+    lines = format_diff({"users": {"alice": 1}}, previous={"users": {}})
+    assert any(_ADD in line and "alice" in line for line in lines)
+
+
+def test_update_path_not_colored_as_add():
+    lines = format_diff({"name": "Bob"}, previous={"name": "Alice"})
+    assert lines
+    assert all(_ADD not in line for line in lines)
+
+
+def test_delete_path_not_colored_as_add():
+    lines = format_diff({"$delete": ["old_key"]}, previous={"old_key": 1})
+    assert any(_DELETE in line for line in lines)
+    assert all(_ADD not in line for line in lines)
+
+
 def test_update():
     lines = format_diff({"name": "Bob"}, previous={"name": "Alice"})
     assert any("Bob" in line for line in lines)

tests/test_kanta_integration.py

@@ -1,4 +1,5 @@
 import asyncio
+import logging
 import sys
 from datetime import UTC, datetime
 from uuid import uuid4
@@ -6,6 +7,7 @@ from uuid import uuid4
 import pytest
 
 from kanta.exceptions import DatabaseError, DataIntegrityError, FileLockError
+from kanta.migrations import MigrationResult
 from kanta.serialization import struct_to_dict
 
 from .support import (
@@ -17,6 +19,9 @@ from .support import (
     change_actions,
     fixed_change,
     make_kanta,
+    make_migrations_module,
+    read_changes,
+    read_last_snapshot,
     seed_single_change,
 )
 
@@ -30,6 +35,63 @@ 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_reopen_without_changes_does_not_force_snapshot(tmp_path, format_config):
+    path = tmp_path / "test.db"
+    kanta = make_kanta(path, Data(counter=5), format_config)
+    await kanta.open()
+    await kanta.close()
+
+    # No snapshot should exist after the initial bootstrap and close.
+    assert read_last_snapshot(path, format_config) is None
+
+    kanta2 = make_kanta(path, Data, format_config)
+    await kanta2.open()
+    assert kanta2.data.counter == 5
+    await kanta2.close()
+
+    # Re-opening without migrations or normalization changes must not force one.
+    assert read_last_snapshot(path, format_config) is None
+
+
 @pytest.mark.asyncio
 async def test_open_overwrites_caller_owned_root_data(tmp_path, format_config):
     path = tmp_path / "test.db"
@@ -404,7 +466,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
@@ -434,6 +496,287 @@ 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_writes_snapshot_and_is_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.close()
+
+        # Empty migrations must not produce empty change records.
+        records = read_changes(path, format_config)
+        migration_records = [r for r in records if r.a.startswith("migrate")]
+        assert not migration_records
+
+        # The version bump is persisted via a snapshot instead.
+        snap = read_last_snapshot(path, format_config)
+        assert snap is not None
+        assert snap.v == 1
+        assert snap.state == {"counter": 0, "users": {}}
+
+        kanta2 = make_kanta(path, Data, format_config, migrations=mod)
+        await kanta2.open()
+        assert kanta2.version == 1
+        await kanta2.close()
+
+        # Re-opening must not create additional migration records or snapshots.
+        records2 = read_changes(path, format_config)
+        assert not [r for r in records2 if r.a.startswith("migrate")]
+    finally:
+        sys.modules.pop("empty_migration_mod", None)
+
+
+@pytest.mark.asyncio
+async def test_migration_with_changes_records_diff_and_snapshot(
+    tmp_path, format_config
+):
+    path = tmp_path / "test.db"
+    seed_single_change(path, fixed_change("init", {"counter": 0}), format_config)
+
+    mod = type(sys)("test_migrations_changes")
+
+    def migrate_v1(d, kanta):
+        d["counter"] = 2
+
+    mod.__dict__["migrate_v1"] = migrate_v1
+
+    kanta = make_kanta(path, Data, format_config, migrations=mod)
+    await kanta.open()
+    assert kanta.version == 1
+    assert kanta.data.counter == 2
+    await kanta.close()
+
+    records = read_changes(path, format_config)
+    migration_records = [r for r in records if r.a.startswith("migrate")]
+    # The version migration and the msgspec normalization that follows it are
+    # grouped into a single migrate:vN record.
+    assert len(migration_records) == 1
+    assert migration_records[0].a == "migrate:v1"
+    assert migration_records[0].v == 1
+    assert migration_records[0].diff == {"counter": 2, "users": {}}
+
+    snap = read_last_snapshot(path, format_config)
+    assert snap is not None
+    assert snap.v == 1
+    assert snap.state == {"counter": 2, "users": {}}
+
+
+@pytest.mark.asyncio
+async def test_migration_summary_log_includes_filename(tmp_path, format_config, caplog):
+    path = tmp_path / "test.db"
+    seed_single_change(path, fixed_change("init", {"counter": 0}), format_config)
+
+    mod = type(sys)("test_migrations_log")
+
+    def migrate_v1(d, kanta):
+        """Bump counter."""
+        d["counter"] = 2
+
+    mod.__dict__["migrate_v1"] = migrate_v1
+
+    with caplog.at_level(logging.INFO, logger="kanta.migration"):
+        kanta = make_kanta(path, Data, format_config, migrations=mod)
+        await kanta.open()
+        assert kanta.version == 1
+        await kanta.close()
+
+    info_messages = [r.message for r in caplog.records if r.levelno == logging.INFO]
+    assert len(info_messages) == 1
+    assert str(path) in info_messages[0]
+    assert "v0 -> v1" in info_messages[0]
+    assert "migrate_v1 (Bump counter)" in info_messages[0]
+
+
+@pytest.mark.asyncio
+async def test_open_log_false_suppresses_migration_log(tmp_path, format_config, caplog):
+    path = tmp_path / "test.db"
+    seed_single_change(path, fixed_change("init", {"counter": 0}), format_config)
+
+    mod = type(sys)("test_migrations_silent")
+
+    def migrate_v1(d, kanta):
+        d["counter"] = 2
+
+    mod.__dict__["migrate_v1"] = migrate_v1
+
+    with caplog.at_level(logging.INFO, logger="kanta.migration"):
+        kanta = make_kanta(path, Data, format_config, migrations=mod)
+        await kanta.open(log=False)
+        await kanta.close()
+
+    info_messages = [r for r in caplog.records if r.levelno == logging.INFO]
+    assert not info_messages
+
+
+@pytest.mark.asyncio
+async def test_open_log_true_logs_bootstrap(tmp_path, format_config, caplog):
+    path = tmp_path / "test.db"
+    kanta = make_kanta(path, Data, format_config)
+
+    with caplog.at_level(logging.INFO, logger="kanta.bootstrap"):
+        await kanta.open()
+        await kanta.close()
+
+    info_messages = [r.message for r in caplog.records if r.levelno == logging.INFO]
+    assert len(info_messages) >= 2
+    assert "created" in info_messages[0]
+    assert "bootstrap" in info_messages[1]
+
+
+@pytest.mark.asyncio
+async def test_open_log_false_suppresses_bootstrap_log(tmp_path, format_config, caplog):
+    path = tmp_path / "test.db"
+    kanta = make_kanta(path, Data, format_config)
+
+    with caplog.at_level(logging.INFO, logger="kanta.bootstrap"):
+        await kanta.open(log=False)
+        await kanta.close()
+
+    info_messages = [r for r in caplog.records if r.levelno == logging.INFO]
+    assert not info_messages
+
+
+@pytest.mark.asyncio
+async def test_open_log_custom_logger_logs_bootstrap(tmp_path, format_config, caplog):
+    path = tmp_path / "test.db"
+    kanta = make_kanta(path, Data, format_config)
+
+    custom_logger = logging.getLogger("custom.bootstrap")
+    custom_logger.setLevel(logging.INFO)
+
+    with caplog.at_level(logging.INFO, logger="custom.bootstrap"):
+        await kanta.open(log=custom_logger)
+        await kanta.close()
+
+    info_messages = [r.message for r in caplog.records if r.levelno == logging.INFO]
+    assert len(info_messages) >= 2
+    assert "created" in info_messages[0]
+    assert "bootstrap" in info_messages[1]
+
+
+@pytest.mark.asyncio
+async def test_open_existing_database_logs_using_on_debug(
+    tmp_path, format_config, caplog
+):
+    path = tmp_path / "test.db"
+    kanta = make_kanta(path, Data, format_config)
+    await kanta.open()
+    await kanta.close()
+
+    kanta2 = make_kanta(path, Data, format_config)
+
+    with caplog.at_level(logging.DEBUG, logger="kanta.bootstrap"):
+        await kanta2.open()
+        await kanta2.close()
+
+    debug_messages = [r.message for r in caplog.records if r.levelno == logging.DEBUG]
+    assert any("opened" in m and str(path.resolve()) in m for m in debug_messages)
+
+
+@pytest.mark.asyncio
+async def test_logmigr_callback_replaces_default_logging(
+    tmp_path, format_config, caplog
+):
+    path = tmp_path / "test.db"
+    seed_single_change(path, fixed_change("init", {"counter": 0}), format_config)
+
+    mod = type(sys)("test_migrations_callback")
+
+    def migrate_v1(d, kanta):
+        """Bump counter."""
+        d["counter"] = 2
+
+    mod.__dict__["migrate_v1"] = migrate_v1
+
+    summaries = []
+
+    kanta = make_kanta(path, Data, format_config, migrations=mod)
+
+    @kanta.logmigr
+    def collect(summary: MigrationResult):
+        summaries.append(summary)
+
+    with caplog.at_level(logging.INFO, logger="kanta.migration"):
+        await kanta.open()
+        await kanta.close()
+
+    assert len(summaries) == 1
+    assert summaries[0].version == 1
+    assert summaries[0].migrations[0].name == "migrate_v1"
+    info_messages = [r for r in caplog.records if r.levelno == logging.INFO]
+    assert not info_messages
+
+
+@pytest.mark.asyncio
+async def test_transaction_log_false_suppresses_log(tmp_path, format_config, caplog):
+    path = tmp_path / "test.db"
+    kanta = make_kanta(path, Data, format_config)
+    await kanta.open()
+
+    with caplog.at_level(logging.INFO, logger="kanta.transaction"):
+        with kanta.transaction(action="inc", log=False) as data:
+            data.counter = 1
+
+    await kanta.close()
+
+    info_messages = [r for r in caplog.records if r.levelno == logging.INFO]
+    assert not info_messages
+
+
+@pytest.mark.asyncio
+async def test_transaction_logdiff_false_logs_header_only(
+    tmp_path, format_config, caplog
+):
+    path = tmp_path / "test.db"
+    kanta = make_kanta(path, Data, format_config)
+    await kanta.open()
+
+    with caplog.at_level(logging.INFO, logger="kanta.transaction"):
+        with kanta.transaction(action="inc", logdiff=False) as data:
+            data.counter = 1
+
+    await kanta.close()
+
+    messages = [r.message for r in caplog.records if r.levelno == logging.INFO]
+    assert len(messages) == 1
+    assert "inc" in messages[0]
+    assert "counter" not in messages[0]
+
+
+@pytest.mark.asyncio
+async def test_transaction_log_custom_logger(tmp_path, format_config, caplog):
+    path = tmp_path / "test.db"
+    kanta = make_kanta(path, Data, format_config)
+    await kanta.open()
+
+    custom_logger = logging.getLogger("custom.transaction")
+    custom_logger.setLevel(logging.INFO)
+
+    with caplog.at_level(logging.INFO, logger="custom.transaction"):
+        with kanta.transaction(action="inc", log=custom_logger) as data:
+            data.counter = 1
+
+    await kanta.close()
+
+    info_messages = [r for r in caplog.records if r.levelno == logging.INFO]
+    assert len(info_messages) >= 1
+    assert "inc" in info_messages[0].message
+
+
 @pytest.mark.asyncio
 async def test_open_locked_file_raises_filelock_error(tmp_path, format_config):
     path = tmp_path / "test.db"
@@ -514,7 +857,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_logemit.py

@@ -0,0 +1,362 @@
+import logging
+import sys
+
+import pytest
+
+from kanta.logging import (
+    LogEvent,
+    bootstrap_logger,
+    configure_logging,
+    emit_event,
+    log_change,
+    migration_logger,
+    transaction_logger,
+)
+from kanta.migrations import MigrationResult
+from tests.support import (
+    Data,
+    fixed_change,
+    make_kanta,
+    seed_single_change,
+)
+
+
+@pytest.fixture(autouse=True)
+def _reset_kanta_loggers():
+    yield
+    for name in (
+        "kanta",
+        "kanta.transaction",
+        "kanta.transaction.diff",
+        "kanta.bootstrap",
+        "kanta.migration",
+    ):
+        logger = logging.getLogger(name)
+        logger.setLevel(logging.NOTSET)
+        logger.propagate = True
+        logger.disabled = False
+        logger.handlers.clear()
+
+
+def _change_event(**kwargs) -> LogEvent:
+    return LogEvent(kind="change", logger=transaction_logger, action="update", **kwargs)
+
+
+def test_emit_event_falsy_return_stops_chain(capsys):
+    logging.getLogger("kanta").handlers.clear()
+    configure_logging()
+    calls = []
+
+    def first(ev):
+        calls.append("first")
+        return None
+
+    def second(ev):
+        calls.append("second")
+
+    emit_event(_change_event(), [first, second])
+    assert calls == ["first"]
+    assert capsys.readouterr().err == ""
+
+
+def test_emit_event_truthy_return_falls_back_to_default(capsys):
+    logging.getLogger("kanta").handlers.clear()
+    configure_logging()
+    emit_event(_change_event(), [lambda ev: True])
+    assert "update" in capsys.readouterr().err
+
+
+def test_emit_event_mutation_reaches_later_handlers_and_default(capsys):
+    logging.getLogger("kanta").handlers.clear()
+    configure_logging()
+    calls = []
+
+    def first(ev):
+        calls.append("first")
+        ev.extra = "tgt"
+        return True
+
+    def second(ev):
+        calls.append(("second", ev.extra))
+        return True
+
+    emit_event(_change_event(), [first, second])
+    assert calls == ["first", ("second", "tgt")]
+    assert "tgt" in capsys.readouterr().err
+
+
+def test_emit_event_handler_error_falls_back_to_default(capsys):
+    logging.getLogger("kanta").handlers.clear()
+    configure_logging()
+
+    def boom(ev):
+        raise RuntimeError("broken")
+
+    emit_event(_change_event(), [boom])
+    assert "update" in capsys.readouterr().err
+
+
+def test_diff_lines_built_lazily(monkeypatch):
+    def _boom(*args, **kwargs):
+        raise AssertionError("format_diff should not be called")
+
+    monkeypatch.setattr("kanta.logging.format_diff", _boom)
+    ev = _change_event(diff={"counter": 1})
+    emit_event(ev, [lambda ev: None])  # handled without touching the diff
+    monkeypatch.undo()
+    assert len(ev.diff_lines) == 1
+    assert "counter" in ev.diff_lines[0]
+
+
+def test_default_emit_created_and_migrated(capsys):
+    logging.getLogger("kanta").handlers.clear()
+    configure_logging()
+    emit_event(LogEvent(kind="created", logger=bootstrap_logger, filename="x.kantadb"))
+    emit_event(
+        LogEvent(
+            kind="migrated",
+            logger=migration_logger,
+            filename="x.kantadb",
+            from_version=0,
+            to_version=1,
+            migrations=["migrate_v1 (rename)"],
+        )
+    )
+    err = capsys.readouterr().err
+    assert "🛢️ x.kantadb created" in err
+    assert "🛢️ x.kantadb migrated v0 -> v1: migrate_v1 (rename)" in err
+
+
+@pytest.mark.asyncio
+async def test_logemit_receives_transaction_events(tmp_path, format_config):
+    path = tmp_path / "test.db"
+    kanta = make_kanta(path, Data, format_config)
+    events = []
+    kanta.logemit(lambda ev: events.append(ev) or True)
+    await kanta.open()
+
+    with kanta.transaction(action="inc", user="u1", extra="x") as data:
+        data.counter = 1
+
+    await kanta.close()
+
+    change = events[-1]
+    assert change.kind == "change"
+    assert change.action == "inc"
+    assert change.user == "u1"
+    assert change.extra == "x"
+    assert change.diff == {"counter": 1}
+    assert change.logger.name == "kanta.transaction"
+
+
+def test_logemit_rejects_classes_and_async(tmp_path, format_config):
+    kanta = make_kanta(tmp_path / "test.db", Data, format_config)
+
+    class NotAFunction:
+        pass
+
+    with pytest.raises(TypeError):
+        kanta.logemit(NotAFunction)
+
+    async def ahandler(ev):
+        return None
+
+    with pytest.raises(TypeError):
+        kanta.logemit(ahandler)
+
+
+def _raise(*args, **kwargs):
+    raise RuntimeError("formatting broken")
+
+
+def test_log_change_never_raises(monkeypatch):
+    monkeypatch.setattr("kanta.logging.format_action_header", _raise)
+    log_change("update", {"counter": 1}, previous={})  # must not raise
+
+
+@pytest.mark.asyncio
+async def test_logging_failure_does_not_break_transaction(
+    tmp_path, format_config, monkeypatch
+):
+    path = tmp_path / "test.db"
+    kanta = make_kanta(path, Data, format_config)
+    kanta.logemit(_raise)
+    monkeypatch.setattr("kanta.logging.format_action_header", _raise)
+    await kanta.open()
+
+    with kanta.transaction(action="inc") as data:
+        data.counter = 1
+
+    await kanta.close()
+
+    kanta2 = make_kanta(path, Data, format_config)
+    kanta2.logemit(_raise)
+    monkeypatch.setattr("kanta.logging.format_action_header", _raise)
+    await kanta2.open()
+    assert kanta2.data.counter == 1
+    await kanta2.close()
+
+
+@pytest.mark.asyncio
+async def test_logfmt_failure_falls_back_to_default(tmp_path, format_config, caplog):
+    path = tmp_path / "test.db"
+    kanta = make_kanta(path, Data, format_config)
+
+    @kanta.logfmt
+    def bad(value: str, path: str) -> str | None:
+        raise RuntimeError("broken")
+
+    await kanta.open()
+    with caplog.at_level(logging.INFO, logger="kanta.transaction"):
+        with kanta.transaction(action="inc", user="alice") as data:
+            data.counter = 1
+    await kanta.close()
+
+    assert kanta.data.counter == 1
+    assert "alice" in caplog.text  # raw rendering used despite the failure
+    assert "counter" in caplog.text
+
+
+@pytest.mark.asyncio
+async def test_logmigr_failure_does_not_break_open(tmp_path, format_config):
+    path = tmp_path / "test.db"
+    seed_single_change(path, fixed_change("init", {"counter": 0}), format_config)
+
+    mod = type(sys)("test_migrations_broken_logmigr")
+
+    def migrate_v1(d, kanta):
+        """Bump counter."""
+        d["counter"] = 2
+
+    mod.__dict__["migrate_v1"] = migrate_v1
+
+    kanta = make_kanta(path, Data, format_config, migrations=mod)
+
+    @kanta.logmigr
+    def bad(summary: MigrationResult) -> None:
+        raise RuntimeError("broken")
+
+    await kanta.open()
+    assert kanta.data.counter == 2
+    await kanta.close()
+
+
+@pytest.mark.asyncio
+async def test_aborted_transaction_emits_event(tmp_path, format_config, caplog):
+    path = tmp_path / "test.db"
+    kanta = make_kanta(path, Data, format_config)
+    events = []
+    kanta.logemit(lambda ev: events.append(ev) or True)
+    await kanta.open()
+
+    with caplog.at_level(logging.WARNING, logger="kanta.transaction"):
+        with pytest.raises(ValueError):
+            with kanta.transaction(action="reset") as data:
+                data.counter = 99
+                raise ValueError("simulated failure")
+
+    await kanta.close()
+
+    aborted = events[-1]
+    assert aborted.kind == "aborted"
+    assert aborted.action == "reset"
+    assert aborted.level == logging.WARNING
+    assert isinstance(aborted.error, ValueError)
+    messages = [r.getMessage() for r in caplog.records if r.levelno == logging.WARNING]
+    assert any("\x1b[1;34mreset" in m for m in messages)  # action color, no quotes
+    assert any(" transaction aborted: simulated failure" in m for m in messages)
+    assert kanta.data.counter == 0  # rolled back
+
+
+@pytest.mark.asyncio
+async def test_aborted_transaction_includes_resolved_user(
+    tmp_path, format_config, caplog
+):
+    path = tmp_path / "test.db"
+    kanta = make_kanta(path, Data, format_config)
+
+    @kanta.logfmt
+    def resolve(value: str, path: str) -> str | None:
+        return "Alice" if value == "u1" else None
+
+    await kanta.open()
+    with caplog.at_level(logging.WARNING, logger="kanta.transaction"):
+        with pytest.raises(ValueError):
+            with kanta.transaction(action="reset", user="u1", extra="exp") as data:
+                data.counter = 99
+                raise ValueError("boom")
+    await kanta.close()
+
+    messages = [r.getMessage() for r in caplog.records if r.levelno == logging.WARNING]
+    assert any("exp" in m for m in messages)
+    assert any(" by " in m and "Alice" in m for m in messages)
+    assert any(" transaction aborted: boom" in m for m in messages)
+
+
+def test_event_header_covers_all_kinds():
+    created = LogEvent(kind="created", logger=transaction_logger, filename="x.db")
+    assert created.header == "🛢️ x.db created"
+
+    migrated = LogEvent(
+        kind="migrated",
+        logger=transaction_logger,
+        filename="x.db",
+        from_version=0,
+        to_version=1,
+        migrations=["migrate_v1 (rename)"],
+    )
+    assert migrated.header == "🛢️ x.db migrated v0 -> v1: migrate_v1 (rename)"
+
+    aborted = LogEvent(
+        kind="aborted",
+        logger=transaction_logger,
+        action="reset",
+        user="alice",
+        error=ValueError("boom"),
+    )
+    assert "transaction aborted: boom" in aborted.header
+    assert "alice" in aborted.header
+
+
+@pytest.mark.asyncio
+async def test_event_carries_kanta_instance(tmp_path, format_config):
+    path = tmp_path / "test.db"
+    kanta = make_kanta(path, Data, format_config)
+    events = []
+    kanta.logemit(lambda ev: events.append(ev) or True)
+    await kanta.open()
+    with kanta.transaction(action="inc") as data:
+        data.counter = 1
+    await kanta.close()
+
+    assert events
+    assert all(ev.kanta is kanta for ev in events)
+
+
+def test_header_is_settable_and_used_by_default_emit(capsys):
+    logging.getLogger("kanta").handlers.clear()
+    configure_logging()
+
+    def restyle(ev):
+        ev.header = f"CUSTOM {ev.action}"
+        return True
+
+    emit_event(_change_event(diff={"counter": 1}, previous={}), [restyle])
+    err = capsys.readouterr().err
+    assert "CUSTOM update" in err
+    assert "counter" in err  # default diff routing still applies
+
+
+@pytest.mark.asyncio
+async def test_ctx_reachable_from_event(tmp_path, format_config):
+    path = tmp_path / "test.db"
+    kanta = make_kanta(path, Data, format_config)
+    kanta.ctx.connection_id = 7
+    seen = []
+    kanta.logemit(lambda ev: seen.append(ev.kanta.ctx.connection_id) or True)
+    await kanta.open()
+    with kanta.transaction(action="inc") as data:
+        data.counter = 1
+    await kanta.close()
+
+    assert seen and all(connection_id == 7 for connection_id in seen)

tests/test_logging.py

@@ -1,17 +1,116 @@
 import logging
 
-from kanta.logging import configure_logging, log_change
-from kanta.logging import logger
+import pytest
+
+from kanta.logging import (
+    configure_logging,
+    format_action_header,
+    log_change,
+)
+from kanta.tty import ESC
 
 
-def test_configure_logging():
+def test_format_action_header():
+    header = format_action_header("update", "alice", "tgt")
+    assert header == (
+        f"{ESC}1;34mupdate{ESC}0m {ESC}38;5;250mtgt{ESC}0m by {ESC}34malice{ESC}0m"
+    )
+
+
+def test_format_action_header_action_only():
+    assert format_action_header("update") == f"{ESC}1;34mupdate{ESC}0m"
+
+
+@pytest.fixture(autouse=True)
+def _reset_kanta_loggers():
+    yield
+    for name in (
+        "kanta",
+        "kanta.transaction",
+        "kanta.transaction.diff",
+        "kanta.bootstrap",
+        "kanta.migration",
+    ):
+        logger = logging.getLogger(name)
+        logger.setLevel(logging.NOTSET)
+        logger.propagate = True
+        logger.disabled = False
+        logger.handlers.clear()
+
+
+def test_configure_logging_defaults():
+    kanta_logger = logging.getLogger("kanta")
     configure_logging()
-    assert logger.level == logging.INFO
+    assert kanta_logger.level == logging.INFO
+    assert not kanta_logger.propagate
+    assert kanta_logger.handlers
+
+
+def test_configure_logging_disables_specific_loggers():
+    configure_logging(bootstrap=False, migration=False, transaction=False)
+    assert not logging.getLogger("kanta.bootstrap").propagate
+    assert not logging.getLogger("kanta.migration").propagate
+    assert not logging.getLogger("kanta.transaction").propagate
+
+
+def test_configure_logging_skiproot_false_leaves_kanta_propagation():
+    kanta_logger = logging.getLogger("kanta")
+    kanta_logger.handlers.clear()
+    configure_logging(bootstrap=False, skiproot=False)
+    assert kanta_logger.propagate
+    assert not kanta_logger.handlers
+    assert not logging.getLogger("kanta.bootstrap").propagate
 
 
 def test_log_change_no_diff(capsys):
-    logger.handlers.clear()
+    kanta_logger = logging.getLogger("kanta")
+    kanta_logger.handlers.clear()
     configure_logging()
     log_change("test", {})
     captured = capsys.readouterr()
     assert "test" in captured.err
+
+
+def test_log_change_appends_extra_string(capsys):
+    kanta_logger = logging.getLogger("kanta")
+    kanta_logger.handlers.clear()
+    configure_logging()
+    log_change("export", {}, extra="mydb.db")
+    captured = capsys.readouterr()
+    assert "export" in captured.err
+    assert f"{ESC}38;5;250mmydb.db{ESC}0m" in captured.err
+
+
+def test_log_change_log_diff_false(capsys, monkeypatch):
+    kanta_logger = logging.getLogger("kanta")
+    kanta_logger.handlers.clear()
+    configure_logging()
+
+    def _boom(*args, **kwargs):
+        raise AssertionError("format_diff should not be called")
+
+    monkeypatch.setattr("kanta.logging.format_diff", _boom)
+    log_change("update", {"counter": 5}, previous={}, log_diff=False)
+    captured = capsys.readouterr()
+    assert "update" in captured.err
+    assert "counter" not in captured.err
+
+
+def test_configure_logging_diff_false(capsys):
+    kanta_logger = logging.getLogger("kanta")
+    kanta_logger.handlers.clear()
+    configure_logging(diff=False)
+    log_change("update", {"counter": 5}, previous={})
+    captured = capsys.readouterr()
+    assert "update" in captured.err
+    assert "counter" not in captured.err
+
+
+def test_configure_logging_diff_true_reenables(capsys):
+    kanta_logger = logging.getLogger("kanta")
+    kanta_logger.handlers.clear()
+    configure_logging(diff=False)
+    configure_logging(diff=True)
+    log_change("update", {"counter": 5}, previous={})
+    captured = capsys.readouterr()
+    assert "counter" in captured.err

tests/test_migrations.py

@@ -1,53 +1,208 @@
-from types import ModuleType
+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)
-    assert new_ver == 2
+    result = reg.apply(state, current_version=0, kanta=kanta)
+    assert result.version == 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)
-    assert new_ver == 1
+    result = reg.apply(state, current_version=1, kanta=kanta)
+    assert result.version == 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)
-    assert new_ver == 2
+    result = reg.apply(state, current_version=0, kanta=kanta)
+    assert result.version == 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 = {}
+    result = reg.apply(state, current_version=0, kanta=kanta)
+    assert result.version == 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 = {}
+    result = reg.apply(state, current_version=0, kanta=kanta)
+    assert result.version == 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)
+
+
+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)
+
+
+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}
+    result = reg.apply(state, current_version=1, kanta=kanta)
+    assert result.version == 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}
+    result = reg.apply(state, current_version=2, kanta=kanta)
+    assert result.version == 3
+    assert state["x"] == 3
+
+
+def test_apply_returns_change_information():
+    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
+
+    result = reg.apply({}, current_version=0, kanta=kanta)
+    assert result.version == 3
+    assert len(result.migrations) == 3
+
+    assert result.migrations[0].name == "migrate_v1"
+    assert result.migrations[0].description == "Set x"
+    assert result.migrations[0].changed is True
+    assert result.migrations[0].diff == {"$replace": {"x": 1}}
+
+    assert result.migrations[1].name == "migrate_v2"
+    assert result.migrations[1].description == "No-op"
+    assert result.migrations[1].changed is False
+    assert result.migrations[1].diff is None
+
+    assert result.migrations[2].name == "migrate_v3"
+    assert result.migrations[2].description == "Set y"
+    assert result.migrations[2].changed is True
+    assert result.migrations[2].diff == {"y": 3}
+
+
+def test_description_defaults_to_version_when_no_docstring():
+    reg = Migrations()
+    kanta = _DummyKanta()
+
+    @reg.register
+    def migrate_v1(d):
+        d["x"] = 1
+
+    result = reg.apply({}, current_version=0, kanta=kanta)
+    assert result.migrations[0].description == "v1"

tests/test_mtime.py

@@ -82,8 +82,9 @@ async def test_transaction_mtime_false_preserves_mtime(tmp_path, format_config):
             continue
         records.append(serializer.decode(payload, type=ChangeRecord))
 
-    assert records[0].m == first_m
-    assert records[1].m is None
+    assert records[0].a == "bootstrap"
+    assert records[1].m == first_m
+    assert records[2].m is None
     assert kanta.mtime == first_m
 
 

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_snapshot.py

@@ -32,3 +32,20 @@ def test_force_writes():
     f = FakeFile()
     ss.maybe_write(f, 1, {"x": 1})
     assert len(f.written) == 1
+
+
+def test_force_bypasses_min_diffs():
+    class FakeFile:
+        def __init__(self):
+            self.written = []
+            self.is_open = True
+
+        def write(self, data: bytes):
+            self.written.append(data)
+
+    ss = SnapshotState(min_diffs=100)
+    ss.record_changes(5)
+    ss.request_force()
+    f = FakeFile()
+    ss.maybe_write(f, 1, {"x": 1})
+    assert len(f.written) == 1

tests/test_tty.py

@@ -0,0 +1,74 @@
+import pytest
+
+from kanta.tty import ESC, Colors, Line, colors, displaywidth, pad, strip_ansi
+
+
+def test_strip_ansi():
+    assert strip_ansi(f"{ESC}1;34mhello{ESC}0m") == "hello"
+
+
+def test_displaywidth_plain_and_ansi():
+    assert displaywidth("hello") == 5
+    assert displaywidth(f"{ESC}38;5;226mhi{ESC}0m") == 2
+
+
+def test_displaywidth_wide_and_combining_chars():
+    assert displaywidth("你好") == 4
+    assert displaywidth("🚀") == 2
+    assert displaywidth("é") == 1
+
+
+def test_pad():
+    assert pad("ab", 4) == "ab  "
+    assert pad("ab", 4, align="right") == "  ab"
+    assert pad("ab", 5, align="center") == " ab  "
+    assert pad("abcdef", 4) == "abcdef"
+    assert pad("你好", 6) == "你好  "
+
+
+def test_line_plain_and_str_conversion():
+    assert str(Line()("n=", 42)) == "n=42"
+
+
+def test_line_color_auto_resets_on_next_call():
+    assert str(Line().user("Alice")(" by ")) == f"{ESC}34mAlice{ESC}0m by "
+
+
+def test_line_str_restores_active_color():
+    assert str(Line().user("Alice")) == f"{ESC}34mAlice{ESC}0m"
+
+
+def test_line_same_color_not_reemitted():
+    assert str(Line().user("a").user("b")) == f"{ESC}34mab{ESC}0m"
+
+
+def test_line_transition_folds_reset_into_one_sequence():
+    # bold blue -> plain blue: the bold clear rides in the same sequence
+    assert str(Line().action("a").user("b")) == f"{ESC}1;34ma{ESC}0;34mb{ESC}0m"
+
+
+def test_line_unknown_color_raises():
+    with pytest.raises(AttributeError, match="unknown color"):
+        Line().nosuchcolor("x")
+
+
+def test_line_palette_addition(monkeypatch):
+    monkeypatch.setattr(colors, "session", "38;5;226", raising=False)
+    assert str(Line().session("3")) == f"{ESC}38;5;226m3{ESC}0m"
+
+
+def test_line_palette_override_takes_effect(monkeypatch):
+    monkeypatch.setattr(colors, "user", "36")
+    assert str(Line().user("x")) == f"{ESC}36mx{ESC}0m"
+
+
+def test_line_custom_palette():
+    palette = Colors()
+    palette.brand = "35"
+    assert str(Line(palette).brand("x")) == f"{ESC}35mx{ESC}0m"
+
+
+def test_line_width_and_align():
+    assert str(Line()("ab", width=4)) == "ab  "
+    assert str(Line()("ab", width=4, align="right")) == "  ab"
+    assert str(Line().user("ab", width=4)) == f"{ESC}34mab  {ESC}0m"