Inject state dicts by name (prev/state), rename MigrationResult to MigrationReport.

State dict injection now works by parameter name (prev/state, annotation
not checked) with DictPrev/DictState tags taking precedence; DictPre/DictPost
remain as aliases. MigrationResult is renamed to MigrationReport with fields
original/version/applied; the old type alias and a deprecated .migrations
property remain. Old symbols stay covered by the original tests; new-style
tests import from the kanta root. Includes some unrelated ruff formatting.

demo/demo.py

@@ -6,9 +6,7 @@ from pathlib import Path
 
 import msgspec
 
-from kanta import Kanta
-from kanta.callbacks import DictPre
-from kanta.logging import configure_logging
+from kanta import Kanta, configure_logging
 
 
 filename = Path(__file__).with_name("demo.kantadb")
@@ -47,11 +45,11 @@ kanta_v1 = Kanta(filename, Data(), migrations=sys.modules[__name__])
 
 
 @kanta_v1.logfmt
-def resolve_user(value: str, path: str, previous: DictPre) -> str | None:
+def resolve_user(value: str, path: str, state: dict) -> 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")
+    return state.get("users", {}).get(value, {}).get("name")
 
 
 async def main() -> None:

docs/database.md

@@ -186,9 +186,13 @@ when they have a default value.
   transaction actor, replacing the old `user_display` parameter.
 - The callback returns `str | None`: a string replaces the default rendering,
   while `None` means "fall through to the next formatter".
-- State dicts can be injected via `DictPre` (`Annotated[dict, "pre"]`)
-  and `DictPost` (`Annotated[dict, "post"]`); the `Kanta` instance can also be
-  injected.
+- State dicts are injected by parameter name or annotation tag, which share
+  the same vocabulary: `prev` receives the previous state dict and `state`
+  the current one.  Matching by name ignores the annotation entirely.  The
+  `DictPrev`/`DictState` aliases (`Annotated[dict, "prev"]` /
+  `Annotated[dict, "state"]`) work under any parameter name, and a tag takes
+  precedence over the name.  `DictPre` and `DictPost` are kept as aliases of
+  `DictPrev` and `DictState`.  The `Kanta` instance can also be injected.
 - Alternatively, a logfmt callback can be a class inheriting from `LogFmt`; the
   framework instantiates it with the state dicts and calls its
   `resolve(value, path) -> str | None` method.
@@ -201,8 +205,8 @@ values at that exact path:
 
 ```python
 @kanta.logfmt(path="$user")
-def resolve_user(value: str, current: DictPost) -> str | None:
-    return current.get("users", {}).get(value, {}).get("name")
+def resolve_user(value: str, state: dict) -> str | None:
+    return state.get("users", {}).get(value, {}).get("name")
 
 @kanta.logfmt(path="users.uuid-1")
 def resolve_user_key(value: str) -> str | None:

kanta/__init__.py

@@ -1,5 +1,17 @@
+from .callbacks import DictPrev, DictState, LogFmt
+from .exceptions import DatabaseError
 from .kanta import Kanta
+from .logging import LogEvent, configure_logging
+from .migrations import MigrationReport
 
 __all__ = [
     "Kanta",
+    "DatabaseError",
+    "configure_logging",
+    # Callback argument types
+    "DictPrev",
+    "DictState",
+    "LogEvent",
+    "LogFmt",
+    "MigrationReport",
 ]

kanta/__main__.py

@@ -0,0 +1,585 @@
+"""Module-level CLI for reading a kantadb file and printing its change log."""
+
+from __future__ import annotations
+
+import argparse
+import asyncio
+import contextlib
+import importlib
+import importlib.util
+import logging
+import sys
+import tempfile
+from pathlib import Path
+from typing import Any
+
+import msgspec
+
+from kanta import Kanta
+from kanta.callbacks import InjectionContext
+from kanta.exceptions import DatabaseError, DataIntegrityError, ReplayError
+from kanta.logging import LogEvent, emit_event, migration_logger
+from kanta.replaylog import (
+    RangeNotFoundError,
+    Selection,
+    SnapshotEvent,
+    _snapshot_lines,
+    end_of_file,
+    record_change_event,
+    record_label,
+    replay_events,
+    scan_events,
+    select,
+)
+from kanta.serialization import Serializer, dict_to_struct, struct_to_dict
+from kanta.structs import ChangeRecord, Snapshot
+from kanta.tty import Line
+
+EXIT_SUCCESS = 0
+EXIT_GENERIC = 1
+EXIT_RANGE_ERROR = 2
+EXIT_PARSE_ERROR = 10
+EXIT_MIGRATION_ERROR = 20
+EXIT_VALIDATION_ERROR = 21
+
+_logger = logging.getLogger(__name__)
+
+
+class _CliError(Exception):
+    """A user-facing error message paired with a process exit code."""
+
+    def __init__(self, message: str, code: int = EXIT_GENERIC) -> None:
+        self.code = code
+        super().__init__(message)
+
+
+def _import_dotted(path: str) -> Any:
+    """Import ``module.submodule.Attr`` or a filesystem path and return the attribute."""
+    if _is_file_path(path):
+        return _import_from_file(path)
+    if "." not in path:
+        raise ValueError(f"dotted path must contain a dot: {path!r}")
+    module_name, attr_name = path.rsplit(".", 1)
+    module = importlib.import_module(module_name)
+    try:
+        return getattr(module, attr_name)
+    except AttributeError as exc:
+        raise ImportError(f"{path!r} not found in {module_name!r}") from exc
+
+
+def _is_file_path(path: str) -> bool:
+    """Return True if *path* looks like a filesystem path rather than a dotted name."""
+    return "/" in path or "\\" in path or ":" in path
+
+
+def _import_from_file(path: str) -> Any:
+    """Import a module or attribute from a filesystem path.
+
+    *path* may be ``path/to/file.py`` (returns the module) or
+    ``path/to/file.py:symbol`` (returns ``symbol`` from the module).
+    """
+    if ":" in path:
+        file_path, symbol = path.rsplit(":", 1)
+    else:
+        file_path, symbol = path, None
+
+    file_path = Path(file_path).resolve()
+    if not file_path.exists():
+        raise ImportError(f"{file_path!r} not found")
+    if not file_path.is_file():
+        raise ImportError(f"{file_path!r} is not a file")
+
+    module_name = f"_kanta_cli_{file_path.stem}_{file_path.stat().st_ino}"
+    spec = importlib.util.spec_from_file_location(module_name, file_path)
+    if spec is None or spec.loader is None:
+        raise ImportError(f"cannot load {file_path!r}")
+    module = importlib.util.module_from_spec(spec)
+
+    file_dir = str(file_path.parent)
+    added_dir = False
+    if file_dir not in sys.path:
+        sys.path.insert(0, file_dir)
+        added_dir = True
+    try:
+        sys.modules[module_name] = module
+        spec.loader.exec_module(module)
+    finally:
+        if added_dir:
+            sys.path.remove(file_dir)
+
+    if symbol is None:
+        return module
+    try:
+        return getattr(module, symbol)
+    except AttributeError as exc:
+        raise ImportError(f"{symbol!r} not found in {file_path!r}") from exc
+
+
+def _import_kanta_object(path: str) -> Any:
+    """Import a Kanta object by module or filesystem path.
+
+    If ``path`` names an importable module, look up an object named
+    ``kanta`` in it; otherwise treat ``path`` as ``module.attr`` or
+    ``path/to/file.py[:kanta]`` referring directly to the object.
+    """
+    if _is_file_path(path):
+        if ":" in path:
+            return _import_from_file(path)
+        module = _import_from_file(path)
+        try:
+            return getattr(module, "kanta")
+        except AttributeError as exc:
+            raise ImportError(f"no 'kanta' object found in {path!r}") from exc
+    try:
+        spec = importlib.util.find_spec(path)
+    except ImportError:
+        spec = None
+    if spec is not None:
+        module = importlib.import_module(path)
+        try:
+            return getattr(module, "kanta")
+        except AttributeError as exc:
+            raise ImportError(f"no 'kanta' object found in module {path!r}") from exc
+    return _import_dotted(path)
+
+
+def _format_ts(dt) -> str:
+    """Return a local-looking timestamp without a timezone offset or microseconds."""
+    return dt.replace(tzinfo=None, microsecond=0).isoformat(sep=" ")
+
+
+def _parse_args(argv: list[str] | None = None) -> argparse.Namespace:
+    parser = argparse.ArgumentParser(
+        prog="kanta",
+        description="Read a kantadb file and print each change record to the console.",
+    )
+    parser.add_argument(
+        "file",
+        help="Path to the kantadb file, or '-' to read from stdin.",
+    )
+    parser.add_argument(
+        "-d",
+        "--data",
+        metavar="MOD",
+        help=(
+            "Dotted path or filesystem path to the root data type."
+            " Examples: myapp.models.Data, myapp/models.py:Data."
+        ),
+    )
+    parser.add_argument(
+        "-m",
+        "--migrations",
+        metavar="MOD",
+        help=(
+            "Dotted path or filesystem path to the migrations module."
+            " Examples: myapp.migrations, myapp/migrations.py."
+        ),
+    )
+    parser.add_argument(
+        "-k",
+        "--kanta",
+        metavar="MOD",
+        help=(
+            "Module path or filesystem path to an existing Kanta object to use."
+            " Either a module containing an object named 'kanta' (e.g. myapp.db),"
+            " a dotted path to the object (e.g. myapp.db.kanta), or a file path"
+            " (e.g. myapp/db.py or myapp/db.py:kanta).  Its type, migrations, and"
+            " logfmt/logemit callbacks are used.  Cannot be combined with -d or -m."
+        ),
+    )
+    parser.add_argument(
+        "-o",
+        "--output",
+        help="Write the final replayed state as JSON to this file, or '-' for stdout.",
+    )
+    parser.add_argument(
+        "-q",
+        "--quiet",
+        action="store_true",
+        help="Suppress normal change/snapshot logs; only print warnings and errors.",
+    )
+    parser.add_argument(
+        "-r",
+        "--range",
+        help=(
+            "Python-style range to process.  Units: plain number = change index,"
+            " lN = line number, sN = snapshot, vN = version.  Negative snapshot"
+            " values count from the end (s-1 is the last snapshot).  Use ':' for"
+            " half-open ranges and '..' for inclusive end ranges.  Examples:"
+            " '2:5', '2..5', 'l10:l20', 's1:s3', 'v0:v2', 's-1:', ':-1', '-1'."
+        ),
+    )
+    args = parser.parse_args(argv)
+    if args.kanta and (args.data or args.migrations):
+        parser.error("-k/--kanta cannot be used together with -d or -m")
+    return args
+
+
+def _print_change_log(
+    label: str,
+    record: ChangeRecord,
+    previous: dict[str, Any],
+    current: dict[str, Any],
+    kanta: Kanta[Any],
+) -> None:
+    """Log a single change record to stderr.
+
+    The record is dispatched as a :class:`LogEvent` through the Kanta
+    object's logemit handlers; the CLI's own rendering (with the ``l<N>``
+    label and timestamp) is the fallback when no handler claims the event.
+    """
+    event = record_change_event(record, previous, current, kanta)
+
+    def render(ev) -> None:
+        ts = _format_ts(record.ts)
+        lines = ev.diff_lines
+        if not lines:
+            print(f"{label} {ts} {ev.header}", file=sys.stderr)
+        elif len(lines) == 1:
+            print(f"{label} {ts} {ev.header}{lines[0]}", file=sys.stderr)
+        else:
+            print(f"{label} {ts} {ev.header}", file=sys.stderr)
+            for line in lines:
+                print(line, file=sys.stderr)
+            print(file=sys.stderr)
+
+    emit_event(
+        event,
+        kanta._impl.callback_registry.logemit_handlers,
+        fallback=render,
+    )
+
+
+def _format_size(n: int) -> str:
+    """Return a human-readable byte size."""
+    if n < 1024:
+        return f"{n} B"
+    if n < 1024 * 1024:
+        return f"{n / 1024:.1f} kB"
+    return f"{n / (1024 * 1024):.1f} MB"
+
+
+def _find_venv_site_packages(start: Path) -> list[Path]:
+    """Return site-packages dirs of ``.venv`` directories from *start* to parents."""
+    py_dir = f"python{sys.version_info.major}.{sys.version_info.minor}"
+    found: list[Path] = []
+    for parent in [start, *start.parents]:
+        venv = parent / ".venv"
+        if not venv.is_dir():
+            continue
+        site_packages = venv / "lib" / py_dir / "site-packages"
+        if site_packages.is_dir():
+            found.append(site_packages)
+            continue
+        # Windows layout
+        win_site = venv / "Lib" / "site-packages"
+        if win_site.is_dir():
+            found.append(win_site)
+    return found
+
+
+@contextlib.contextmanager
+def _extra_import_paths():
+    """Temporarily add current dir and nearby venv site-packages to ``sys.path``.
+
+    The current directory is inserted first, then local ``.venv`` site-packages,
+    then any parent ``.venv`` site-packages.  Only paths that were not already
+    present are added, and only those added paths are removed on exit.
+    """
+    paths_to_add = [str(Path.cwd())]
+    paths_to_add.extend(str(p) for p in _find_venv_site_packages(Path.cwd()))
+    added: list[str] = []
+    for path in reversed(paths_to_add):
+        if path not in sys.path:
+            sys.path.insert(0, path)
+            added.append(path)
+    try:
+        yield
+    finally:
+        for path in added:
+            if path in sys.path:
+                sys.path.remove(path)
+
+
+def _print_snapshot_indicator(
+    label: str,
+    snap: Snapshot,
+    index: int,
+    serializer: Serializer,
+) -> None:
+    """Print a snapshot indicator line to stderr.
+
+    ``snapshot s<N>`` is rendered in bright white; the version, optional mtime
+    and data size are printed in normal and dark colors respectively.
+    """
+    ts = _format_ts(snap.ts)
+    line = Line().snapshot("snapshot").snapshot(f" s{index}")
+    line.target(f" v{snap.v}")
+    if snap.m is not None:
+        line.target(f" {_format_ts(snap.m)}")
+    size = len(serializer.encode(snap.state))
+    line.path_prefix(f" {_format_size(size)}")
+    print(f"{label} {ts} {line}", file=sys.stderr)
+
+
+async def _log_migration(
+    kanta: Kanta[Any],
+    filename: Path,
+    result,
+    previous_version: int,
+    quiet: bool,
+) -> None:
+    """Log an applied migration through the Kanta instance's callbacks.
+
+    Routes to the object's logmigr callbacks when registered (like
+    :meth:`KantaImpl._handle_migration_log`); otherwise emits a ``migrated``
+    event through its logemit handlers, falling back to a stderr line.
+    """
+    registry = kanta._impl.callback_registry
+    if registry.has("logmigr"):
+        try:
+            await registry.invoke(
+                "logmigr",
+                InjectionContext(kanta=kanta, report=result),
+            )
+        except Exception:
+            _logger.exception("logmigr callback failed")
+        return
+    if quiet:
+        return
+    descriptions = [f"{m.name} ({m.description})" for m in result.applied if m.changed]
+    emit_event(
+        LogEvent(
+            kind="migrated",
+            logger=migration_logger,
+            kanta=kanta,
+            filename=str(filename),
+            from_version=previous_version,
+            to_version=result.version,
+            migrations=descriptions,
+        ),
+        registry.logemit_handlers,
+        fallback=lambda ev: print(ev.header, file=sys.stderr),
+    )
+
+
+def _get_kanta(args: argparse.Namespace, filename: Path) -> tuple[Kanta[Any], bool]:
+    """Return the Kanta instance to work with, and whether the CLI owns it.
+
+    With ``-k`` the existing object is used as-is (and never closed by us);
+    otherwise an instance is constructed with an empty dict state.
+    """
+    if args.kanta:
+        try:
+            obj = _import_kanta_object(args.kanta)
+        except (ImportError, ValueError) as exc:
+            raise _CliError(f"Invalid --kanta value: {exc}") from exc
+        if not isinstance(obj, Kanta):
+            raise _CliError(
+                f"Invalid --kanta value: {args.kanta!r} is not a Kanta object"
+            )
+        return obj, False
+    try:
+        return Kanta(filename, {}, type=dict, migrations=args.migrations), True
+    except Exception as exc:
+        if args.migrations:
+            raise _CliError(f"Migration error: {exc}", EXIT_MIGRATION_ERROR) from exc
+        raise _CliError(f"Failed to initialize database: {exc}") from exc
+
+
+async def _run(args: argparse.Namespace) -> int:
+    cleanup_path: Path | None = None
+    if args.file == "-":
+        content = sys.stdin.buffer.read()
+        tmp = tempfile.NamedTemporaryFile(delete=False, suffix=".kantadb")
+        tmp.write(content)
+        tmp.close()
+        filename = Path(tmp.name)
+        cleanup_path = filename
+    else:
+        filename = Path(args.file)
+        if not filename.exists():
+            raise _CliError(f"File not found: {filename}")
+        content = filename.read_bytes()
+
+    data_type: type[Any] | None = None
+    if args.data:
+        with _extra_import_paths():
+            try:
+                data_type = _import_dotted(args.data)
+            except (ImportError, ValueError) as exc:
+                raise _CliError(f"Invalid --data value: {exc}") from exc
+
+    kanta: Kanta[Any] | None = None
+    kanta_owned = False
+    kanta_typed: Kanta[Any] | None = None
+    try:
+        with _extra_import_paths():
+            kanta, kanta_owned = _get_kanta(args, filename)
+        if data_type is None and args.kanta and kanta._impl.data_type is not dict:
+            data_type = kanta._impl.data_type
+
+        # Decode and validate the whole file into positioned events.
+        try:
+            events, change_count = scan_events(content, kanta)
+        except ReplayError as exc:
+            raise _CliError(str(exc), EXIT_PARSE_ERROR) from exc
+        except Exception as exc:
+            raise _CliError(
+                f"Failed to replay records from {filename}: {exc}",
+                EXIT_PARSE_ERROR,
+            ) from exc
+
+        snapshot_line_to_index = {
+            line: idx for idx, line in enumerate(_snapshot_lines(events))
+        }
+
+        # Resolve -r into a line range or a single snapshot selection.
+        try:
+            selection = (
+                select(args.range, events, change_count)
+                if args.range is not None
+                else Selection(0, end_of_file(events))
+            )
+        except RangeNotFoundError as exc:
+            raise _CliError(str(exc), EXIT_RANGE_ERROR) from exc
+        except ValueError as exc:
+            raise _CliError(f"Invalid --range value: {exc}") from exc
+
+        if selection.snapshot is not None:
+            snap_event = selection.snapshot
+            state = snap_event.snap.state
+            version = snap_event.snap.v
+            if not args.quiet:
+                _print_snapshot_indicator(
+                    record_label(snap_event.line_number, snap_event.record_index),
+                    snap_event.snap,
+                    snapshot_line_to_index[snap_event.line_number],
+                    kanta._impl.serializer,
+                )
+                print(file=sys.stderr)
+        else:
+            # Replay up to the range end, printing logs within the range.
+            state = {}
+            version = 0
+            printed = False
+            for event, previous, current in replay_events(events, selection.end_line):
+                state = current
+                version = event.version
+                if event.line_number < selection.start_line or args.quiet:
+                    continue
+                label = record_label(event.line_number, event.record_index)
+                if isinstance(event, SnapshotEvent):
+                    _print_snapshot_indicator(
+                        label,
+                        event.snap,
+                        snapshot_line_to_index[event.line_number],
+                        kanta._impl.serializer,
+                    )
+                else:
+                    assert previous is not None
+                    _print_change_log(label, event.record, previous, current, kanta)
+                printed = True
+            if printed:
+                print(file=sys.stderr)
+
+        # Apply optional migrations to the range-end state.
+        if kanta._impl.migrations is not None:
+            try:
+                previous_version = version
+                result = kanta._impl.migrations.apply(state, version, kanta)
+                version = result.version
+            except Exception as exc:
+                raise _CliError(
+                    f"Migration error: {exc}", EXIT_MIGRATION_ERROR
+                ) from exc
+            if version != previous_version:
+                await _log_migration(
+                    kanta, filename, result, previous_version, args.quiet
+                )
+
+        output_state: dict[str, Any]
+        if data_type is not None:
+            try:
+                data = dict_to_struct(
+                    state, data_type, serializer=kanta._impl.serializer
+                )
+            except (
+                msgspec.ValidationError,
+                msgspec.DecodeError,
+                TypeError,
+                ValueError,
+            ) as exc:
+                raise _CliError(
+                    f"Validation error: {exc}", EXIT_VALIDATION_ERROR
+                ) from exc
+
+            if args.kanta:
+                # The file was already fully decoded and validated above with
+                # the object's own serializer, and its migrations were applied
+                # to the state; no need to re-open through a new instance.
+                print(f"{data}", file=sys.stderr)
+                output_state = struct_to_dict(data, serializer=kanta._impl.serializer)
+            else:
+                kanta_typed = Kanta(
+                    filename, data, type=data_type, migrations=args.migrations
+                )
+                try:
+                    await kanta_typed.open(create=False, readonly=True, log=False)
+                    print(f"{data}", file=sys.stderr)
+                except (msgspec.ValidationError, msgspec.DecodeError) as exc:
+                    raise _CliError(
+                        f"Validation error: {exc}", EXIT_VALIDATION_ERROR
+                    ) from exc
+                except DataIntegrityError as exc:
+                    raise _CliError(f"Parse error: {exc}", EXIT_PARSE_ERROR) from exc
+                except DatabaseError as exc:
+                    if not args.migrations or exc.cause_type == "ReplayError":
+                        raise _CliError(
+                            f"Parse error: {exc}", EXIT_PARSE_ERROR
+                        ) from exc
+                    raise _CliError(
+                        f"Migration error: {exc}", EXIT_MIGRATION_ERROR
+                    ) from exc
+                except Exception as exc:
+                    if args.migrations:
+                        raise _CliError(
+                            f"Migration error: {exc}", EXIT_MIGRATION_ERROR
+                        ) from exc
+                    raise _CliError(f"Failed to open {filename}: {exc}") from exc
+                output_state = kanta_typed._impl.statedict
+        else:
+            output_state = state
+
+        if args.output:
+            try:
+                out_bytes = msgspec.json.encode(output_state)
+                if args.output == "-":
+                    sys.stdout.buffer.write(out_bytes)
+                else:
+                    Path(args.output).write_bytes(out_bytes)
+            except Exception as exc:  # pragma: no cover
+                raise _CliError(f"Failed to write output: {exc}") from exc
+
+        return EXIT_SUCCESS
+    finally:
+        if kanta_typed is not None:
+            await kanta_typed.close()
+        if kanta is not None and kanta_owned:
+            await kanta.close()
+        if cleanup_path is not None:
+            cleanup_path.unlink(missing_ok=True)
+
+
+def main(argv: list[str] | None = None) -> int:
+    """Entry point for ``python -m kanta``."""
+    args = _parse_args(argv)
+    try:
+        return asyncio.run(_run(args))
+    except _CliError as exc:
+        print(exc, file=sys.stderr)
+        return exc.code
+
+
+if __name__ == "__main__":
+    sys.exit(main())

kanta/callbacks.py

@@ -1,8 +1,8 @@
 """Unified decorator-based callback registry for Kanta.
 
-Callbacks are registered once and invoked with arguments filled by their
-annotation types.  Unknown arguments are only permitted when they have a
-default value.
+Callbacks are registered once and invoked with arguments filled from their
+parameter names (state dicts: ``prev`` / ``state``) and annotation types.
+Unknown arguments are only permitted when they have a default value.
 
 Log formatters are a special case: they are called per value being rendered
 and receive the value plus an optional ``path`` string.  They return
@@ -23,10 +23,36 @@ from dataclasses import dataclass
 from typing import Annotated, Any, Union, get_args, get_origin
 
 from kanta.exceptions import DatabaseError
-from kanta.migrations import MigrationResult
+from kanta.migrations import MigrationReport
+
+DictPrev = DictPre = Annotated[dict, "prev"]
+DictState = DictPost = Annotated[dict, "state"]
+
+# State-dict injection keys, shared by parameter names and annotation tags:
+# a callback parameter named *or* tagged ``prev``/``state`` receives the
+# previous or current state dict respectively.  Matching by name does not
+# check the annotation; an explicit tag takes precedence over the name.
+_STATE_KINDS = {"prev": "previous_state", "state": "current_state"}
+
+
+def _state_key(text: Any) -> str | None:
+    """Return the state kind for a parameter name or annotation tag."""
+    return text if text in _STATE_KINDS else None
+
+
+def _state_tag(ann: Any) -> str | None:
+    """Return the state tag of an ``Annotated[dict, ...]`` annotation, if any."""
+    if get_origin(ann) is not Annotated:
+        return None
+    args = get_args(ann)
+    if not args or args[0] is not dict:
+        return None
+    for meta in args[1:]:
+        key = _state_key(meta)
+        if key is not None:
+            return key
+    return None
 
-DictPre = Annotated[dict, "pre"]
-DictPost = Annotated[dict, "post"]
 
 _logger = logging.getLogger(__name__)
 
@@ -35,16 +61,18 @@ class LogFmt:
     """Base class for stateful logfmt callbacks.
 
     Subclasses only need to override :meth:`resolve`.  The framework injects
-    ``previous_state`` and ``current_state`` through ``__init__``.
+    the previous and current state dicts through ``__init__`` and exposes them
+    as ``previous_state`` and ``state``.
     """
 
     def __init__(
         self,
-        previous: DictPre | None = None,
-        current: DictPost | None = None,
+        prev: dict | None = None,
+        state: dict | None = None,
     ) -> None:
-        self.previous_state = previous
-        self.current_state = current
+        self.previous_state = prev
+        self.state = state
+        self.current_state = state  # deprecated alias for ``state``
 
     def __call__(self, value: Any, path: str) -> str | None:
         return self.resolve(value, path)
@@ -67,7 +95,7 @@ class InjectionContext:
     error: DatabaseError | None = None
     previous_state: dict | None = None
     current_state: dict | None = None
-    migration_result: MigrationResult | None = None
+    report: MigrationReport | None = None
 
 
 @dataclass
@@ -261,6 +289,9 @@ class CallbackRegistry:
                 )
 
             if param.annotation is inspect.Parameter.empty:
+                if _state_key(name) is not None:
+                    params.append((name, dict))
+                    continue
                 if param.default is inspect.Parameter.empty:
                     raise TypeError(
                         f"{kind} callback {callback.__name__} has parameter "
@@ -269,6 +300,11 @@ class CallbackRegistry:
                 continue
 
             ann = self._resolve_raw_annotation(param.annotation, callback)
+            if _state_key(name) is not None:
+                # The name alone selects state injection; an explicit tag
+                # still overrides it.  The annotation is not checked.
+                params.append((name, ann))
+                continue
             if not self._is_allowed(kind, ann):
                 if param.default is inspect.Parameter.empty:
                     raise TypeError(
@@ -329,6 +365,9 @@ class CallbackRegistry:
                     f"*args or **kwargs"
                 )
             if param.annotation is inspect.Parameter.empty:
+                if _state_key(name) is not None:
+                    inject_params.append((name, dict))
+                    continue
                 if param.default is inspect.Parameter.empty:
                     raise TypeError(
                         f"logfmt callback {callback.__name__} has parameter "
@@ -337,6 +376,11 @@ class CallbackRegistry:
                 continue
 
             ann = self._resolve_raw_annotation(param.annotation, callback)
+            if _state_key(name) is not None:
+                # The name alone selects state injection; an explicit tag
+                # still overrides it.  The annotation is not checked.
+                inject_params.append((name, ann))
+                continue
             if name == "path" and self._unwrap_optional(ann) is str:
                 has_path = True
                 continue
@@ -392,6 +436,9 @@ class CallbackRegistry:
                     f"*args or **kwargs"
                 )
             if param.annotation is inspect.Parameter.empty:
+                if _state_key(name) is not None:
+                    inject_params.append((name, dict))
+                    continue
                 if param.default is inspect.Parameter.empty:
                     raise TypeError(
                         f"logfmt class {cls.__name__}.__init__ has parameter "
@@ -400,6 +447,11 @@ class CallbackRegistry:
                 continue
 
             ann = self._resolve_raw_annotation(param.annotation, cls.__init__)
+            if _state_key(name) is not None:
+                # The name alone selects state injection; an explicit tag
+                # still overrides it.  The annotation is not checked.
+                inject_params.append((name, ann))
+                continue
             if self._is_allowed("logfmt", ann):
                 inject_params.append((name, ann))
                 continue
@@ -464,7 +516,7 @@ class CallbackRegistry:
     ) -> dict[str, Any]:
         kwargs: dict[str, Any] = {}
         for name, ann in params:
-            value = self._resolve_annotation(ann, ctx)
+            value = self._resolve_annotation(name, ann, ctx)
             if value is _UNRESOLVED:
                 raise RuntimeError(f"no value available for annotation {ann!r}")
             kwargs[name] = value
@@ -472,13 +524,11 @@ class CallbackRegistry:
 
     def _is_allowed(self, kind: str, ann: Any) -> bool:
         bare = self._unwrap_optional(ann)
-        if self._matches_state_annotation(bare, "pre"):
-            return kind == "logfmt"
-        if self._matches_state_annotation(bare, "post"):
+        if _state_tag(bare) is not None:
             return kind == "logfmt"
         if bare is DatabaseError:
             return kind == "fatal_error"
-        if bare is MigrationResult:
+        if bare is MigrationReport:
             return kind == "logmigr"
         if self._data_type is not None and bare is self._data_type:
             return kind == "bootstrap"
@@ -502,22 +552,25 @@ class CallbackRegistry:
         if kind == "fatal_error":
             parts.append("DatabaseError")
         if kind == "logmigr":
-            parts.append("MigrationResult")
+            parts.append("MigrationReport")
         if kind == "logfmt":
-            parts.append("Annotated[dict, 'pre']")
-            parts.append("Annotated[dict, 'post']")
+            parts.append("prev: dict")
+            parts.append("state: dict")
         return ", ".join(parts) if parts else "none"
 
-    def _resolve_annotation(self, ann: Any, ctx: InjectionContext) -> Any:
+    def _resolve_annotation(self, name: str, ann: Any, ctx: InjectionContext) -> Any:
         bare = self._unwrap_optional(ann)
-        if self._matches_state_annotation(bare, "pre"):
-            return ctx.previous_state
-        if self._matches_state_annotation(bare, "post"):
-            return ctx.current_state
+        # An explicit tag takes precedence over the parameter name.
+        tag = _state_tag(bare)
+        if tag is not None:
+            return getattr(ctx, _STATE_KINDS[tag])
+        key = _state_key(name)
+        if key is not None:
+            return getattr(ctx, _STATE_KINDS[key])
         if bare is DatabaseError:
             return ctx.error
-        if bare is MigrationResult:
-            return ctx.migration_result
+        if bare is MigrationReport:
+            return ctx.report
         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:
@@ -539,16 +592,6 @@ class CallbackRegistry:
                 ) from exc
         return raw_ann
 
-    @staticmethod
-    def _matches_state_annotation(ann: Any, marker: str) -> bool:
-        origin = get_origin(ann)
-        if origin is not Annotated:
-            return False
-        args = get_args(ann)
-        if not args:
-            return False
-        return args[0] is dict and marker in args[1:]
-
     @staticmethod
     def _unwrap_optional(ann: Any) -> Any:
         origin = get_origin(ann)

kanta/kanta.py

@@ -281,7 +281,7 @@ class Kanta(Generic[T]):
         """Register a migration logging callback.
 
         Can be used as ``@kanta.logmigr``.
-        The callback receives a :class:`kanta.migrations.MigrationResult` and
+        The callback receives a :class:`kanta.migrations.MigrationReport` and
         may be sync or async. If registered, it replaces the default migration
         logger output; the application is responsible for emitting any log
         messages.

kanta/kantaimpl.py

@@ -19,7 +19,7 @@ from kanta.logging import (
     emit_event,
     migration_logger,
 )
-from kanta.migrations import MigrationResult, Migrations
+from kanta.migrations import MigrationReport, Migrations
 from kanta.persistence import PersistenceMixin
 from kanta.serialization import restore_data_in_place, struct_to_dict
 from kanta.serialization.base import replay
@@ -97,19 +97,18 @@ class KantaImpl(PersistenceMixin, Generic[T]):
 
     async def _handle_migration_log(
         self,
-        migration_result: MigrationResult,
-        previous_version: int,
+        report: MigrationReport,
         log: bool | logging.Logger,
     ) -> None:
         """Route migration logging to callback or default logger."""
-        assert isinstance(migration_result, MigrationResult)
+        assert isinstance(report, MigrationReport)
 
         if self.callback_registry.has("logmigr"):
             await self.callback_registry.invoke(
                 "logmigr",
                 InjectionContext(
                     kanta=self._kanta,
-                    migration_result=migration_result,
+                    report=report,
                 ),
                 on_error=_log_callback_error,
             )
@@ -120,7 +119,7 @@ class KantaImpl(PersistenceMixin, Generic[T]):
 
         migration_log = log if isinstance(log, logging.Logger) else migration_logger
 
-        changed = [m for m in migration_result.migrations if m.changed]
+        changed = [m for m in report.applied if m.changed]
         if not changed:
             return
 
@@ -131,8 +130,8 @@ class KantaImpl(PersistenceMixin, Generic[T]):
                 logger=migration_log,
                 kanta=self._kanta,
                 filename=str(self.filename),
-                from_version=previous_version,
-                to_version=migration_result.version,
+                from_version=report.original,
+                to_version=report.version,
                 migrations=descriptions,
             ),
             self.callback_registry.logemit_handlers,
@@ -211,15 +210,15 @@ class KantaImpl(PersistenceMixin, Generic[T]):
                     cause_type=type(e).__name__,
                 ) from e
 
-            migration_result = None
+            migration_report = 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(
+                migration_report = self.migrations.apply(
                     rr.state, rr.version, self._kanta
                 )
-                rr.version = migration_result.version
+                rr.version = migration_report.version
 
             migrations_ran = rr.version != previous_version
 
@@ -267,10 +266,8 @@ class KantaImpl(PersistenceMixin, Generic[T]):
                 )
                 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 migrations_ran and migration_report is not None:
+                    await self._handle_migration_log(migration_report, log)
                 if (
                     record is not None
                     and log is not False

kanta/logging.py

@@ -124,19 +124,22 @@ class LogEvent(msgspec.Struct, kw_only=True):
 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, :func:`default_emit` renders the event
-    with the built-in formatting.
+    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 built-in formatting, and a failure in
-    the built-in formatting itself is reported and swallowed.
+    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:
@@ -146,7 +149,7 @@ def emit_event(
                 break
             if not proceed:
                 return
-        default_emit(ev)
+        render(ev)
     except Exception:
         _logger.exception("failed to emit %s log event", ev.kind)
 

kanta/migrations.py

@@ -34,11 +34,20 @@ class MigrationInfo:
 
 
 @dataclass
-class MigrationResult:
-    """Result of applying migrations."""
+class MigrationReport:
+    """Report of applying migrations."""
 
     version: int
-    migrations: list[MigrationInfo]
+    original: int
+    applied: list[MigrationInfo]
+
+    @property
+    def migrations(self) -> list[MigrationInfo]:
+        """Deprecated alias for :attr:`applied`."""
+        return self.applied
+
+
+MigrationResult = MigrationReport  # deprecated alias for MigrationReport
 
 
 class Migrations:
@@ -57,13 +66,13 @@ class Migrations:
         def migrate_v2(d: dict) -> None:
             d.setdefault("version", 2)
 
-        result = migrations.apply(state, current_version=0, kanta=kanta)
-        new_version = result.version
+        report = migrations.apply(state, current_version=0, kanta=kanta)
+        new_version = report.version
 
     Or load from a module::
 
         migrations = Migrations.from_module("myapp.migrations")
-        result = migrations.apply(state, current_version=0, kanta=kanta)
+        report = migrations.apply(state, current_version=0, kanta=kanta)
     """
 
     def __init__(self) -> None:
@@ -137,7 +146,7 @@ class Migrations:
         data_dict: dict[str, Any],
         current_version: int,
         kanta: Any,
-    ) -> MigrationResult:
+    ) -> MigrationReport:
         """Apply pending migrations to *data_dict* in place.
 
         Missing intermediate migration steps are silently skipped.
@@ -146,8 +155,8 @@ class Migrations:
             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.
+        Returns a :class:`MigrationReport` describing the original and new
+        versions and every migration that ran.
         """
         if current_version > self.dbver:
             raise DatabaseError(
@@ -161,6 +170,7 @@ class Migrations:
             )
 
         migrations: list[MigrationInfo] = []
+        original = current_version
         for version in sorted(self._migrations.keys()):
             if version <= current_version:
                 continue
@@ -181,4 +191,6 @@ class Migrations:
                     before=before,
                 )
             )
-        return MigrationResult(version=current_version, migrations=migrations)
+        return MigrationReport(
+            version=current_version, original=original, applied=migrations
+        )

kanta/replaylog.py

@@ -0,0 +1,378 @@
+"""Line-oriented replay and range selection for kantadb files.
+
+Support machinery for the ``python -m kanta`` CLI: decoding a file into
+positioned events, resolving ``-r`` range specifications to line numbers,
+replaying state over a line range, and building change log events.  Internal
+for now; not part of the public API.
+"""
+
+from __future__ import annotations
+
+import copy
+import dataclasses
+from collections.abc import Iterator
+from typing import TYPE_CHECKING, Any, Union
+
+import msgspec
+
+from kanta.callbacks import InjectionContext
+from kanta.diff import patch_state
+from kanta.exceptions import ReplayError
+from kanta.logging import _USER_PATH, LogEvent, transaction_logger
+from kanta.structs import ChangeRecord, Snapshot
+
+if TYPE_CHECKING:
+    from kanta import Kanta
+
+
+@dataclasses.dataclass
+class SnapshotEvent:
+    """A snapshot record positioned in the file."""
+
+    line_number: int
+    byte_pos: int
+    record_index: int
+    snap: Snapshot
+
+    @property
+    def version(self) -> int:
+        return self.snap.v
+
+
+@dataclasses.dataclass
+class ChangeEvent:
+    """A change record positioned in the file."""
+
+    line_number: int
+    byte_pos: int
+    record_index: int
+    record: ChangeRecord
+
+    @property
+    def version(self) -> int:
+        return self.record.v
+
+
+Event = Union[SnapshotEvent, ChangeEvent]
+
+
+class RangeNotFoundError(Exception):
+    """A single-item range specification that does not exist in the file.
+
+    The message is fully formatted for display, including the offending
+    input and how many items of that kind the file contains.
+    """
+
+
+@dataclasses.dataclass
+class Selection:
+    """A resolved range specification.
+
+    Either a ``[start_line, end_line)`` line range, or a single snapshot
+    (``snapshot`` set), used to show the snapshot state without replaying
+    further records.
+    """
+
+    start_line: int
+    end_line: int
+    snapshot: SnapshotEvent | None = None
+
+
+def record_label(line_number: int, record_index: int) -> str:
+    """Return a padded record label based on line number, falling back to record index."""
+    number = line_number if line_number else record_index
+    return f"l{number:<3}"
+
+
+def scan_events(content: bytes, kanta: Kanta[Any]) -> tuple[list[Event], int]:
+    """Decode all records, validating snapshot consistency.
+
+    Uses the Kanta instance's serializer and framer.  Returns the events in
+    file order and the number of change records.  Raises :class:`ReplayError`
+    with a located, display-ready message on decode failures or when a
+    snapshot does not match the replayed state.
+    """
+    impl = kanta._impl
+    state: dict[str, Any] = {}
+    events: list[Event] = []
+    change_count = 0
+    for is_snapshot, payload, line_number, byte_pos in impl.framer.iter_records(
+        content, 0
+    ):
+        record_index = len(events) + 1
+        label = record_label(line_number, record_index)
+        try:
+            if is_snapshot:
+                snap = impl.serializer.decode(payload, type=Snapshot)
+                if record_index > 1 and state != snap.state:
+                    raise ReplayError(
+                        f"Snapshot mismatch at {label}: replayed state"
+                        " does not equal the snapshot state.",
+                        line_number=line_number,
+                        byte_pos=byte_pos,
+                        record_type="snapshot",
+                    )
+                state = snap.state
+                events.append(SnapshotEvent(line_number, byte_pos, record_index, snap))
+            else:
+                record = impl.serializer.decode(payload, type=ChangeRecord)
+                state = patch_state(state, record.diff)
+                events.append(ChangeEvent(line_number, byte_pos, record_index, record))
+                change_count += 1
+        except msgspec.DecodeError as exc:
+            raise ReplayError(
+                f"Parse error at {label}: {exc}",
+                line_number=line_number,
+                byte_pos=byte_pos,
+            ) from exc
+    return events, change_count
+
+
+def replay_events(
+    events: list[Event], end_line: int
+) -> Iterator[tuple[Event, dict[str, Any] | None, dict[str, Any]]]:
+    """Replay events with line numbers below ``end_line``.
+
+    Yields ``(event, previous, state)`` per event: ``previous`` is the state
+    before a change (``None`` for snapshots) and ``state`` the state after
+    the event.
+    """
+    state: dict[str, Any] = {}
+    for event in events:
+        if event.line_number >= end_line:
+            break
+        if isinstance(event, SnapshotEvent):
+            state = event.snap.state
+            yield event, None, state
+        else:
+            previous = copy.deepcopy(state)
+            state = patch_state(state, event.record.diff)
+            yield event, previous, state
+
+
+def record_change_event(
+    record: ChangeRecord,
+    previous: dict[str, Any],
+    current: dict[str, Any],
+    kanta: Kanta[Any],
+) -> LogEvent:
+    """Build a change :class:`LogEvent` for a replayed record.
+
+    The Kanta instance's logfmt callbacks are used for value formatting and
+    for resolving the user/actor name; the event can then be dispatched with
+    :func:`kanta.logging.emit_event` and the instance's logemit handlers.
+    """
+    registry = kanta._impl.callback_registry
+    logfmt = registry.build_logfmt(
+        InjectionContext(
+            kanta=kanta,
+            previous_state=previous,
+            current_state=current,
+        )
+    )
+    user = record.u
+    if user is not None:
+        resolved = logfmt(user, _USER_PATH)
+        if resolved is not None:
+            user = resolved
+    return LogEvent(
+        kind="change",
+        logger=transaction_logger,
+        kanta=kanta,
+        action=record.a,
+        user=user,
+        diff=record.diff,
+        previous=previous,
+        logfmt=logfmt,
+    )
+
+
+def _plural(count: int, word: str) -> str:
+    """Return e.g. ``1 snapshot`` or ``2 snapshots``."""
+    return f"{count} {word}{'' if count == 1 else 's'}"
+
+
+def end_of_file(events: list[Event]) -> int:
+    """Return the sentinel line number just past the last line of the file."""
+    return events[-1].line_number + 1 if events else 0
+
+
+def _change_lines(events: list[Event]) -> list[int]:
+    """Return the line numbers of all change records, in file order."""
+    return [e.line_number for e in events if isinstance(e, ChangeEvent)]
+
+
+def _snapshot_lines(events: list[Event]) -> list[int]:
+    """Return the line numbers addressed by s0, s1, ...
+
+    If the file begins with a snapshot, s0 is that snapshot (l1) and s1 is
+    the next snapshot.  Otherwise the file begins with change records (empty
+    initial state): s0 is l0, the position before the start of the file, and
+    s1 is the first snapshot.
+    """
+    lines = [e.line_number for e in events if isinstance(e, SnapshotEvent)]
+    if events and isinstance(events[0], SnapshotEvent):
+        return lines
+    return [0, *lines]
+
+
+def _version_lines(events: list[Event]) -> dict[int, int]:
+    """Map each version to the line where it first appears; v0 is l0."""
+    lines: dict[int, int] = {0: 0}
+    for event in events:
+        lines.setdefault(event.version, event.line_number)
+    return lines
+
+
+def _event_at_line(events: list[Event], line: int) -> Event | None:
+    """Return the event whose file line number exactly matches ``line``."""
+    for event in events:
+        if event.line_number == line:
+            return event
+    return None
+
+
+def _parse_bound(bound_str: str) -> tuple[str, int | None]:
+    """Parse a range bound with optional unit prefix (l, s, v) or change index."""
+    if not bound_str:
+        return "change", None
+    unit_map = {"l": "line", "s": "snapshot", "v": "version"}
+    if bound_str[0] in unit_map:
+        unit = unit_map[bound_str[0]]
+        rest = bound_str[1:]
+        if not rest:
+            raise ValueError(f"empty value in {bound_str!r}")
+        return unit, int(rest)
+    return "change", int(bound_str)
+
+
+def _bound_to_line(
+    unit: str,
+    value: int | None,
+    events: list[Event],
+    total: int,
+    is_start: bool,
+) -> int:
+    """Convert a range bound to a line number.
+
+    Out-of-range values are truncated to l0 (before the first line) or to
+    the line just past the end of the file rather than erroring; a missing
+    bound means the corresponding file end.
+    """
+    eof = end_of_file(events)
+    if value is None:
+        return 0 if is_start else eof
+
+    if unit == "change":
+        lines = _change_lines(events)
+        if value < 0:
+            value = total + value
+        value = max(0, min(value, total))
+        return lines[value] if value < total else eof
+
+    if unit == "line":
+        if value < 0:
+            raise ValueError("line numbers do not support negative indexing")
+        return value
+
+    if unit == "snapshot":
+        lines = _snapshot_lines(events)
+        idx = len(lines) + value if value < 0 else value
+        if idx < 0:
+            return 0
+        return lines[idx] if idx < len(lines) else eof
+
+    if unit == "version":
+        if value < 0:
+            raise ValueError("version numbers do not support negative indexing")
+        return _version_lines(events).get(value, eof)
+
+    raise ValueError(f"unknown range unit: {unit}")
+
+
+def _resolve_range(range_str: str, events: list[Event], total: int) -> tuple[int, int]:
+    """Parse a range string into a [start_line, end_line) line range."""
+    sep = ".." if ".." in range_str else ":"
+    start_str, end_str = range_str.split(sep, 1)
+    start_unit, start_val = _parse_bound(start_str)
+    end_unit, end_val = _parse_bound(end_str)
+    start_line = _bound_to_line(start_unit, start_val, events, total, is_start=True)
+    end_line = _bound_to_line(end_unit, end_val, events, total, is_start=False)
+    # ``..`` makes the end bound inclusive.
+    if sep == ".." and end_val is not None:
+        end_line += 1
+    return min(start_line, end_line), end_line
+
+
+def _negative_check(unit: str, value: int) -> None:
+    if value < 0:
+        raise ValueError(f"{unit} numbers do not support negative indexing")
+
+
+def select(spec: str, events: list[Event], total: int) -> Selection:
+    """Resolve a range specification against the scanned events.
+
+    ``total`` is the number of change records.  Returns a :class:`Selection`:
+    a line range, or a single snapshot for snapshot selections (``sN``, or
+    ``lN`` pointing at a snapshot).  Ranges truncate out-of-bounds values;
+    a single index must exist and raises :class:`RangeNotFoundError`
+    otherwise.  Syntax errors raise :class:`ValueError`.
+    """
+    if ":" in spec or ".." in spec:
+        start_line, end_line = _resolve_range(spec, events, total)
+        return Selection(start_line, end_line)
+
+    # A single index must exist; out-of-bounds is an error.
+    unit, value = _parse_bound(spec)
+    if value is None:
+        raise ValueError("single bound must not be empty")
+
+    if unit == "snapshot":
+        lines = _snapshot_lines(events)
+        idx = len(lines) + value if value < 0 else value
+        n_snapshots = sum(isinstance(e, SnapshotEvent) for e in events)
+        count = _plural(n_snapshots, "snapshot")
+        if not 0 <= idx < len(lines):
+            raise RangeNotFoundError(f"Snapshot {spec!r} not found in file ({count})")
+        event = _event_at_line(events, lines[idx])
+        if event is None:
+            # s0 with an empty initial state (l0): not a real record, so it
+            # cannot be selected as a single item.
+            raise RangeNotFoundError(
+                f"Snapshot {spec!r} not found in file: the file starts"
+                f" with an empty initial state ({count})"
+            )
+        assert isinstance(event, SnapshotEvent)
+        return Selection(event.line_number, event.line_number + 1, event)
+
+    if unit == "line":
+        _negative_check(unit, value)
+        event = _event_at_line(events, value)
+        if event is None:
+            n_lines = events[-1].line_number if events else 0
+            raise RangeNotFoundError(
+                f"Line {spec!r} not found in file ({_plural(n_lines, 'line')})"
+            )
+        if isinstance(event, SnapshotEvent):
+            return Selection(value, value + 1, event)
+        return Selection(value, value + 1)
+
+    if unit == "version":
+        _negative_check(unit, value)
+        lines = _version_lines(events)
+        if value not in lines:
+            raise RangeNotFoundError(
+                f"Version {spec!r} not found in file ({_plural(len(lines), 'version')})"
+            )
+        start_line = lines[value]
+        later = [line for line in lines.values() if line > start_line]
+        return Selection(start_line, min(later) if later else end_of_file(events))
+
+    lines = _change_lines(events)
+    idx = total + value if value < 0 else value
+    if not 0 <= idx < total:
+        raise RangeNotFoundError(
+            f"Change index {spec!r} not found in file ({_plural(total, 'change')})"
+        )
+    end_line = lines[idx + 1] if idx + 1 < total else end_of_file(events)
+    return Selection(lines[idx], end_line)

kanta/tty.py

@@ -70,6 +70,7 @@ class Colors:
     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
+    snapshot = "97"  # Bright white for snapshot indicator text
     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

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/test_callbacks.py

@@ -2,7 +2,7 @@ from typing import Any, Optional, Union
 
 import pytest
 
-from kanta import Kanta
+from kanta import DictPrev, DictState, Kanta
 from kanta.callbacks import DictPost, DictPre, LogFmt
 from kanta.exceptions import DatabaseError
 
@@ -166,6 +166,106 @@ async def test_logfmt_injects_states(tmp_path, format_config, caplog):
     assert "Alice" in caplog.text
 
 
+@pytest.mark.asyncio
+async def test_logfmt_injects_states_by_name(tmp_path, format_config, caplog):
+    import logging
+
+    caplog.set_level(logging.INFO, logger="kanta.transaction")
+    path = tmp_path / "test.db"
+    kanta = make_kanta(path, Data, format_config)
+
+    @kanta.logfmt
+    def resolve_users(value: str, prev, state: dict | None) -> str | None:
+        assert prev == {}
+        assert state is not None
+        return state.get("users", {}).get(value, {}).get("name")
+
+    await kanta.open()
+
+    with kanta.transaction(action="create_user") as data:
+        data.users["uuid-9"] = User(name="Carol")
+
+    await kanta.close()
+
+    assert "Carol" in caplog.text
+
+
+@pytest.mark.asyncio
+async def test_logfmt_state_name_ignores_annotation(tmp_path, format_config, caplog):
+    import logging
+
+    caplog.set_level(logging.INFO, logger="kanta.transaction")
+    path = tmp_path / "test.db"
+    kanta = make_kanta(path, Data, format_config)
+
+    # Matching by name does not check the annotation.
+    @kanta.logfmt
+    def resolve_users(value: str, state: int) -> str | None:
+        return state.get("users", {}).get(value, {}).get("name")
+
+    await kanta.open()
+
+    with kanta.transaction(action="create_user") as data:
+        data.users["uuid-10"] = User(name="Dave")
+
+    await kanta.close()
+
+    assert "Dave" in caplog.text
+
+
+@pytest.mark.asyncio
+async def test_logfmt_tag_takes_precedence_over_name(tmp_path, format_config, caplog):
+    import logging
+
+    caplog.set_level(logging.INFO, logger="kanta.transaction")
+    path = tmp_path / "test.db"
+    kanta = make_kanta(path, Data, format_config)
+
+    @kanta.logfmt
+    def check_prev(value: str, anything: DictPrev) -> str | None:
+        assert anything == {}
+        return None
+
+    @kanta.logfmt
+    def resolve_users(value: str, prev: DictState) -> str | None:
+        # The tag wins: prev receives the current state despite its name.
+        return prev.get("users", {}).get(value, {}).get("name")
+
+    await kanta.open()
+
+    with kanta.transaction(action="create_user") as data:
+        data.users["uuid-11"] = User(name="Erin")
+
+    await kanta.close()
+
+    assert "Erin" in caplog.text
+
+
+@pytest.mark.asyncio
+async def test_logfmt_class_state_attribute(tmp_path, format_config, caplog):
+    import logging
+
+    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.state.get("users", {}).get(value, {}).get("name")
+
+    await kanta.open()
+
+    with kanta.transaction(action="create_user") as data:
+        data.users["uuid-12"] = User(name="Fred")
+
+    await kanta.close()
+
+    assert "Fred" in caplog.text
+
+
 @pytest.mark.asyncio
 async def test_logfmt_class_injection(tmp_path, format_config, caplog):
     import logging

tests/test_cli.py

@@ -0,0 +1,129 @@
+"""Tests for the ``python -m kanta`` CLI output formatting."""
+
+import sys
+from datetime import UTC, datetime
+
+from kanta.__main__ import (
+    _extra_import_paths,
+    _format_ts,
+    _import_dotted,
+    _import_kanta_object,
+    main,
+)
+from kanta.serialization import JsonSerializer
+from kanta.serialization.framing import LineFramer
+from kanta.structs import ChangeRecord, Snapshot
+
+
+def test_format_ts_strips_microseconds():
+    """Timestamps are rendered without microsecond precision."""
+    dt = datetime(2026, 8, 12, 10, 6, 52, 375398, tzinfo=UTC)
+    assert _format_ts(dt) == "2026-08-12 10:06:52"
+
+
+def test_extra_import_paths_are_temporary(tmp_path, monkeypatch):
+    """CWD and nearby venv site-packages are added only for the import block."""
+    parent_dir = tmp_path / "parent"
+    cwd = parent_dir / "child"
+    venv_site = (
+        cwd
+        / ".venv"
+        / "lib"
+        / f"python{sys.version_info.major}.{sys.version_info.minor}"
+        / "site-packages"
+    )
+    venv_site.mkdir(parents=True)
+    parent_venv_site = (
+        parent_dir
+        / ".venv"
+        / "lib"
+        / f"python{sys.version_info.major}.{sys.version_info.minor}"
+        / "site-packages"
+    )
+    parent_venv_site.mkdir(parents=True)
+
+    monkeypatch.chdir(cwd)
+    cwd_str = str(cwd)
+    venv = str(venv_site)
+    parent_venv = str(parent_venv_site)
+
+    before = sys.path.copy()
+    with _extra_import_paths():
+        during = sys.path.copy()
+        assert cwd_str in during
+        assert venv in during
+        assert parent_venv in during
+        assert during.index(cwd_str) < during.index(venv) < during.index(parent_venv)
+    assert sys.path == before
+
+
+def test_extra_import_paths_ignores_other_python_versions(tmp_path, monkeypatch):
+    """Only the site-packages for the running Python version is picked up."""
+    current_site = (
+        tmp_path
+        / ".venv"
+        / "lib"
+        / f"python{sys.version_info.major}.{sys.version_info.minor}"
+        / "site-packages"
+    )
+    other_site = tmp_path / ".venv" / "lib" / "python9.9" / "site-packages"
+    current_site.mkdir(parents=True)
+    other_site.mkdir(parents=True)
+
+    monkeypatch.chdir(tmp_path)
+    with _extra_import_paths():
+        assert str(current_site) in sys.path
+        assert str(other_site) not in sys.path
+
+
+def test_cli_snapshot_line_format(tmp_path, capsys):
+    """Snapshot lines are timestamped and colored with metadata."""
+    path = tmp_path / "test.kantadb"
+    ts = datetime(2026, 8, 12, 10, 6, 52, 375398, tzinfo=UTC)
+    mtime = datetime(2026, 8, 12, 9, 0, 0, tzinfo=UTC)
+    serializer = JsonSerializer()
+    framer = LineFramer()
+
+    snapshot = Snapshot(ts=ts, v=1, m=mtime, state={"counter": 5})
+    change = ChangeRecord(ts=ts, a="inc", v=1, u="user1", diff={"counter": 6})
+    data = framer.frame_snapshot(
+        serializer.encode(snapshot), record_offset=0
+    ) + framer.frame_change(serializer.encode(change), record_offset=0)
+    path.write_bytes(data)
+
+    code = main([str(path)])
+    assert code == 0
+
+    err = capsys.readouterr().err
+    # No microsecond precision anywhere.
+    assert "10:06:52" in err
+    assert "10:06:52.375398" not in err
+
+    # Snapshot line: bright white snapshot/sN, white version/mtime, dark size.
+    assert "\x1b[97msnapshot s0" in err
+    assert "\x1b[38;5;250m v1 2026-08-12 09:00:00" in err
+    assert "\x1b[38;5;242m 13 B" in err
+
+
+def test_import_dotted_from_file_path(tmp_path):
+    """--data can be a filesystem path with an optional colon-separated symbol."""
+    module = tmp_path / "models.py"
+    module.write_text("class Data:\n    pass\n")
+    result = _import_dotted(f"{module}:Data")
+    assert result.__name__ == "Data"
+
+
+def test_import_kanta_object_from_file_path(tmp_path):
+    """--kanta can be a filesystem path; default symbol is ``kanta``."""
+    module = tmp_path / "database.py"
+    module.write_text("class Kanta:\n    pass\nkanta = Kanta()\n")
+    result = _import_kanta_object(str(module))
+    assert type(result).__name__ == "Kanta"
+
+
+def test_import_kanta_object_from_file_path_with_symbol(tmp_path):
+    """--kanta can be a filesystem path with an explicit colon-separated symbol."""
+    module = tmp_path / "database.py"
+    module.write_text("class CustomKanta:\n    pass\nmy_kanta = CustomKanta()\n")
+    result = _import_kanta_object(f"{module}:my_kanta")
+    assert type(result).__name__ == "CustomKanta"

tests/test_kanta_integration.py

@@ -721,6 +721,41 @@ async def test_logmigr_callback_replaces_default_logging(
     assert not info_messages
 
 
+@pytest.mark.asyncio
+async def test_logmigr_callback_report(tmp_path, format_config, caplog):
+    import logging
+
+    from kanta import MigrationReport
+
+    path = tmp_path / "test.db"
+    seed_single_change(path, fixed_change("init", {"counter": 0}), format_config)
+
+    mod = type(sys)("test_migrations_report")
+
+    def migrate_v1(d):
+        """Bump counter."""
+        d["counter"] = 2
+
+    mod.__dict__["migrate_v1"] = migrate_v1
+
+    reports = []
+
+    kanta = make_kanta(path, Data, format_config, migrations=mod)
+
+    @kanta.logmigr
+    def collect(report: MigrationReport):
+        reports.append(report)
+
+    with caplog.at_level(logging.INFO, logger="kanta.migration"):
+        await kanta.open()
+        await kanta.close()
+
+    assert len(reports) == 1
+    assert reports[0].original == 0
+    assert reports[0].version == 1
+    assert [m.name for m in reports[0].applied] == ["migrate_v1"]
+
+
 @pytest.mark.asyncio
 async def test_transaction_log_false_suppresses_log(tmp_path, format_config, caplog):
     path = tmp_path / "test.db"

tests/test_migrations.py

@@ -206,3 +206,22 @@ def test_description_defaults_to_version_when_no_docstring():
 
     result = reg.apply({}, current_version=0, kanta=kanta)
     assert result.migrations[0].description == "v1"
+
+
+def test_report_fields():
+    from kanta import MigrationReport
+
+    reg = Migrations()
+    kanta = _DummyKanta()
+
+    @reg.register
+    def migrate_v1(d):
+        d["x"] = 1
+
+    report = reg.apply({"x": 0}, current_version=0, kanta=kanta)
+    assert isinstance(report, MigrationReport)
+    assert report.original == 0
+    assert report.version == 1
+    assert [m.name for m in report.applied] == ["migrate_v1"]
+    # Deprecated alias still works.
+    assert report.migrations is report.applied