Support filesystem paths in --data, --migrations, and --kanta arguments.

demo/demo.py

@@ -85,7 +85,9 @@ async def main() -> None:
                 data.total = 99
                 raise ValueError("simulated failure")
         except ValueError:
-            print(f"\nReset rolled back: {data.total=} (we can always read data without tx)\n")
+            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"]

kanta/__main__.py

@@ -0,0 +1,601 @@
+"""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, migration_result=result),
+            )
+        except Exception:
+            _logger.exception("logmigr callback failed")
+        return
+    if quiet:
+        return
+    descriptions = [
+        f"{m.name} ({m.description})" for m in result.migrations 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/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
+    handler.  When all handlers pass, the *fallback* renders the event;
-    with the built-in formatting.
+    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
+    and the chain falls back to the fallback rendering, and a failure in
-    the built-in formatting itself is reported and swallowed.
+    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)
 
@@ -187,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,
@@ -209,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:
@@ -235,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
 
 
@@ -361,15 +369,21 @@ def _format_change_lines(
         path_str = _format_path(path, logfmt, final_color="add")
         if isinstance(value, dict) and value:
             lines = [str(Line()("  ", path_str, " ").sep("="))]
-            formatted_items = []
             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)
+                keys.append((k, _format_value(k, key_path, max_len=30, logfmt=logfmt)))
-                v_str = _format_value(v, key_path, max_len=30, logfmt=logfmt)
+            field_width = max(displaywidth(kd) for _, kd in keys)
-                formatted_items.append((key_display, v_str))
-            field_width = max(displaywidth(k) for k, _ in formatted_items)
             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))
             return lines + [
                 str(
                     Line()("    ", k).sep(":")(

kanta/replaylog.py

@@ -0,0 +1,390 @@
+"""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"
+                f" ({_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"
+            f" ({_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,11 +70,13 @@ 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
     add = "32"  # Green for additions
     delete = "1;31"  # Bold red for deletions
+    ellipsis = "38;5;242"  # Dark grey for the truncation ellipsis
 
 
 colors = Colors()

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

@@ -0,0 +1,130 @@
+"""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

@@ -562,13 +562,12 @@ async def test_migration_with_changes_records_diff_and_snapshot(
 
     records = read_changes(path, format_config)
     migration_records = [r for r in records if r.a.startswith("migrate")]
-    assert len(migration_records) == 2
+    # 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}
+    assert migration_records[0].diff == {"counter": 2, "users": {}}
-    assert migration_records[1].a == "migrate:msgspec"
-    assert migration_records[1].v == 1
-    assert migration_records[1].diff == {"users": {}}
 
     snap = read_last_snapshot(path, format_config)
     assert snap is not None