Repeatable -r: plain numbers slice filtered output, unit windows intersect; add --context

-r is now repeatable and restrictive: unit specs (l/s/v/r) intersect
their database line windows, while plain-number specs index into the
filtered output items left after unit windows and --grep, acting as a
pure display filter (negatives from the end, single index errors with
exit 2, ranges truncate).  --context N (default 0) then expands each
surviving item GNU grep-style by N surrounding in-window records, free
of the grep/plain filters, merging overlaps.

docs/database.md

@@ -73,6 +73,17 @@ A revision number identifies a *state position between diffs*, not a diff:
 
 `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. Snapshot indicator lines show the snapshot's revision.
 
+## 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.

kanta/__main__.py

@@ -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,15 +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, rN = revision"
-            " (rN is the record that produced state N, with its state)."
-            "  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', 'r10:r20', 's-1:',"
-            " ':-1', '-1'."
+            " ':-1', '-1', '-r r1:r5 -r 0'."
         ),
     )
     parser.add_argument(
@@ -271,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
@@ -506,12 +530,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
@@ -534,17 +559,20 @@ 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):
+                    is_match = True
                     if grep_patterns:
                         logfmt = kanta._impl.callback_registry.build_logfmt(
                             InjectionContext(
@@ -553,18 +581,11 @@ async def _run(args: argparse.Namespace) -> int:
                                 current_state=current,
                             )
                         )
-                        if not matches_snapshot(current, grep_patterns, logfmt=logfmt):
-                            continue
-                    _print_snapshot_indicator(
-                        label,
-                        event.snap,
-                        snapshot_line_to_index[event.line_number],
-                        event.revision,
-                        kanta._impl.serializer,
-                    )
+                        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.
@@ -578,13 +599,45 @@ async def _run(args: argparse.Namespace) -> int:
                         highlight = evaluate(
                             event.record, previous, grep_patterns, logfmt=logfmt
                         )
-                        if highlight is None:
-                            continue
+                    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],
+                        event.revision,
+                        kanta._impl.serializer,
+                    )
+                else:
+                    assert previous is not None
                     _print_change_log(
                         label, event.record, previous, current, kanta, highlight
                     )
-                printed = True
-            if printed:
+            if items:
                 _print()
 
         # Apply optional migrations to the range-end state.

kanta/replaylog.py

@@ -431,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)

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