Support time-based -r ranges (relative and ISO)

docs/database.md

@@ -78,11 +78,25 @@ A revision number identifies a *state position between diffs*, not a diff:
 `-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.
+- **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. Negative values count from the end of the filtered output.
+- **Time specs** — a relative `N` + unit (`s`, `min`, `h`, `d`, `w`, e.g. `15min` or `2w`) selects the last N units ending now; a UTC ISO point (`2026-09-28`, `2026-09-28T10:30`, or compact `20260928T1030`) selects the whole natural period its precision denotes (that day, hour, minute, or second); an ISO range `start/end` is inclusive of the end's whole period, and either end may be left open. All times are UTC. Time specs match records individually by their timestamp rather than resolving to a line window, so non-monotonic timestamps simply match wherever they occur, and multiple time specs intersect like the others.
 
 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.
+`--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 whole file — expanding whatever would be shown past any range restriction, so `-r s3 --context 2` shows that snapshot plus the two records before and after it. Overlapping expansions merge so nearby matches do not duplicate context records.
+
+The JSON dump (`-o`) and migrations operate on the state after the **last printed record**, whatever filtering produced the printout — so a plain `-r 2:5` dump ends at the 5th filtered record, and `--grep` or `--context` shift the dump to their last printed record as well. When nothing is printed (an empty filter result or `-q`), the dump falls back to the state at the end of the unit-spec window (or the end of the file).
+
+## CLI Search
+
+`--grep PATTERN` (repeatable) prints only records matching the pattern, structurally and case-insensitively; matched regions get a yellow background, and a matched change record is printed in full, not just its matching line. Patterns match the record's structure, never rendered text:
+
+- A bare pattern matches the action or user as a substring, a dotted path by element (each element in full unless it uses shell wildcards: `users` matches `users` anywhere but not `foousers`; `us*` does; elements match as a contiguous sequence, so `users.*.email` works), or a value (strings by substring, other scalars in full: `true` matches a boolean, `tru` does not).
+- The `path=value` form requires both sides to match within the same change line; `=value` and `path=` match values or paths only.
+- Repeated patterns are ANDed within the same record, but different patterns may match different lines of it.
+- With `-k`, logfmt-prettified values, users, and path elements match alongside the raw ones, and the displayed (prettified) text is what gets highlighted.
+
+Snapshots are matched against their flattened state with the same rules. A matched snapshot shows the matched records of its state as compact `path.to.record = {value}` lines under the indicator — normal path coloring, logfmt-prettified elements, matched regions highlighted — while unmatched parts of the state stay hidden. In the filter pipeline, grep applies after unit `-r` windows and before plain-number slicing and `--context`.
 
 ## Meta Format Migrations
 

kanta/__main__.py

@@ -10,6 +10,7 @@ import importlib.metadata
 import importlib.util
 import sys
 import tempfile
+from datetime import UTC, datetime
 from pathlib import Path
 from typing import Any
 
@@ -18,7 +19,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, matches_snapshot
+from kanta.grep import GrepPattern, evaluate, evaluate_snapshot, snapshot_record_lines
 from kanta.logging import (
     LogEvent,
     emit_event,
@@ -33,6 +34,7 @@ from kanta.replaylog import (
     SnapshotEvent,
     _snapshot_lines,
     end_of_file,
+    parse_time_spec,
     record_change_event,
     record_label,
     replay_events,
@@ -244,12 +246,18 @@ def _parse_args(argv: list[str] | None = None) -> argparse.Namespace:
             " 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"
+            " all unit ranges and --grep patterns.  The -o dump and"
+            " migrations operate on the state after the last printed"
+            " record.  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"
+            " filtered output.  Time specs (UTC) filter records by"
+            " timestamp: relative '15min', '1h', '7d' (N plus one of s, min,"
+            " h, d, w) cover the last N units ending now; an ISO point"
+            " ('2026-09-28', '2026-09-28T10:30', '20260928T1030') covers its"
+            " natural period by precision; an ISO range 'a/b' is inclusive"
+            " and either side may be open ('2026-09-24/').  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', '-r r1:r5 -r 0'."
@@ -264,7 +272,9 @@ def _parse_args(argv: list[str] | None = None) -> argparse.Namespace:
             "Print only change records matching PATTERN, structurally and"
             " case-insensitively; matched regions get a yellow background,"
             " and on a match the whole record is printed, not just the"
-            " matching line.  Repeatable: every pattern must match somewhere"
+            " matching line.  Snapshots are matched against their state and"
+            " show the matched records of it as compact lines.  Repeatable:"
+            " every pattern must match somewhere"
             " in the same record, but different patterns may match different"
             " lines of it.  A bare pattern matches the action or user as a"
             " substring, a dotted path by element (each element in full,"
@@ -288,9 +298,10 @@ def _parse_args(argv: list[str] | None = None) -> argparse.Namespace:
         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."
+            " -r slicing: context records are drawn from the whole file,"
+            " expanding whatever would be shown past any range restriction,"
+            " and overlapping expansions merge.  Without any other filter"
+            " this has no visible effect."
         ),
     )
     args = parser.parse_args(argv)
@@ -421,6 +432,31 @@ def _print_snapshot_indicator(
     _print(f"{label} {ts} {line}")
 
 
+def _in_time_windows(ts: datetime, windows: list) -> bool:
+    """Return whether a record timestamp falls within every time window."""
+    if ts.tzinfo is None:
+        ts = ts.replace(tzinfo=UTC)
+    return all(
+        (start is None or ts >= start) and (end is None or ts < end)
+        for start, end in windows
+    )
+
+
+def _print_snapshot_matches(
+    state: dict[str, Any], kanta: Kanta[Any], highlight: Any
+) -> None:
+    """Print the grep-matched records of a snapshot's state, change-log style.
+
+    Each matched record is one ``path.to.record = {compact value}`` line
+    with normal path coloring and the matched regions highlighted.
+    """
+    logfmt = kanta._impl.callback_registry.build_logfmt(
+        InjectionContext(kanta=kanta, previous_state=state, current_state=state)
+    )
+    for line in snapshot_record_lines(state, highlight, logfmt):
+        _print(line)
+
+
 async def _log_migration(
     kanta: Kanta[Any],
     filename: Path,
@@ -532,13 +568,16 @@ async def _run(args: argparse.Namespace) -> int:
         }
 
         # Resolve -r specs into a line range (unit specs) or a single
-        # snapshot selection, plus plain specs applied to the output items.
+        # snapshot selection, plus plain specs applied to the output items
+        # and time specs matched per record.
         try:
-            selection, plain_specs = (
+            selection, plain_specs, time_specs = (
                 select_all(args.range, events, change_count)
                 if args.range
-                else (Selection(0, end_of_file(events)), [])
+                else (Selection(0, end_of_file(events)), [], [])
             )
+            now = datetime.now(UTC)
+            time_windows = [parse_time_spec(spec, now) for spec in time_specs]
         except RangeNotFoundError as exc:
             raise _CliError(str(exc), EXIT_RANGE_ERROR) from exc
         except ValueError as exc:
@@ -546,34 +585,60 @@ async def _run(args: argparse.Namespace) -> int:
 
         grep_patterns = [GrepPattern.parse(p) for p in args.grep or ()]
 
-        if selection.snapshot is not None:
+        if selection.snapshot is not None and not args.context and not time_windows:
             snap_event = selection.snapshot
             state = snap_event.snap.state
             version = snap_event.snap.v
             if not args.quiet:
-                _print_snapshot_indicator(
-                    record_label(snap_event.line_number, snap_event.record_index),
-                    snap_event.snap,
-                    snapshot_line_to_index[snap_event.line_number],
-                    kanta._impl.serializer,
-                )
-                _print()
+                highlight = None
+                if grep_patterns:
+                    logfmt = kanta._impl.callback_registry.build_logfmt(
+                        InjectionContext(
+                            kanta=kanta, previous_state=state, current_state=state
+                        )
+                    )
+                    highlight = evaluate_snapshot(state, grep_patterns, logfmt=logfmt)
+                if not grep_patterns or highlight is not None:
+                    _print_snapshot_indicator(
+                        record_label(snap_event.line_number, snap_event.record_index),
+                        snap_event.snap,
+                        snapshot_line_to_index[snap_event.line_number],
+                        kanta._impl.serializer,
+                    )
+                    if highlight is not None:
+                        _print_snapshot_matches(state, kanta, highlight)
+                    _print()
         else:
-            # Replay up to the range end, collecting the in-window item pool
-            # and tracking which items survive --grep.
+            # Replay, collecting the item pool and tracking which items
+            # survive the range window and --grep.  With --context the pool
+            # spans the whole file, so context can expand whatever would be
+            # shown past any range restriction; otherwise the pool is
+            # limited to the window.
             state = {}
             version = 0
+            window_state: dict[str, Any] = {}
+            window_version = 0
+            pool_end = end_of_file(events) if args.context else selection.end_line
             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):
+            for event, previous, current in replay_events(events, pool_end):
+                if event.line_number < selection.end_line:
+                    window_state = current
+                    window_version = event.version
                 state = current
                 version = event.version
-                if event.line_number < selection.start_line or args.quiet:
+                if (
+                    args.quiet
+                    or event.line_number < selection.start_line
+                    or event.line_number >= selection.end_line
+                ):
+                    if not args.quiet and args.context:
+                        pool.append((event, previous, current, None))
                     continue
                 highlight = None
                 if isinstance(event, SnapshotEvent):
-                    is_match = True
-                    if grep_patterns:
+                    is_match = _in_time_windows(event.snap.ts, time_windows)
+                    if is_match and grep_patterns:
                         logfmt = kanta._impl.callback_registry.build_logfmt(
                             InjectionContext(
                                 kanta=kanta,
@@ -581,9 +646,10 @@ async def _run(args: argparse.Namespace) -> int:
                                 current_state=current,
                             )
                         )
-                        is_match = matches_snapshot(
+                        highlight = evaluate_snapshot(
                             current, grep_patterns, logfmt=logfmt
                         )
+                        is_match = highlight is not None
                 else:
                     assert previous is not None
                     if grep_patterns:
@@ -599,7 +665,9 @@ async def _run(args: argparse.Namespace) -> int:
                         highlight = evaluate(
                             event.record, previous, grep_patterns, logfmt=logfmt
                         )
-                    is_match = highlight is not None or not grep_patterns
+                    is_match = (
+                        highlight is not None or not grep_patterns
+                    ) and _in_time_windows(event.record.ts, time_windows)
                 pool.append((event, previous, current, highlight if is_match else None))
                 if is_match:
                     matched.append(len(pool) - 1)
@@ -622,6 +690,15 @@ async def _run(args: argparse.Namespace) -> int:
                     expanded.update(range(lo, hi))
                 matched = sorted(expanded)
             items = [pool[idx] for idx in matched]
+            # The dump and migrations operate on the state after the last
+            # printed item; an empty display keeps the window-end state.
+            if items:
+                last_event, _, last_current, _ = items[-1]
+                state = last_current
+                version = last_event.version
+            else:
+                state = window_state
+                version = window_version
             for event, previous, current, highlight in items:
                 label = record_label(event.line_number, event.record_index)
                 if isinstance(event, SnapshotEvent):
@@ -631,6 +708,8 @@ async def _run(args: argparse.Namespace) -> int:
                         snapshot_line_to_index[event.line_number],
                         kanta._impl.serializer,
                     )
+                    if highlight is not None:
+                        _print_snapshot_matches(current, kanta, highlight)
                 else:
                     assert previous is not None
                     _print_change_log(
@@ -645,7 +724,7 @@ async def _run(args: argparse.Namespace) -> int:
             if items:
                 _print()
 
-        # Apply optional migrations to the range-end state.
+        # Apply optional migrations to the selected end state.
         if kanta._impl.migrations is not None:
             try:
                 previous_version = version

kanta/grep.py

@@ -1,9 +1,11 @@
-"""Structural grep matching and match highlighting for change records.
+"""Structural grep matching and match highlighting for the kanta CLI.
 
 Backs the ``--grep`` option of the ``python -m kanta`` CLI.  Patterns
 match against the structure of a transaction — its action, user, and the
-dotted paths and values of its diff — never against rendered output
-text.  Matching is case-insensitive.
+dotted paths and values of its diff — or against a snapshot's flattened
+state, never against rendered output text.  Matching is
+case-insensitive.  A matched change record is printed in full; a matched
+snapshot shows only the matched lines of its state, rendered diff-style.
 
 Path terms match element-wise: each dotted element of the term must match
 a whole path element (``users`` matches ``users`` anywhere in the path
@@ -29,29 +31,60 @@ from __future__ import annotations
 import dataclasses
 import fnmatch
 import json
+import re
 from collections.abc import Iterator
 from typing import Any
 
-from kanta.logging import _USER_PATH, _collect_changes, _get_nested
+from kanta.logging import (
+    _USER_PATH,
+    _collect_changes,
+    _format_path,
+    _format_value,
+    _get_nested,
+)
 from kanta.serialization.base import unmarshal
 from kanta.structs import ChangeRecord
-from kanta.tty import ANSI_RE, ESC, strip_ansi
+from kanta.tty import (
+    ANSI_RE,
+    Line,
+    _parse_sgr,
+    _sgr_transition,
+    strip_ansi,
+)
 
 _GLOB_CHARS = frozenset("*?[")
 
-_MARK_BG = f"{ESC}48;5;220m"  # yellow background (xterm256 #ffd700) for matches
-_UNMARK_BG = f"{ESC}49m"  # back to the default background, foreground untouched
+# Marked regions: xterm256 black text (16, not theme-dependent "black") on
+# xterm256 yellow (220 ~ #ffd700).
+_MARK_FG = "38;5;16"
+_MARK_BG = "48;5;220"
+_SGR_RE = re.compile(r"\x1b\[([0-9;]*)m")
+
+
+def _apply_sgr(active: dict[str, str], escape: str) -> None:
+    """Fold one escape sequence into the active SGR state; non-SGR ignored."""
+    match = _SGR_RE.fullmatch(escape)
+    if not match:
+        return
+    spec = match.group(1)
+    if not spec:
+        active.clear()
+        return
+    if "0" in spec.split(";"):
+        active.clear()
+    active.update(_parse_sgr(spec))
 
 
 def mark_spans(styled: str, spans: list[tuple[int, int]]) -> str:
-    """Wrap the given visible-text spans of a styled string with the mark color.
+    """Wrap the given visible-text spans of a styled string with the mark colors.
 
     ``spans`` are ``(start, end)`` offsets into the visible text of
     *styled*; ANSI sequences are not counted.  Overlapping and adjacent
     spans are merged first, so overlapping matches from different patterns
-    produce one continuous highlight.  The set/clear codes are inserted at
-    the mapped positions in *styled*; only the background attribute is
-    touched, leaving foreground colors intact.
+    produce one continuous highlight.  Marked regions get black text on a
+    yellow background; the active SGR state is tracked through *styled* so
+    the original styling — foreground included — is restored exactly where
+    each mark ends.
     """
     spans = _merge_spans(spans)
     if not spans:
@@ -59,12 +92,16 @@ def mark_spans(styled: str, spans: list[tuple[int, int]]) -> str:
     out: list[str] = []
     plain_pos = 0
     prev_end = 0
+    active: dict[str, str] = {}
     for match in ANSI_RE.finditer(styled):
-        out.append(_wrap_run(styled[prev_end : match.start()], plain_pos, spans))
+        out.append(
+            _wrap_run(styled[prev_end : match.start()], plain_pos, spans, active)
+        )
         plain_pos += match.start() - prev_end
         out.append(match.group(0))
+        _apply_sgr(active, match.group(0))
         prev_end = match.end()
-    out.append(_wrap_run(styled[prev_end:], plain_pos, spans))
+    out.append(_wrap_run(styled[prev_end:], plain_pos, spans, active))
     return "".join(out)
 
 
@@ -81,10 +118,17 @@ def _merge_spans(spans: list[tuple[int, int]]) -> list[tuple[int, int]]:
     return [(start, end) for start, end in merged]
 
 
-def _wrap_run(run: str, plain_start: int, spans: list[tuple[int, int]]) -> str:
-    """Wrap the intersections of *spans* with one escape-free text run."""
+def _wrap_run(
+    run: str, plain_start: int, spans: list[tuple[int, int]], active: dict[str, str]
+) -> str:
+    """Wrap the intersections of *spans* with one escape-free text run.
+
+    *active* is the SGR state in effect at this run; each mark sets the
+    black-on-yellow mark state and each unmark restores *active* exactly.
+    """
     if not run:
         return run
+    mark_state = {**active, "fg": _MARK_FG, "bg": _MARK_BG}
     out: list[str] = []
     pos = 0
     for start, end in spans:
@@ -93,7 +137,9 @@ def _wrap_run(run: str, plain_start: int, spans: list[tuple[int, int]]) -> str:
         if s >= e or e <= pos:
             continue
         out.append(run[pos:s])
-        out.append(f"{_MARK_BG}{run[s:e]}{_UNMARK_BG}")
+        out.append(_sgr_transition(active, mark_state))
+        out.append(run[s:e])
+        out.append(_sgr_transition(mark_state, active))
         pos = e
     out.append(run[pos:])
     return "".join(out)
@@ -114,6 +160,31 @@ def _find_spans(text: str, needle: str) -> list[tuple[int, int]]:
     return spans
 
 
+def snapshot_record_lines(
+    state: dict, highlighter: GrepHighlighter, logfmt: Any = None
+) -> list[str]:
+    """Render the records of a snapshot state that matched grep patterns.
+
+    Each matched entry is shown as one ``path.to.record = {compact value}``
+    line, where the record is the top-level key plus one subkey — mirroring
+    change-record rendering, with normal path coloring, logfmt-prettified
+    elements, and the matched regions highlighted.
+    """
+    records: dict[tuple[str, ...], None] = {}
+    for path in highlighter.matched_paths:
+        record = tuple(path[:2]) if len(path) > 1 else tuple(path)
+        records.setdefault(record)
+    lines = []
+    for record in records:
+        path = list(record)
+        dotted = ".".join(path)
+        value = _get_nested(state, path)
+        path_str = _format_path(path, logfmt, highlight=highlighter)
+        value_str = _format_value(value, dotted, logfmt=logfmt, highlight=highlighter)
+        lines.append(str(Line()("  ", path_str, " ").sep("=")(" ", value_str)))
+    return lines
+
+
 @dataclasses.dataclass(frozen=True)
 class GrepPattern:
     """One parsed ``--grep`` pattern.
@@ -306,6 +377,7 @@ class GrepHighlighter:
         self._lit_deletes: set[str] = set()
         self._value_marks: dict[str, list[_ValueMark]] = {}
         self._meta_marks: dict[str, list[_ValueMark]] = {}
+        self.matched_paths: list[list[str]] = []
 
     def _light_elements(self, path: list[str], indices) -> None:
         for i in indices:
@@ -318,6 +390,7 @@ class GrepHighlighter:
         vmark: _ValueMark | None,
     ) -> None:
         """Record one entry's match: lit element indices and/or a value mark."""
+        self.matched_paths.append(entry.path)
         marked = vmark is not None and (vmark.needle or vmark.whole)
         if entry.anchor is None:
             if elements:
@@ -376,6 +449,27 @@ class GrepHighlighter:
         return self._apply_marks(text, self._meta_marks.get(field, []))
 
 
+def _match_path_dual(term: str, path: list[str], logfmt: Any) -> frozenset[int] | None:
+    """Match a path term against raw and logfmt-prettified elements.
+
+    A logfmt formatter may display an element differently (e.g. resolving a
+    user id to a name); the term matches either form, and the matched
+    element indices are lit so the displayed text gets marked.
+    """
+    elements = _match_path(term, path)
+    if logfmt is not None:
+        pretty = []
+        for i, element in enumerate(path):
+            resolved = logfmt(element, ".".join(path[: i + 1]))
+            pretty.append(resolved if resolved is not None else element)
+        pretty_elements = _match_path(term, pretty)
+        if pretty_elements is not None:
+            elements = (
+                pretty_elements if elements is None else elements | pretty_elements
+            )
+    return elements
+
+
 def _match_entry(
     pattern: GrepPattern, entry: _Entry, logfmt: Any
 ) -> tuple[frozenset[int] | None, _ValueMark | None] | None:
@@ -389,33 +483,48 @@ def _match_entry(
     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)
+        elements = _match_path_dual(pattern.term, entry.path, logfmt)
         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)
+        elements = _match_path_dual(pattern.path_term or "", entry.path, logfmt)
         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(
+def evaluate_snapshot(
     state: dict, patterns: list[GrepPattern], logfmt: Any = None
-) -> bool:
-    """Match *patterns* against a snapshot's full state.
+) -> GrepHighlighter | None:
+    """Match *patterns* against a snapshot's full state, returning its matched regions.
 
     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.
+    ``None`` when any pattern matches nowhere in the state.
     """
     entries = list(_leaf_entries([], unmarshal(state), None))
+    highlighter = GrepHighlighter()
     for pattern in patterns:
-        if not any(_match_entry(pattern, entry, logfmt) for entry in entries):
-            return False
-    return True
+        matched = False
+        for entry in entries:
+            match = _match_entry(pattern, entry, logfmt)
+            if match is None:
+                continue
+            matched = True
+            highlighter.add(entry, *match)
+        if not matched:
+            return None
+    return highlighter
+
+
+def matches_snapshot(
+    state: dict, patterns: list[GrepPattern], logfmt: Any = None
+) -> bool:
+    """Return whether every pattern matched somewhere in a snapshot's state."""
+    return evaluate_snapshot(state, patterns, logfmt) is not None
 
 
 def evaluate(

kanta/replaylog.py

@@ -10,7 +10,9 @@ from __future__ import annotations
 
 import copy
 import dataclasses
+import re
 from collections.abc import Iterator
+from datetime import UTC, datetime, timedelta
 from typing import TYPE_CHECKING, Any, Union
 
 import msgspec
@@ -433,6 +435,97 @@ def select(spec: str, events: list[Event], total: int) -> Selection:
     return Selection(lines[idx], end_line)
 
 
+_REL_RE = re.compile(r"(\d+)(s|min|h|d|w)")
+_REL_UNITS = {"s": 1, "min": 60, "h": 3600, "d": 86400, "w": 604800}
+
+# ISO 8601, UTC: hyphenated date with optional T-time (colons optional), or a
+# compact date requiring T.  Precision (day/hour/minute/second) selects the
+# period the point denotes.
+_ISO_NORM = re.compile(
+    r"(\d{4})-(\d{2})-(\d{2})(?:T(\d{2})(?::?(\d{2}))?(?::?(\d{2}))?)?"
+)
+_ISO_COMPACT = re.compile(r"(\d{4})(\d{2})(\d{2})T(\d{2})(?:(\d{2}))?(?:(\d{2}))?")
+
+
+def _parse_iso_point(text: str) -> tuple[datetime, timedelta]:
+    """Parse an ISO point into its UTC start and the period it denotes."""
+    match = _ISO_NORM.fullmatch(text) or _ISO_COMPACT.fullmatch(text)
+    if match is None:
+        raise ValueError(
+            f"invalid time {text!r} (UTC ISO like 2026-09-28, 2026-09-28T10:30"
+            " or 20260928T1030)"
+        )
+    year, month, day, hour, minute, second = (
+        int(group) if group else None for group in match.groups()
+    )
+    assert year is not None and month is not None and day is not None
+    try:
+        base = datetime(
+            year, month, day, hour or 0, minute or 0, second or 0, tzinfo=UTC
+        )
+    except ValueError:
+        raise ValueError(
+            f"invalid time {text!r} (UTC ISO like 2026-09-28, 2026-09-28T10:30"
+            " or 20260928T1030)"
+        ) from None
+    if second is not None:
+        period = timedelta(seconds=1)
+    elif minute is not None:
+        period = timedelta(minutes=1)
+    elif hour is not None:
+        period = timedelta(hours=1)
+    else:
+        period = timedelta(days=1)
+    return base, period
+
+
+def is_time_spec(spec: str) -> bool:
+    """Return True if the spec is a time spec rather than a position spec.
+
+    Time specs are relative (``15min``, ``1h``, ``7d`` — a trailing unit
+    letter), ISO points or ranges (a hyphenated date, a ``T`` time, or a
+    ``/`` separator).  A leading hyphen stays a negative plain index.
+    """
+    if _REL_RE.fullmatch(spec):
+        return True
+    if "/" in spec:
+        return True
+    return bool(
+        _ISO_NORM.fullmatch(spec) or _ISO_COMPACT.fullmatch(spec)
+    ) or _looks_like_iso(spec)
+
+
+def _looks_like_iso(spec: str) -> bool:
+    """Heuristic for malformed ISO specs, so they error as times, not indexes."""
+    return bool(re.match(r"\d{4}-", spec) or re.match(r"\d{8}T", spec))
+
+
+def parse_time_spec(
+    spec: str, now: datetime
+) -> tuple[datetime | None, datetime | None]:
+    """Resolve a time spec into a UTC [start, end) window; None = unbounded.
+
+    A relative spec covers the last N units ending at ``now``.  An ISO point
+    covers its natural period (day/hour/minute/second by precision).  An ISO
+    range ``a/b`` runs from the start of a's period through the end of b's
+    (inclusive); either side may be left open.
+    """
+    match = _REL_RE.fullmatch(spec)
+    if match:
+        delta = timedelta(seconds=int(match.group(1)) * _REL_UNITS[match.group(2)])
+        return now - delta, now
+    if "/" in spec:
+        start_str, end_str = spec.split("/", 1)
+        start = _parse_iso_point(start_str)[0] if start_str else None
+        end = None
+        if end_str:
+            end_base, end_period = _parse_iso_point(end_str)
+            end = end_base + end_period
+        return start, end
+    base, period = _parse_iso_point(spec)
+    return base, base + period
+
+
 def is_plain_spec(spec: str) -> bool:
     """Return True if every bound of the spec is a plain number (no unit letter).
 
@@ -449,24 +542,33 @@ def is_plain_spec(spec: str) -> bool:
 
 def select_all(
     specs: list[str], events: list[Event], total: int
-) -> tuple[Selection, list[str]]:
+) -> tuple[Selection, list[str], 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.
+    shortcut applies only to a single unit spec.  Plain-number specs and
+    time 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)]
+    time_specs: list[str] = []
+    plain_specs: list[str] = []
+    unit_specs: list[str] = []
+    for spec in specs:
+        if is_time_spec(spec):
+            time_specs.append(spec)
+        elif is_plain_spec(spec):
+            plain_specs.append(spec)
+        else:
+            unit_specs.append(spec)
     if not unit_specs:
-        return Selection(0, end_of_file(events)), plain_specs
+        return Selection(0, end_of_file(events)), plain_specs, time_specs
     selections = [select(spec, events, total) for spec in unit_specs]
     if len(selections) == 1:
-        return selections[0], plain_specs
+        return selections[0], plain_specs, time_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
+    return Selection(min(start, end), end), plain_specs, time_specs
 
 
 def _clamp_index(value: int, count: int) -> int:

pyproject.toml

@@ -31,6 +31,7 @@ bin = [
 [project.urls]
 Homepage = "https://vasanko.com/coders/kanta"
 Repository = "https://git.zi.fi/LeoVasanko/kanta"
+Issues = "https://github.com/LeoVasanko/kanta"
 
 [dependency-groups]
 dev = [

tests/test_grep.py

@@ -10,8 +10,10 @@ from kanta.grep import (
     _match_path,
     _match_value,
     evaluate,
+    evaluate_snapshot,
     mark_spans,
     matches_snapshot,
+    snapshot_record_lines,
 )
 from kanta.logging import _USER_PATH
 from kanta.serialization import JsonSerializer
@@ -21,8 +23,8 @@ from kanta.tty import strip_ansi
 
 TS = datetime(2026, 1, 1, tzinfo=UTC)
 
-MARK = "\x1b[48;5;220m"
-UNMARK = "\x1b[49m"
+MARK = "\x1b[38;5;16;48;5;220m"  # xterm256 black text on yellow background
+UNMARK = "\x1b[0m"  # full reset (unstyled context); styled runs restore their style
 
 
 def test_match_path_whole_elements_anywhere():
@@ -74,13 +76,14 @@ def test_mark_spans_merges_overlaps_and_adjacents():
 
 def test_mark_spans_offsets_into_styled_text():
     styled = "\x1b[32mal\x1b[0mlice"
+    restore_green = "\x1b[0;32m"  # the mark restores the active style exactly
     # A span crossing an escape sequence wraps each escape-free run.
     assert mark_spans(styled, [(0, 3)]) == (
-        f"\x1b[32m{MARK}al{UNMARK}\x1b[0m{MARK}l{UNMARK}ice"
+        f"\x1b[32m{MARK}al{restore_green}\x1b[0m{MARK}l{UNMARK}ice"
     )
     # A span covering everything wraps each escape-free run separately.
     assert mark_spans(styled, [(0, 6)]) == (
-        f"\x1b[32m{MARK}al{UNMARK}\x1b[0m{MARK}lice{UNMARK}"
+        f"\x1b[32m{MARK}al{restore_green}\x1b[0m{MARK}lice{UNMARK}"
     )
     assert mark_spans(styled, []) == styled
 
@@ -251,6 +254,41 @@ def test_matches_snapshot_supports_logfmt_prettified_values():
     assert not matches_snapshot(state, _patterns("2026"))
 
 
+def test_snapshot_record_lines_render_matched_records():
+    """Matched entries render as one compact record line, change-log style."""
+    state = {
+        "users": {
+            "alice": {"email": "alice@example.com", "age": 30},
+            "bob": {"email": "bob@example.com"},
+        },
+        "counter": 1,
+    }
+    highlighter = evaluate_snapshot(state, _patterns("=alice@example.com"))
+    assert highlighter is not None
+    lines = snapshot_record_lines(state, highlighter)
+    # The whole alice record is shown; bob and counter are not.
+    assert [strip_ansi(line) for line in lines] == [
+        "  users.alice = {email: alice@example.com, age: 30}"
+    ]
+    assert MARK + "alice@example.com" + UNMARK in lines[0]
+
+
+def test_snapshot_record_lines_pretty_path_and_value():
+    """A logfmt-prettified path element matches and is marked, like values."""
+    state = {"users": {"userid002": {"name": "Bob", "role": "editor"}}}
+
+    def logfmt(value, path):
+        return "Bob" if value == "userid002" else None
+
+    highlighter = evaluate_snapshot(state, _patterns("bob"), logfmt=logfmt)
+    assert highlighter is not None
+    lines = snapshot_record_lines(state, highlighter, logfmt)
+    assert len(lines) == 1
+    assert strip_ansi(lines[0]) == "  users.Bob = {name: Bob, role: editor}"
+    # Both displayed Bobs — the prettified path element and the name value.
+    assert lines[0].count(MARK) == 2
+
+
 def test_entry_dataclass_holds_anchor_for_deletes():
     entry = _Entry(["a", "b"], 1, "1", anchor=["a"])
     assert entry.anchor == ["a"]
@@ -421,6 +459,84 @@ def test_cli_grep_snapshot_prints_only_when_state_matches(
     assert "snapshot s0" not in err
 
 
+def test_cli_grep_snapshot_prints_matched_state(tmp_path, capsys, monkeypatch):
+    """A matched snapshot shows the matched part of its state, diff-style."""
+    state = {
+        "users": {
+            "alice": {"email": "alice@example.com", "age": 30},
+            "bob": {"email": "bob@example.com"},
+        }
+    }
+
+    err = _run_cli(
+        tmp_path,
+        capsys,
+        monkeypatch,
+        _sample_changes(),
+        "--grep",
+        "=alice@example.com",
+        state=state,
+    )
+    assert "snapshot s0" in err
+    # The matched state line is printed under the snapshot indicator;
+    # unmatched parts of the state (bob) are not shown.
+    assert "alice@example.com" in err
+    assert "bob@example.com" not in err
+
+    # The matched region carries the yellow highlight when color is on.
+    err = _run_cli(
+        tmp_path,
+        capsys,
+        monkeypatch,
+        _sample_changes(),
+        "--grep",
+        "=alice@example.com",
+        state=state,
+        color=True,
+    )
+    assert f"{MARK}alice@example.com{UNMARK}" in err
+
+
+def test_cli_grep_snapshot_selection_shortcut(tmp_path, capsys, monkeypatch):
+    """-r sN applies --grep like the replay path: matched state is shown."""
+    state = {
+        "users": {
+            "alice": {"email": "alice@example.com", "age": 30},
+            "bob": {"email": "bob@example.com"},
+        }
+    }
+
+    err = _run_cli(
+        tmp_path,
+        capsys,
+        monkeypatch,
+        _sample_changes(),
+        "-r",
+        "s0",
+        "--grep",
+        "=alice@example.com",
+        state=state,
+    )
+    assert "snapshot s0" in err
+    assert "users.alice" in err
+    assert "alice@example.com" in err
+    assert "bob@example.com" not in err
+
+    # A non-matching pattern suppresses the explicitly selected snapshot.
+    err = _run_cli(
+        tmp_path,
+        capsys,
+        monkeypatch,
+        _sample_changes(),
+        "-r",
+        "s0",
+        "--grep",
+        "zzz",
+        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_range.py

@@ -2,13 +2,21 @@
 
 import asyncio
 import json
+from datetime import UTC, datetime, timedelta
 
+import pytest
+
+from kanta.replaylog import is_time_spec, parse_time_spec
 from kanta.serialization import JsonSerializer
+from kanta.structs import META_FORMAT, ChangeRecord, Snapshot
 from tests.support import Data, make_kanta
 
 # The CLI reads files with its own JSON Kanta instance unless -k is given.
 JSON = ("json", JsonSerializer)
 
+T0 = datetime(2026, 1, 1, tzinfo=UTC)
+NOW = datetime(2026, 9, 28, 12, 0, 0, tzinfo=UTC)
+
 
 def _run_cli(path, *args):
     from kanta.__main__ import main
@@ -28,8 +36,8 @@ def _build(path, increments=3):
     asyncio.run(build())
 
 
-def test_plain_range_filters_output_not_state(tmp_path, capsys):
-    """Plain numbers slice the printed items; -o state is unaffected."""
+def test_plain_range_restricts_dump_to_last_printed(tmp_path, capsys):
+    """The JSON dump holds the state after the last printed item."""
     path = tmp_path / "data.kantadb"
     _build(path)
     capsys.readouterr()
@@ -41,8 +49,8 @@ def test_plain_range_filters_output_not_state(tmp_path, capsys):
     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}
+    # The dump is the state after inc1, the last printed record.
+    assert json.loads(out.out) == {"users": {}, "counter": 2}
 
     # Inclusive end.
     assert _run_cli(path, "-r", "0..1") == 0
@@ -143,12 +151,12 @@ def test_unit_and_plain_combine(tmp_path, capsys):
     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}
+    # The dump is the state after the last printed item (inc0).
+    assert json.loads(out.out) == {"users": {}, "counter": 1}
 
 
 def test_context_expands_after_grep(tmp_path, capsys):
-    """Context records come from the in-window pool, free of --grep."""
+    """Context records come from the whole file, free of --grep."""
     path = tmp_path / "data.kantadb"
     _build(path, increments=5)
     capsys.readouterr()
@@ -168,7 +176,7 @@ def test_context_expands_after_grep(tmp_path, capsys):
     assert "inc1" not in err
     assert "inc2" in err
 
-    # Context is clamped at the start of the range.
+    # Context is clamped at the start of the file.
     assert _run_cli(path, "--grep", "bootstrap", "--context", "2") == 0
     err = capsys.readouterr().err
     assert "bootstrap" in err
@@ -177,6 +185,62 @@ def test_context_expands_after_grep(tmp_path, capsys):
     assert "inc2" not in err
 
 
+def test_context_expands_past_unit_window(tmp_path, capsys):
+    """Context expands whatever a unit selection would show, past the window."""
+    path = tmp_path / "data.kantadb"
+    _build(path)
+    capsys.readouterr()
+
+    # -r r2 alone shows only inc0 (the record producing state 2); context 1
+    # expands past the window to bootstrap and inc1, and the dump follows
+    # the last printed record.
+    assert _run_cli(path, "-r", "r2", "--context", "1", "-o", "-") == 0
+    out = capsys.readouterr()
+    assert "bootstrap" in out.err
+    assert "inc0" in out.err
+    assert "inc1" in out.err
+    assert "inc2" not in out.err
+    assert json.loads(out.out) == {"users": {}, "counter": 2}
+
+
+def test_context_expands_snapshot_selection(tmp_path, capsys):
+    """-r sN --context N shows the snapshot plus its neighbors."""
+    path = tmp_path / "data.kantadb"
+    serializer = JsonSerializer()
+    framer = serializer.framer_cls()
+    out_bytes = bytearray()
+    out_bytes += framer.frame_change(
+        serializer.encode(ChangeRecord(ts=T0, a="inc0", v=0, diff={"counter": 1})),
+        record_offset=len(out_bytes),
+    )
+    snap = Snapshot(ts=T0, v=0, state={"counter": 1}, r=1, k=META_FORMAT)
+    out_bytes += framer.frame_snapshot(
+        serializer.encode(snap), record_offset=len(out_bytes)
+    )
+    out_bytes += framer.frame_change(
+        serializer.encode(ChangeRecord(ts=T0, a="inc1", v=0, diff={"counter": 2})),
+        record_offset=len(out_bytes),
+    )
+    path.write_bytes(bytes(out_bytes))
+    capsys.readouterr()
+
+    # Without context, s1 (the file starts with a change, so s0 is l0)
+    # shows just the snapshot and its state.
+    assert _run_cli(path, "-r", "s1", "-o", "-") == 0
+    out = capsys.readouterr()
+    assert "snapshot s1" in out.err
+    assert "inc0" not in out.err
+    assert json.loads(out.out) == {"counter": 1}
+
+    # With context, the selection expands past the snapshot line.
+    assert _run_cli(path, "-r", "s1", "--context", "1", "-o", "-") == 0
+    out = capsys.readouterr()
+    assert "inc0" in out.err
+    assert "snapshot s1" in out.err
+    assert "inc1" in out.err
+    assert json.loads(out.out) == {"counter": 2}
+
+
 def test_context_merges_overlaps(tmp_path, capsys):
     """Nearby matches share context instead of duplicating records."""
     path = tmp_path / "data.kantadb"
@@ -191,3 +255,227 @@ def test_context_merges_overlaps(tmp_path, capsys):
         assert err.count(action) == 1
     assert "bootstrap" not in err
     assert "inc4" not in err
+
+
+def test_grep_and_context_shift_dump_to_last_printed(tmp_path, capsys):
+    """Grep and --context are display filters, but the dump follows the
+    last printed record, so they move it too."""
+    path = tmp_path / "data.kantadb"
+    _build(path)
+    capsys.readouterr()
+
+    # Only inc1 prints; the dump is the state after inc1.
+    assert _run_cli(path, "--grep", "inc1", "-o", "-") == 0
+    out = capsys.readouterr()
+    assert "inc0" not in out.err
+    assert json.loads(out.out) == {"users": {}, "counter": 2}
+
+    # Context extends the printout past the match; the dump follows it.
+    assert _run_cli(path, "--grep", "inc1", "--context", "1", "-o", "-") == 0
+    out = capsys.readouterr()
+    assert "inc2" in out.err
+    assert json.loads(out.out) == {"users": {}, "counter": 3}
+
+    # Nothing printed: the dump falls back to the range-end state.
+    assert _run_cli(path, "--grep", "zzz", "-o", "-") == 0
+    out = capsys.readouterr()
+    assert json.loads(out.out) == {"users": {}, "counter": 3}
+
+
+def _write_timed(path, entries):
+    """Write one change record per (timestamp, action, counter) entry."""
+    serializer = JsonSerializer()
+    framer = serializer.framer_cls()
+    out = bytearray()
+    for ts, action, counter in entries:
+        out += framer.frame_change(
+            serializer.encode(
+                ChangeRecord(ts=ts, a=action, v=0, diff={"counter": counter})
+            ),
+            record_offset=len(out),
+        )
+    path.write_bytes(bytes(out))
+
+
+def test_is_time_spec_classification():
+    for spec in [
+        "30s",
+        "15min",
+        "1h",
+        "7d",
+        "2w",
+        "2026-09-28",
+        "2026-09-28T10",
+        "2026-09-28T10:30",
+        "20260928T103045",
+        "2026-09-24/2026-09-28",
+        "2026-09-24/",
+        "/2026-09-28",
+        "2026-13-45",  # malformed ISO still classifies as time, then errors
+    ]:
+        assert is_time_spec(spec), spec
+    for spec in ["5", "-1", "2:5", "2..5", ":-1", "r3", "s-1", "l10", "v0", "r1:r4"]:
+        assert not is_time_spec(spec), spec
+
+
+def test_parse_time_spec_relative():
+    assert parse_time_spec("1h", NOW) == (NOW - timedelta(hours=1), NOW)
+    assert parse_time_spec("15min", NOW) == (NOW - timedelta(minutes=15), NOW)
+    assert parse_time_spec("30s", NOW) == (NOW - timedelta(seconds=30), NOW)
+    assert parse_time_spec("7d", NOW) == (NOW - timedelta(days=7), NOW)
+    assert parse_time_spec("2w", NOW) == (NOW - timedelta(weeks=2), NOW)
+
+
+def test_parse_time_spec_iso_periods():
+    assert parse_time_spec("2026-09-28", NOW) == (
+        datetime(2026, 9, 28, tzinfo=UTC),
+        datetime(2026, 9, 29, tzinfo=UTC),
+    )
+    assert parse_time_spec("2026-09-28T10", NOW) == (
+        datetime(2026, 9, 28, 10, tzinfo=UTC),
+        datetime(2026, 9, 28, 11, tzinfo=UTC),
+    )
+    assert parse_time_spec("2026-09-28T10:30", NOW) == (
+        datetime(2026, 9, 28, 10, 30, tzinfo=UTC),
+        datetime(2026, 9, 28, 10, 31, tzinfo=UTC),
+    )
+    assert parse_time_spec("20260928T103045", NOW) == (
+        datetime(2026, 9, 28, 10, 30, 45, tzinfo=UTC),
+        datetime(2026, 9, 28, 10, 30, 46, tzinfo=UTC),
+    )
+
+
+def test_parse_time_spec_iso_ranges():
+    # Inclusive of the end's whole period.
+    assert parse_time_spec("2026-09-24/2026-09-28", NOW) == (
+        datetime(2026, 9, 24, tzinfo=UTC),
+        datetime(2026, 9, 29, tzinfo=UTC),
+    )
+    # Open ends are unbounded.
+    assert parse_time_spec("2026-09-24/", NOW) == (
+        datetime(2026, 9, 24, tzinfo=UTC),
+        None,
+    )
+    assert parse_time_spec("/2026-09-28", NOW) == (
+        None,
+        datetime(2026, 9, 29, tzinfo=UTC),
+    )
+    with pytest.raises(ValueError, match="invalid time"):
+        parse_time_spec("2026-13-45", NOW)
+
+
+def test_cli_time_iso_filters_items(tmp_path, capsys):
+    path = tmp_path / "data.kantadb"
+    _write_timed(
+        path,
+        [
+            (datetime(2026, 9, 24, 10, tzinfo=UTC), "inc0", 1),
+            (datetime(2026, 9, 28, 8, tzinfo=UTC), "inc1", 2),
+            (datetime(2026, 9, 28, 23, tzinfo=UTC), "inc2", 3),
+        ],
+    )
+    capsys.readouterr()
+
+    # A date denotes the full UTC day; the dump follows the last printed item.
+    assert _run_cli(path, "-r", "2026-09-28", "-o", "-") == 0
+    out = capsys.readouterr()
+    assert "inc0" not in out.err
+    assert "inc1" in out.err
+    assert "inc2" in out.err
+    assert json.loads(out.out) == {"counter": 3}
+
+    # Hour precision.
+    assert _run_cli(path, "-r", "2026-09-28T08") == 0
+    err = capsys.readouterr().err
+    assert "inc1" in err
+    assert "inc2" not in err
+
+    # Compact minute precision.
+    assert _run_cli(path, "-r", "20260924T1000") == 0
+    err = capsys.readouterr().err
+    assert "inc0" in err
+    assert "inc1" not in err
+
+    # An inclusive ISO range covers the end's whole period.
+    assert _run_cli(path, "-r", "2026-09-24/2026-09-28") == 0
+    err = capsys.readouterr().err
+    assert "inc0" in err and "inc2" in err
+
+    # Open-ended range.
+    assert _run_cli(path, "-r", "/2026-09-25") == 0
+    err = capsys.readouterr().err
+    assert "inc0" in err
+    assert "inc1" not in err
+
+
+def test_cli_time_combines_with_plain_and_grep(tmp_path, capsys):
+    path = tmp_path / "data.kantadb"
+    _write_timed(
+        path,
+        [
+            (datetime(2026, 9, 24, 10, tzinfo=UTC), "old", 1),
+            (datetime(2026, 9, 28, 8, tzinfo=UTC), "new_a", 2),
+            (datetime(2026, 9, 28, 9, tzinfo=UTC), "new_b", 3),
+        ],
+    )
+    capsys.readouterr()
+
+    # Time filter first, then the plain slice over what remains.
+    assert _run_cli(path, "-r", "2026-09-28", "-r", "0", "-o", "-") == 0
+    out = capsys.readouterr()
+    assert "new_a" in out.err
+    assert "new_b" not in out.err
+    assert json.loads(out.out) == {"counter": 2}
+
+    # Time and grep are both restrictive.
+    assert _run_cli(path, "-r", "2026-09-28", "--grep", "new_b") == 0
+    err = capsys.readouterr().err
+    assert "new_a" not in err
+    assert "new_b" in err
+
+
+def test_cli_time_matches_items_not_windows(tmp_path, capsys):
+    """Non-monotonic timestamps: items match individually, not by position."""
+    path = tmp_path / "data.kantadb"
+    _write_timed(
+        path,
+        [
+            (datetime(2026, 9, 28, 12, tzinfo=UTC), "noon", 1),
+            (datetime(2026, 9, 28, 8, tzinfo=UTC), "morning", 2),
+            (datetime(2026, 9, 28, 10, tzinfo=UTC), "ten", 3),
+        ],
+    )
+    capsys.readouterr()
+
+    assert _run_cli(path, "-r", "2026-09-28T10") == 0
+    err = capsys.readouterr().err
+    assert "ten" in err
+    assert "noon" not in err
+    assert "morning" not in err
+
+
+def test_cli_time_relative(tmp_path, capsys):
+    path = tmp_path / "data.kantadb"
+    now = datetime.now(UTC)
+    _write_timed(
+        path,
+        [
+            (now - timedelta(hours=2), "old", 1),
+            (now - timedelta(minutes=30), "recent", 2),
+        ],
+    )
+    capsys.readouterr()
+
+    assert _run_cli(path, "-r", "1h") == 0
+    err = capsys.readouterr().err
+    assert "recent" in err
+    assert "old" not in err
+
+
+def test_cli_time_invalid_is_error(tmp_path, capsys):
+    path = tmp_path / "data.kantadb"
+    _write_timed(path, [(T0, "inc0", 1)])
+    capsys.readouterr()
+
+    assert _run_cli(path, "-r", "2026-13-45") == 1
+    assert "Invalid --range value" in capsys.readouterr().err