README

Kanta output is now visible without any application logging setup: the
event loggers Kanta owns (kanta.bootstrap/migration/transaction) are
set to INFO instead of inheriting the root level, which defaults to
WARNING and hid INFO transaction logs.  The plain kanta diagnostics
logger stays unset so root controls its verbosity.  Tests that relied
on INFO events being suppressed now scope their caplog assertions to
the relevant logger.

README.md

@@ -15,6 +15,8 @@ This design is often preferable when you want low-latency local persistence with
 
 Queries and updates on native data are far faster than over an SQL server connection, and we can can provide fully synchronous operation. The limitation is that you can only use the same database within a single process at a time, but this is well suited for async programming.
 
+See also: [SQL, NoSQL, and why I started Kanta DB](https://vasanko.com/coders/kanta)
+
 ## Quick Start
 
 ```python

docs/database.md

@@ -51,7 +51,7 @@ Fields:
 Snapshot lines are prefixed with `SNAPSHOT `, followed by JSON:
 
 ```text
-SNAPSHOT {"ts":"2026-06-10T00:00:00Z","v":5,"state":{"users":{}},"m":"2026-06-10T00:00:00Z"}
+SNAPSHOT {"ts":"2026-06-10T00:00:00Z","v":5,"state":{"users":{}},"m":"2026-06-10T00:00:00Z","r":12,"k":"0.9.2"}
 ```
 
 Fields:
@@ -59,13 +59,49 @@ Fields:
 - `v`: schema version represented by the snapshot.
 - `state`: full state dictionary.
 - `m`: optional domain modification timestamp.
+- `r`: revision of the snapshotted state — the total number of change records that occurred before this snapshot.
+- `k`: meta format version (integer), the module-level constant `kanta.structs.META_FORMAT` at write time. It is bumped manually when the meta format changes. `0` (a missing field in old files) marks a legacy snapshot whose `r` cannot be trusted; readers then count change records from the start of the file instead. k1 introduced revision numbers.
+
+## Revision Numbers
+
+A revision number identifies a *state position between diffs*, not a diff:
+
+- The empty initial state is revision 0.
+- Each change record — including the bootstrap record — advances the position by one, so after the first record the state is at revision 1.
+- Change records carry no revision of their own; the revision of a state is counted from the preceding snapshot (`snapshot.r + n` records after it), or from the start of the file.
+- Revisions are absolute and survive rotation: the leading snapshot of a rotated file carries the revision of its cutoff state, so `r` values keep referring to the same state positions across history rewrites.
+
+`kanta.rev` exposes the current revision at runtime: the number of change records applied since revision 0, advanced in memory as transactions queue records. Reading it before `open()` raises `DataIntegrityError`. The CLI accepts `rN` range bounds to select records by revision (e.g. `-r r10:r20`): a bound `rN` addresses state N together with the record that produced it — the change record resulting in state N, or, when that change is no longer in the file, the snapshot materializing state N. `-r r5` and `-r r5:r6` therefore show the same record and output state 5, while `-r r5..r6` also includes the next change and outputs state 6. Ranges truncate Python-style instead of erroring: on a rotated file whose leading snapshot holds revision 3, `-r r1:r4` starts at the beginning of the file and shows that snapshot. Change record headers show the revision of the state the record produces as `rN`.
+
+## CLI Range Filtering
+
+`-r` is repeatable and its specs fall into two classes:
+
+- **Unit specs** — any spec with a letter-prefixed bound (`l`, `s`, `v`, `r`, including mixed forms like `2:s3`) — index into the database file and resolve to a line window. Multiple unit specs intersect restrictively: only the common line range is processed.
+- **Plain-number specs** — both bounds (or the single index) are plain numbers — index into the *filtered output*: the records left after all unit specs and all `--grep` patterns have been applied. They are a pure display filter and do not affect replay, migrations, or the `-o` output state. Negative values count from the end of the filtered output.
+
+Like repeated `--grep` patterns (which must all match), repeated ranges are restrictive: `-r r1:r5 -r 0:2` prints only the first two records of those in the revision window, regardless of flag order. Plain ranges truncate Python-style and print nothing when the intersection is empty; a single plain index must exist and fails with exit code 2 otherwise.
+
+`--context N` (default 0) adds GNU grep-style context as the final step: each printed item is expanded by N surrounding records drawn from the records in the processed range, free of the `--grep` and plain-number filters, with overlapping expansions merged so nearby matches do not duplicate context records.
+
+## Meta Format Migrations
+
+The snapshot `k` field carries an integer meta format version (`kanta.structs.META_FORMAT`, currently 1), bumped manually whenever the record-level format around the payloads changes. k1 introduced revision numbers.
+
+When a writable `open()` finds a legacy (k0) last snapshot, the file is rewritten in full under the lock, before replay results are used further (meta migration k1, `kanta/metamigr.py`):
+
+1. A leading snapshot, when the file begins with one, is preserved with its data and timestamps, re-stamped k1 with `r = 0` (the only count knowable from the file).
+2. All change records follow in order, re-framed; redundant intermediate snapshots are dropped.
+3. A closing k1 snapshot captures the final state with `r` equal to the total number of change records.
+
+After the rewrite every snapshot in the file is k1 with a trustworthy `r`, so readers never deal with mixed r/k meanings in one file. Rotation composes with this: it runs first and counts revisions from record positions regardless of `k`, so rotating a legacy file already produces k1 snapshots and no separate migration is needed. A read-only open computes the same revision the migration would have produced but leaves the file untouched.
 
 ## Replay Model
 
 1. Find the last snapshot in the file, if present.
-2. Initialize replay state from snapshot state (or `{}` if none).
-3. Replay subsequent change records in order using patch application.
-4. The final replay state becomes in-memory `kanta.data`.
+2. Initialize replay state from snapshot state (or `{}` if none) and the base revision from the snapshot's `r` (or a count of the preceding change records when the snapshot is a legacy one without `k`).
+3. Replay subsequent change records in order using patch application, counting them toward the revision.
+4. The final replay state becomes in-memory `kanta.data`, and the final count becomes `kanta.rev`.
 
 This model provides fast startup for large logs while retaining append-only history.
 
@@ -174,7 +210,7 @@ def resolve_user_key(value: str) -> str | None:
 - By default a transaction is logged with an `action by user` header followed by the diff lines. Added paths are colored green, deleted paths red. ANSI color codes are stripped after formatting when the standard error stream does not support color: `NO_COLOR` disables colors, `FORCE_COLOR` forces them, otherwise a tty check and a journald (`JOURNAL_STREAM`) check decide. The CLI (`python -m kanta`) strips its output the same way.
 - `kanta.transaction(..., extra=...)` accepts a display-only value that is shown after the action in the header. Anything other than `None` is printed str-converted (colored by Kanta), unless a custom logemit handler does something else with it; it is never persisted in the `ChangeRecord`.
 - `kanta.transaction(..., logdiff=False)` skips building and printing the diff body and logs only the header, which is useful for large or noisy changesets. Diff output can also be disabled globally with `configure_logging(diff=False)`; diff lines are emitted on the `kanta.transaction.diff` child logger so applications can route or silence them separately from the headers.
-- The event loggers `kanta.bootstrap`, `kanta.migration` and `kanta.transaction` are configured at import time (via `configure_logging()`, callable again to change the toggles): a plain stderr handler with no prefix and `propagate = False`, since Kanta renders this output itself. No levels are set, so they inherit the effective root level — a framework switching root between INFO in development and WARNING in production governs Kanta output too. Operational diagnostics (integrity errors, flush failures, rotation notes) use the plain `kanta` logger instead, propagating to the root logger and following the application's normal logging configuration.
+- The event loggers `kanta.bootstrap`, `kanta.migration` and `kanta.transaction` are configured at import time (via `configure_logging()`, callable again to change the toggles): a plain stderr handler with no prefix and `propagate = False`, since Kanta renders this output itself. They are set to INFO so Kanta output is visible without any application logging setup (`debug=True` lifts the DEBUG-emitting bootstrap/migration loggers to DEBUG); only loggers Kanta owns are touched. Operational diagnostics (integrity errors, flush failures, rotation notes) use the plain `kanta` logger instead, which is left unset — propagating to the root logger and following the application's normal logging configuration.
 
 #### Log Emitters
 

docs/rotation.md

@@ -43,13 +43,13 @@ Let `cutoff = now - retention`. All planning happens on the in-memory bytes of t
 4. **Copy the original aside.** The main file is copied with `shutil.copy2` directly to its final `{stem}@{ts}.kantadb` name. On filesystems with copy-on-write this performs a cheap reflink copy. A failure here aborts rotation with the original file intact.
 
 5. **Rewrite the main file in place.** On the locked file descriptor the content is replaced (seek 0, truncate, write, fsync) with, in order:
-   1. A **snapshot of the state at the cutoff**, stamped with the schema version and modification time in effect at the cutoff. Its `ts` is the `ts` of the last pre-cutoff record — the same timestamp used in the rotated filename. This snapshot is the new replay base and carries the version forward so migrations are not re-run; it is always written.
+   1. A **snapshot of the state at the cutoff**, stamped with the schema version and modification time in effect at the cutoff. Its `ts` is the `ts` of the last pre-cutoff record — the same timestamp used in the rotated filename. Its `r` is the absolute revision of the cutoff state (the number of change records before the cutoff position in the original file), so revision numbers keep referring to the same state positions after the rotation. This snapshot is the new replay base and carries the version forward so migrations are not re-run; it is always written.
    2. The retained change records (`ts >= cutoff`), recreated record by record — no internal snapshots are carried over.
-   3. A **final snapshot** of the state after the last retained record, written only when enough changes were retained to warrant one (the same policy as regular snapshot writes). If no records survived the cutoff, the new file consists of the single leading snapshot and nothing else — the steady state for databases whose history has fully aged out.
+   3. A **final snapshot** of the state after the last retained record, written only when enough changes were retained to warrant one (the same policy as regular snapshot writes); its `r` is the total number of change records in the original file. If no records survived the cutoff, the new file consists of the single leading snapshot and nothing else — the steady state for databases whose history has fully aged out.
 
 6. **Trim the rotated copy.** The rotated file is truncated to `cutoff_end` bytes, so it contains only the dropped history and does not duplicate the records retained in the main file. The cut is at a frame boundary, so the rotated file remains a valid, replayable database on its own. This happens only after step 5's fsync, so until then the rotated file still holds the complete original content as a crash-recovery anchor.
 
-7. **Continue normal open.** Replay and migrations proceed on the same locked file. Because the leading snapshot carries the current version, migrations run exactly as they would have against the old content.
+7. **Continue normal open.** Replay and migrations proceed on the same locked file. Because the leading snapshot carries the current version, migrations run exactly as they would have against the old content. The rotation's snapshots are stamped with the current meta format version (`k`), so rotating a legacy (pre-revision) file also performs the k1 meta migration — revision numbers are counted from record positions before the rotation (see [Database format](database.md#meta-format-migrations)).
 
 Failure rule: any error before step 5 leaves the main file byte-identical (only an extra copy exists). A crash during step 5 may leave the main file torn, but the rotated copy still holds the complete original content — recovery is copying it back. After step 6 the split is complete and both files are consistent.
 

kanta/__main__.py

@@ -18,7 +18,7 @@ import msgspec
 from kanta import Kanta
 from kanta.callbacks import InjectionContext, callback_error_reporter
 from kanta.exceptions import DatabaseError, DataIntegrityError, ReplayError
-from kanta.grep import GrepPattern, evaluate
+from kanta.grep import GrepPattern, evaluate, matches_snapshot
 from kanta.logging import (
     LogEvent,
     emit_event,
@@ -27,6 +27,7 @@ from kanta.logging import (
     migration_logger,
 )
 from kanta.replaylog import (
+    Event,
     RangeNotFoundError,
     Selection,
     SnapshotEvent,
@@ -36,7 +37,8 @@ from kanta.replaylog import (
     record_label,
     replay_events,
     scan_events,
-    select,
+    select_all,
+    slice_items,
 )
 from kanta.serialization import Serializer, dict_to_struct, struct_to_dict
 from kanta.structs import ChangeRecord, Snapshot
@@ -235,12 +237,22 @@ def _parse_args(argv: list[str] | None = None) -> argparse.Namespace:
     parser.add_argument(
         "-r",
         "--range",
+        action="append",
         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"
+            "Python-style range to process; repeatable.  Unit bounds index"
+            " into the database file: lN = line number, sN = snapshot,"
+            " vN = version, rN = revision (rN is the record that produced"
+            " state N, with its state).  Plain numbers instead index into"
+            " the filtered output: they select among the records left after"
+            " all unit ranges and --grep patterns, like a display filter,"
+            " without affecting the output state.  Repeated ranges are"
+            " restrictive: only the common subset is printed.  Negative"
+            " snapshot values count from the end (s-1 is the last"
+            " snapshot); negative plain numbers count from the end of the"
+            " filtered output.  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'."
+            " '2:5', '2..5', 'l10:l20', 's1:s3', 'v0:v2', 'r10:r20', 's-1:',"
+            " ':-1', '-1', '-r r1:r5 -r 0'."
         ),
     )
     parser.add_argument(
@@ -268,7 +280,22 @@ def _parse_args(argv: list[str] | None = None) -> argparse.Namespace:
             " '@example.com'."
         ),
     )
+    parser.add_argument(
+        "--context",
+        type=int,
+        default=0,
+        metavar="N",
+        help=(
+            "Print N surrounding records around each printed item, GNU grep"
+            " style.  Applied last, after --grep filtering and plain-number"
+            " -r slicing: context records are drawn from the records in the"
+            " processed range, free of those filters, and overlapping"
+            " expansions merge.  Without --grep this has no visible effect."
+        ),
+    )
     args = parser.parse_args(argv)
+    if args.context < 0:
+        parser.error("--context must not be negative")
     if args.kanta and (args.data or args.migrations):
         parser.error("-k/--kanta cannot be used together with -d or -m")
     return args
@@ -280,6 +307,7 @@ def _print_change_log(
     previous: dict[str, Any],
     current: dict[str, Any],
     kanta: Kanta[Any],
+    revision: int,
     highlight: Any = None,
 ) -> None:
     """Log a single change record to stderr.
@@ -287,8 +315,10 @@ def _print_change_log(
     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.
-    ``highlight`` is an optional :class:`kanta.grep.GrepHighlighter` with
-    the record's matched regions, applied to the fallback rendering.
+    The fallback header shows the revision of the state the record produces
+    as ``r<N>``.  ``highlight`` is an optional
+    :class:`kanta.grep.GrepHighlighter` with the record's matched regions,
+    applied to the fallback rendering.
     """
     event = record_change_event(record, previous, current, kanta)
 
@@ -301,6 +331,7 @@ def _print_change_log(
             lines = format_diff(ev.diff, ev.previous, ev.logfmt, highlight=highlight)
         else:
             header, lines = ev.header, ev.diff_lines
+        header += str(Line().target(f" r{revision}"))
         if not lines:
             _print(f"{label} {ts} {header}")
         elif len(lines) == 1:
@@ -377,8 +408,8 @@ def _print_snapshot_indicator(
 ) -> 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.
+    ``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}")
@@ -500,12 +531,13 @@ async def _run(args: argparse.Namespace) -> int:
             line: idx for idx, line in enumerate(_snapshot_lines(events))
         }
 
-        # Resolve -r into a line range or a single snapshot selection.
+        # Resolve -r specs into a line range (unit specs) or a single
+        # snapshot selection, plus plain specs applied to the output items.
         try:
-            selection = (
-                select(args.range, events, change_count)
-                if args.range is not None
-                else Selection(0, end_of_file(events))
+            selection, plain_specs = (
+                select_all(args.range, events, change_count)
+                if args.range
+                else (Selection(0, end_of_file(events)), [])
             )
         except RangeNotFoundError as exc:
             raise _CliError(str(exc), EXIT_RANGE_ERROR) from exc
@@ -527,26 +559,33 @@ async def _run(args: argparse.Namespace) -> int:
                 )
                 _print()
         else:
-            # Replay up to the range end, printing logs within the range.
+            # Replay up to the range end, collecting the in-window item pool
+            # and tracking which items survive --grep.
             state = {}
             version = 0
-            printed = False
+            pool: list[tuple[Event, dict[str, Any] | None, dict[str, Any], Any]] = []
+            matched: list[int] = []
             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)
+                highlight = None
                 if isinstance(event, SnapshotEvent):
-                    _print_snapshot_indicator(
-                        label,
-                        event.snap,
-                        snapshot_line_to_index[event.line_number],
-                        kanta._impl.serializer,
-                    )
+                    is_match = True
+                    if grep_patterns:
+                        logfmt = kanta._impl.callback_registry.build_logfmt(
+                            InjectionContext(
+                                kanta=kanta,
+                                previous_state=current,
+                                current_state=current,
+                            )
+                        )
+                        is_match = matches_snapshot(
+                            current, grep_patterns, logfmt=logfmt
+                        )
                 else:
                     assert previous is not None
-                    highlight = None
                     if grep_patterns:
                         # Build the same logfmt the rendering uses, so both
                         # raw and prettified values are matched.
@@ -560,13 +599,50 @@ async def _run(args: argparse.Namespace) -> int:
                         highlight = evaluate(
                             event.record, previous, grep_patterns, logfmt=logfmt
                         )
-                        if highlight is None:
-                            continue
-                    _print_change_log(
-                        label, event.record, previous, current, kanta, highlight
+                    is_match = highlight is not None or not grep_patterns
+                pool.append((event, previous, current, highlight if is_match else None))
+                if is_match:
+                    matched.append(len(pool) - 1)
+            # Plain -r specs slice the filtered output items.
+            if plain_specs:
+                try:
+                    item_start, item_end = slice_items(plain_specs, len(matched))
+                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
+                matched = matched[item_start:item_end]
+            # --context expands around the surviving items within the pool,
+            # free of the grep and plain-spec filters; overlaps merge.
+            if args.context and matched:
+                expanded: set[int] = set()
+                for idx in matched:
+                    lo = max(0, idx - args.context)
+                    hi = min(len(pool), idx + args.context + 1)
+                    expanded.update(range(lo, hi))
+                matched = sorted(expanded)
+            items = [pool[idx] for idx in matched]
+            for event, previous, current, highlight in items:
+                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,
                     )
-                printed = True
-            if printed:
+                else:
+                    assert previous is not None
+                    _print_change_log(
+                        label,
+                        event.record,
+                        previous,
+                        current,
+                        kanta,
+                        event.revision,
+                        highlight,
+                    )
+            if items:
                 _print()
 
         # Apply optional migrations to the range-end state.

kanta/grep.py

@@ -376,6 +376,48 @@ class GrepHighlighter:
         return self._apply_marks(text, self._meta_marks.get(field, []))
 
 
+def _match_entry(
+    pattern: GrepPattern, entry: _Entry, logfmt: Any
+) -> tuple[frozenset[int] | None, _ValueMark | None] | None:
+    """Match one pattern against one entry, returning its match marks.
+
+    Returns ``None`` when the entry does not match.  Otherwise returns the
+    matched path-element indices and/or the value mark, ready for
+    :meth:`GrepHighlighter.add`.
+    """
+    pretty = None
+    if logfmt is not None:
+        pretty = logfmt(entry.raw, ".".join(entry.path))
+    if pattern.term is not None:
+        elements = _match_path(pattern.term, entry.path)
+        vmark = _dual_mark(pattern.term, entry.raw, entry.text, pretty)
+        if elements is None and vmark is None:
+            return None
+    else:
+        elements = _match_path(pattern.path_term or "", entry.path)
+        vmark = _dual_mark(pattern.value_term or "", entry.raw, entry.text, pretty)
+        if elements is None or vmark is None:
+            return None
+    return elements, vmark
+
+
+def matches_snapshot(
+    state: dict, patterns: list[GrepPattern], logfmt: Any = None
+) -> bool:
+    """Match *patterns* against a snapshot's full state.
+
+    The state is flattened into the same ``(path, value)`` entries change
+    records are matched against, so path and value matching semantics are
+    identical; a snapshot simply has no action or user to match.  Returns
+    whether every pattern matched somewhere in the state.
+    """
+    entries = list(_leaf_entries([], unmarshal(state), None))
+    for pattern in patterns:
+        if not any(_match_entry(pattern, entry, logfmt) for entry in entries):
+            return False
+    return True
+
+
 def evaluate(
     record: ChangeRecord,
     previous: dict | None,
@@ -411,23 +453,11 @@ def evaluate(
                     highlighter.add_meta("user", mark)
                     matched = True
         for entry in entries:
-            pretty = None
-            if logfmt is not None:
-                pretty = logfmt(entry.raw, ".".join(entry.path))
-            if pattern.term is not None:
-                elements = _match_path(pattern.term, entry.path)
-                vmark = _dual_mark(pattern.term, entry.raw, entry.text, pretty)
-                if elements is None and vmark is None:
-                    continue
-            else:
-                elements = _match_path(pattern.path_term or "", entry.path)
-                vmark = _dual_mark(
-                    pattern.value_term or "", entry.raw, entry.text, pretty
-                )
-                if elements is None or vmark is None:
-                    continue
+            match = _match_entry(pattern, entry, logfmt)
+            if match is None:
+                continue
             matched = True
-            highlighter.add(entry, elements, vmark)
+            highlighter.add(entry, *match)
         if not matched:
             return None
     return highlighter

kanta/kanta.py

@@ -7,6 +7,7 @@ from pathlib import Path
 from types import ModuleType, SimpleNamespace
 from typing import Any, Generic, TypeVar
 
+from kanta.exceptions import DataIntegrityError
 from kanta.kantaimpl import KantaImpl
 from kanta.serialization import JsonSerializer, Serializer
 from kanta.transaction import transaction as _transaction
@@ -146,6 +147,30 @@ class Kanta(Generic[T]):
         """
         return self._impl.ctx
 
+    @property
+    def rev(self) -> int:
+        """Current revision: the state position as a count of change records.
+
+        Revision numbers identify state positions *between* diffs, not the
+        diffs themselves: the empty initial state is revision 0, and each
+        change record (including the bootstrap record) advances the position
+        by one.  Snapshots persist the revision of the state they contain in
+        their ``r`` field; change records carry no revision of their own.
+
+        Returns:
+            The number of change records applied since revision 0.
+
+        Raises:
+            kanta.exceptions.DataIntegrityError: If the instance is not open.
+        """
+        if not self._impl.opened:
+            raise DataIntegrityError(
+                "Kanta instance must be opened before reading rev",
+                db_path=self._impl.filename,
+                action="rev",
+            )
+        return self._impl.rev
+
     @property
     def mtime(self) -> datetime | None:
         """Last modification time carried forward from change records.

kanta/kantaimpl.py

@@ -19,6 +19,7 @@ from kanta.logging import (
     emit_event,
     migration_logger,
 )
+from kanta.metamigr import migrate_k1
 from kanta.migrations import MigrationReport, Migrations
 from kanta.persistence import PersistenceMixin
 from kanta.rotation import execute_rotation, plan_rotation
@@ -217,6 +218,26 @@ class KantaImpl(PersistenceMixin, Generic[T]):
                     cause_type=type(e).__name__,
                 ) from e
 
+            # Meta migration k0 → k1: a file whose last snapshot is legacy
+            # is rewritten in full so every snapshot gains a trustworthy r
+            # (see kanta/metamigr.py).  Rotation, when it ran, already
+            # produced k1 snapshots with positionally counted revisions, so
+            # this only fires when no rotation happened.  Read-only opens
+            # skip the rewrite; replay counted the same revision either way.
+            meta_migrated_at: datetime | None = None
+            if rr.legacy_snapshot and not self.readonly:
+                meta_migrated_at = self.now()
+                new_content = migrate_k1(
+                    content,
+                    framer=self.framer,
+                    serializer=self.serializer,
+                    state=rr.state,
+                    version=rr.version,
+                    m=rr.m,
+                    now=meta_migrated_at,
+                )
+                await asyncio.to_thread(self.file.replace_content, new_content)
+
             migration_report = None
             state_before_migrations = None
             previous_version = rr.version
@@ -230,9 +251,13 @@ class KantaImpl(PersistenceMixin, Generic[T]):
             migrations_ran = rr.version != previous_version
 
             self.snapshot.ts = (
-                datetime.fromtimestamp(rr.last_snapshot_mtime, UTC)
-                if rr.last_snapshot_mtime is not None
-                else None
+                meta_migrated_at
+                if meta_migrated_at is not None
+                else (
+                    datetime.fromtimestamp(rr.last_snapshot_mtime, UTC)
+                    if rr.last_snapshot_mtime is not None
+                    else None
+                )
             )
 
             self.statedict = copy.deepcopy(
@@ -258,6 +283,7 @@ class KantaImpl(PersistenceMixin, Generic[T]):
                     raise
             self.version = rr.version
             self.mtime = rr.m
+            self.rev = rr.rev
             if log is not False and not migrations_ran:
                 logger = log if isinstance(log, logging.Logger) else bootstrap_logger
                 emit_event(
@@ -314,6 +340,7 @@ class KantaImpl(PersistenceMixin, Generic[T]):
                         self.statedict,
                         m=self.mtime,
                         now=self.now,
+                        r=self.rev,
                     )
         elif self.readonly:
             self.opened = False

kanta/logging.py

@@ -605,10 +605,11 @@ def configure_logging(
     and ``kanta.transaction`` carry Kanta-rendered output (colored headers,
     diffs) and print it bare through a plain stderr handler with
     ``propagate = False``.  Diagnostic messages use the plain ``kanta``
-    logger and follow the application's root logging configuration.
+    logger, which is left untouched so the application's root logging
+    configuration controls its verbosity.
 
-    No levels are set by default: the event loggers inherit the effective
-    level of the root logger.
+    The event loggers are set to ``INFO`` so Kanta output is visible without
+    any application logging setup; only loggers Kanta owns are touched.
 
     Args:
         skiproot: If ``True`` (default), event loggers print through Kanta's
@@ -625,7 +626,7 @@ def configure_logging(
         debug: Whether to set the event loggers that emit DEBUG-level output
             (bootstrap and migration) to ``DEBUG``, revealing output such as
             the file-opened summary and migration diffs.  ``False`` resets
-            them to inheriting the root level.
+            all event loggers to ``INFO``.
     """
     logging.getLogger("kanta.transaction.diff").disabled = not diff
 
@@ -636,11 +637,11 @@ def configure_logging(
     ):
         logging.getLogger(name).disabled = not enabled
 
-    for name in _DEBUG_LOGGERS:
-        logging.getLogger(name).setLevel(logging.DEBUG if debug else logging.NOTSET)
-
     for name in EVENT_LOGGERS:
         logger = logging.getLogger(name)
+        logger.setLevel(
+            logging.DEBUG if debug and name in _DEBUG_LOGGERS else logging.INFO
+        )
         if skiproot:
             logger.propagate = False
             _ensure_plain_handler(logger)

kanta/metamigr.py

@@ -0,0 +1,75 @@
+"""Meta format migrations: rewriting files between snapshot ``k`` versions.
+
+The k0 → k1 migration introduces revision numbers.  Like a rotation, the
+whole file is rewritten under the lock so that every snapshot in it is a
+k1 snapshot with a trustworthy ``r`` — mixed r/k meanings never persist in
+one file.  Unlike a rotation, all change records are kept; redundant
+intermediate snapshots are dropped.
+"""
+
+from __future__ import annotations
+
+from datetime import datetime
+from typing import Any
+
+from kanta.structs import META_FORMAT, ChangeRecord, Snapshot
+from kanta.serialization.base import Serializer
+from kanta.serialization.framing import Framer
+
+
+def migrate_k1(
+    content: bytes,
+    *,
+    framer: Framer,
+    serializer: Serializer,
+    state: dict[str, Any],
+    version: int,
+    m: datetime | None,
+    now: datetime,
+) -> bytes:
+    """Rewrite *content* wholly into k1 form.
+
+    The leading snapshot, when the file begins with one (the file start is
+    not the database start), is preserved with its data and timestamps,
+    re-stamped k1; its ``r`` is 0, the only count knowable from the file.
+    All change records follow, re-framed in order, and a closing snapshot
+    captures the replayed final state with ``r`` equal to the total number
+    of change records.  *state*, *version* and *m* are the replayed final
+    values (pre-schema-migration, so migrations still run afterwards).
+
+    The caller gates on a legacy last snapshot and performs the in-place
+    rewrite under the file lock.
+    """
+    entries = []
+    for is_snapshot, payload, _line, _pos in framer.iter_records(content, 0):
+        entries.append(
+            (
+                is_snapshot,
+                serializer.decode(
+                    payload, type=Snapshot if is_snapshot else ChangeRecord
+                ),
+            )
+        )
+
+    out = bytearray()
+    if entries and entries[0][0]:
+        snap = entries[0][1]
+        assert isinstance(snap, Snapshot)
+        leading = Snapshot(
+            ts=snap.ts, v=snap.v, state=snap.state, m=snap.m, r=0, k=META_FORMAT
+        )
+        out += framer.frame_snapshot(serializer.encode(leading), record_offset=0)
+
+    n_changes = 0
+    for is_snapshot, record in entries:
+        if is_snapshot:
+            continue
+        out += framer.frame_change(serializer.encode(record), record_offset=len(out))
+        n_changes += 1
+
+    if n_changes:
+        closing = Snapshot(
+            ts=now, v=version, state=state, m=m, r=n_changes, k=META_FORMAT
+        )
+        out += framer.frame_snapshot(serializer.encode(closing), record_offset=len(out))
+    return bytes(out)

kanta/persistence.py

@@ -40,6 +40,7 @@ class PersistenceMixin:
     background_error: DatabaseError | None
     flush_interval: float
     version: int
+    rev: int
     opened: bool
     readonly: bool
     mtime: datetime | None
@@ -64,6 +65,7 @@ class PersistenceMixin:
         self.background_error = None
         self.flush_interval = flush_interval
         self.version = 0
+        self.rev = 0
         self.mtime: datetime | None = None
         self.clock: Callable[[], datetime] | None = None
 
@@ -123,7 +125,12 @@ class PersistenceMixin:
     def maybe_snapshot(self) -> None:
         """Evaluate and possibly write a snapshot from current state."""
         self.snapshot.maybe_write(
-            self.file, self.version, self.statedict, m=self.mtime, now=self.now
+            self.file,
+            self.version,
+            self.statedict,
+            m=self.mtime,
+            now=self.now,
+            r=self.rev,
         )
 
     def queue_change(
@@ -180,6 +187,7 @@ class PersistenceMixin:
         )
         self.pending_changes.append(record)
         self.statedict = copy.deepcopy(current)
+        self.rev += 1
         if m is not None:
             self.mtime = m
         return record

kanta/replaylog.py

@@ -33,6 +33,7 @@ class SnapshotEvent:
     byte_pos: int
     record_index: int
     snap: Snapshot
+    revision: int = 0
 
     @property
     def version(self) -> int:
@@ -47,6 +48,7 @@ class ChangeEvent:
     byte_pos: int
     record_index: int
     record: ChangeRecord
+    revision: int = 0
 
     @property
     def version(self) -> int:
@@ -96,6 +98,7 @@ def scan_events(content: bytes, kanta: Kanta[Any]) -> tuple[list[Event], int]:
     state: dict[str, Any] = {}
     events: list[Event] = []
     change_count = 0
+    revision = 0
     for is_snapshot, payload, line_number, byte_pos in impl.framer.iter_records(
         content, 0
     ):
@@ -113,12 +116,22 @@ def scan_events(content: bytes, kanta: Kanta[Any]) -> tuple[list[Event], int]:
                         record_type="snapshot",
                     )
                 state = snap.state
-                events.append(SnapshotEvent(line_number, byte_pos, record_index, snap))
+                # A k-marked snapshot carries an absolute revision, which may
+                # start past zero (rotated files).  Legacy snapshots leave the
+                # file-relative count in effect, which is exact for them.
+                if snap.k:
+                    revision = snap.r
+                events.append(
+                    SnapshotEvent(line_number, byte_pos, record_index, snap, revision)
+                )
             else:
                 record = impl.serializer.decode(payload, type=ChangeRecord)
                 state = patch(state, record.diff)
-                events.append(ChangeEvent(line_number, byte_pos, record_index, record))
                 change_count += 1
+                revision += 1
+                events.append(
+                    ChangeEvent(line_number, byte_pos, record_index, record, revision)
+                )
         except msgspec.DecodeError as exc:
             raise ReplayError(
                 f"Parse error at {label}: {exc}",
@@ -224,6 +237,21 @@ def _version_lines(events: list[Event]) -> dict[int, int]:
     return lines
 
 
+def _revision_lines(events: list[Event]) -> dict[int, int]:
+    """Map each selectable revision to the line of the record that produced it.
+
+    A revision bound ``rN`` addresses state N together with the record that
+    produced it: the change record whose resulting revision is N, or — when
+    that change is no longer in the file — the snapshot that materializes
+    state N.  A change takes precedence over a redundant snapshot of the
+    same revision (file order).
+    """
+    lines: dict[int, int] = {}
+    for event in events:
+        lines.setdefault(event.revision, 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:
@@ -236,7 +264,7 @@ 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"}
+    unit_map = {"l": "line", "s": "snapshot", "v": "version", "r": "revision"}
     if bound_str[0] in unit_map:
         unit = unit_map[bound_str[0]]
         rest = bound_str[1:]
@@ -287,6 +315,16 @@ def _bound_to_line(
             raise ValueError("version numbers do not support negative indexing")
         return _version_lines(events).get(value, eof)
 
+    if unit == "revision":
+        if value < 0:
+            raise ValueError("revision numbers do not support negative indexing")
+        lines = _revision_lines(events)
+        if value in lines:
+            return lines[value]
+        if not lines or value < min(lines):
+            return 0
+        return eof
+
     raise ValueError(f"unknown range unit: {unit}")
 
 
@@ -368,6 +406,23 @@ def select(spec: str, events: list[Event], total: int) -> Selection:
         later = [line for line in lines.values() if line > start_line]
         return Selection(start_line, min(later) if later else end_of_file(events))
 
+    if unit == "revision":
+        _negative_check(unit, value)
+        lines = _revision_lines(events)
+        if value not in lines:
+            raise RangeNotFoundError(
+                f"Revision {spec!r} not found in file"
+                f" ({_plural(len(lines), 'revision')})"
+            )
+        start_line = lines[value]
+        event = _event_at_line(events, start_line)
+        if isinstance(event, SnapshotEvent):
+            # The producing change is no longer in the file; the snapshot is
+            # the record materializing this state.
+            return Selection(start_line, start_line + 1, event)
+        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:
@@ -376,3 +431,76 @@ def select(spec: str, events: list[Event], total: int) -> Selection:
         )
     end_line = lines[idx + 1] if idx + 1 < total else end_of_file(events)
     return Selection(lines[idx], end_line)
+
+
+def is_plain_spec(spec: str) -> bool:
+    """Return True if every bound of the spec is a plain number (no unit letter).
+
+    Plain specs index into the filtered output items rather than the file.
+    """
+    if ".." in spec:
+        bounds = spec.split("..", 1)
+    elif ":" in spec:
+        bounds = spec.split(":", 1)
+    else:
+        bounds = [spec]
+    return all(_parse_bound(b)[0] == "change" for b in bounds)
+
+
+def select_all(
+    specs: list[str], events: list[Event], total: int
+) -> tuple[Selection, list[str]]:
+    """Resolve repeated ``-r`` specs into a database window plus plain specs.
+
+    Unit specs (``l``, ``s``, ``v``, ``r``, and mixed specs) resolve to line
+    windows via :func:`select` and intersect restrictively.  The snapshot
+    shortcut applies only to a single unit spec.  Plain-number specs are
+    returned unresolved for the caller to apply against the filtered output.
+    """
+    plain_specs = [spec for spec in specs if is_plain_spec(spec)]
+    unit_specs = [spec for spec in specs if not is_plain_spec(spec)]
+    if not unit_specs:
+        return Selection(0, end_of_file(events)), plain_specs
+    selections = [select(spec, events, total) for spec in unit_specs]
+    if len(selections) == 1:
+        return selections[0], plain_specs
+    start = max(sel.start_line for sel in selections)
+    end = min(sel.end_line for sel in selections)
+    return Selection(min(start, end), end), plain_specs
+
+
+def _clamp_index(value: int, count: int) -> int:
+    """Clamp an item index Python-slice style; negatives count from the end."""
+    if value < 0:
+        value = count + value
+    return max(0, min(value, count))
+
+
+def slice_items(specs: list[str], count: int) -> tuple[int, int]:
+    """Resolve plain item specs into ``[start, end)`` indices over ``count`` items.
+
+    Multiple specs intersect restrictively; ranges truncate Python-slice
+    style.  A single index must exist and raises :class:`RangeNotFoundError`
+    otherwise.  Syntax errors raise :class:`ValueError`.
+    """
+    start, end = 0, count
+    for spec in specs:
+        if ":" in spec or ".." in spec:
+            sep = ".." if ".." in spec else ":"
+            start_str, end_str = spec.split(sep, 1)
+            item_start = _clamp_index(int(start_str), count) if start_str else 0
+            item_end = _clamp_index(int(end_str), count) if end_str else count
+            if sep == ".." and end_str:
+                item_end = min(item_end + 1, count)
+        else:
+            value = int(spec)
+            idx = count + value if value < 0 else value
+            if not 0 <= idx < count:
+                raise RangeNotFoundError(
+                    f"Item index {spec!r} not found in filtered output"
+                    f" ({_plural(count, 'item')})"
+                )
+            item_start, item_end = idx, idx + 1
+        start = max(start, item_start)
+        end = min(end, item_end)
+    return start, max(start, end)

kanta/rotation.py

@@ -16,7 +16,7 @@ from pathlib import Path
 from typing import Any
 
 from kanta.exceptions import DatabaseError
-from kanta.structs import ChangeRecord, Snapshot
+from kanta.structs import META_FORMAT, ChangeRecord, Snapshot
 from kanta.serialization.base import Serializer, apply_diff
 from kanta.serialization.framing import Framer
 
@@ -174,10 +174,18 @@ def plan_rotation(
     # Build the new content: leading cutoff snapshot, retained changes
     # re-framed at fresh offsets, and a final snapshot when enough changes
     # survived to warrant one (mirrors the regular snapshot policy).
+    # Revisions are absolute positions counted from the start of the
+    # original file, so they survive the rotation unchanged.
+    r_at_cutoff = sum(1 for e in changes if e.byte_pos < cutoff_end)
     out = bytearray()
     leading = serializer.encode(
         Snapshot(
-            ts=rotated_ts, v=version_at_cutoff, state=state_at_cutoff, m=m_at_cutoff
+            ts=rotated_ts,
+            v=version_at_cutoff,
+            state=state_at_cutoff,
+            m=m_at_cutoff,
+            r=r_at_cutoff,
+            k=META_FORMAT,
         )
     )
     out += framer.frame_snapshot(leading, record_offset=0)
@@ -186,7 +194,14 @@ def plan_rotation(
         out += framer.frame_change(payload, record_offset=len(out))
     if len(retained) >= min_diffs:
         closing = serializer.encode(
-            Snapshot(ts=now, v=final_version, state=state, m=final_m)
+            Snapshot(
+                ts=now,
+                v=final_version,
+                state=state,
+                m=final_m,
+                r=len(changes),
+                k=META_FORMAT,
+            )
         )
         out += framer.frame_snapshot(closing, record_offset=len(out))
 

kanta/serialization/base.py

@@ -25,12 +25,16 @@ class ReplayResult:
         has_migration: bool = False,
         last_snapshot_mtime: float | None = None,
         m: datetime | None = None,
+        rev: int = 0,
+        legacy_snapshot: bool = False,
     ):
         self.state = state
         self.version = version
         self.has_migration = has_migration
         self.last_snapshot_mtime = last_snapshot_mtime
         self.m = m
+        self.rev = rev
+        self.legacy_snapshot = legacy_snapshot
 
 
 class Serializer(Protocol):
@@ -61,6 +65,8 @@ def replay(
     last_snapshot_mtime: float | None = None
     m: datetime | None = None
     has_migration = False
+    r_base = 0
+    legacy_snapshot = False
 
     if snap_payload is not None:
         try:
@@ -76,7 +82,21 @@ def replay(
         version = snap.v
         last_snapshot_mtime = snap.ts.timestamp()
         m = snap.m
+        if snap.k == 0:
+            # Legacy snapshot: r cannot be trusted, so count change records
+            # from the start of the file (framing only, no payload decode).
+            legacy_snapshot = True
+            r_base = sum(
+                1
+                for is_snapshot, _payload, _line, byte_pos in framer.iter_records(
+                    data, 0
+                )
+                if not is_snapshot and byte_pos < snap_byte_pos
+            )
+        else:
+            r_base = snap.r
 
+    changes = 0
     for is_snapshot, payload, line_number, byte_pos in framer.iter_records(
         data, offset
     ):
@@ -111,12 +131,15 @@ def replay(
             m = change.m
         version = change.v
         state = _patch_state(state, change.diff)
+        changes += 1
     return ReplayResult(
         state=state,
         version=version,
         has_migration=has_migration,
         last_snapshot_mtime=last_snapshot_mtime,
         m=m,
+        rev=r_base + changes,
+        legacy_snapshot=legacy_snapshot,
     )
 
 

kanta/snapshot.py

@@ -6,7 +6,7 @@ import logging
 from collections.abc import Callable
 from datetime import UTC, datetime
 
-from kanta.structs import Snapshot
+from kanta.structs import META_FORMAT, Snapshot
 from kanta.serialization import JsonSerializer, Serializer
 from kanta.serialization.framing import Framer, LineFramer
 
@@ -50,6 +50,7 @@ class SnapshotState:
         state: dict,
         m: datetime | None = None,
         now: Callable[[], datetime] | None = None,
+        r: int = 0,
     ) -> None:
         """Write snapshot when thresholds/time policy allows it."""
         force = self._force_pending
@@ -66,16 +67,24 @@ class SnapshotState:
         if not file.is_open:
             return
         try:
-            self._write(file, version, state, ts, m=m)
+            self._write(file, version, state, ts, m=m, r=r)
             self._force_pending = False
         except Exception as exc:
             _logger.error("Kanta snapshot failed: %r", exc)
 
     def _write(
-        self, file, version: int, state: dict, now: datetime, m: datetime | None = None
+        self,
+        file,
+        version: int,
+        state: dict,
+        now: datetime,
+        m: datetime | None = None,
+        r: int = 0,
     ) -> None:
         """Write a snapshot and update internal state."""
-        payload = self._serializer.encode(Snapshot(ts=now, v=version, state=state, m=m))
+        payload = self._serializer.encode(
+            Snapshot(ts=now, v=version, state=state, m=m, r=r, k=META_FORMAT)
+        )
         record_offset = file.size() if hasattr(file, "size") else 0
         file.write(self._framer.frame_snapshot(payload, record_offset=record_offset))
         self.changes = 0

kanta/structs.py

@@ -5,6 +5,11 @@ from typing import Any
 
 import msgspec
 
+# On-disk meta format version, stored in the snapshot ``k`` field and bumped
+# manually when the meta format changes.  k0 (missing field) predates
+# revision numbers; k1 introduces them (see docs/database.md).
+META_FORMAT = 1
+
 
 class ChangeRecord(msgspec.Struct, omit_defaults=True, kw_only=True):
     """A single change record in the JSONL log.
@@ -27,9 +32,23 @@ class ChangeRecord(msgspec.Struct, omit_defaults=True, kw_only=True):
 
 
 class Snapshot(msgspec.Struct, omit_defaults=True):
-    """Full state snapshot embedded in the JSONL log."""
+    """Full state snapshot embedded in the JSONL log.
+
+    Attributes:
+        ts: Snapshot creation time.
+        v: Schema version represented by the snapshot.
+        state: Full state dictionary.
+        m: Optional domain modification timestamp.
+        r: Revision of the snapshotted state: the total number of change
+            records that occurred before this snapshot.
+        k: Meta format version (``META_FORMAT`` when written).  0 marks a
+            legacy snapshot whose ``r`` cannot be trusted; readers count
+            change records from the start of the file instead.
+    """
 
     ts: datetime
     v: int
     state: dict[str, Any]
     m: datetime | None = None
+    r: int = 0
+    k: int = 0

tests/test_cli.py

@@ -108,6 +108,9 @@ def test_cli_snapshot_line_format(tmp_path, capsys, monkeypatch):
     assert "\x1b[38;5;250m v1 2026-08-12 09:00:00" in err
     assert "\x1b[38;5;242m 13 B" in err
 
+    # The change record header carries the revision it produces.
+    assert "\x1b[38;5;250m r1" in err
+
 
 def test_cli_strips_ansi_without_color_support(tmp_path, capsys, monkeypatch):
     """Without a tty and with NO_COLOR set, output contains no ANSI codes."""

tests/test_grep.py

@@ -11,6 +11,7 @@ from kanta.grep import (
     _match_value,
     evaluate,
     mark_spans,
+    matches_snapshot,
 )
 from kanta.logging import _USER_PATH
 from kanta.serialization import JsonSerializer
@@ -227,6 +228,29 @@ def test_highlighter_meta_marks_action_and_user():
     assert hl.meta("extra", "other") == "extra"
 
 
+def test_matches_snapshot_uses_change_record_matching():
+    state = {"users": {"alice": {"email": "alice@example.com", "age": 30}}}
+    assert matches_snapshot(state, _patterns("users.alice"))
+    assert matches_snapshot(state, _patterns("alice@example"))
+    assert matches_snapshot(state, _patterns("age=30"))
+    assert matches_snapshot(state, _patterns("users.alice", "=30"))
+    assert not matches_snapshot(state, _patterns("bob"))
+    assert not matches_snapshot(state, _patterns("3"))  # not a full number match
+    assert not matches_snapshot(state, _patterns("alice", "missing"))
+    assert not matches_snapshot({}, _patterns("alice"))
+
+
+def test_matches_snapshot_supports_logfmt_prettified_values():
+    state = {"when": 1767225600}
+
+    def logfmt(value, path):
+        return "2026-01-01" if value == 1767225600 else None
+
+    assert matches_snapshot(state, _patterns("2026"), logfmt=logfmt)
+    assert matches_snapshot(state, _patterns("1767225600"), logfmt=logfmt)
+    assert not matches_snapshot(state, _patterns("2026"))
+
+
 def test_entry_dataclass_holds_anchor_for_deletes():
     entry = _Entry(["a", "b"], 1, "1", anchor=["a"])
     assert entry.anchor == ["a"]
@@ -269,7 +293,7 @@ def _sample_changes():
     ]
 
 
-def _run_cli(tmp_path, capsys, monkeypatch, changes, *args, color=False):
+def _run_cli(tmp_path, capsys, monkeypatch, changes, *args, color=False, state=None):
     if color:
         monkeypatch.setenv("FORCE_COLOR", "1")
         monkeypatch.delenv("NO_COLOR", raising=False)
@@ -277,7 +301,7 @@ def _run_cli(tmp_path, capsys, monkeypatch, changes, *args, color=False):
         monkeypatch.setenv("NO_COLOR", "1")
         monkeypatch.delenv("FORCE_COLOR", raising=False)
     path = tmp_path / "test.kantadb"
-    _write_db(path, changes)
+    _write_db(path, changes, state=state)
     code = main([str(path), *args])
     assert code == 0
     return capsys.readouterr().err
@@ -335,13 +359,68 @@ def test_cli_grep_repeated_patterns_must_all_match(tmp_path, capsys, monkeypatch
 
 
 def test_cli_grep_no_match_exits_zero(tmp_path, capsys, monkeypatch):
-    """No matching records is not an error; snapshot lines still print."""
+    """No matching records is not an error; non-matching snapshots are hidden."""
     err = _run_cli(tmp_path, capsys, monkeypatch, _sample_changes(), "--grep", "nobody")
-    assert "snapshot s0" in err
+    assert "snapshot s0" not in err
     assert "create_alice" not in err
     assert "create_bob" not in err
 
 
+def test_cli_grep_snapshot_prints_only_when_state_matches(
+    tmp_path, capsys, monkeypatch
+):
+    """Snapshots are matched against their full state like change records."""
+    state = {"users": {"alice": {"email": "alice@example.com", "age": 30}}}
+
+    err = _run_cli(
+        tmp_path,
+        capsys,
+        monkeypatch,
+        _sample_changes(),
+        "--grep",
+        "alice",
+        state=state,
+    )
+    assert "snapshot s0" in err
+
+    err = _run_cli(
+        tmp_path,
+        capsys,
+        monkeypatch,
+        _sample_changes(),
+        "--grep",
+        "users.alice.age=30",
+        state=state,
+    )
+    assert "snapshot s0" in err
+
+    # A pattern matching nothing in the state suppresses the snapshot.
+    err = _run_cli(
+        tmp_path,
+        capsys,
+        monkeypatch,
+        _sample_changes(),
+        "--grep",
+        "bob",
+        state=state,
+    )
+    assert "snapshot s0" not in err
+
+    # Repeated patterns are ANDed within the snapshot state too.
+    err = _run_cli(
+        tmp_path,
+        capsys,
+        monkeypatch,
+        _sample_changes(),
+        "--grep",
+        "alice",
+        "--grep",
+        "missing",
+        state=state,
+    )
+    assert "snapshot s0" not in err
+
+
 def test_cli_grep_path_value_forms(tmp_path, capsys, monkeypatch):
     """The 'path=value', 'path=' and '=value' forms restrict the match side."""
     err = _run_cli(

tests/test_kanta_integration.py

@@ -769,7 +769,11 @@ async def test_transaction_log_false_suppresses_log(tmp_path, format_config, cap
 
     await kanta.close()
 
-    info_messages = [r for r in caplog.records if r.levelno == logging.INFO]
+    info_messages = [
+        r
+        for r in caplog.records
+        if r.levelno == logging.INFO and r.name.startswith("kanta.transaction")
+    ]
     assert not info_messages
 
 
@@ -787,7 +791,11 @@ async def test_transaction_logdiff_false_logs_header_only(
 
     await kanta.close()
 
-    messages = [r.message for r in caplog.records if r.levelno == logging.INFO]
+    messages = [
+        r.message
+        for r in caplog.records
+        if r.levelno == logging.INFO and r.name.startswith("kanta.transaction")
+    ]
     assert len(messages) == 1
     assert "inc" in messages[0]
     assert "counter" not in messages[0]
@@ -808,7 +816,11 @@ async def test_transaction_log_custom_logger(tmp_path, format_config, caplog):
 
     await kanta.close()
 
-    info_messages = [r for r in caplog.records if r.levelno == logging.INFO]
+    info_messages = [
+        r
+        for r in caplog.records
+        if r.levelno == logging.INFO and r.name.startswith("custom.transaction")
+    ]
     assert len(info_messages) >= 1
     assert "inc" in info_messages[0].message
 

tests/test_logging.py

@@ -42,11 +42,18 @@ def test_configure_logging_defaults():
     configure_logging()
     for name in ("kanta.bootstrap", "kanta.migration", "kanta.transaction"):
         logger = logging.getLogger(name)
-        assert logger.level == logging.NOTSET  # inherits the root level
+        assert logger.level == logging.INFO  # visible without app logging setup
         assert not logger.propagate
         assert logger.handlers
 
 
+def test_configure_logging_debug_lifts_debug_loggers():
+    configure_logging(debug=True)
+    assert logging.getLogger("kanta.bootstrap").level == logging.DEBUG
+    assert logging.getLogger("kanta.migration").level == logging.DEBUG
+    assert logging.getLogger("kanta.transaction").level == logging.INFO
+
+
 def test_configure_logging_disables_specific_loggers():
     configure_logging(bootstrap=False, migration=False, transaction=False)
     assert logging.getLogger("kanta.bootstrap").disabled
@@ -63,14 +70,8 @@ def test_configure_logging_skiproot_false_routes_via_root():
 
 
 def _setup_logging(**kwargs):
-    """Default kanta logging with the event loggers lifted to INFO.
-
-    Event loggers inherit the root level (WARNING under pytest); output
-    assertions need INFO.
-    """
+    """Default kanta logging; the event loggers are INFO out of the box."""
     configure_logging(**kwargs)
-    for name in ("kanta.bootstrap", "kanta.migration", "kanta.transaction"):
-        logging.getLogger(name).setLevel(logging.INFO)
 
 
 def test_log_change_no_diff(capsys):
@@ -114,7 +115,6 @@ def test_configure_logging_diff_false(capsys):
 def test_configure_logging_diff_true_reenables(capsys):
     _setup_logging(diff=False)
     configure_logging(diff=True)
-    logging.getLogger("kanta.transaction").setLevel(logging.INFO)
     log_change("update", {"counter": 5}, previous={})
     captured = capsys.readouterr()
     assert "counter" in captured.err

tests/test_range.py

@@ -0,0 +1,193 @@
+"""Tests for repeated -r ranges: unit window intersection, plain item slicing."""
+
+import asyncio
+import json
+
+from kanta.serialization import JsonSerializer
+from tests.support import Data, make_kanta
+
+# The CLI reads files with its own JSON Kanta instance unless -k is given.
+JSON = ("json", JsonSerializer)
+
+
+def _run_cli(path, *args):
+    from kanta.__main__ import main
+
+    return main([str(path), *args])
+
+
+def _build(path, increments=3):
+    async def build():
+        kanta = make_kanta(path, Data, JSON)
+        await kanta.open(log=False)
+        for i in range(increments):
+            with kanta.transaction(f"inc{i}", log=False) as data:
+                data.counter += 1
+        await kanta.close()
+
+    asyncio.run(build())
+
+
+def test_plain_range_filters_output_not_state(tmp_path, capsys):
+    """Plain numbers slice the printed items; -o state is unaffected."""
+    path = tmp_path / "data.kantadb"
+    _build(path)
+    capsys.readouterr()
+
+    # Items are bootstrap, inc0, inc1, inc2: 1:3 prints inc0 and inc1.
+    assert _run_cli(path, "-r", "1:3", "-o", "-") == 0
+    out = capsys.readouterr()
+    assert "bootstrap" not in out.err
+    assert "inc0" in out.err
+    assert "inc1" in out.err
+    assert "inc2" not in out.err
+    # Display filter only: the output state is the full replayed state.
+    assert json.loads(out.out) == {"users": {}, "counter": 3}
+
+    # Inclusive end.
+    assert _run_cli(path, "-r", "0..1") == 0
+    err = capsys.readouterr().err
+    assert "bootstrap" in err
+    assert "inc0" in err
+    assert "inc1" not in err
+
+
+def test_plain_single_index(tmp_path, capsys):
+    path = tmp_path / "data.kantadb"
+    _build(path)
+    capsys.readouterr()
+
+    assert _run_cli(path, "-r", "0") == 0
+    err = capsys.readouterr().err
+    assert "bootstrap" in err
+    assert "inc0" not in err
+
+    # Negative index counts from the end of the filtered output.
+    assert _run_cli(path, "-r", "-1") == 0
+    err = capsys.readouterr().err
+    assert "inc2" in err
+    assert "inc1" not in err
+
+    # A single index must exist; out of range is exit 2.
+    assert _run_cli(path, "-r", "99") == 2
+    assert "99" in capsys.readouterr().err
+
+
+def test_plain_ranges_intersect(tmp_path, capsys):
+    path = tmp_path / "data.kantadb"
+    _build(path)
+    capsys.readouterr()
+
+    # [1:) and [:3) intersect to items 1 and 2.
+    assert _run_cli(path, "-r", "1:", "-r", ":3") == 0
+    err = capsys.readouterr().err
+    assert "bootstrap" not in err
+    assert "inc0" in err
+    assert "inc1" in err
+    assert "inc2" not in err
+
+    # An empty intersection prints nothing, exit 0 (truncation, not error).
+    assert _run_cli(path, "-r", ":1", "-r", "3:") == 0
+    err = capsys.readouterr().err
+    assert "inc" not in err
+    assert "bootstrap" not in err
+
+
+def test_plain_applies_after_grep(tmp_path, capsys):
+    """Plain numbers index into what survives --grep, order-independently."""
+    path = tmp_path / "data.kantadb"
+    _build(path)
+    capsys.readouterr()
+
+    # Grep leaves inc0..inc2; plain 0 picks the first of those.
+    assert _run_cli(path, "--grep", "inc", "-r", "0") == 0
+    first = capsys.readouterr().err
+    assert "inc0" in first
+    assert "inc1" not in first
+    assert "bootstrap" not in first
+
+    assert _run_cli(path, "-r", "0", "--grep", "inc") == 0
+    assert capsys.readouterr().err == first
+
+    # Negative index from the filtered end.
+    assert _run_cli(path, "--grep", "inc", "-r", "-1") == 0
+    err = capsys.readouterr().err
+    assert "inc2" in err
+    assert "inc1" not in err
+
+
+def test_unit_specs_intersect(tmp_path, capsys):
+    path = tmp_path / "data.kantadb"
+    _build(path)
+    capsys.readouterr()
+
+    # r1:r4 = bootstrap..inc1, r2:r5 = inc0..inc2; intersection is inc0, inc1.
+    assert _run_cli(path, "-r", "r1:r4", "-r", "r2:r5", "-o", "-") == 0
+    out = capsys.readouterr()
+    assert "bootstrap" not in out.err
+    assert "inc0" in out.err
+    assert "inc1" in out.err
+    assert "inc2" not in out.err
+    # The replay end is the intersected window end (state after inc1).
+    assert json.loads(out.out) == {"users": {}, "counter": 2}
+
+
+def test_unit_and_plain_combine(tmp_path, capsys):
+    """Unit window first, then plain slicing over the window's items."""
+    path = tmp_path / "data.kantadb"
+    _build(path)
+    capsys.readouterr()
+
+    # Window r2:r4 yields inc0, inc1; plain item 0 prints only inc0.
+    assert _run_cli(path, "-r", "r2:r4", "-r", "0", "-o", "-") == 0
+    out = capsys.readouterr()
+    assert "inc0" in out.err
+    assert "inc1" not in out.err
+    # The plain spec does not narrow the output state.
+    assert json.loads(out.out) == {"users": {}, "counter": 2}
+
+
+def test_context_expands_after_grep(tmp_path, capsys):
+    """Context records come from the in-window pool, free of --grep."""
+    path = tmp_path / "data.kantadb"
+    _build(path, increments=5)
+    capsys.readouterr()
+
+    assert _run_cli(path, "--grep", "inc2", "--context", "1") == 0
+    err = capsys.readouterr().err
+    assert "inc1" in err
+    assert "inc2" in err
+    assert "inc3" in err
+    assert "inc0" not in err
+    assert "inc4" not in err
+    assert "bootstrap" not in err
+
+    # Context 0 (default) changes nothing.
+    assert _run_cli(path, "--grep", "inc2") == 0
+    err = capsys.readouterr().err
+    assert "inc1" not in err
+    assert "inc2" in err
+
+    # Context is clamped at the start of the range.
+    assert _run_cli(path, "--grep", "bootstrap", "--context", "2") == 0
+    err = capsys.readouterr().err
+    assert "bootstrap" in err
+    assert "inc0" in err
+    assert "inc1" in err
+    assert "inc2" not in err
+
+
+def test_context_merges_overlaps(tmp_path, capsys):
+    """Nearby matches share context instead of duplicating records."""
+    path = tmp_path / "data.kantadb"
+    _build(path, increments=5)
+    capsys.readouterr()
+
+    # Grep leaves inc0..inc4; plain 1:3 selects inc1, inc2; context 1
+    # expands to inc0..inc3, each printed exactly once (overlaps merged).
+    assert _run_cli(path, "--grep", "inc", "-r", "1:3", "--context", "1") == 0
+    err = capsys.readouterr().err
+    for action in ["inc0", "inc1", "inc2", "inc3"]:
+        assert err.count(action) == 1
+    assert "bootstrap" not in err
+    assert "inc4" not in err

tests/test_revision.py

@@ -0,0 +1,480 @@
+"""Tests for revision numbers: runtime counter, snapshot ``r`` field, CLI rN."""
+
+import asyncio
+import json
+from datetime import UTC, datetime, timedelta
+
+import pytest
+
+from kanta.exceptions import DataIntegrityError
+from kanta.serialization import JsonSerializer
+from kanta.structs import META_FORMAT, ChangeRecord, Snapshot
+from tests.support import Data, make_kanta, read_last_snapshot
+from tests.test_rotation import make_clock, read_all, write_history
+
+T0 = datetime(2026, 1, 1, tzinfo=UTC)
+
+# The CLI reads files with its own JSON Kanta instance unless -k is given.
+JSON = ("json", JsonSerializer)
+
+
+@pytest.mark.asyncio
+async def test_revision_requires_open(tmp_path, format_config):
+    kanta = make_kanta(tmp_path / "data.kantadb", Data, format_config)
+    with pytest.raises(DataIntegrityError):
+        kanta.rev
+
+
+@pytest.mark.asyncio
+async def test_revision_counts_change_records(tmp_path, format_config):
+    path = tmp_path / "data.kantadb"
+    kanta = make_kanta(path, Data, format_config)
+    await kanta.open(log=False)
+    # The bootstrap record is the first diff: state is at revision 1.
+    assert kanta.rev == 1
+
+    with kanta.transaction("inc", log=False) as data:
+        data.counter += 1
+    assert kanta.rev == 2
+
+    # A no-op transaction queues no record and does not advance the revision.
+    with kanta.transaction("noop", log=False):
+        pass
+    assert kanta.rev == 2
+    await kanta.close()
+
+
+@pytest.mark.asyncio
+async def test_revision_restored_on_reopen(tmp_path, format_config):
+    path = tmp_path / "data.kantadb"
+    kanta = make_kanta(path, Data, format_config)
+    await kanta.open(log=False)
+    for _ in range(3):
+        with kanta.transaction("inc", log=False) as data:
+            data.counter += 1
+    await kanta.close()
+
+    kanta2 = make_kanta(path, Data, format_config)
+    await kanta2.open(log=False)
+    assert kanta2.rev == 4
+    await kanta2.close()
+
+
+@pytest.mark.asyncio
+async def test_snapshot_carries_revision(tmp_path, format_config):
+    path = tmp_path / "data.kantadb"
+    kanta = make_kanta(path, Data, format_config)
+    await kanta.open(log=False)
+    for _ in range(2):
+        with kanta.transaction("inc", log=False) as data:
+            data.counter += 1
+    await kanta.flush()
+    kanta.request_snapshot()
+    kanta._impl.maybe_snapshot()
+    await kanta.close()
+
+    snap = read_last_snapshot(path, format_config)
+    assert snap is not None
+    assert snap.r == 3
+    assert snap.k == META_FORMAT
+
+    # Reopening from the snapshot keeps the absolute revision.
+    kanta2 = make_kanta(path, Data, format_config)
+    await kanta2.open(log=False)
+    assert kanta2.rev == 3
+    with kanta2.transaction("inc", log=False) as data:
+        data.counter += 1
+    assert kanta2.rev == 4
+    await kanta2.close()
+
+
+@pytest.mark.asyncio
+async def test_legacy_snapshot_recounted_and_rewritten(tmp_path, format_config):
+    """Meta migration k1 rewrites a k0 file in full (see kanta/metamigr.py)."""
+    path = tmp_path / "data.kantadb"
+    kanta = make_kanta(path, Data, format_config)
+    await kanta.open(log=False)
+    for _ in range(2):
+        with kanta.transaction("inc", log=False) as data:
+            data.counter += 1
+    await kanta.flush()
+    kanta.request_snapshot()
+    kanta._impl.maybe_snapshot()
+    await kanta.close()
+
+    # Simulate a pre-revision database: rewrite the last snapshot without k,
+    # with a wrong r, then append one more change record.
+    _, serializer_cls = format_config
+    serializer = serializer_cls()
+    framer = serializer.framer_cls()
+    records = read_all(path, format_config)
+    out = bytearray()
+    for rec in records:
+        if isinstance(rec, Snapshot):
+            rec = Snapshot(ts=rec.ts, v=rec.v, state=rec.state, m=rec.m, r=0)
+            out += framer.frame_snapshot(
+                serializer.encode(rec), record_offset=len(out)
+            )
+        else:
+            out += framer.frame_change(
+                serializer.encode(rec), record_offset=len(out)
+            )
+    extra = ChangeRecord(ts=T0, a="inc", v=0, diff={"counter": 3})
+    out += framer.frame_change(serializer.encode(extra), record_offset=len(out))
+    path.write_bytes(bytes(out))
+
+    kanta2 = make_kanta(path, Data, format_config)
+    await kanta2.open(log=False)
+    # bootstrap + 2 changes + trailing change = revision 4, despite r=0 above.
+    assert kanta2.rev == 4
+    assert kanta2.data.counter == 3
+    await kanta2.close()
+
+    # The file was rewritten in full: all change records preserved in order,
+    # the legacy intermediate snapshot dropped, and a k1 closing snapshot
+    # carrying the total revision appended.
+    records = read_all(path, format_config)
+    changes = [r for r in records if isinstance(r, ChangeRecord)]
+    assert [c.a for c in changes] == ["bootstrap", "inc", "inc", "inc"]
+    snapshots = [r for r in records if isinstance(r, Snapshot)]
+    assert len(snapshots) == 1
+    assert isinstance(records[-1], Snapshot)
+    closing = snapshots[0]
+    assert closing.k == META_FORMAT
+    assert closing.r == 4
+    assert closing.state["counter"] == 3
+
+    # The next open trusts the stored r (file unchanged this time).
+    content = path.read_bytes()
+    kanta3 = make_kanta(path, Data, format_config)
+    await kanta3.open(log=False)
+    assert kanta3.rev == 4
+    await kanta3.close()
+    assert path.read_bytes() == content
+
+
+@pytest.mark.asyncio
+async def test_legacy_migration_preserves_leading_snapshot(tmp_path, format_config):
+    """A legacy file starting with a snapshot keeps it, re-stamped k1."""
+    path = tmp_path / "data.kantadb"
+    _, serializer_cls = format_config
+    serializer = serializer_cls()
+    framer = serializer.framer_cls()
+    snap_ts = T0 - timedelta(days=40)
+    snap_m = T0 - timedelta(days=41)
+    leading = Snapshot(ts=snap_ts, v=3, state={"users": {}, "counter": 7}, m=snap_m)
+    out = bytearray()
+    out += framer.frame_snapshot(serializer.encode(leading), record_offset=0)
+    for i in range(2):
+        rec = ChangeRecord(ts=T0, a=f"inc{i}", v=3, diff={"counter": 8 + i})
+        out += framer.frame_change(serializer.encode(rec), record_offset=len(out))
+    path.write_bytes(bytes(out))
+
+    kanta = make_kanta(path, Data, format_config)
+    await kanta.open(log=False)
+    # Only the in-file history is countable: 2 change records.
+    assert kanta.rev == 2
+    assert kanta.data.counter == 9
+    await kanta.close()
+
+    records = read_all(path, format_config)
+    assert isinstance(records[0], Snapshot)
+    assert records[0].k == META_FORMAT
+    assert records[0].r == 0
+    assert records[0].ts == snap_ts
+    assert records[0].m == snap_m
+    assert records[0].v == 3
+    assert records[0].state == {"users": {}, "counter": 7}
+    changes = [r for r in records if isinstance(r, ChangeRecord)]
+    assert [c.a for c in changes] == ["inc0", "inc1"]
+    assert isinstance(records[-1], Snapshot)
+    assert records[-1].k == META_FORMAT
+    assert records[-1].r == 2
+    assert records[-1].state["counter"] == 9
+
+
+@pytest.mark.asyncio
+async def test_rotation_of_legacy_file_migrates(tmp_path, format_config):
+    """Rotation and meta migration coincide: rotation already produces k1
+    snapshots with revisions counted from pre-rotation record positions."""
+    path = tmp_path / "data.kantadb"
+    _, serializer_cls = format_config
+    serializer = serializer_cls()
+    framer = serializer.framer_cls()
+
+    def change(action, day, counter):
+        return ChangeRecord(
+            ts=T0 + day * timedelta(days=1), a=action, v=0, diff={"counter": counter}
+        )
+
+    legacy_snap = Snapshot(
+        ts=T0 - 40 * timedelta(days=1), v=0, state={"users": {}, "counter": 1}, r=0
+    )
+    out = bytearray()
+    out += framer.frame_change(
+        serializer.encode(change("bootstrap", -41, 0)), record_offset=len(out)
+    )
+    out += framer.frame_change(
+        serializer.encode(change("day-40", -40, 1)), record_offset=len(out)
+    )
+    out += framer.frame_snapshot(
+        serializer.encode(legacy_snap), record_offset=len(out)
+    )
+    out += framer.frame_change(
+        serializer.encode(change("day-20", -20, 2)), record_offset=len(out)
+    )
+    out += framer.frame_change(
+        serializer.encode(change("day-5", -5, 3)), record_offset=len(out)
+    )
+    path.write_bytes(bytes(out))
+
+    cell = [T0]
+    kanta = make_kanta(path, Data, format_config, retention=30 * timedelta(days=1))
+    kanta.clock(make_clock(cell))
+    await kanta.open(log=False)
+    assert kanta.rev == 4
+    assert kanta.data.counter == 3
+    await kanta.close()
+
+    # The rotation's leading snapshot is k1 with the pre-rotation revision.
+    records = read_all(path, format_config)
+    assert isinstance(records[0], Snapshot)
+    assert records[0].k == META_FORMAT
+    assert records[0].r == 2
+    assert [r.a for r in records if isinstance(r, ChangeRecord)] == [
+        "day-20",
+        "day-5",
+    ]
+
+    # No further rewriting on the next open.
+    content = path.read_bytes()
+    kanta2 = make_kanta(path, Data, format_config)
+    await kanta2.open(log=False)
+    assert kanta2.rev == 4
+    await kanta2.close()
+    assert path.read_bytes() == content
+
+
+@pytest.mark.asyncio
+async def test_legacy_snapshot_readonly_recounts_without_writing(
+    tmp_path, format_config
+):
+    path = tmp_path / "data.kantadb"
+    kanta = make_kanta(path, Data, format_config)
+    await kanta.open(log=False)
+    with kanta.transaction("inc", log=False) as data:
+        data.counter += 1
+    await kanta.close()
+
+    # Inject a legacy snapshot (no k marker) at the start; the whole file is
+    # re-framed with a single framer so binary checksums stay valid.
+    _, serializer_cls = format_config
+    serializer = serializer_cls()
+    framer = serializer.framer_cls()
+    legacy = Snapshot(ts=T0, v=0, state={"users": {}, "counter": 0}, r=0)
+    out = bytearray()
+    out += framer.frame_snapshot(serializer.encode(legacy), record_offset=0)
+    for rec in read_all(path, format_config):
+        out += framer.frame_change(serializer.encode(rec), record_offset=len(out))
+    path.write_bytes(bytes(out))
+    content = path.read_bytes()
+
+    kanta2 = make_kanta(path, Data, format_config)
+    await kanta2.open(readonly=True, log=False)
+    assert kanta2.rev == 2
+    await kanta2.close()
+    assert path.read_bytes() == content
+
+
+@pytest.mark.asyncio
+async def test_rotation_preserves_revisions(tmp_path, format_config):
+    path = tmp_path / "data.kantadb"
+    await write_history(path, format_config, days=[-40, -20, -5])
+
+    cell = [T0]
+    kanta = make_kanta(path, Data, format_config, retention=30 * timedelta(days=1))
+    kanta.clock(make_clock(cell))
+    await kanta.open(log=False)
+    # bootstrap + 3 changes = revision 4 before and after the rotation.
+    assert kanta.rev == 4
+    await kanta.close()
+
+    records = read_all(path, format_config)
+    leading = records[0]
+    assert isinstance(leading, Snapshot)
+    # One change was dropped (bootstrap + day-40 are before the cutoff
+    # position... bootstrap is a change record too: 2 records before).
+    assert leading.r == 2
+    assert leading.k == META_FORMAT
+
+    kanta2 = make_kanta(path, Data, format_config)
+    await kanta2.open(log=False)
+    assert kanta2.rev == 4
+    with kanta2.transaction("inc", log=False) as data:
+        data.counter += 1
+    assert kanta2.rev == 5
+    await kanta2.close()
+
+
+def test_cli_revision_snapshot_base_file(tmp_path, capsys):
+    """File = snapshot {r: 2} + one change; the two producing changes are gone."""
+    path = tmp_path / "data.kantadb"
+    serializer = JsonSerializer()
+    framer = serializer.framer_cls()
+    snap = Snapshot(
+        ts=T0, v=0, state={"foo": 1, "bar": 1}, r=2, k=META_FORMAT
+    )
+    out = bytearray()
+    out += framer.frame_snapshot(serializer.encode(snap), record_offset=0)
+    change = ChangeRecord(ts=T0, a="add_baz", v=0, diff={"baz": 1})
+    out += framer.frame_change(serializer.encode(change), record_offset=len(out))
+    path.write_bytes(bytes(out))
+    capsys.readouterr()
+
+    # r1's producing change is gone and no snapshot holds state 1: error.
+    assert _run_cli(path, "-r", "r1") == 2
+    assert "r1" in capsys.readouterr().err
+
+    # r2 prints the snapshot line (it literally carries r: 2) and its state;
+    # revisions are shown on change records, not on snapshot indicators.
+    assert _run_cli(path, "-r", "r2", "-o", "-") == 0
+    out = capsys.readouterr()
+    assert "snapshot s0" in out.err
+    assert "r2" not in out.err
+    assert json.loads(out.out) == {"foo": 1, "bar": 1}
+
+    # r3 is the change, replayed on the snapshot base; state includes baz.
+    assert _run_cli(path, "-r", "r3", "-o", "-") == 0
+    out = capsys.readouterr()
+    assert "add_baz" in out.err
+    assert " r3" in out.err
+    assert json.loads(out.out) == {"foo": 1, "bar": 1, "baz": 1}
+
+    # Single index and half-open range agree.
+    assert _run_cli(path, "-r", "r2") == 0
+    single = capsys.readouterr().err
+    assert _run_cli(path, "-r", "r2:r3") == 0
+    assert capsys.readouterr().err == single
+
+
+def _run_cli(path, *args):
+    from kanta.__main__ import main
+
+    return main([str(path), *args])
+
+
+def test_cli_revision_ranges(tmp_path, capsys):
+    path = tmp_path / "data.kantadb"
+
+    async def build():
+        kanta = make_kanta(path, Data, JSON)
+        await kanta.open(log=False)
+        for i in range(3):
+            with kanta.transaction(f"inc{i}", log=False) as data:
+                data.counter += 1
+        await kanta.close()
+
+    asyncio.run(build())
+    capsys.readouterr()
+
+    # A revision selects the record that produced that state, with the
+    # state: r1 and r1:r2 both show only the bootstrap and output state 1.
+    assert _run_cli(path, "-r", "r1") == 0
+    single = capsys.readouterr().err
+    assert "bootstrap" in single
+    # Change record headers carry the revision of the state they produce.
+    assert " r1" in single
+    assert "inc0" not in single
+
+    assert _run_cli(path, "-r", "r1:r2") == 0
+    assert capsys.readouterr().err == single
+
+    assert _run_cli(path, "-r", "r1", "-o", "-") == 0
+    assert json.loads(capsys.readouterr().out) == {"users": {}, "counter": 0}
+
+    # r1..r2 (inclusive end) adds the next change and outputs state 2.
+    assert _run_cli(path, "-r", "r1..r2", "-o", "-") == 0
+    out = capsys.readouterr()
+    assert "inc0" in out.err
+    assert json.loads(out.out) == {"users": {}, "counter": 1}
+
+    # r0 is not meaningful: no record produces the empty initial state.
+    assert _run_cli(path, "-r", "r0") == 2
+    assert "r0" in capsys.readouterr().err
+
+    # Range by revision.
+    assert _run_cli(path, "-r", "r2:r4") == 0
+    err = capsys.readouterr().err
+    assert "inc0" in err
+    assert "inc1" in err
+    assert "inc2" not in err
+
+    # Negative revisions are rejected.
+    assert _run_cli(path, "-r", "r-1") == 1
+    assert "negative" in capsys.readouterr().err
+
+    # Out-of-range single revision is a range error (exit 2).
+    assert _run_cli(path, "-r", "r99") == 2
+    assert "r99" in capsys.readouterr().err
+
+    # The same bounds as a range truncate Python-style instead: an empty
+    # display and the closest known state, not an error.
+    assert _run_cli(path, "-r", "r99:r100", "-o", "-") == 0
+    out = capsys.readouterr()
+    assert "bootstrap" not in out.err
+    assert "inc" not in out.err
+    assert json.loads(out.out) == {"users": {}, "counter": 3}
+
+    # Only the end beyond the last revision: truncated to end of file.
+    assert _run_cli(path, "-r", "r3:r99", "-o", "-") == 0
+    out = capsys.readouterr()
+    assert "inc1" in out.err
+    assert "inc2" in out.err
+    assert json.loads(out.out) == {"users": {}, "counter": 3}
+
+
+def test_cli_revision_after_rotation(tmp_path, capsys):
+    path = tmp_path / "data.kantadb"
+
+    async def build():
+        await write_history(path, JSON, days=[-40, -20, -5])
+        cell = [T0]
+        kanta = make_kanta(
+            path, Data, JSON, retention=30 * timedelta(days=1)
+        )
+        kanta.clock(make_clock(cell))
+        await kanta.open(log=False)
+        await kanta.close()
+
+    asyncio.run(build())
+    capsys.readouterr()
+
+    # The leading snapshot holds revision 2 (bootstrap + day-40).  Since the
+    # change that produced state 2 is gone, r2 shows the snapshot itself and
+    # outputs the snapshot state.
+    assert _run_cli(path, "-r", "r2", "-o", "-") == 0
+    out = capsys.readouterr()
+    assert "snapshot s0" in out.err
+    assert "r2" not in out.err
+    assert "day-20" not in out.err
+    assert json.loads(out.out) == {"users": {}, "counter": 1}
+
+    # r1 has no record in this file: a single index does not truncate.
+    assert _run_cli(path, "-r", "r1") == 2
+    assert "r1" in capsys.readouterr().err
+
+    # A range truncates like Python slicing instead: r1:r3 starts at the
+    # beginning of the file and stops before the r3 record, showing just the
+    # leading snapshot.
+    assert _run_cli(path, "-r", "r1:r3") == 0
+    err = capsys.readouterr().err
+    assert "snapshot s0" in err
+    assert "r2" not in err
+    assert "day-20" not in err
+
+    # Absolute revisions reach the retained change records.
+    assert _run_cli(path, "-r", "r3:r4") == 0
+    err = capsys.readouterr().err
+    assert "day-20" in err
+    assert "snapshot" not in err
+    assert "day-5" not in err