Pretty-print snapshot lines and drop microseconds from CLI timestamps.

kanta/__main__.py

@@ -0,0 +1,490 @@
+"""Module-level CLI for reading a kantadb file and printing its change log."""
+
+from __future__ import annotations
+
+import argparse
+import asyncio
+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`` and return the attribute."""
+    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 _import_kanta_object(path: str) -> Any:
+    """Import a Kanta object by module path.
+
+    If ``path`` names an importable module, look up an object named
+    ``kanta`` in it; otherwise treat ``path`` as ``module.attr`` referring
+    directly to the object.
+    """
+    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 to the root data type (e.g. myapp.models.Data).",
+    )
+    parser.add_argument(
+        "-m",
+        "--migrations",
+        metavar="MOD",
+        help="Dotted path to the migrations module (e.g. myapp.migrations).",
+    )
+    parser.add_argument(
+        "-k",
+        "--kanta",
+        metavar="MOD",
+        help=(
+            "Module path to an existing Kanta object to use: either a module"
+            " containing an object named 'kanta' (e.g. myapp.db) or a dotted"
+            " path to the object itself (e.g. myapp.db.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 _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:
+        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:
+        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/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,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

tests/test_cli.py

@@ -0,0 +1,44 @@
+"""Tests for the ``python -m kanta`` CLI output formatting."""
+
+from datetime import UTC, datetime
+
+from kanta.__main__ import _format_ts, 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_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