Filter snapshot lines through --grep like change records

Snapshot indicator lines printed unconditionally under --grep, suggesting
the snapshot matched. Now the snapshot's full state is flattened into the
same (path, value) entries change records are matched against, and the
snapshot line is suppressed unless every pattern matches. Snapshots
explicitly selected with -r s<N> still print unconditionally.

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,
@@ -538,6 +538,16 @@ async def _run(args: argparse.Namespace) -> int:
                     continue
                 label = record_label(event.line_number, event.record_index)
                 if isinstance(event, SnapshotEvent):
+                    if grep_patterns:
+                        logfmt = kanta._impl.callback_registry.build_logfmt(
+                            InjectionContext(
+                                kanta=kanta,
+                                previous_state=current,
+                                current_state=current,
+                            )
+                        )
+                        if not matches_snapshot(current, grep_patterns, logfmt=logfmt):
+                            continue
                     _print_snapshot_indicator(
                         label,
                         event.snap,

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:
+            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

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(