Clarify diagnostic log messages; simplify describe_callback fallback

Unnamed callables (partials, callable instances) are described by type
name only: docstrings are shown only for named callables, avoiding
misleading class docstrings in failure messages.

Diagnostic messages revised for clarity when mixed with application
logs; logger.exception() messages no longer repeat the exception text,
which the traceback already shows.

docs/database.md

@@ -171,9 +171,10 @@ def resolve_user_key(value: str) -> str | None:
 
 #### Transaction Log Headers
 
-- 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.
+- 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.
 
 #### Log Emitters
 
@@ -203,6 +204,7 @@ def emit(ev: LogEvent):
   - `colors`: the mutable color palette. Colors are bare SGR parameter strings (e.g. `"1;34"`, `"38;5;226"`) without escape framing. Attributes are read at render time, so assignments (`colors.action = "36"`) and additions (`colors.session = "38;5;226"`) take effect immediately.
   - `Line`: builds a terminal string part by part. Calling it appends content (`str`-converted); `.<colorname>` arms a palette color for the next call only, and the reset is folded into a single escape sequence with whatever color comes next. `width=`/`align=` pad by display width; `str(line)` finishes the line and restores default colors.
   - `strip_ansi`, `displaywidth` (wide chars and emoji count correctly) and `pad` for working with pre-colored strings.
+  - `use_color(stream)`: the color-support test used by Kanta's own output — honors `NO_COLOR`/`FORCE_COLOR`, then `stream.isatty()`, then the journald `JOURNAL_STREAM` device/inode match.
 
 ## Migrations
 

kanta/__init__.py

@@ -1,4 +1,5 @@
 from .callbacks import DictPrev, DictState, LogFmt
+from .diff import diff, patch
 from .exceptions import DatabaseError
 from .kanta import Kanta
 from .logging import LogEvent, configure_logging
@@ -8,6 +9,8 @@ __all__ = [
     "Kanta",
     "DatabaseError",
     "configure_logging",
+    "diff",
+    "patch",
     # Callback argument types
     "DictPrev",
     "DictState",

kanta/__main__.py

@@ -6,8 +6,8 @@ import argparse
 import asyncio
 import contextlib
 import importlib
+import importlib.metadata
 import importlib.util
-import logging
 import sys
 import tempfile
 from pathlib import Path
@@ -16,9 +16,16 @@ from typing import Any
 import msgspec
 
 from kanta import Kanta
-from kanta.callbacks import InjectionContext
+from kanta.callbacks import InjectionContext, callback_error_reporter
 from kanta.exceptions import DatabaseError, DataIntegrityError, ReplayError
-from kanta.logging import LogEvent, emit_event, migration_logger
+from kanta.grep import GrepPattern, evaluate
+from kanta.logging import (
+    LogEvent,
+    emit_event,
+    format_action_header,
+    format_diff,
+    migration_logger,
+)
 from kanta.replaylog import (
     RangeNotFoundError,
     Selection,
@@ -33,7 +40,7 @@ from kanta.replaylog import (
 )
 from kanta.serialization import Serializer, dict_to_struct, struct_to_dict
 from kanta.structs import ChangeRecord, Snapshot
-from kanta.tty import Line
+from kanta.tty import Line, strip_ansi, use_color
 
 EXIT_SUCCESS = 0
 EXIT_GENERIC = 1
@@ -42,7 +49,18 @@ EXIT_PARSE_ERROR = 10
 EXIT_MIGRATION_ERROR = 20
 EXIT_VALIDATION_ERROR = 21
 
-_logger = logging.getLogger(__name__)
+
+def _print(*args: Any) -> None:
+    """Print to stderr, stripping ANSI codes when the stream has no color support.
+
+    Color detection runs per call so redirected or reassigned ``sys.stderr``
+    (and environment changes) are honored; ANSI codes are stripped after
+    formatting, not by formatting differently.
+    """
+    text = " ".join(str(arg) for arg in args)
+    if not use_color():
+        text = strip_ansi(text)
+    print(text, file=sys.stderr)
 
 
 class _CliError(Exception):
@@ -148,10 +166,26 @@ def _format_ts(dt) -> str:
     return dt.replace(tzinfo=None, microsecond=0).isoformat(sep=" ")
 
 
+def _package_version() -> str:
+    """Return the installed package version, or ``"unknown"`` from a source tree."""
+    try:
+        return importlib.metadata.version("kanta")
+    except importlib.metadata.PackageNotFoundError:
+        return "unknown"
+
+
 def _parse_args(argv: list[str] | None = None) -> argparse.Namespace:
     parser = argparse.ArgumentParser(
         prog="kanta",
-        description="Read a kantadb file and print each change record to the console.",
+        description=(
+            f"kanta {_package_version()} - read a kantadb file and print each"
+            " change record to the console."
+        ),
+    )
+    parser.add_argument(
+        "--version",
+        action="version",
+        version=f"%(prog)s {_package_version()}",
     )
     parser.add_argument(
         "file",
@@ -209,6 +243,31 @@ def _parse_args(argv: list[str] | None = None) -> argparse.Namespace:
             " '2:5', '2..5', 'l10:l20', 's1:s3', 'v0:v2', 's-1:', ':-1', '-1'."
         ),
     )
+    parser.add_argument(
+        "-g",
+        "--grep",
+        action="append",
+        metavar="PATTERN",
+        help=(
+            "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"
+            " 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,"
+            " unless it uses wildcards: 'users' matches 'users' anywhere but"
+            " not 'foousers', 'us*' does), or a value (strings by substring,"
+            " other values in full: 'true' matches a boolean, 'tru' does"
+            " not).  The 'path=value' form requires the path and the value"
+            " to match within the same change line; use '=value' or 'path='"
+            " to match values or paths only.  With -k, logfmt-prettified"
+            " values and users match alongside the raw ones.  Examples:"
+            " --grep alice, --grep 'users.*.email', --grep"
+            " 'users.alice.admin=true', --grep create_user --grep"
+            " '@example.com'."
+        ),
+    )
     args = parser.parse_args(argv)
     if args.kanta and (args.data or args.migrations):
         parser.error("-k/--kanta cannot be used together with -d or -m")
@@ -221,27 +280,36 @@ def _print_change_log(
     previous: dict[str, Any],
     current: dict[str, Any],
     kanta: Kanta[Any],
+    highlight: Any = None,
 ) -> None:
     """Log a single change record to stderr.
 
     The record is dispatched as a :class:`LogEvent` through the Kanta
     object's logemit handlers; the CLI's own rendering (with the ``l<N>``
     label and timestamp) is the fallback when no handler claims the event.
+    ``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)
 
     def render(ev) -> None:
         ts = _format_ts(record.ts)
-        lines = ev.diff_lines
+        if highlight is not None:
-        if not lines:
+            header = format_action_header(
-            print(f"{label} {ts} {ev.header}", file=sys.stderr)
+                ev.action or "", ev.user, ev.extra, highlight=highlight
-        elif len(lines) == 1:
+            )
-            print(f"{label} {ts} {ev.header}{lines[0]}", file=sys.stderr)
+            lines = format_diff(ev.diff, ev.previous, ev.logfmt, highlight=highlight)
         else:
-            print(f"{label} {ts} {ev.header}", file=sys.stderr)
+            header, lines = ev.header, ev.diff_lines
+        if not lines:
+            _print(f"{label} {ts} {header}")
+        elif len(lines) == 1:
+            _print(f"{label} {ts} {header}{lines[0]}")
+        else:
+            _print(f"{label} {ts} {header}")
             for line in lines:
-                print(line, file=sys.stderr)
+                _print(line)
-            print(file=sys.stderr)
+            _print()
 
     emit_event(
         event,
@@ -319,7 +387,7 @@ def _print_snapshot_indicator(
         line.target(f" {_format_ts(snap.m)}")
     size = len(serializer.encode(snap.state))
     line.path_prefix(f" {_format_size(size)}")
-    print(f"{label} {ts} {line}", file=sys.stderr)
+    _print(f"{label} {ts} {line}")
 
 
 async def _log_migration(
@@ -337,13 +405,11 @@ async def _log_migration(
     """
     registry = kanta._impl.callback_registry
     if registry.has("logmigr"):
-        try:
+        await registry.invoke(
-            await registry.invoke(
+            "logmigr",
-                "logmigr",
+            InjectionContext(kanta=kanta, report=result),
-                InjectionContext(kanta=kanta, report=result),
+            on_error=callback_error_reporter("logmigr"),
-            )
+        )
-        except Exception:
-            _logger.exception("logmigr callback failed")
         return
     if quiet:
         return
@@ -359,7 +425,7 @@ async def _log_migration(
             migrations=descriptions,
         ),
         registry.logemit_handlers,
-        fallback=lambda ev: print(ev.header, file=sys.stderr),
+        fallback=lambda ev: _print(ev.header),
     )
 
 
@@ -446,6 +512,8 @@ async def _run(args: argparse.Namespace) -> int:
         except ValueError as exc:
             raise _CliError(f"Invalid --range value: {exc}") from exc
 
+        grep_patterns = [GrepPattern.parse(p) for p in args.grep or ()]
+
         if selection.snapshot is not None:
             snap_event = selection.snapshot
             state = snap_event.snap.state
@@ -457,7 +525,7 @@ async def _run(args: argparse.Namespace) -> int:
                     snapshot_line_to_index[snap_event.line_number],
                     kanta._impl.serializer,
                 )
-                print(file=sys.stderr)
+                _print()
         else:
             # Replay up to the range end, printing logs within the range.
             state = {}
@@ -478,10 +546,28 @@ async def _run(args: argparse.Namespace) -> int:
                     )
                 else:
                     assert previous is not None
-                    _print_change_log(label, event.record, previous, current, kanta)
+                    highlight = None
+                    if grep_patterns:
+                        # Build the same logfmt the rendering uses, so both
+                        # raw and prettified values are matched.
+                        logfmt = kanta._impl.callback_registry.build_logfmt(
+                            InjectionContext(
+                                kanta=kanta,
+                                previous_state=previous,
+                                current_state=current,
+                            )
+                        )
+                        highlight = evaluate(
+                            event.record, previous, grep_patterns, logfmt=logfmt
+                        )
+                        if highlight is None:
+                            continue
+                    _print_change_log(
+                        label, event.record, previous, current, kanta, highlight
+                    )
                 printed = True
             if printed:
-                print(file=sys.stderr)
+                _print()
 
         # Apply optional migrations to the range-end state.
         if kanta._impl.migrations is not None:
@@ -518,7 +604,7 @@ async def _run(args: argparse.Namespace) -> int:
                 # The file was already fully decoded and validated above with
                 # the object's own serializer, and its migrations were applied
                 # to the state; no need to re-open through a new instance.
-                print(f"{data}", file=sys.stderr)
+                _print(f"{data}")
                 output_state = struct_to_dict(data, serializer=kanta._impl.serializer)
             else:
                 kanta_typed = Kanta(
@@ -526,7 +612,7 @@ async def _run(args: argparse.Namespace) -> int:
                 )
                 try:
                     await kanta_typed.open(create=False, readonly=True, log=False)
-                    print(f"{data}", file=sys.stderr)
+                    _print(f"{data}")
                 except (msgspec.ValidationError, msgspec.DecodeError) as exc:
                     raise _CliError(
                         f"Validation error: {exc}", EXIT_VALIDATION_ERROR
@@ -577,7 +663,7 @@ def main(argv: list[str] | None = None) -> int:
     try:
         return asyncio.run(_run(args))
     except _CliError as exc:
-        print(exc, file=sys.stderr)
+        _print(exc)
         return exc.code
 
 

kanta/callbacks.py

@@ -54,7 +54,40 @@ def _state_tag(ann: Any) -> str | None:
     return None
 
 
-_logger = logging.getLogger(__name__)
+_logger = logging.getLogger("kanta")
+
+
+def describe_callback(callback: Callable[..., Any]) -> str:
+    """Return ``name (docstring first line)`` identifying *callback*.
+
+    Used in failure messages so a bare log line names the function that
+    failed, e.g. ``myformatter (Concise log formatter)``.  Callables without
+    a ``__name__`` (partials, callable instances, ...) are described by
+    their type name only: less information, but never wrong information.
+    """
+    name = getattr(callback, "__name__", None)
+    if not isinstance(name, str):
+        return type(callback).__name__
+    doc = inspect.getdoc(callback)
+    if doc:
+        return f"{name} ({doc.splitlines()[0]})"
+    return name
+
+
+def callback_error_reporter(
+    kind: str,
+) -> Callable[[Exception, Callable[..., Any]], None]:
+    """Return an ``on_error`` reporter for :meth:`CallbackRegistry.invoke`.
+
+    The returned callable logs ``Kanta.<kind> <name (docstring)> failed``
+    with the traceback for each failing callback; invoke continues with
+    the rest.
+    """
+
+    def _report(callback_error: Exception, callback: Callable[..., Any]) -> None:
+        _logger.exception("Kanta.%s %s failed", kind, describe_callback(callback))
+
+    return _report
 
 
 class LogFmt:
@@ -245,12 +278,14 @@ class CallbackRegistry:
 
     def build_logfmt(self, ctx: InjectionContext) -> Callable[[Any, str], str | None]:
         """Build a chained formatter from registered logfmt callbacks."""
-        formatters: list[tuple[Callable[[Any, str], str | None], str | None]] = []
+        formatters: list[
+            tuple[Callable[[Any, str], str | None], str | None, Callable[..., Any]]
+        ] = []
         for spec in self._logfmt_callbacks:
             if isinstance(spec, _LogFmtClassSpec):
                 kwargs = self._build_kwargs(spec.inject_params, ctx)
                 instance: Callable[[Any, str], str | None] = spec.cls(**kwargs)
-                formatters.append((instance, spec.path))
+                formatters.append((instance, spec.path, spec.cls))
             else:
                 kwargs = self._build_kwargs(spec.inject_params, ctx)
 
@@ -270,10 +305,10 @@ class CallbackRegistry:
 
                     return formatter
 
-                formatters.append((make_formatter(), spec.path))
+                formatters.append((make_formatter(), spec.path, spec.callback))
 
         def format_value(value: Any, path: str) -> str | None:
-            for fn, pattern in formatters:
+            for fn, pattern, callback in formatters:
                 if pattern is not None and path != pattern:
                     continue
                 try:
@@ -281,7 +316,9 @@ class CallbackRegistry:
                 except Exception:
                     # Formatting must never break functionality; a failing
                     # callback is reported and treated as a fall-through.
-                    _logger.exception("logfmt callback %r failed", fn)
+                    _logger.exception(
+                        "Kanta.logfmt %s failed", describe_callback(callback)
+                    )
                     continue
                 if resolved is not None:
                     return resolved

kanta/diff.py

@@ -49,16 +49,16 @@ def _diff(previous, current):
     return current
 
 
-def compute_diff(previous: dict, current: dict) -> dict | None:
+def diff(previous: dict, current: dict) -> dict | None:
     """Compute a marshaled diff between two state dicts.
 
     Returns None if there is no difference.
     """
-    diff = _diff(previous, current)
+    result = _diff(previous, current)
-    return diff if diff is not _UNCHANGED else None
+    return result if result is not _UNCHANGED else None
 
 
-def patch_state(state: dict, diff: dict) -> dict:
+def patch(state: dict, diff: dict) -> dict:
     """Apply a marshaled diff to a state dict."""
     return apply_diff(state, diff)
 

kanta/filelock.py

@@ -16,7 +16,7 @@ from pathlib import Path
 
 from kanta.exceptions import FileLockError
 
-_logger = logging.getLogger(__name__)
+_logger = logging.getLogger("kanta")
 
 
 def _fatal(msg: str, *, db_path: Path | None = None) -> None:

kanta/grep.py

@@ -0,0 +1,433 @@
+"""Structural grep matching and match highlighting for change records.
+
+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.
+
+Path terms match element-wise: each dotted element of the term must match
+a whole path element (``users`` matches ``users`` anywhere in the path
+but not ``foousers``), unless the element uses shell wildcards
+(``us*``).  The term's elements match as a contiguous sequence, so
+``users.*.email`` matches the path ``users.alice.email``.  Only the
+matched elements are highlighted.
+
+Value terms match strings by substring and all other values (booleans,
+numbers, null) only in full: ``true`` matches a boolean but ``tru`` does
+not.  A term with wildcards matches the whole value text.  Matched
+substrings — or the whole scalar — are highlighted.  When a logfmt
+formatter (``-k``) prettifies a value or the user, both the raw and the
+prettified form are matched.
+
+A matched record carries a :class:`GrepHighlighter`, which the
+:kanta.logging formatters consult to wrap exactly the matched regions
+with a yellow background.
+"""
+
+from __future__ import annotations
+
+import dataclasses
+import fnmatch
+import json
+from collections.abc import Iterator
+from typing import Any
+
+from kanta.logging import _USER_PATH, _collect_changes, _get_nested
+from kanta.serialization.base import unmarshal
+from kanta.structs import ChangeRecord
+from kanta.tty import ANSI_RE, ESC, 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
+
+
+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.
+
+    ``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.
+    """
+    spans = _merge_spans(spans)
+    if not spans:
+        return styled
+    out: list[str] = []
+    plain_pos = 0
+    prev_end = 0
+    for match in ANSI_RE.finditer(styled):
+        out.append(_wrap_run(styled[prev_end : match.start()], plain_pos, spans))
+        plain_pos += match.start() - prev_end
+        out.append(match.group(0))
+        prev_end = match.end()
+    out.append(_wrap_run(styled[prev_end:], plain_pos, spans))
+    return "".join(out)
+
+
+def _merge_spans(spans: list[tuple[int, int]]) -> list[tuple[int, int]]:
+    """Return *spans* sorted, with overlapping and adjacent spans merged."""
+    merged: list[list[int]] = []
+    for start, end in sorted(spans):
+        if start >= end:
+            continue
+        if merged and start <= merged[-1][1]:
+            merged[-1][1] = max(end, merged[-1][1])
+        else:
+            merged.append([start, end])
+    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."""
+    if not run:
+        return run
+    out: list[str] = []
+    pos = 0
+    for start, end in spans:
+        s = max(start - plain_start, 0)
+        e = min(end - plain_start, len(run))
+        if s >= e or e <= pos:
+            continue
+        out.append(run[pos:s])
+        out.append(f"{_MARK_BG}{run[s:e]}{_UNMARK_BG}")
+        pos = e
+    out.append(run[pos:])
+    return "".join(out)
+
+
+def _find_spans(text: str, needle: str) -> list[tuple[int, int]]:
+    """Return visible-text spans of every case-insensitive occurrence of *needle*."""
+    if not needle:
+        return []
+    haystack = strip_ansi(text).lower()
+    needle = needle.lower()
+    spans = []
+    pos = 0
+    while (found := haystack.find(needle, pos)) >= 0:
+        end = found + len(needle)
+        spans.append((found, end))
+        pos = end
+    return spans
+
+
+@dataclasses.dataclass(frozen=True)
+class GrepPattern:
+    """One parsed ``--grep`` pattern.
+
+    The bare form (``term`` set) matches the action, the user, or any
+    dotted path or value in the diff.  The ``path=value`` form
+    (``path_term`` and ``value_term`` set) requires both sides to match
+    within the same change line; either side may be left empty to match
+    values only (``=value``) or paths only (``path=``).
+    """
+
+    raw: str
+    term: str | None = None
+    path_term: str | None = None
+    value_term: str | None = None
+
+    @classmethod
+    def parse(cls, raw: str) -> GrepPattern:
+        """Parse a pattern, splitting the ``path=value`` form on the first ``=``."""
+        if "=" in raw:
+            path_term, value_term = raw.split("=", 1)
+            return cls(raw, path_term=path_term, value_term=value_term)
+        return cls(raw, term=raw)
+
+
+@dataclasses.dataclass(frozen=True)
+class _ValueMark:
+    """A value-side match.
+
+    ``needle`` is the text to locate in the displayed value ("" = a match
+    that marks nothing, e.g. from an empty term).  ``whole`` marks the
+    entire displayed value: used when the raw value matched but a logfmt
+    formatter displays something else, so no needle can be located.
+    """
+
+    needle: str = ""
+    whole: bool = False
+
+
+@dataclasses.dataclass
+class _Entry:
+    """One matchable ``(path, value)`` line of a record's flattened diff.
+
+    ``anchor`` is set for deleted content: the deleted path whose line is
+    displayed for this entry (the entry itself may sit below it).
+    """
+
+    path: list[str]
+    raw: Any
+    text: str
+    anchor: list[str] | None = None
+
+
+def _element_matches(pattern: str, element: str) -> bool:
+    """Match one path element: in full, or as a glob when it uses wildcards."""
+    pattern = pattern.casefold()
+    element = element.casefold()
+    if any(char in pattern for char in _GLOB_CHARS):
+        return fnmatch.fnmatchcase(element, pattern)
+    return element == pattern
+
+
+def _match_path(term: str, path: list[str]) -> frozenset[int] | None:
+    """Match a dotted term against *path* as a contiguous element sequence.
+
+    Returns the indices of the matched elements, or ``None``.  An empty
+    term matches anything and marks no elements.
+    """
+    if not term:
+        return frozenset()
+    patterns = term.split(".")
+    for start in range(len(path) - len(patterns) + 1):
+        if all(
+            _element_matches(pattern, path[start + i])
+            for i, pattern in enumerate(patterns)
+        ):
+            return frozenset(range(start, start + len(patterns)))
+    return None
+
+
+def _match_value(term: str, raw: Any, text: str) -> _ValueMark | None:
+    """Match a term against a value.
+
+    String values match by substring; containers do not match (their
+    leaves are matched individually); all other values match only in
+    full.  A term with wildcards matches the whole value text.
+    """
+    if not term:
+        return _ValueMark()
+    needle = term.casefold()
+    haystack = text.casefold()
+    if any(char in needle for char in _GLOB_CHARS):
+        return _ValueMark(text) if fnmatch.fnmatchcase(haystack, needle) else None
+    if isinstance(raw, str):
+        return _ValueMark(term) if needle in haystack else None
+    if isinstance(raw, (dict, list)):
+        return None
+    return _ValueMark(text) if needle == haystack else None
+
+
+def _dual_mark(term: str, raw: Any, text: str, pretty: str | None) -> _ValueMark | None:
+    """Match a term against both the raw and the prettified form of a value.
+
+    *pretty* is the logfmt-resolved display text, which is what the log
+    shows when set.  A match on the displayed form locates its needle
+    there; a match on the raw form alone marks the whole displayed value.
+    """
+    raw_mark = _match_value(term, raw, text)
+    if pretty is None:
+        return raw_mark
+    pretty_mark = _match_value(term, pretty, pretty)
+    if pretty_mark is not None:
+        return pretty_mark
+    if raw_mark is not None and (raw_mark.needle or raw_mark.whole):
+        return _ValueMark(whole=True)
+    return raw_mark
+
+
+def _value_text(value: Any) -> str:
+    """Render a value as matchable text, following the display conventions."""
+    if value is None:
+        return "null"
+    if isinstance(value, bool):
+        return "true" if value else "false"
+    if isinstance(value, (int, float)):
+        return str(value)
+    if isinstance(value, str):
+        return value
+    if isinstance(value, (dict, list)):
+        try:
+            return json.dumps(value, default=str)
+        except (TypeError, ValueError):
+            pass
+    return str(value)
+
+
+def _leaf_entries(
+    path: list[str], value: Any, anchor: list[str] | None
+) -> Iterator[_Entry]:
+    """Yield an entry for every value under *value*.
+
+    Added, replaced or deleted containers are a single change line in the
+    log but hold many values; descending into them lets patterns match
+    their content.  List elements are addressed by index (``users.0``).
+    """
+    items: Iterator[tuple[str, Any]]
+    if isinstance(value, dict):
+        items = ((str(key), item) for key, item in value.items())
+    elif isinstance(value, list):
+        items = ((str(index), item) for index, item in enumerate(value))
+    else:
+        return
+    for key, item in items:
+        item_path = [*path, key]
+        yield _Entry(item_path, item, _value_text(item), anchor)
+        yield from _leaf_entries(item_path, item, anchor)
+
+
+def _record_entries(record: ChangeRecord, previous: dict | None) -> list[_Entry]:
+    """Flatten a record's diff into matchable entries.
+
+    Uses the same traversal as the change-log rendering, so paths match
+    what the log shows; deleted paths carry their previous value.
+    """
+    changes: list[tuple[str, list[str], Any]] = []
+    _collect_changes(unmarshal(record.diff), [], changes, previous)
+    entries: list[_Entry] = []
+    for change_type, path, value in changes:
+        anchor = None
+        if change_type == "delete":
+            value = _get_nested(previous, path)
+            anchor = path
+        entries.append(_Entry(path, value, _value_text(value), anchor))
+        entries.extend(_leaf_entries(path, value, anchor))
+    return entries
+
+
+class GrepHighlighter:
+    """The matched regions of one record, wrapping rendered text on demand.
+
+    Implements the highlighter hook of the :mod:`kanta.logging`
+    formatters: ``path`` for path elements, ``value`` for values,
+    ``meta`` for header fields and ``delete`` for deletion markers.
+    All wrapping goes through :func:`mark_spans`, so overlapping matches
+    merge into one highlight.
+    """
+
+    def __init__(self) -> None:
+        self._lit_paths: set[str] = set()
+        self._lit_deletes: set[str] = set()
+        self._value_marks: dict[str, list[_ValueMark]] = {}
+        self._meta_marks: dict[str, list[_ValueMark]] = {}
+
+    def _light_elements(self, path: list[str], indices) -> None:
+        for i in indices:
+            self._lit_paths.add(".".join(path[: i + 1]))
+
+    def add(
+        self,
+        entry: _Entry,
+        elements: frozenset[int] | None,
+        vmark: _ValueMark | None,
+    ) -> None:
+        """Record one entry's match: lit element indices and/or a value mark."""
+        marked = vmark is not None and (vmark.needle or vmark.whole)
+        if entry.anchor is None:
+            if elements:
+                self._light_elements(entry.path, elements)
+            if marked:
+                key = ".".join(entry.path)
+                self._value_marks.setdefault(key, []).append(vmark)
+            return
+        # Deleted content: only the anchor path line is displayed.  Light
+        # the genuinely matched elements within it; a match on the removed
+        # value or below the anchor marks the deletion marker (✗) instead.
+        if elements:
+            shown = {i for i in elements if i < len(entry.anchor)}
+            if shown:
+                self._light_elements(entry.path, shown)
+            if len(shown) != len(elements):
+                self._lit_deletes.add(".".join(entry.anchor))
+        if marked:
+            self._lit_deletes.add(".".join(entry.anchor))
+
+    def add_meta(self, field: str, mark: _ValueMark) -> None:
+        """Record a header match on ``field`` (``"action"`` or ``"user"``)."""
+        if mark.needle or mark.whole:
+            self._meta_marks.setdefault(field, []).append(mark)
+
+    def path(self, text: str, path: str) -> str:
+        """Wrap a rendered path element when its element matched."""
+        if text and path in self._lit_paths:
+            return mark_spans(text, [(0, len(strip_ansi(text)))])
+        return text
+
+    def delete(self, text: str, path: str) -> str:
+        """Wrap the deletion marker when the removed content matched."""
+        if text and path in self._lit_deletes:
+            return mark_spans(text, [(0, len(strip_ansi(text)))])
+        return text
+
+    @staticmethod
+    def _apply_marks(text: str, marks: list[_ValueMark]) -> str:
+        if not text or not marks:
+            return text
+        spans: list[tuple[int, int]] = []
+        for mark in marks:
+            if mark.whole:
+                spans.append((0, len(strip_ansi(text))))
+            else:
+                spans.extend(_find_spans(text, mark.needle))
+        return mark_spans(text, spans)
+
+    def value(self, text: str, path: str) -> str:
+        """Wrap the matched regions of a rendered value."""
+        return self._apply_marks(text, self._value_marks.get(path, []))
+
+    def meta(self, text: str, field: str) -> str:
+        """Wrap the matched regions of a rendered header field."""
+        return self._apply_marks(text, self._meta_marks.get(field, []))
+
+
+def evaluate(
+    record: ChangeRecord,
+    previous: dict | None,
+    patterns: list[GrepPattern],
+    logfmt: Any = None,
+) -> GrepHighlighter | None:
+    """Match *patterns* against a record, returning its matched regions.
+
+    Returns ``None`` when any pattern matches nowhere in the transaction.
+    Otherwise every pattern contributed its matches — action, user, or
+    change lines — to the returned highlighter; different patterns may
+    match different lines of the same record.
+
+    ``logfmt`` is the optional composed logfmt callable; when given, both
+    the raw and the prettified form of each value (and of the user) are
+    matched.
+    """
+    entries = _record_entries(record, previous)
+    highlighter = GrepHighlighter()
+    for pattern in patterns:
+        matched = False
+        if pattern.term is not None:
+            mark = _match_value(pattern.term, record.a, record.a)
+            if mark is not None:
+                highlighter.add_meta("action", mark)
+                matched = True
+            if record.u:
+                pretty_user = (
+                    logfmt(record.u, _USER_PATH) if logfmt is not None else None
+                )
+                mark = _dual_mark(pattern.term, record.u, record.u, pretty_user)
+                if mark is not None:
+                    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
+            matched = True
+            highlighter.add(entry, elements, vmark)
+        if not matched:
+            return None
+    return highlighter

kanta/kantaimpl.py

@@ -10,7 +10,7 @@ from datetime import UTC, datetime, timedelta
 from types import SimpleNamespace
 from typing import Any, Generic, TypeVar
 
-from kanta.callbacks import CallbackRegistry, InjectionContext
+from kanta.callbacks import CallbackRegistry, InjectionContext, callback_error_reporter
 from kanta.exceptions import DatabaseError, DataIntegrityError, ReplayError
 from kanta.logging import (
     _USER_PATH,
@@ -25,16 +25,11 @@ from kanta.rotation import execute_rotation, plan_rotation
 from kanta.serialization import restore_data_in_place, struct_to_dict
 from kanta.serialization.base import replay
 
-_logger = logging.getLogger(__name__)
+_logger = logging.getLogger("kanta")
 
 T = TypeVar("T")
 
 
-def _log_callback_error(callback_error, callback):
-    """Report a failing logging callback and continue with the next one."""
-    _logger.exception("Log callback %r failed: %s", callback, callback_error)
-
-
 class KantaImpl(PersistenceMixin, Generic[T]):
     """Internal state and logic for Kanta."""
 
@@ -119,7 +114,7 @@ class KantaImpl(PersistenceMixin, Generic[T]):
                     kanta=self._kanta,
                     report=report,
                 ),
-                on_error=_log_callback_error,
+                on_error=callback_error_reporter("logmigr"),
             )
             return
 

kanta/logging.py

@@ -5,6 +5,8 @@ through :func:`emit_event`, which runs any registered ``logemit`` callbacks
 and falls back to :func:`default_emit` for the built-in formatting.  Diff
 output is formatted in a human-readable path notation style with color
 coding; see :mod:`kanta.tty` for the color palette and line builder.
+ANSI codes are stripped at emit time when the standard error stream does
+not support color (``NO_COLOR``/``FORCE_COLOR``, tty and journald checks).
 """
 
 import logging
@@ -15,14 +17,25 @@ from typing import Any
 
 import msgspec
 
+from kanta.callbacks import describe_callback
 from kanta.serialization.base import _apply, unmarshal
-from kanta.tty import Line, displaywidth
+from kanta.tty import Line, displaywidth, strip_ansi, use_color
 
 transaction_logger = logging.getLogger("kanta.transaction")
 bootstrap_logger = logging.getLogger("kanta.bootstrap")
 migration_logger = logging.getLogger("kanta.migration")
 
-_logger = logging.getLogger(__name__)
+# Event loggers carry Kanta-rendered content (colored headers, diffs) and are
+# configured at import time; diagnostics from Kanta's internals use the plain
+# "kanta" logger so they follow the application's root logging configuration.
+EVENT_LOGGERS = ("kanta.bootstrap", "kanta.migration", "kanta.transaction")
+
+# Loggers that emit DEBUG-level events (file-opened summary, migration diffs).
+_DEBUG_LOGGERS = ("kanta.bootstrap", "kanta.migration")
+
+_PLAIN_HANDLER_NAME = "kanta.plain"
+
+_logger = logging.getLogger("kanta")
 
 # Pattern to match control characters and bidirectional overrides
 _UNSAFE_CHARS = re.compile(
@@ -146,13 +159,21 @@ def emit_event(
             try:
                 proceed = handler(ev)
             except Exception:
-                _logger.exception("logemit callback failed, using default formatting")
+                _logger.exception(
+                    "Kanta.logemit %s failed, using default formatting",
+                    describe_callback(handler),
+                )
                 break
             if not proceed:
                 return
         render(ev)
     except Exception:
-        _logger.exception("failed to emit %s log event", ev.kind)
+        _logger.exception("Kanta failed to emit %s log event", ev.kind)
+
+
+def _maybe_strip(text: str) -> str:
+    """Strip ANSI codes from *text* when stderr has no color support."""
+    return text if use_color() else strip_ansi(text)
 
 
 def default_emit(ev: LogEvent) -> None:
@@ -163,25 +184,28 @@ def default_emit(ev: LogEvent) -> None:
     logger so it can be silenced or routed separately from the headers.
     This is what runs when no logemit callback handles the event; custom
     callbacks may call it to delegate events they do not care about.
+
+    ANSI color codes are stripped after formatting when the standard error
+    stream does not support color (see :func:`kanta.tty.use_color`).
     """
     if ev.kind != "change":
-        ev.logger.log(ev.level, ev.header)
+        ev.logger.log(ev.level, _maybe_strip(ev.header))
         return
 
     diff_logger = logging.getLogger(f"{ev.logger.name}.diff")
     lines = ev.diff_lines if ev.show_diff and diff_logger.isEnabledFor(ev.level) else []
 
     if not lines:
-        ev.logger.log(ev.level, ev.header)
+        ev.logger.log(ev.level, _maybe_strip(ev.header))
         return
 
     if len(lines) == 1:
-        diff_logger.log(ev.level, f"{ev.header}{lines[0]}")
+        diff_logger.log(ev.level, _maybe_strip(f"{ev.header}{lines[0]}"))
         return
 
-    ev.logger.log(ev.level, ev.header)
+    ev.logger.log(ev.level, _maybe_strip(ev.header))
     for line in lines:
-        diff_logger.log(ev.level, line)
+        diff_logger.log(ev.level, _maybe_strip(line))
 
 
 def _join_path(path: str, key: str) -> str:
@@ -202,56 +226,80 @@ def _format_value(
     *,
     max_len: int = 60,
     logfmt: Callable[[Any, str], str | None] | None = None,
+    highlight: Any = None,
 ) -> str:
-    """Format a value for display, truncating if needed."""
+    """Format a value for display, truncating if needed.
+
+    ``highlight`` is an optional hook with ``path(text, path)`` and
+    ``value(text, path)`` methods (see :class:`kanta.grep.GrepHighlighter`);
+    it wraps matched keys and scalar values, and recurses into containers.
+    """
     if logfmt is not None:
         resolved = logfmt(value, path)
         if resolved is not None:
+            if highlight is not None:
+                resolved = highlight.value(resolved, path)
             return resolved
 
+    def keyed(key: Any, key_path: str) -> str:
+        display = _format_value(key, key_path, max_len=30, logfmt=logfmt)
+        if highlight is not None:
+            display = highlight.path(display, key_path)
+        return display
+
     if value is None:
-        return "null"
+        text = "null"
-    if isinstance(value, bool):
+    elif isinstance(value, bool):
-        return "true" if value else "false"
+        text = "true" if value else "false"
-    if isinstance(value, (int, float)):
+    elif isinstance(value, (int, float)):
-        return str(value)
+        text = str(value)
-    if isinstance(value, str):
+    elif isinstance(value, str):
-        value = _UNSAFE_CHARS.sub("", value)
+        text = _UNSAFE_CHARS.sub("", value)
-        if len(value) > max_len:
+        if len(text) > max_len:
-            return value[: max_len - 1] + _dim_ellipsis()
+            text = text[: max_len - 1] + _dim_ellipsis()
-        return value
+    elif isinstance(value, dict):
-    if isinstance(value, dict):
         if not value:
             return "{}"
         all_true = all(v is True for v in value.values())
         parts = []
         for k, v in value.items():
             key_path = _join_path(path, str(k))
-            key_display = _format_value(k, key_path, max_len=30, logfmt=logfmt)
+            key_display = keyed(k, key_path)
             if all_true:
                 parts.append(key_display)
             else:
-                val_display = _format_value(v, key_path, max_len=30, logfmt=logfmt)
+                val_display = _format_value(
+                    v, key_path, max_len=30, logfmt=logfmt, highlight=highlight
+                )
                 parts.append(f"{key_display}: {val_display}")
         return "{" + ", ".join(parts) + "}"
-    if isinstance(value, list):
+    elif isinstance(value, list):
         if not value:
             return "[]"
         parts = []
         for i, v in enumerate(value):
             item_path = _join_path(path, str(i))
-            parts.append(_format_value(v, item_path, max_len=30, logfmt=logfmt))
+            parts.append(
+                _format_value(
+                    v, item_path, max_len=30, logfmt=logfmt, highlight=highlight
+                )
+            )
         return "[" + ", ".join(parts) + "]"
-    text = str(value)
+    else:
-    if len(text) > max_len:
+        text = str(value)
-        text = text[: max_len - 1] + _dim_ellipsis()
+        if len(text) > max_len:
+            text = text[: max_len - 1] + _dim_ellipsis()
+    if highlight is not None:
+        text = highlight.value(text, path)
     return text
 
 
 def _format_path_components(
-    path: list[str], logfmt: Callable[[Any, str], str | None] | None
+    path: list[str],
+    logfmt: Callable[[Any, str], str | None] | None,
+    highlight: Any = None,
 ) -> list[str]:
-    """Return path components after applying formatters."""
+    """Return path components after applying formatters and match highlights."""
     if not path:
         return []
     result = []
@@ -262,6 +310,8 @@ def _format_path_components(
             resolved = logfmt(component, prefix_path)
             if resolved is not None:
                 display = resolved
+        if highlight is not None:
+            display = highlight.path(display, prefix_path)
         result.append(display)
     return result
 
@@ -270,12 +320,13 @@ def _format_path(
     path: list[str],
     logfmt: Callable[[Any, str], str | None] | None,
     final_color: str = "path_final",
+    highlight: Any = None,
 ) -> str:
     """Format a path as dot notation with prefix in dark grey, final colored.
 
     *final_color* names a color in the :data:`kanta.tty.colors` palette.
     """
-    components = _format_path_components(path, logfmt)
+    components = _format_path_components(path, logfmt, highlight)
     if not components:
         return ""
     line = Line()
@@ -370,25 +421,32 @@ def _format_change_lines(
     path: list[str],
     value: Any,
     logfmt: Callable[[Any, str], str | None] | None = None,
+    highlight: Any = None,
 ) -> list[str]:
     """Format a single change as one or more lines."""
     if change_type == "delete":
-        components = _format_path_components(path, logfmt)
+        components = _format_path_components(path, logfmt, highlight)
         line = Line()("  ")
         if len(components) > 1:
             line.path_prefix(".".join(components[:-1]) + ".")
-        line.delete(components[-1], " ✗")
+        marker = "✗"
+        if highlight is not None:
+            marker = highlight.delete(marker, ".".join(path))
+        line.delete(components[-1], " ", marker)
         return [str(line)]
 
     if change_type == "add":
-        path_str = _format_path(path, logfmt, final_color="add")
+        path_str = _format_path(path, logfmt, final_color="add", highlight=highlight)
         if isinstance(value, dict) and value:
             lines = [str(Line()("  ", path_str, " ").sep("="))]
             base_path = ".".join(path)
             keys = []
             for k in value:
                 key_path = _join_path(base_path, str(k))
-                keys.append((k, _format_value(k, key_path, max_len=30, logfmt=logfmt)))
+                key_display = _format_value(k, key_path, max_len=30, logfmt=logfmt)
+                if highlight is not None:
+                    key_display = highlight.path(key_display, key_path)
+                keys.append((k, key_display))
             field_width = max(displaywidth(kd) for _, kd in keys)
             field_width = max(field_width, 12)
             # Each item line is "    {key:{field_width}}: {value}"; budget the
@@ -397,7 +455,9 @@ def _format_change_lines(
             formatted_items = []
             for (k, key_display), v in zip(keys, value.values()):
                 key_path = _join_path(base_path, str(k))
-                v_str = _format_value(v, key_path, max_len=value_width, logfmt=logfmt)
+                v_str = _format_value(
+                    v, key_path, max_len=value_width, logfmt=logfmt, highlight=highlight
+                )
                 formatted_items.append((key_display, v_str))
             return lines + [
                 str(
@@ -407,11 +467,13 @@ def _format_change_lines(
                 )
                 for k, v in formatted_items
             ]
-        value_str = _format_value(value, ".".join(path), logfmt=logfmt)
+        value_str = _format_value(
+            value, ".".join(path), logfmt=logfmt, highlight=highlight
+        )
         return [str(Line()("  ", path_str, " ").sep("=")(" ", value_str))]
 
-    value_str = _format_value(value, ".".join(path), logfmt=logfmt)
+    value_str = _format_value(value, ".".join(path), logfmt=logfmt, highlight=highlight)
-    path_str = _format_path(path, logfmt=logfmt)
+    path_str = _format_path(path, logfmt=logfmt, highlight=highlight)
     return [str(Line()("  ", path_str, " ").sep("=")(" ", value_str))]
 
 
@@ -419,6 +481,7 @@ def format_diff(
     diff: dict,
     previous: dict | None = None,
     logfmt: Callable[[Any, str], str | None] | None = None,
+    highlight: Any = None,
 ) -> list[str]:
     """Format a JSON diff as human-readable lines.
 
@@ -429,6 +492,8 @@ def format_diff(
             ``path`` is a dot-notation string; ``"$user"`` is used for the
             transaction actor.  If the callable returns ``None``, default
             formatting is used.
+        highlight: Optional match highlighter hook (see
+            :class:`kanta.grep.GrepHighlighter`) wrapping matched regions.
 
     Returns a list of formatted lines (without newlines).
     """
@@ -438,7 +503,7 @@ def format_diff(
         return []
     lines = []
     for change_type, path, value in changes:
-        lines.extend(_format_change_lines(change_type, path, value, logfmt))
+        lines.extend(_format_change_lines(change_type, path, value, logfmt, highlight))
     return lines
 
 
@@ -446,12 +511,22 @@ def format_action_header(
     action: str,
     user: str | None = None,
     extra: Any = None,
+    highlight: Any = None,
 ) -> str:
-    """Format the default action header line."""
+    """Format the default action header line.
+
+    ``highlight`` is an optional hook with a ``meta(text, field)`` method
+    (see :class:`kanta.grep.GrepHighlighter`) wrapping matched regions of
+    the action and user fields.
+    """
+    if highlight is not None:
+        action = highlight.meta(action, "action")
     line = Line().action(action)
     if extra is not None and (extra := f"{extra}"):
         line(" ").target(extra)
     if user is not None and (user := f"{user}"):
+        if highlight is not None:
+            user = highlight.meta(user, "user")
         line(" by ").user(user)
     return str(line)
 
@@ -505,6 +580,15 @@ def log_change(
     )
 
 
+def _ensure_plain_handler(logger: logging.Logger) -> None:
+    """Attach Kanta's no-prefix stderr handler to *logger* if it has none."""
+    if not logger.handlers:
+        handler = logging.StreamHandler(sys.stderr)
+        handler.setFormatter(logging.Formatter("%(message)s"))
+        handler.name = _PLAIN_HANDLER_NAME
+        logger.addHandler(handler)
+
+
 def configure_logging(
     *,
     skiproot: bool = True,
@@ -516,13 +600,21 @@ def configure_logging(
 ) -> None:
     """Configure Kanta's default logging output.
 
+    Called once at import time with default arguments; call again to change
+    the toggles.  The event loggers ``kanta.bootstrap``, ``kanta.migration``
+    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.
+
+    No levels are set by default: the event loggers inherit the effective
+    level of the root logger.
+
     Args:
-        skiproot: If ``True`` (default), attach a no-prefix stderr handler to
+        skiproot: If ``True`` (default), event loggers print through Kanta's
-            the ``kanta`` logger and set ``kanta.propagate = False`` so Kanta
+            own plain handler without propagating to the root logger.  If
-            output is rendered directly without propagating to the root logger.
+            ``False``, Kanta's handler is removed and propagation enabled so
-            If ``False``, the child logger enable flags are still applied, but
+            the application's root logger renders event output instead.
-            no handler is added and ``kanta`` propagation is left untouched so
-            the application's root logger handles Kanta output.
         bootstrap: Whether bootstrap logs are enabled.
         migration: Whether migration logs are enabled.
         transaction: Whether transaction logs are enabled.
@@ -530,13 +622,10 @@ def configure_logging(
             only transaction headers are printed and diff formatting is
             skipped.  Per transaction this is controlled by the ``logdiff``
             argument of :meth:`Kanta.transaction`.
-        debug: Whether to set the ``kanta`` logger level to ``DEBUG`` instead
+        debug: Whether to set the event loggers that emit DEBUG-level output
-            of ``INFO``.  This reveals debug-level output such as migration
+            (bootstrap and migration) to ``DEBUG``, revealing output such as
-            diffs, which are hidden by default.
+            the file-opened summary and migration diffs.  ``False`` resets
-
+            them to inheriting the root level.
-    This helper is not called automatically; applications that want Kanta's
-    default output can call it, but most applications will configure logging
-    themselves.
     """
     logging.getLogger("kanta.transaction.diff").disabled = not diff
 
@@ -545,16 +634,21 @@ def configure_logging(
         ("kanta.migration", migration),
         ("kanta.transaction", transaction),
     ):
-        logging.getLogger(name).propagate = enabled
+        logging.getLogger(name).disabled = not enabled
 
-    if not skiproot:
+    for name in _DEBUG_LOGGERS:
-        return
+        logging.getLogger(name).setLevel(logging.DEBUG if debug else logging.NOTSET)
 
-    target = logging.getLogger("kanta")
+    for name in EVENT_LOGGERS:
-    target.propagate = False
+        logger = logging.getLogger(name)
+        if skiproot:
+            logger.propagate = False
+            _ensure_plain_handler(logger)
+        else:
+            logger.propagate = True
+            logger.handlers[:] = [
+                h for h in logger.handlers if h.name != _PLAIN_HANDLER_NAME
+            ]
 
-    if not target.handlers:
+
-        handler = logging.StreamHandler(sys.stderr)
+configure_logging()  # Import-time default setup; call again to reconfigure.
-        handler.setFormatter(logging.Formatter("%(message)s"))
-        target.addHandler(handler)
-    target.setLevel(logging.DEBUG if debug else logging.INFO)

kanta/migrations.py

@@ -13,7 +13,7 @@ from dataclasses import dataclass
 from types import ModuleType
 from typing import Any
 
-from kanta.diff import compute_diff
+from kanta.diff import diff
 from kanta.exceptions import DatabaseError
 
 # Cache registries by imported module object so that many Kanta instances using
@@ -179,7 +179,7 @@ class Migrations:
             self._call_migration(fn, data_dict, kanta)
             current_version = version
             changed = before != data_dict
-            diff = compute_diff(before, data_dict) if changed else None
+            delta = diff(before, data_dict) if changed else None
             desc = (fn.__doc__ or f"v{version}").split("\n")[0].rstrip(".")
             migrations.append(
                 MigrationInfo(
@@ -187,7 +187,7 @@ class Migrations:
                     description=desc,
                     version=version,
                     changed=changed,
-                    diff=diff,
+                    diff=delta,
                     before=before,
                 )
             )

kanta/persistence.py

@@ -12,8 +12,8 @@ from datetime import UTC, datetime
 from pathlib import Path
 from typing import Any
 
-from kanta.callbacks import CallbackRegistry, InjectionContext
+from kanta.callbacks import CallbackRegistry, InjectionContext, callback_error_reporter
-from kanta.diff import compute_diff
+from kanta.diff import diff
 from kanta.exceptions import DatabaseError, DataIntegrityError
 from kanta.filelock import LockedFile
 from kanta.structs import ChangeRecord
@@ -21,7 +21,7 @@ from kanta.serialization import JsonSerializer, Serializer
 from kanta.serialization.framing import Framer
 from kanta.snapshot import SnapshotState
 
-_logger = logging.getLogger(__name__)
+_logger = logging.getLogger("kanta")
 
 
 class PersistenceMixin:
@@ -110,20 +110,14 @@ class PersistenceMixin:
                 break
             except DatabaseError as e:
                 self.background_error = e
-
-                def _log_callback_error(callback_error, callback):
-                    _logger.exception(
-                        "Background error callback %r failed: %s",
-                        callback,
-                        callback_error,
-                    )
-
                 await self.callback_registry.invoke(
                     "fatal_error",
                     InjectionContext(error=e, kanta=self._kanta),
-                    on_error=_log_callback_error,
+                    on_error=callback_error_reporter("fatal_error"),
+                )
+                _logger.error(
+                    "Kanta background flush failed; automatic flushing stopped: %s", e
                 )
-                _logger.error("Background flush loop stopped: %s", e)
                 break
 
     def maybe_snapshot(self) -> None:
@@ -158,11 +152,11 @@ class PersistenceMixin:
             The queued :class:`ChangeRecord`, or ``None`` if the diff was empty
             and *force* is ``False``.
         """
-        diff = compute_diff(self.statedict, current)
+        delta = diff(self.statedict, current)
-        if not diff:
+        if not delta:
             if not force:
                 return None
-            diff = {}
+            delta = {}
 
         # The clock is only read when a record is actually queued.
         now = self.now()
@@ -182,7 +176,7 @@ class PersistenceMixin:
             v=self.version,
             u=user,
             m=m,
-            diff=diff,
+            diff=delta,
         )
         self.pending_changes.append(record)
         self.statedict = copy.deepcopy(current)

kanta/replaylog.py

@@ -16,7 +16,7 @@ from typing import TYPE_CHECKING, Any, Union
 import msgspec
 
 from kanta.callbacks import InjectionContext
-from kanta.diff import patch_state
+from kanta.diff import patch
 from kanta.exceptions import ReplayError
 from kanta.logging import _USER_PATH, LogEvent, transaction_logger
 from kanta.structs import ChangeRecord, Snapshot
@@ -116,7 +116,7 @@ def scan_events(content: bytes, kanta: Kanta[Any]) -> tuple[list[Event], int]:
                 events.append(SnapshotEvent(line_number, byte_pos, record_index, snap))
             else:
                 record = impl.serializer.decode(payload, type=ChangeRecord)
-                state = patch_state(state, record.diff)
+                state = patch(state, record.diff)
                 events.append(ChangeEvent(line_number, byte_pos, record_index, record))
                 change_count += 1
         except msgspec.DecodeError as exc:
@@ -146,7 +146,7 @@ def replay_events(
             yield event, None, state
         else:
             previous = copy.deepcopy(state)
-            state = patch_state(state, event.record.diff)
+            state = patch(state, event.record.diff)
             yield event, previous, state
 
 

kanta/rotation.py

@@ -20,7 +20,7 @@ from kanta.structs import ChangeRecord, Snapshot
 from kanta.serialization.base import Serializer, apply_diff
 from kanta.serialization.framing import Framer
 
-_logger = logging.getLogger(__name__)
+_logger = logging.getLogger("kanta")
 
 
 @dataclass
@@ -216,7 +216,8 @@ def execute_rotation(
         f.truncate(plan.cutoff_end)
     if log:
         _logger.info(
-            "rotated %s: kept %d change record(s), history before %s moved to %s",
+            "Rotated database %s: kept %d change record(s), "
+            "moved history before %s to %s",
             path,
             plan.retained_changes,
             plan.rotated_ts.isoformat(),

kanta/snapshot.py

@@ -10,7 +10,7 @@ from kanta.structs import Snapshot
 from kanta.serialization import JsonSerializer, Serializer
 from kanta.serialization.framing import Framer, LineFramer
 
-_logger = logging.getLogger(__name__)
+_logger = logging.getLogger("kanta")
 
 MINDIFFS = 100
 
@@ -69,7 +69,7 @@ class SnapshotState:
             self._write(file, version, state, ts, m=m)
             self._force_pending = False
         except Exception as exc:
-            _logger.error("snapshot: failed to write snapshot: %r", exc)
+            _logger.error("Kanta snapshot failed: %r", exc)
 
     def _write(
         self, file, version: int, state: dict, now: datetime, m: datetime | None = None

kanta/transaction.py

@@ -7,13 +7,13 @@ from contextlib import contextmanager
 from datetime import datetime
 from typing import Any
 
-from kanta.diff import compute_diff
+from kanta.diff import diff
 from kanta.exceptions import DataIntegrityError
 from kanta.callbacks import InjectionContext
 from kanta.logging import _USER_PATH, LogEvent, emit_event, transaction_logger
 from kanta.serialization import restore_data_in_place, struct_to_dict
 
-_logger = logging.getLogger(__name__)
+_logger = logging.getLogger("kanta")
 
 
 def _build_logfmt(impl, previous: dict, current: dict):
@@ -65,19 +65,19 @@ def transaction(
     if current_dict != impl.statedict:
         is_bootstrap = action in {"bootstrap"}
         if not (is_bootstrap and not impl.statedict):
-            diff = compute_diff(impl.statedict, current_dict)
+            delta = diff(impl.statedict, current_dict)
-            if diff:
+            if delta:
                 _logger.critical(
                     "Database state modified outside of transaction! "
                     "This indicates a bug where changes occurred without a transaction wrapper.\n"
                     "Changes detected: %s",
-                    diff,
+                    delta,
                 )
                 raise DataIntegrityError(
                     "Database state modified outside of transaction",
                     db_path=impl.db_path,
                     action=action,
-                    diff=diff,
+                    diff=delta,
                 )
 
     impl.in_transaction = True
@@ -86,8 +86,8 @@ def transaction(
     try:
         yield impl.data
         new_dict = struct_to_dict(impl.data, serializer=impl.serializer)
-        diff = compute_diff(impl.statedict, new_dict)
+        delta = diff(impl.statedict, new_dict)
-        if diff:
+        if delta:
             if impl.callback_registry.has("validate"):
                 impl.callback_registry.invoke_sync(
                     "validate",

kanta/tty.py

@@ -10,8 +10,12 @@ color instead of emitting a separate one.
 
 from __future__ import annotations
 
+import io
+import os
 import re
+import sys
 import unicodedata
+from contextlib import suppress
 from typing import Any
 
 ESC = "\x1b["
@@ -25,6 +29,21 @@ def strip_ansi(text: str) -> str:
     return ANSI_RE.sub("", text)
 
 
+def use_color(stream: io.TextIOBase = sys.stderr) -> bool:
+    """Test if the stream supports color codes."""
+    if os.environ.get("NO_COLOR"):  # Non empty means no (no-color.org)
+        return False
+    if os.environ.get("FORCE_COLOR", "") not in {"", "0"}:  # force-color.org, node
+        return True
+    if hasattr(stream, "isatty") and stream.isatty():
+        return True
+    with suppress(KeyError, ValueError, OSError):  # Journald does color (-ocat)
+        dev, ino = map(int, os.environ["JOURNAL_STREAM"].split(":", 1))
+        st = os.fstat(stream.fileno())
+        return st.st_dev == dev and st.st_ino == ino
+    return False
+
+
 def displaywidth(text: str) -> int:
     """Return the terminal column width of *text*, ignoring ANSI sequences.
 

tests/conftest.py

@@ -1,3 +1,5 @@
+import logging
+
 import pytest
 
 from kanta.serialization import JsonSerializer, MsgPackSerializer
@@ -12,3 +14,20 @@ from kanta.serialization import JsonSerializer, MsgPackSerializer
 )
 def format_config(request):
     return request.param
+
+
+@pytest.fixture(autouse=True)
+def _kanta_event_loggers_propagate():
+    """Let kanta's event loggers propagate so caplog captures their records.
+
+    Kanta configures them with ``propagate = False`` at import time, which
+    would hide their records from pytest's root-logger capture handler.
+    """
+    names = ("kanta.bootstrap", "kanta.migration", "kanta.transaction")
+    loggers = [logging.getLogger(name) for name in names]
+    previous = [logger.propagate for logger in loggers]
+    for logger in loggers:
+        logger.propagate = True
+    yield
+    for logger, propagate in zip(loggers, previous):
+        logger.propagate = propagate

tests/test_cli.py

@@ -3,6 +3,8 @@
 import sys
 from datetime import UTC, datetime
 
+import pytest
+
 from kanta.__main__ import (
     _extra_import_paths,
     _format_ts,
@@ -76,8 +78,10 @@ def test_extra_import_paths_ignores_other_python_versions(tmp_path, monkeypatch)
         assert str(other_site) not in sys.path
 
 
-def test_cli_snapshot_line_format(tmp_path, capsys):
+def test_cli_snapshot_line_format(tmp_path, capsys, monkeypatch):
     """Snapshot lines are timestamped and colored with metadata."""
+    monkeypatch.setenv("FORCE_COLOR", "1")
+    monkeypatch.delenv("NO_COLOR", raising=False)
     path = tmp_path / "test.kantadb"
     ts = datetime(2026, 8, 12, 10, 6, 52, 375398, tzinfo=UTC)
     mtime = datetime(2026, 8, 12, 9, 0, 0, tzinfo=UTC)
@@ -105,6 +109,47 @@ def test_cli_snapshot_line_format(tmp_path, capsys):
     assert "\x1b[38;5;242m 13 B" 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."""
+    monkeypatch.setenv("NO_COLOR", "1")
+    monkeypatch.delenv("FORCE_COLOR", raising=False)
+    path = tmp_path / "test.kantadb"
+    ts = datetime(2026, 8, 12, 10, 6, 52, 375398, tzinfo=UTC)
+    serializer = JsonSerializer()
+    framer = LineFramer()
+
+    snapshot = Snapshot(ts=ts, v=1, m=None, state={"counter": 5})
+    change = ChangeRecord(ts=ts, a="inc", v=1, u="user1", diff={"counter": 6})
+    data = framer.frame_snapshot(
+        serializer.encode(snapshot), record_offset=0
+    ) + framer.frame_change(serializer.encode(change), record_offset=0)
+    path.write_bytes(data)
+
+    code = main([str(path)])
+    assert code == 0
+
+    err = capsys.readouterr().err
+    assert "\x1b[" not in err
+    assert "snapshot s0" in err
+
+
+def test_cli_version_on_help_and_version_flag(capsys):
+    """--help and --version print the installed package version."""
+    import importlib.metadata
+
+    version = importlib.metadata.version("kanta")
+
+    with pytest.raises(SystemExit) as help_exit:
+        main(["--help"])
+    assert help_exit.value.code == 0
+    assert f"kanta {version}" in capsys.readouterr().out
+
+    with pytest.raises(SystemExit) as version_exit:
+        main(["--version"])
+    assert version_exit.value.code == 0
+    assert capsys.readouterr().out.strip() == f"kanta {version}"
+
+
 def test_import_dotted_from_file_path(tmp_path):
     """--data can be a filesystem path with an optional colon-separated symbol."""
     module = tmp_path / "models.py"

tests/test_diff.py

@@ -3,7 +3,7 @@
 jsondiff is a dev dependency used only here, to verify that:
 
 - jsondiff.patch(..., marshal=True) can apply patches produced by
-  compute_diff (our format is a subset of jsondiff's marshaled syntax);
+  diff (our format is a subset of jsondiff's marshaled syntax);
 - apply_diff can apply patches produced by jsondiff.diff(..., marshal=True),
   including positional $insert/$delete list edits and per-index nested diffs.
 """
@@ -11,88 +11,88 @@ jsondiff is a dev dependency used only here, to verify that:
 import jsondiff
 import pytest
 
-from kanta.diff import compute_diff, patch_state
+from kanta.diff import diff, patch
 from kanta.logging import format_diff
 from kanta.serialization.base import apply_diff
 
-# --- Producer: compute_diff ------------------------------------------------
+# --- Producer: diff ------------------------------------------------
 
 
 def test_no_diff():
-    assert compute_diff({"a": 1}, {"a": 1}) is None
+    assert diff({"a": 1}, {"a": 1}) is None
-    assert compute_diff({}, {}) is None
+    assert diff({}, {}) is None
 
 
 def test_simple_diff():
-    diff = compute_diff({"a": 1}, {"a": 2})
+    delta = diff({"a": 1}, {"a": 2})
-    assert diff is not None
+    assert delta is not None
-    assert diff == {"a": 2}
+    assert delta == {"a": 2}
 
 
 def test_nested_diff():
-    diff = compute_diff({"x": {"y": 1}}, {"x": {"y": 2}})
+    delta = diff({"x": {"y": 1}}, {"x": {"y": 2}})
-    assert diff == {"x": {"y": 2}}
+    assert delta == {"x": {"y": 2}}
 
 
 def test_key_added():
-    assert compute_diff({"a": 1}, {"a": 1, "b": 2}) == {"b": 2}
+    assert diff({"a": 1}, {"a": 1, "b": 2}) == {"b": 2}
 
 
 def test_key_removed():
-    assert compute_diff({"a": 1, "b": 2}, {"a": 1}) == {"$delete": ["b"]}
+    assert diff({"a": 1, "b": 2}, {"a": 1}) == {"$delete": ["b"]}
 
 
 def test_last_key_removed_is_delete_not_replace():
     # jsondiff's minimal-diff search emits {"$replace": {}} here; we emit
     # what actually happened: the key was deleted.
-    assert compute_diff({"a": 1}, {}) == {"$delete": ["a"]}
+    assert diff({"a": 1}, {}) == {"$delete": ["a"]}
-    assert compute_diff({"x": {"y": 1}}, {"x": {}}) == {"x": {"$delete": ["y"]}}
+    assert diff({"x": {"y": 1}}, {"x": {}}) == {"x": {"$delete": ["y"]}}
 
 
 def test_list_changes_are_full_assignment():
     # No $insert/$delete positional edits: lists are replaced wholesale.
-    assert compute_diff({"l": [1, 2]}, {"l": [1, 2, 3]}) == {"l": [1, 2, 3]}
+    assert diff({"l": [1, 2]}, {"l": [1, 2, 3]}) == {"l": [1, 2, 3]}
-    assert compute_diff({"l": [1, 2, 3]}, {"l": [1, 3]}) == {"l": [1, 3]}
+    assert diff({"l": [1, 2, 3]}, {"l": [1, 3]}) == {"l": [1, 3]}
-    assert compute_diff({"l": [1]}, {"l": []}) == {"l": []}
+    assert diff({"l": [1]}, {"l": []}) == {"l": []}
 
 
 def test_list_with_unchanged_prefix_is_full_assignment():
-    diff = compute_diff({"l": ["a", "b", "c"]}, {"l": ["a", "x", "b", "c"]})
+    delta = diff({"l": ["a", "b", "c"]}, {"l": ["a", "x", "b", "c"]})
-    assert diff == {"l": ["a", "x", "b", "c"]}
+    assert delta == {"l": ["a", "x", "b", "c"]}
 
 
 def test_type_changes_are_full_assignment():
-    assert compute_diff({"a": {"x": 1}}, {"a": [1]}) == {"a": [1]}
+    assert diff({"a": {"x": 1}}, {"a": [1]}) == {"a": [1]}
     # A dict replacing a non-dict is a plain assignment too: the consumer
     # sees from the old value whether to patch (dict) or replace.
-    assert compute_diff({"a": [1]}, {"a": {"x": 1}}) == {"a": {"x": 1}}
+    assert diff({"a": [1]}, {"a": {"x": 1}}) == {"a": {"x": 1}}
-    assert compute_diff({"a": 1}, {"a": None}) == {"a": None}
+    assert diff({"a": 1}, {"a": None}) == {"a": None}
 
 
 def test_new_dict_value_assigned_wholesale():
-    assert compute_diff({}, {"a": {"x": 1}}) == {"a": {"x": 1}}
+    assert diff({}, {"a": {"x": 1}}) == {"a": {"x": 1}}
 
 
 def test_dollar_keys_escaped():
-    assert compute_diff({}, {"$weird": 1}) == {"$$weird": 1}
+    assert diff({}, {"$weird": 1}) == {"$$weird": 1}
-    assert compute_diff({"$weird": 1}, {"$weird": 2}) == {"$$weird": 2}
+    assert diff({"$weird": 1}, {"$weird": 2}) == {"$$weird": 2}
-    assert compute_diff({"$weird": 1}, {}) == {"$delete": ["$$weird"]}
+    assert diff({"$weird": 1}, {}) == {"$delete": ["$$weird"]}
 
 
 def test_dollar_values_not_escaped():
     # Only keys are escaped; values are stored verbatim, even "$delete".
-    assert compute_diff({"s": 1}, {"s": "$y"}) == {"s": "$y"}
+    assert diff({"s": 1}, {"s": "$y"}) == {"s": "$y"}
-    assert compute_diff({"s": 1}, {"s": "$delete"}) == {"s": "$delete"}
+    assert diff({"s": 1}, {"s": "$delete"}) == {"s": "$delete"}
-    assert compute_diff({}, {"o": {"s": "$y", "l": ["$z"]}}) == {
+    assert diff({}, {"o": {"s": "$y", "l": ["$z"]}}) == {
         "o": {"s": "$y", "l": ["$z"]}
     }
 
 
-# --- Consumer: apply_diff / patch_state -------------------------------------
+# --- Consumer: apply_diff / patch -------------------------------------
 
 
-def test_patch_state_delegates():
+def test_patch_delegates():
-    assert patch_state({"a": 1}, {"a": 2}) == {"a": 2}
+    assert patch({"a": 1}, {"a": 2}) == {"a": 2}
 
 
 def test_apply_scalar_and_add():
@@ -218,9 +218,9 @@ JSONDIFF_APPLIES_CASES = [
     "name,old,new", JSONDIFF_APPLIES_CASES, ids=[c[0] for c in JSONDIFF_APPLIES_CASES]
 )
 def test_jsondiff_applies_our_patches(name, old, new):
-    diff = compute_diff(old, new)
+    delta = diff(old, new)
-    assert diff is not None
+    assert delta is not None
-    assert jsondiff.patch(old, diff, marshal=True) == new
+    assert jsondiff.patch(old, delta, marshal=True) == new
 
 
 @pytest.mark.parametrize("name,old,new", COMPAT_CASES, ids=[c[0] for c in COMPAT_CASES])
@@ -231,15 +231,15 @@ def test_we_apply_jsondiff_patches(name, old, new):
 
 @pytest.mark.parametrize("name,old,new", COMPAT_CASES, ids=[c[0] for c in COMPAT_CASES])
 def test_our_own_round_trip(name, old, new):
-    diff = compute_diff(old, new)
+    delta = diff(old, new)
-    assert diff is not None
+    assert delta is not None
-    assert apply_diff(old, diff) == new
+    assert apply_diff(old, delta) == new
 
 
 def test_no_diff_means_equal_states():
     for _name, old, new in COMPAT_CASES:
-        assert compute_diff(old, new) is not None  # cases really differ
+        assert diff(old, new) is not None  # cases really differ
-    assert compute_diff({"a": [1, {"b": "$x"}]}, {"a": [1, {"b": "$x"}]}) is None
+    assert diff({"a": [1, {"b": "$x"}]}, {"a": [1, {"b": "$x"}]}) is None
 
 
 # --- Logging ----------------------------------------------------------------

tests/test_grep.py

@@ -0,0 +1,610 @@
+"""Tests for structural ``--grep`` matching and match highlighting."""
+
+from datetime import UTC, datetime
+
+from kanta.__main__ import main
+from kanta.grep import (
+    GrepPattern,
+    _Entry,
+    _find_spans,
+    _match_path,
+    _match_value,
+    evaluate,
+    mark_spans,
+)
+from kanta.logging import _USER_PATH
+from kanta.serialization import JsonSerializer
+from kanta.serialization.framing import LineFramer
+from kanta.structs import ChangeRecord, Snapshot
+from kanta.tty import strip_ansi
+
+TS = datetime(2026, 1, 1, tzinfo=UTC)
+
+MARK = "\x1b[48;5;220m"
+UNMARK = "\x1b[49m"
+
+
+def test_match_path_whole_elements_anywhere():
+    assert _match_path("users", ["users"]) == frozenset({0})
+    assert _match_path("users", ["data", "users", "alice"]) == frozenset({1})
+    # Partial element matches require explicit wildcards.
+    assert _match_path("users", ["foousers"]) is None
+    assert _match_path("us*rs", ["foousers"]) is None
+    assert _match_path("*users", ["foousers"]) == frozenset({0})
+
+
+def test_match_path_contiguous_element_sequence():
+    assert _match_path("alice.email", ["users", "alice", "email"]) == frozenset({1, 2})
+    # The sequence must be contiguous.
+    assert _match_path("users.email", ["users", "alice", "email"]) is None
+    assert _match_path("users.*.email", ["users", "alice", "email"]) == frozenset(
+        {0, 1, 2}
+    )
+    # Matching is case-insensitive and empty terms match without marking.
+    assert _match_path("USERS", ["users"]) == frozenset({0})
+    assert _match_path("", ["anything"]) == frozenset()
+
+
+def test_match_value_substring_for_strings_full_for_scalars():
+    assert _match_value("lice", "alice@example.com", "alice@example.com")
+    assert not _match_value("bob", "alice@example.com", "alice@example.com")
+    # Booleans, numbers and null match only in full.
+    assert _match_value("true", True, "true")
+    assert not _match_value("tru", True, "true")
+    assert _match_value("42", 42, "42")
+    assert not _match_value("4", 42, "42")
+    assert _match_value("null", None, "null")
+    # Wildcards match the whole value text of any type.
+    assert _match_value("tru*", True, "true")
+    assert _match_value("4*", 42, "42")
+    # Containers do not match; their leaves are matched individually.
+    assert not _match_value("email", {"email": "a@b.c"}, '{"email": "a@b.c"}')
+    # An empty term matches anything.
+    assert _match_value("", True, "true")
+
+
+def test_mark_spans_merges_overlaps_and_adjacents():
+    assert mark_spans("alice@example.com", [(0, 5), (3, 11)]) == (
+        f"{MARK}alice@examp{UNMARK}le.com"
+    )
+    assert mark_spans("aab", [(0, 1), (1, 3)]) == f"{MARK}aab{UNMARK}"
+    assert mark_spans("ab", [(1, 1), (2, 2)]) == "ab"
+
+
+def test_mark_spans_offsets_into_styled_text():
+    styled = "\x1b[32mal\x1b[0mlice"
+    # 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"
+    )
+    # 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}"
+    )
+    assert mark_spans(styled, []) == styled
+
+
+def test_find_spans_case_insensitive_occurrences():
+    assert _find_spans("Alice likes ALICE", "alice") == [(0, 5), (12, 17)]
+    assert _find_spans("nope", "alice") == []
+    assert _find_spans("anything", "") == []
+
+
+def test_pattern_parse_forms():
+    bare = GrepPattern.parse("alice")
+    assert bare.term == "alice" and bare.path_term is None
+
+    pair = GrepPattern.parse("users.alice.age=30")
+    assert pair.term is None
+    assert pair.path_term == "users.alice.age"
+    assert pair.value_term == "30"
+
+    value_only = GrepPattern.parse("=alice@example.com")
+    assert value_only.path_term == "" and value_only.value_term == "alice@example.com"
+
+    path_only = GrepPattern.parse("users.alice=")
+    assert path_only.path_term == "users.alice" and path_only.value_term == ""
+
+    # Split on the first '=' only; the value may contain '='.
+    multi = GrepPattern.parse("key=a=b")
+    assert multi.path_term == "key" and multi.value_term == "a=b"
+
+
+def _patterns(*raws: str) -> list[GrepPattern]:
+    return [GrepPattern.parse(raw) for raw in raws]
+
+
+def test_evaluate_bare_term_against_path_value_action_user():
+    record = ChangeRecord(
+        ts=TS, a="create_user", u="admin", diff={"users": {"alice": {"age": 30}}}
+    )
+    assert evaluate(record, {}, _patterns("users.alice"))
+    assert evaluate(record, {}, _patterns("30"))
+    assert evaluate(record, {}, _patterns("CREATE_user"))
+    assert evaluate(record, {}, _patterns("admin"))
+    assert evaluate(record, {}, _patterns("age=30"))
+    assert not evaluate(record, {}, _patterns("bob"))
+    assert not evaluate(record, {}, _patterns("3"))  # not a full number match
+
+
+def test_evaluate_unescapes_dollar_keys():
+    record = ChangeRecord(ts=TS, a="set", diff={"$$config": 5})
+    assert evaluate(record, {}, _patterns("$config=5"))
+
+
+def test_evaluate_path_value_form_requires_same_line():
+    record = ChangeRecord(ts=TS, a="set", diff={"a": {"x": 1}, "b": {"y": 2}})
+    previous = {"a": {"x": 0}, "b": {"y": 0}}
+    assert evaluate(record, previous, _patterns("a.x=1"))
+    # 'a' matches one line's path, '2' another line's value: no match.
+    assert not evaluate(record, previous, _patterns("a=2"))
+
+
+def test_evaluate_all_patterns_same_record_any_line():
+    record = ChangeRecord(ts=TS, a="set", diff={"a": {"x": 1}, "b": {"y": 2}})
+    previous = {"a": {"x": 0}, "b": {"y": 0}}
+    assert evaluate(record, previous, _patterns("a.x", "=2"))
+    assert not evaluate(record, previous, _patterns("a.x", "=2", "missing"))
+
+
+def test_evaluate_matches_deleted_content_by_previous_value():
+    record = ChangeRecord(ts=TS, a="delete_user", diff={"users": {"$delete": "alice"}})
+    previous = {"users": {"alice": {"email": "alice@example.com"}}}
+    assert evaluate(record, previous, _patterns("alice@example.com"))
+    assert evaluate(record, previous, _patterns("users.alice.email"))
+    assert not evaluate(record, {}, _patterns("alice@example.com"))
+
+
+def test_highlighter_marks_exactly_the_matched_regions():
+    record = ChangeRecord(
+        ts=TS,
+        a="set",
+        diff={
+            "users": {"alice": {"email": "same@x.com"}, "bob": {"email": "same@x.com"}}
+        },
+    )
+    previous = {
+        "users": {"alice": {"email": "old@x.com"}, "bob": {"email": "same@x.com"}}
+    }
+    hl = evaluate(record, previous, _patterns("users.alice.email"))
+    assert hl is not None
+    # Path-side match: the matched elements are lit, the value is not.
+    assert hl.path("alice", "users.alice") == f"{MARK}alice{UNMARK}"
+    assert hl.value("same@x.com", "users.alice.email") == "same@x.com"
+    # Bob's identical value is a different path: untouched.
+    assert hl.path("bob", "users.bob") == "bob"
+    assert hl.value("same@x.com", "users.bob.email") == "same@x.com"
+
+    hl = evaluate(record, previous, _patterns("=same@x.com"))
+    assert hl is not None
+    # Both lines genuinely match the value: both are marked.
+    assert hl.value("same@x.com", "users.alice.email") == f"{MARK}same@x.com{UNMARK}"
+    assert hl.value("same@x.com", "users.bob.email") == f"{MARK}same@x.com{UNMARK}"
+    assert hl.path("alice", "users.alice") == "alice"
+
+
+def test_highlighter_merges_overlapping_needles_from_different_patterns():
+    record = ChangeRecord(ts=TS, a="set", diff={"email": "alice@example.com"})
+    hl = evaluate(record, {}, _patterns("alice", "lice@exam"))
+    assert hl is not None
+    assert hl.value("alice@example.com", "email") == (
+        f"{MARK}alice@exam{UNMARK}ple.com"
+    )
+
+
+def test_highlighter_marks_delete_marker_on_removed_content_match():
+    record = ChangeRecord(ts=TS, a="delete_user", diff={"users": {"$delete": "bob"}})
+    previous = {"users": {"bob": {"email": "bob@example.com"}}}
+    hl = evaluate(record, previous, _patterns("bob@example.com"))
+    assert hl is not None
+    # The matched content is gone: the deletion marker is lit, not the path.
+    assert hl.path("users", "users") == "users"
+    assert hl.path("bob", "users.bob") == "bob"
+    assert hl.delete("✗", "users.bob") == f"{MARK}✗{UNMARK}"
+    assert hl.delete("✗", "users.alice") == "✗"
+
+    # A path-side match lights the genuinely matched elements of the anchor,
+    # and the marker for the element below it.
+    hl = evaluate(record, previous, _patterns("users.bob.email"))
+    assert hl is not None
+    assert hl.path("users", "users") == f"{MARK}users{UNMARK}"
+    assert hl.path("bob", "users.bob") == f"{MARK}bob{UNMARK}"
+    assert hl.delete("✗", "users.bob") == f"{MARK}✗{UNMARK}"
+
+    # A path-side match of one displayed element lights only that element.
+    hl = evaluate(record, previous, _patterns("bob"))
+    assert hl is not None
+    assert hl.path("users", "users") == "users"
+    assert hl.path("bob", "users.bob") == f"{MARK}bob{UNMARK}"
+
+
+def test_highlighter_meta_marks_action_and_user():
+    record = ChangeRecord(ts=TS, a="create_user", u="admin", diff={"x": 1})
+    hl = evaluate(record, {}, _patterns("create", "ADM"))
+    assert hl is not None
+    assert hl.meta("create_user", "action") == f"{MARK}create{UNMARK}_user"
+    assert hl.meta("admin", "user") == f"{MARK}adm{UNMARK}in"
+    assert hl.meta("extra", "other") == "extra"
+
+
+def test_entry_dataclass_holds_anchor_for_deletes():
+    entry = _Entry(["a", "b"], 1, "1", anchor=["a"])
+    assert entry.anchor == ["a"]
+
+
+def _write_db(path, changes, state=None):
+    serializer = JsonSerializer()
+    framer = LineFramer()
+    snapshot = Snapshot(ts=TS, v=1, state=state or {})
+    data = framer.frame_snapshot(serializer.encode(snapshot), record_offset=0)
+    for change in changes:
+        data += framer.frame_change(serializer.encode(change), record_offset=0)
+    path.write_bytes(data)
+
+
+def _sample_changes():
+    return [
+        ChangeRecord(
+            ts=TS,
+            a="create_alice",
+            u="admin",
+            diff={
+                "users": {
+                    "alice": {"email": "alice@example.com", "admin": True},
+                }
+            },
+        ),
+        ChangeRecord(
+            ts=TS,
+            a="update_alice",
+            u="admin",
+            diff={"users": {"alice": {"age": 30}}},
+        ),
+        ChangeRecord(
+            ts=TS,
+            a="create_bob",
+            u="bob",
+            diff={"users": {"bob": {"email": "bob@example.com"}}},
+        ),
+    ]
+
+
+def _run_cli(tmp_path, capsys, monkeypatch, changes, *args, color=False):
+    if color:
+        monkeypatch.setenv("FORCE_COLOR", "1")
+        monkeypatch.delenv("NO_COLOR", raising=False)
+    else:
+        monkeypatch.setenv("NO_COLOR", "1")
+        monkeypatch.delenv("FORCE_COLOR", raising=False)
+    path = tmp_path / "test.kantadb"
+    _write_db(path, changes)
+    code = main([str(path), *args])
+    assert code == 0
+    return capsys.readouterr().err
+
+
+def test_cli_grep_filters_records(tmp_path, capsys, monkeypatch):
+    """Only matching change records are printed."""
+    err = _run_cli(tmp_path, capsys, monkeypatch, _sample_changes(), "--grep", "alice")
+    assert "create_alice" in err
+    assert "update_alice" in err
+    assert "create_bob" not in err
+    assert "bob@example.com" not in err
+
+
+def test_cli_grep_prints_entire_transaction(tmp_path, capsys, monkeypatch):
+    """A match prints the whole record, not just the matching line."""
+    err = _run_cli(
+        tmp_path, capsys, monkeypatch, _sample_changes(), "--grep", "users.alice.email"
+    )
+    assert "create_alice" in err
+    # Non-matching lines of the same record are printed too.
+    assert "admin" in err and "true" in err
+    # The update record does not contain the path.
+    assert "update_alice" not in err
+
+
+def test_cli_grep_repeated_patterns_must_all_match(tmp_path, capsys, monkeypatch):
+    """Repeated --grep options are ANDed within the same record."""
+    err = _run_cli(
+        tmp_path,
+        capsys,
+        monkeypatch,
+        _sample_changes(),
+        "--grep",
+        "alice",
+        "--grep",
+        "=alice@example.com",
+    )
+    assert "create_alice" in err
+    # update_alice matches 'alice' but has no email value.
+    assert "update_alice" not in err
+
+    err = _run_cli(
+        tmp_path,
+        capsys,
+        monkeypatch,
+        _sample_changes(),
+        "--grep",
+        "alice",
+        "--grep",
+        "bob",
+    )
+    assert "create_alice" not in err
+    assert "create_bob" not in err
+
+
+def test_cli_grep_no_match_exits_zero(tmp_path, capsys, monkeypatch):
+    """No matching records is not an error; snapshot lines still print."""
+    err = _run_cli(tmp_path, capsys, monkeypatch, _sample_changes(), "--grep", "nobody")
+    assert "snapshot s0" in err
+    assert "create_alice" not in err
+    assert "create_bob" 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(
+        tmp_path, capsys, monkeypatch, _sample_changes(), "--grep", "users.*.admin=true"
+    )
+    assert "create_alice" in err
+    assert "create_bob" not in err
+
+    err = _run_cli(
+        tmp_path, capsys, monkeypatch, _sample_changes(), "--grep", "users.alice.age="
+    )
+    assert "update_alice" in err
+    assert "create_alice" not in err
+
+    err = _run_cli(
+        tmp_path, capsys, monkeypatch, _sample_changes(), "--grep", "=bob@example.com"
+    )
+    assert "create_bob" in err
+    assert "create_alice" not in err
+
+
+def test_cli_grep_path_elements_match_in_full(tmp_path, capsys, monkeypatch):
+    """'users' does not match a 'foousers' element; 'us*' does."""
+    changes = [ChangeRecord(ts=TS, a="trap", diff={"foousers": {"note": "x"}})]
+    err = _run_cli(tmp_path, capsys, monkeypatch, changes, "--grep", "users")
+    assert "trap" not in err
+
+    err = _run_cli(tmp_path, capsys, monkeypatch, changes, "--grep", "*users")
+    assert "trap" in err
+
+
+def _shared_email_changes():
+    return [
+        ChangeRecord(
+            ts=TS,
+            a="create_alice",
+            u="admin",
+            diff={"users": {"alice": {"email": "shared@example.com", "admin": True}}},
+        ),
+        ChangeRecord(
+            ts=TS,
+            a="create_bob",
+            u="admin",
+            diff={"users": {"bob": {"email": "shared@example.com"}}},
+        ),
+        ChangeRecord(
+            ts=TS,
+            a="delete_bob",
+            u="admin",
+            diff={"users": {"$delete": "bob"}},
+        ),
+    ]
+
+
+def _lines_with(err: str, text: str) -> list[str]:
+    return [line for line in err.splitlines() if text in strip_ansi(line)]
+
+
+def test_cli_highlight_marks_exactly_the_matched_regions(tmp_path, capsys, monkeypatch):
+    """A path-side match lights only the matched elements, not the value."""
+    err = _run_cli(
+        tmp_path,
+        capsys,
+        monkeypatch,
+        _shared_email_changes(),
+        "--grep",
+        "users.alice.email",
+        color=True,
+    )
+    (alice_line,) = _lines_with(err, "shared@example.com")
+    assert f"{MARK}alice{UNMARK}" in alice_line
+    assert f"{MARK}email{UNMARK}" in alice_line
+    # The value itself did not match, so it is not highlighted.
+    assert f"{MARK}shared@example.com{UNMARK}" not in alice_line
+    (header_line,) = _lines_with(err, "users =")
+    assert f"{MARK}users{UNMARK}" in header_line
+    assert "create_bob" not in err
+
+
+def test_cli_highlight_value_matches_on_all_matching_lines(
+    tmp_path, capsys, monkeypatch
+):
+    """A value-side match lights the value on every line that genuinely matched."""
+    err = _run_cli(
+        tmp_path,
+        capsys,
+        monkeypatch,
+        _shared_email_changes(),
+        "--grep",
+        "=shared@example.com",
+        color=True,
+    )
+    lines = _lines_with(err, "shared@example.com")
+    assert len(lines) == 2  # alice's and bob's records both matched
+    for line in lines:
+        assert f"{MARK}shared@example.com{UNMARK}" in line
+    # The deletion of bob matched by its previous (removed) value: the
+    # deletion marker is lit, not the deleted path.
+    (delete_line,) = _lines_with(err, "✗")
+    assert f"{MARK}✗{UNMARK}" in delete_line
+    assert f"{MARK}bob{UNMARK}" not in delete_line
+    assert f"{MARK}users{UNMARK}" not in delete_line
+
+
+def test_cli_highlight_merges_overlapping_matches(tmp_path, capsys, monkeypatch):
+    """Overlapping matches from different patterns form one continuous mark."""
+    err = _run_cli(
+        tmp_path,
+        capsys,
+        monkeypatch,
+        _shared_email_changes(),
+        "--grep",
+        "shared",
+        "--grep",
+        "red@exam",
+        color=True,
+    )
+    lines = _lines_with(err, "shared@example.com")
+    assert len(lines) == 2  # both records genuinely match both patterns
+    for line in lines:
+        assert f"{MARK}shared@exam{UNMARK}ple.com" in line
+
+
+def test_cli_highlight_full_scalar_and_header_fields(tmp_path, capsys, monkeypatch):
+    """Scalars are marked in full; matched action/user substrings are marked."""
+    err = _run_cli(
+        tmp_path,
+        capsys,
+        monkeypatch,
+        _shared_email_changes(),
+        "--grep",
+        "admin=true",
+        "--grep",
+        "create_al",
+        color=True,
+    )
+    (line,) = _lines_with(err, "shared@example.com")
+    assert f"{MARK}true{UNMARK}" in line
+    (header,) = _lines_with(err, "create_alice")
+    assert f"{MARK}create_al{UNMARK}ice" in header
+    # Only the record matching both patterns is printed.
+    assert "create_bob" not in err
+    assert "delete_bob" not in err
+
+
+def test_evaluate_matches_prettified_and_raw_value_forms():
+    """With logfmt, both the raw and the prettified form of a value match."""
+    record = ChangeRecord(ts=TS, a="set", diff={"when": 1767225600})
+
+    def logfmt(value, path):
+        return "2026-01-01" if value == 1767225600 else None
+
+    # The prettified form matches; the needle is located in the display text.
+    hl = evaluate(record, {}, _patterns("2026"), logfmt=logfmt)
+    assert hl is not None
+    assert hl.value("2026-01-01", "when") == f"{MARK}2026{UNMARK}-01-01"
+    # The raw form matches too; its needle is absent from the displayed
+    # text, so the whole displayed value is marked.
+    hl = evaluate(record, {}, _patterns("1767225600"), logfmt=logfmt)
+    assert hl is not None
+    assert hl.value("2026-01-01", "when") == f"{MARK}2026-01-01{UNMARK}"
+    # Without logfmt the raw value is matched and marked precisely.
+    hl = evaluate(record, {}, _patterns("1767225600"))
+    assert hl.value("1767225600", "when") == f"{MARK}1767225600{UNMARK}"
+    # Neither form matches.
+    assert evaluate(record, {}, _patterns("1999"), logfmt=logfmt) is None
+
+
+def test_evaluate_matches_prettified_user():
+    """The user field matches both the raw id and the logfmt-resolved name."""
+
+    def logfmt(value, path):
+        return "Alice Admin" if path == _USER_PATH else None
+
+    record = ChangeRecord(ts=TS, a="set", u="u123", diff={"x": 1})
+    hl = evaluate(record, {}, _patterns("alice"), logfmt=logfmt)
+    assert hl is not None
+    assert hl.meta("Alice Admin", "user") == f"{MARK}Alice{UNMARK} Admin"
+    # A raw-only user match marks the whole displayed name.
+    hl = evaluate(record, {}, _patterns("u123"), logfmt=logfmt)
+    assert hl is not None
+    assert hl.meta("Alice Admin", "user") == f"{MARK}Alice Admin{UNMARK}"
+    # The action is never prettified.
+    assert hl.meta("set", "action") == "set"
+
+
+def test_evaluate_without_logfmt_matches_raw_only():
+    record = ChangeRecord(ts=TS, a="set", u="u123", diff={"when": 1767225600})
+    assert evaluate(record, {}, _patterns("1767225600"))
+    assert evaluate(record, {}, _patterns("2026")) is None
+    assert evaluate(record, {}, _patterns("alice")) is None
+
+
+def test_cli_grep_matches_prettified_forms_with_kanta_object(
+    tmp_path, capsys, monkeypatch
+):
+    """End to end: a -k object's logfmt formatter doubles the match surface."""
+    db_path = tmp_path / "test.kantadb"
+    module = tmp_path / "dbmod.py"
+    module.write_text(
+        "from typing import Any\n"
+        "from kanta import Kanta\n"
+        f"kanta = Kanta({str(db_path)!r}, {{}}, type=dict)\n"
+        "@kanta.logfmt\n"
+        "def pretty(value: Any, path: str) -> str | None:\n"
+        "    if path == 'when':\n"
+        "        return 'Nov 3, 2025'\n"
+        "    if path == '$user':\n"
+        "        return 'Alice Admin'\n"
+        "    return None\n"
+    )
+    changes = [ChangeRecord(ts=TS, a="set", u="u123", diff={"when": 1767225600})]
+    kanta_args = ("-k", str(module))
+
+    # A term matching only the prettified value finds the record.
+    err = _run_cli(
+        tmp_path, capsys, monkeypatch, changes, *kanta_args, "--grep", "nov", color=True
+    )
+    (line,) = _lines_with(err, "Nov 3, 2025")
+    assert f"{MARK}Nov{UNMARK} 3, 2025" in line
+
+    # A term matching only the raw value marks the whole prettified display.
+    err = _run_cli(
+        tmp_path,
+        capsys,
+        monkeypatch,
+        changes,
+        *kanta_args,
+        "--grep",
+        "1767225600",
+        color=True,
+    )
+    (line,) = _lines_with(err, "Nov 3, 2025")
+    assert f"{MARK}Nov 3, 2025{UNMARK}" in line
+
+    # The prettified user matches, and the raw id marks the whole display.
+    err = _run_cli(
+        tmp_path,
+        capsys,
+        monkeypatch,
+        changes,
+        *kanta_args,
+        "--grep",
+        "alice",
+        color=True,
+    )
+    (header,) = _lines_with(err, "Alice Admin")
+    assert f"{MARK}Alice{UNMARK} Admin" in header
+    err = _run_cli(
+        tmp_path,
+        capsys,
+        monkeypatch,
+        changes,
+        *kanta_args,
+        "--grep",
+        "u123",
+        color=True,
+    )
+    (header,) = _lines_with(err, "Alice Admin")
+    assert f"{MARK}Alice Admin{UNMARK}" in header
+
+    # Without -k there is no prettified form to match.
+    err = _run_cli(tmp_path, capsys, monkeypatch, changes, "--grep", "nov")
+    assert not _lines_with(err, "u123")
+    err = _run_cli(tmp_path, capsys, monkeypatch, changes, "--grep", "1767225600")
+    assert _lines_with(err, "u123")

tests/test_logemit.py

@@ -38,13 +38,23 @@ def _reset_kanta_loggers():
         logger.handlers.clear()
 
 
+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.
+    """
+    configure_logging(**kwargs)
+    for name in ("kanta.bootstrap", "kanta.migration", "kanta.transaction"):
+        logging.getLogger(name).setLevel(logging.INFO)
+
+
 def _change_event(**kwargs) -> LogEvent:
     return LogEvent(kind="change", logger=transaction_logger, action="update", **kwargs)
 
 
 def test_emit_event_falsy_return_stops_chain(capsys):
-    logging.getLogger("kanta").handlers.clear()
+    _setup_logging()
-    configure_logging()
     calls = []
 
     def first(ev):
@@ -60,15 +70,13 @@ def test_emit_event_falsy_return_stops_chain(capsys):
 
 
 def test_emit_event_truthy_return_falls_back_to_default(capsys):
-    logging.getLogger("kanta").handlers.clear()
+    _setup_logging()
-    configure_logging()
     emit_event(_change_event(), [lambda ev: True])
     assert "update" in capsys.readouterr().err
 
 
 def test_emit_event_mutation_reaches_later_handlers_and_default(capsys):
-    logging.getLogger("kanta").handlers.clear()
+    _setup_logging()
-    configure_logging()
     calls = []
 
     def first(ev):
@@ -86,8 +94,7 @@ def test_emit_event_mutation_reaches_later_handlers_and_default(capsys):
 
 
 def test_emit_event_handler_error_falls_back_to_default(capsys):
-    logging.getLogger("kanta").handlers.clear()
+    _setup_logging()
-    configure_logging()
 
     def boom(ev):
         raise RuntimeError("broken")
@@ -109,8 +116,7 @@ def test_diff_lines_built_lazily(monkeypatch):
 
 
 def test_default_emit_created_and_migrated(capsys):
-    logging.getLogger("kanta").handlers.clear()
+    _setup_logging()
-    configure_logging()
     emit_event(LogEvent(kind="created", logger=bootstrap_logger, filename="x.kantadb"))
     emit_event(
         LogEvent(
@@ -127,6 +133,23 @@ def test_default_emit_created_and_migrated(capsys):
     assert "🛢️ x.kantadb migrated v0 -> v1: migrate_v1 (rename)" in err
 
 
+def test_default_emit_strips_ansi_without_color_support(capsys, monkeypatch):
+    """NO_COLOR output contains no ANSI codes; FORCE_COLOR keeps them."""
+    _setup_logging()
+    monkeypatch.setenv("NO_COLOR", "1")
+    monkeypatch.delenv("FORCE_COLOR", raising=False)
+    emit_event(_change_event(diff={"counter": 1}))
+    err = capsys.readouterr().err
+    assert "\x1b[" not in err
+    assert "counter" in err
+
+    _setup_logging()
+    monkeypatch.setenv("FORCE_COLOR", "1")
+    monkeypatch.delenv("NO_COLOR", raising=False)
+    emit_event(_change_event(diff={"counter": 1}))
+    assert "\x1b[" in capsys.readouterr().err
+
+
 @pytest.mark.asyncio
 async def test_logemit_receives_transaction_events(tmp_path, format_config):
     path = tmp_path / "test.db"
@@ -242,7 +265,11 @@ async def test_logmigr_failure_does_not_break_open(tmp_path, format_config):
 
 
 @pytest.mark.asyncio
-async def test_aborted_transaction_emits_event(tmp_path, format_config, caplog):
+async def test_aborted_transaction_emits_event(
+    tmp_path, format_config, caplog, monkeypatch
+):
+    monkeypatch.setenv("FORCE_COLOR", "1")
+    monkeypatch.delenv("NO_COLOR", raising=False)
     path = tmp_path / "test.db"
     kanta = make_kanta(path, Data, format_config)
     events = []
@@ -334,8 +361,7 @@ async def test_event_carries_kanta_instance(tmp_path, format_config):
 
 
 def test_header_is_settable_and_used_by_default_emit(capsys):
-    logging.getLogger("kanta").handlers.clear()
+    _setup_logging()
-    configure_logging()
 
     def restyle(ev):
         ev.header = f"CUSTOM {ev.action}"

tests/test_logging.py

@@ -39,42 +39,51 @@ def _reset_kanta_loggers():
 
 
 def test_configure_logging_defaults():
-    kanta_logger = logging.getLogger("kanta")
     configure_logging()
-    assert kanta_logger.level == logging.INFO
+    for name in ("kanta.bootstrap", "kanta.migration", "kanta.transaction"):
-    assert not kanta_logger.propagate
+        logger = logging.getLogger(name)
-    assert kanta_logger.handlers
+        assert logger.level == logging.NOTSET  # inherits the root level
+        assert not logger.propagate
+        assert logger.handlers
 
 
 def test_configure_logging_disables_specific_loggers():
     configure_logging(bootstrap=False, migration=False, transaction=False)
-    assert not logging.getLogger("kanta.bootstrap").propagate
+    assert logging.getLogger("kanta.bootstrap").disabled
-    assert not logging.getLogger("kanta.migration").propagate
+    assert logging.getLogger("kanta.migration").disabled
-    assert not logging.getLogger("kanta.transaction").propagate
+    assert logging.getLogger("kanta.transaction").disabled
 
 
-def test_configure_logging_skiproot_false_leaves_kanta_propagation():
+def test_configure_logging_skiproot_false_routes_via_root():
-    kanta_logger = logging.getLogger("kanta")
+    configure_logging(skiproot=False)
-    kanta_logger.handlers.clear()
+    for name in ("kanta.bootstrap", "kanta.migration", "kanta.transaction"):
-    configure_logging(bootstrap=False, skiproot=False)
+        logger = logging.getLogger(name)
-    assert kanta_logger.propagate
+        assert logger.propagate
-    assert not kanta_logger.handlers
+        assert not logger.handlers
-    assert not logging.getLogger("kanta.bootstrap").propagate
+
+
+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.
+    """
+    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):
-    kanta_logger = logging.getLogger("kanta")
+    _setup_logging()
-    kanta_logger.handlers.clear()
-    configure_logging()
     log_change("test", {})
     captured = capsys.readouterr()
     assert "test" in captured.err
 
 
-def test_log_change_appends_extra_string(capsys):
+def test_log_change_appends_extra_string(capsys, monkeypatch):
-    kanta_logger = logging.getLogger("kanta")
+    monkeypatch.setenv("FORCE_COLOR", "1")
-    kanta_logger.handlers.clear()
+    monkeypatch.delenv("NO_COLOR", raising=False)
-    configure_logging()
+    _setup_logging()
     log_change("export", {}, extra="mydb.db")
     captured = capsys.readouterr()
     assert "export" in captured.err
@@ -82,9 +91,7 @@ def test_log_change_appends_extra_string(capsys):
 
 
 def test_log_change_log_diff_false(capsys, monkeypatch):
-    kanta_logger = logging.getLogger("kanta")
+    _setup_logging()
-    kanta_logger.handlers.clear()
-    configure_logging()
 
     def _boom(*args, **kwargs):
         raise AssertionError("format_diff should not be called")
@@ -97,9 +104,7 @@ def test_log_change_log_diff_false(capsys, monkeypatch):
 
 
 def test_configure_logging_diff_false(capsys):
-    kanta_logger = logging.getLogger("kanta")
+    _setup_logging(diff=False)
-    kanta_logger.handlers.clear()
-    configure_logging(diff=False)
     log_change("update", {"counter": 5}, previous={})
     captured = capsys.readouterr()
     assert "update" in captured.err
@@ -107,10 +112,9 @@ def test_configure_logging_diff_false(capsys):
 
 
 def test_configure_logging_diff_true_reenables(capsys):
-    kanta_logger = logging.getLogger("kanta")
+    _setup_logging(diff=False)
-    kanta_logger.handlers.clear()
-    configure_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