hermes_state_search.py

"""Full-text / trigram / CJK message search and FTS maintenance for SessionDB.

Plain mixin for ``hermes_state.SessionDB`` (no ``__init__``/state of its own).
Must never import hermes_state (cycle); shared constants live in hermes_state_common.
"""

import contextlib
import logging
import re
import sqlite3
import time
from typing import Any, Callable, Collection, Dict, List, Optional, Tuple

from agent.skill_commands import describe_skill_invocation
from hermes_state_errors import is_malformed_db_error
from hermes_state_common import (
    FTS_CJK_STALE_KEY, FTS_SQL, FTS_STALE_KEY, FTS_STORAGE_VERSION, FTS_TOOL_CONTENT_PREFIX_CHARS,
    FTS_TRIGRAM_EXCLUDED_SOURCES, FTS_TRIGRAM_SQL,
    MAX_FTS5_QUERY_CHARS, SCHEMA_VERSION, _FTS_CJK_TRIGGERS,
    escape_like as _escape_like, fts_rebuild_admission, fts_trigram_session_sql, routed_sessions_setting,
)

# Pre-split logger identity so log filtering/capture is unchanged.
logger = logging.getLogger("hermes_state")


def _search_slow_ms() -> float:
    """``sessions.search_slow_ms`` for the served profile (default 1000; 0 logs every call)."""
    value = routed_sessions_setting("search_slow_ms", "HERMES_SEARCH_SLOW_MS")
    try:
        return 1000.0 if value is None or str(value).strip() == "" else float(value)
    except (TypeError, ValueError):
        return 1000.0

# Characters FTS5's query grammar rejects outside a quoted phrase (anything missing
# reaches MATCH raw and raises -> zero results). ``%`` is deliberately excluded: the
# CJK LIKE fallback needs it literal (that path escapes wildcards itself).
_FTS5_SPECIAL_CHARS = '+{}():"^@/#&|~[]<>,;!?$=\\\''
_FTS5_SPECIAL_RE = re.compile(f"[{re.escape(_FTS5_SPECIAL_CHARS)}]")

_FTS_OPERATORS = frozenset({"AND", "OR", "NOT"})
_LIKE_SKIP_TOKENS = _FTS_OPERATORS | {"NEAR"}
_LIKE_TOKEN_RE = re.compile(r'"[^"]+"|\S+')
_QUOTED_PHRASE_RE = re.compile(r'"[^"]*"')

# Column list shared by every search route (snippet + metadata, never content).
_SEARCH_SELECT_TAIL = "m.timestamp, m.tool_name, s.source, s.model, s.started_at AS session_started"
_LIKE_SNIPPET_SQL = "substr(m.content, max(1, instr(m.content, ?) - 40), 120) AS snippet"
_LIKE_ANY_COLUMN_SQL = (
    "(m.content LIKE ? ESCAPE '\\' OR m.tool_name LIKE ? ESCAPE '\\' OR m.tool_calls LIKE ? ESCAPE '\\')"
)
_LIKE_COALESCED_COLUMN_SQL = (
    "(COALESCE(m.content, '') LIKE ? ESCAPE '\\' OR "
    "COALESCE(m.tool_name, '') LIKE ? ESCAPE '\\' OR "
    "COALESCE(m.tool_calls, '') LIKE ? ESCAPE '\\')"
)
# ``sort`` -> ORDER BY for the FTS routes; unknown values are rank-only (user input passes through).
_FTS_ORDER_BY = {"newest": "ORDER BY m.timestamp DESC, rank", "oldest": "ORDER BY m.timestamp ASC, rank"}
# Indexed neighbor seeks avoid scanning whole sessions for a sparse set of hits.
_CONTEXT_WINDOW_SQL = """WITH target AS (
    SELECT session_id, timestamp, id FROM messages WHERE id IN ({ids})
)
SELECT t.id AS match_id, m.role, m.content
FROM target t JOIN messages m ON m.id IN (
    t.id,
    (SELECT p.id FROM messages p
     WHERE p.session_id = t.session_id AND (p.timestamp, p.id) < (t.timestamp, t.id)
     ORDER BY p.timestamp DESC, p.id DESC LIMIT 1),
    (SELECT n.id FROM messages n
     WHERE n.session_id = t.session_id AND (n.timestamp, n.id) > (t.timestamp, t.id)
     ORDER BY n.timestamp, n.id LIMIT 1)
)
ORDER BY t.id, m.timestamp, m.id"""
# Unified Ideographs, Extension A, Extension B, CJK Symbols, Hiragana, Katakana, Hangul Syllables.
_CJK_RANGES = (
    (0x4E00, 0x9FFF), (0x3400, 0x4DBF), (0x20000, 0x2A6DF), (0x3000, 0x303F), (0x3040, 0x309F),
    (0x30A0, 0x30FF), (0xAC00, 0xD7AF),
)


def _meta_row(conn, key: str) -> Optional[sqlite3.Row]:
    """Point-read one ``state_meta`` row (``None`` when absent)."""
    return conn.execute("SELECT value FROM state_meta WHERE key = ?", (key,)).fetchone()


def _delete_meta(conn, *keys: str) -> None:
    conn.execute(f"DELETE FROM state_meta WHERE key IN ({','.join('?' for _ in keys)})", keys)


def _is_cjk(cp: int) -> bool:
    return any(lo <= cp <= hi for lo, hi in _CJK_RANGES)


def _non_operator_tokens(raw_query: str) -> list[str]:
    return [t for t in raw_query.split() if t.upper() not in _FTS_OPERATORS]


def _quote_fts_tokens(raw_query: str) -> str:
    """Quote each non-operator token (neutralising FTS5 special characters), keeping AND/OR/NOT."""
    return " ".join(
        tok if tok.upper() in _FTS_OPERATORS else '"' + tok.replace('"', '""') + '"' for tok in raw_query.split()
    )


def _like_params(term: str) -> list[str]:
    """One ``%term%`` bind per column of ``_LIKE_ANY_COLUMN_SQL``."""
    return [f"%{_escape_like(term)}%"] * 3


def _strip_cjk_wildcards(raw_query: str) -> str:
    """Drop the trailing prefix wildcard callers append for ASCII ("nimb" -> "nimb*").

    None of the CJK routes can honour that star: the bigram and trigram routes
    quote every token before MATCH (so ``*`` matches a literal asterisk) and
    LIKE has no ``*`` wildcard at all (only ``%``/``_``). Left in place, every
    CJK search arriving from the web/desktop search box — which appends the
    star to each unquoted token so partial English words match — searches for
    a term ending in a literal ``*`` and returns nothing (#90636). Only
    TRAILING stars go: a star written inside a quoted phrase is the user's
    own text, and a token that is ALL stars keeps its original form so it
    cannot degrade to a match-everything empty term.
    """
    if "*" not in raw_query:
        return raw_query
    stripped: list[str] = []
    for token in raw_query.split():
        if token.upper() in _FTS_OPERATORS:
            stripped.append(token)
        else:
            stripped.append(token.rstrip("*") or token)
    return " ".join(stripped) or raw_query


def _flatten_text(decoded: Any) -> str:
    """Multimodal part list -> joined text (or the placeholder); str passes through; else ''."""
    if isinstance(decoded, list):
        parts = [p.get("text", "") for p in decoded if isinstance(p, dict) and p.get("type") == "text"]
        return " ".join(t for t in parts if t).strip() or "[multimodal content]"
    return decoded if isinstance(decoded, str) else ""


def _positive_int(name: str, value: Any) -> None:
    if isinstance(value, bool) or not isinstance(value, int):
        raise TypeError(f"{name} must be an integer")
    if value <= 0:
        raise ValueError(f"{name} must be greater than zero")


def _search_select_sql(snippet_sql: str, from_sql: str, where: list[str], order_by: str, limit_sql: str) -> str:
    """Result-row SELECT shared by the FTS and LIKE routes (SQL text is pinned)."""
    return f"""
            SELECT m.id, m.session_id, m.role,
                   {snippet_sql},
                   {_SEARCH_SELECT_TAIL}
            FROM {from_sql}
            JOIN sessions s ON s.id = m.session_id
            WHERE {' AND '.join(where)}
            {order_by}
            {limit_sql}
        """


def _search_filter_clauses(
    where: list[str], params: list, *, include_inactive: bool, source_filter: Optional[list[str]],
    exclude_sources: Optional[list[str]], role_filter: Optional[list[str]],
    after_ts: Optional[int] = None, before_ts: Optional[int] = None) -> None:
    """Append the visibility/source/role/session-start predicates every search route shares. Live
    rows (active=1) AND compaction-archived rows (compacted=1) are discoverable; only
    rewind/undo rows (active=0, compacted=0) are hidden. ``after_ts``/``before_ts`` bound
    ``sessions.started_at`` (inclusive / exclusive) inside the query so LIMIT cannot be
    filled by out-of-window hits."""
    if not include_inactive:
        where.append("(m.active = 1 OR m.compacted = 1)")
    # display_kind="hidden" rows are model-facing scaffolding the person never saw; a hit would confuse.
    where.append("COALESCE(m.display_kind, '') <> 'hidden'")
    if source_filter is not None:
        where.append(f"s.source IN ({','.join('?' for _ in source_filter)})")
        params.extend(source_filter)
    if exclude_sources is not None:
        where.append(f"s.source NOT IN ({','.join('?' for _ in exclude_sources)})")
        params.extend(exclude_sources)
    if role_filter:
        where.append(f"m.role IN ({','.join('?' for _ in role_filter)})")
        params.extend(role_filter)
    if after_ts is not None:
        where.append("s.started_at >= ?")
        params.append(int(after_ts))
    if before_ts is not None:
        where.append("s.started_at < ?")
        params.append(int(before_ts))


class SessionSearchMixin:
    """See module docstring — mixin for SessionDB (Search cluster)."""

    _SEARCH_MESSAGE_RESULT_FIELDS = (
        "id", "session_id", "role", "snippet", "timestamp", "tool_name", "source", "model", "session_started", "context"
    )

    @classmethod
    def _search_message_fields(cls, fields: Optional[Collection[str]]) -> Optional[tuple[str, ...]]:
        """Validate and canonically order an optional result projection."""
        if fields is None:
            return None
        if isinstance(fields, str):
            raise TypeError("search fields must be a collection of field names, not a string")
        requested = set(fields)
        unknown = requested.difference(cls._SEARCH_MESSAGE_RESULT_FIELDS)
        if unknown:
            raise ValueError(f"unknown search result field(s): {', '.join(sorted(unknown))}")
        return tuple(field for field in cls._SEARCH_MESSAGE_RESULT_FIELDS if field in requested)

    def _try_incremental_merge_fts(self) -> None:
        """One bounded FTS5 merge pass that never fails the already-committed write (a caller
        must never replay an ambiguous, possibly-durable write — even on the bare
        SystemError CPython's sqlite3 layer can raise under cross-thread errmsg scrambling)."""
        if not self._fts_enabled:
            return
        try:
            self._merge_fts_incrementally(max_pages=self._FTS_MERGE_MAX_PAGES_PER_INDEX)
        except Exception as exc:
            # The canonical write is already committed before this cadence runs. No maintenance failure —
            # including the bare SystemError the CPython sqlite3 layer can raise under cross-thread errmsg
            # scrambling — may escape and make the caller replay an ambiguous, possibly-durable write
            # (#90734, #85079).
            logger.warning("FTS incremental merge failed after commit: %s", exc)

    # ── Deferred rebuild engine (base + CJK backfills) ─────────────────────

    def fts_rebuild_status(self) -> Optional[dict[str, Any]]:
        """Deferred-rebuild progress ``{"pending", "total", "indexed", "percent"}`` or None. Reads
        via the pooled reader (not get_meta/self._lock) so search never blocks on the writer."""
        return self._rebuild_status("fts_rebuild")

    def fts_cjk_rebuild_status(self) -> Optional[dict[str, Any]]:
        """CJK-index backfill progress, or None when none is pending."""
        return self._rebuild_status("fts_cjk_rebuild")

    def _rebuild_status(self, prefix: str) -> Optional[dict[str, Any]]:
        rows = self._read_all("SELECT key, value FROM state_meta WHERE key IN (?, ?)",
                              (f"{prefix}_high_water", f"{prefix}_progress"))
        meta = {r["key"]: r["value"] for r in rows}
        high_water = meta.get(f"{prefix}_high_water")
        if high_water is None or int(high_water) <= 0:
            return None
        total, progress = int(high_water), int(meta.get(f"{prefix}_progress") or 0)
        return {"pending": True, "total": total, "indexed": progress, "percent": min(100, int(100 * progress / total))}

    # Re-index rows in an id window the index is missing. docsize has one row
    # per indexed doc, so the anti-join is exact. Params: (lo, hi).
    # NOTE: with the aligned projection (FTS_STORAGE_VERSION 3) every writer
    # of the index — this sweep, the chunked backfill, and the sync triggers —
    # feeds ``messages_fts`` through the ONE stable per-row expression:
    # tool rows are truncated to FTS_TOOL_CONTENT_PREFIX_CHARS, everything
    # else is verbatim, and nothing consults a moving state_meta marker.
    _BOUNDARY_SWEEP_SQL = (
        "INSERT INTO {table}(rowid, content, tool_name, tool_calls) "
        "SELECT m.id, m.content, m.tool_name, m.tool_calls FROM messages m WHERE m.id > ? AND m.id <= ? {extra}"
        "AND NOT EXISTS (SELECT 1 FROM {table}_docsize d WHERE d.id = m.id)"
    )
    _BASE_BOUNDARY_SWEEP_SQL = (
        "INSERT INTO messages_fts(rowid, content, tool_name, tool_calls) "
        "SELECT m.id, CASE WHEN m.role = 'tool' THEN substr(COALESCE(m.content, ''), 1, ?) "
        "ELSE m.content END, m.tool_name, m.tool_calls FROM messages m WHERE m.id > ? AND m.id <= ? "
        "AND NOT EXISTS (SELECT 1 FROM messages_fts_docsize d WHERE d.id = m.id)"
    )
    # Trigram excludes tool rows and FTS_TRIGRAM_EXCLUDED_SOURCES sessions; no tool_calls column.
    _TRIGRAM_BOUNDARY_SWEEP_SQL = (
        "INSERT INTO messages_fts_trigram(rowid, content, tool_name) "
        "SELECT m.id, m.content, m.tool_name FROM messages m JOIN sessions s ON s.id = m.session_id "
        f"WHERE m.id > ? AND m.id <= ? AND m.role <> 'tool' AND {fts_trigram_session_sql('s')} "
        "AND NOT EXISTS (SELECT 1 FROM messages_fts_trigram_docsize d WHERE d.id = m.id)"
    )
    _CHUNK_INSERT_SQL = (
        "INSERT INTO {table}(rowid, content, tool_name, tool_calls) "
        "SELECT id, CASE WHEN role = 'tool' "
        f"THEN substr(COALESCE(content, ''), 1, {FTS_TOOL_CONTENT_PREFIX_CHARS}) "
        "ELSE content END, tool_name, tool_calls FROM messages WHERE id > ? AND id <= ?{extra}"
    )
    _TRIGRAM_CHUNK_INSERT_SQL = (
        "INSERT INTO messages_fts_trigram(rowid, content, tool_name) "
        "SELECT m.id, m.content, m.tool_name FROM messages m JOIN sessions s ON s.id = m.session_id "
        f"WHERE m.id > ? AND m.id <= ? AND m.role <> 'tool' AND {fts_trigram_session_sql('s')}"
    )

    def _fts_rebuild_finish(self) -> None:
        """Finalize the deferred rebuild: boundary sweep + clear markers. The sweep is cheap
        insurance against a write that slipped between high_water capture and trigger
        activation. The trigram half is gated on ``_trigram_available``: without the
        tokenizer/table an unconditional INSERT raises and aborts the whole rebuild."""
        sweeps = [(self._BASE_BOUNDARY_SWEEP_SQL, True)]
        if self._trigram_available:
            sweeps.append((self._TRIGRAM_BOUNDARY_SWEEP_SQL, False))
        self._rebuild_finish("fts_rebuild", sweeps)
        logger.info("Deferred FTS rebuild complete — all messages indexed.")

    def _fts_cjk_rebuild_finish(self) -> None:
        """Boundary sweep + clear the cjk markers; index becomes servable."""
        sweep = self._BOUNDARY_SWEEP_SQL.format(table="messages_fts_cjk", extra="AND m.role <> 'tool' ")
        self._rebuild_finish("fts_cjk_rebuild", [(sweep, False)])
        self._fts_cjk_available = True
        logger.info("CJK FTS index backfill complete — serving CJK search.")

    def _rebuild_finish(self, prefix: str, sweep_sqls: list[tuple[str, bool]]) -> None:
        """Sweep a generous window around the high-water boundary, then clear the markers.
        ``(sql, bounded)``: a bounded sweep takes the tool-content prefix_chars param first."""
        def _do(conn):
            hw_row = _meta_row(conn, f"{prefix}_high_water")
            if hw_row is not None:
                hw = int(hw_row[0])
                for sql, bounded in sweep_sqls:
                    params = (FTS_TOOL_CONTENT_PREFIX_CHARS,) if bounded else ()
                    conn.execute(sql, (*params, hw - 1000, hw + 1000))
            _delete_meta(conn, f"{prefix}_high_water", f"{prefix}_progress")
        self._execute_write(_do)

    def fts_rebuild_step(self) -> bool:
        """Backfill one chunk of the deferred FTS rebuild; True while work remains. Chunks are
        claimed atomically inside the write transaction, so concurrent processes interleave
        instead of duplicating rows."""
        if not self._fts_enabled:
            return False
        inserts = [self._CHUNK_INSERT_SQL.format(table="messages_fts", extra="")]
        if self._trigram_available:
            inserts.append(self._TRIGRAM_CHUNK_INSERT_SQL)
        return self._rebuild_step("fts_rebuild", inserts, fail_msg="FTS rebuild chunk failed (will retry): %s",
                                  finish=self._fts_rebuild_finish, finish_when_empty=True)

    def fts_cjk_rebuild_step(self) -> bool:
        """Backfill one chunk of the CJK index. True while work remains."""
        if not self._fts_enabled or not self._fts_cjk_loaded:
            return False
        insert = self._CHUNK_INSERT_SQL.format(table="messages_fts_cjk", extra=" AND role <> 'tool'")
        return self._rebuild_step("fts_cjk_rebuild", [insert], finish=self._fts_cjk_rebuild_finish,
                                  fail_msg="CJK FTS rebuild chunk failed (will retry): %s")

    def _rebuild_step(self, prefix: str, insert_sqls: list[str], *, fail_msg: str, finish,
                      finish_when_empty: bool = False) -> bool:
        """Shared chunk engine for the base and CJK deferred backfills. ``finish_when_empty``
        finalizes a high_water <= 0 marker (empty messages table) instead of leaving it pending."""
        high_water_raw = self.get_meta(f"{prefix}_high_water")
        if high_water_raw is None:
            return False
        high_water = int(high_water_raw)
        chunk = self._FTS_REBUILD_CHUNK_ROWS

        def _do(conn):
            # Re-reading progress inside the BEGIN IMMEDIATE held by _execute_write IS
            # the claim: two workers cannot read the same value.
            row = _meta_row(conn, f"{prefix}_progress")
            if row is None:
                return False  # finished (or cleared) by another process
            progress = int(row[0])
            if progress >= high_water:
                return False
            # Upper bound is an id, not a row count, so deleted-row gaps don't shrink chunks.
            upper = min(progress + chunk, high_water)
            for sql in insert_sqls:
                conn.execute(sql, (progress, upper))
            # Progress lands in the same transaction as its rows (crash-atomic).
            conn.execute("UPDATE state_meta SET value = ? WHERE key = ?", (str(upper), f"{prefix}_progress"))
            return upper < high_water

        try:
            more = self._execute_write(_do)
        except sqlite3.OperationalError as exc:
            logger.debug(fail_msg, exc)
            return True  # transient (lock contention) — caller retries
        if more is False:
            status = self._rebuild_status(prefix)
            if (finish_when_empty and high_water <= 0) or (
                status is not None and status["indexed"] >= status["total"]
            ):
                finish()
            return False
        return bool(more)

    def _fts_teardown_trash_step(self) -> bool:
        """Tear down one chunk of a demoted v22 FTS shadow table (a PLAIN table now); True while
        work remains. INTEGER single-column-key tables drain with a high-water marker so
        each chunk's scan is bounded (restarting the scan was O(n²)); compound-key tables
        keep the chunked ``LIMIT`` delete — they are small by construction.

        Single-column-key trash tables (the common shape — FTS shadow tables carry a rowid/integer PK) are
        drained with a high-water marker mirroring :meth:`fts_rebuild_step`: each chunk deletes only rows
        after the previously-drained key, so the per-chunk scan is bounded instead of re-scanning from the
        start of the table every chunk (O(n²) total on large trash tables, #79324).
        """
        with self._read_ctx() as conn:
            trash = [r[0] for r in conn.execute(
                "SELECT name FROM sqlite_master WHERE type = 'table' AND name LIKE ? ESCAPE '\\'",
                (self._FTS_TRASH_PREFIX.replace("_", "\\_") + "%",),
            ).fetchall()]
        if not trash:
            return False
        tbl = trash[0]

        def _do(conn):
            pk_info = [(r[1], (r[2] or "").upper()) for r in conn.execute(f"PRAGMA table_info({tbl})") if r[5] > 0]
            key = ", ".join(name for name, _typ in pk_info) if pk_info else "rowid"
            if len(pk_info) == 1 and pk_info[0][1] == "INTEGER":
                # High-water drain; marker read/written in the same BEGIN IMMEDIATE as the
                # DELETE so concurrent callers claim disjoint ranges. Only INTEGER pks
                # anchor the comparison (the TEXT-pk config shadow table falls through).
                marker_key = f"fts_teardown_{tbl}_progress"
                row = _meta_row(conn, marker_key)
                high_water = int(row[0]) if row is not None else 0
                # Claim the LAST row of the LIMIT window so a full chunk goes per step.
                upper_rows = conn.execute(
                    f"SELECT {key} FROM {tbl} WHERE {key} > ? ORDER BY {key} LIMIT {self._FTS_REBUILD_CHUNK_ROWS}",
                    (high_water,),
                ).fetchall()
                if not upper_rows:
                    return _drop(conn, marker_key)
                upper = upper_rows[-1][0]
                cur = conn.execute(f"DELETE FROM {tbl} WHERE {key} > ? AND {key} <= ?", (high_water, upper))
                if cur.rowcount > 0:
                    self.set_meta(marker_key, str(upper), cursor=conn)
                return True
            # Compound-key or rowid trash table: legacy chunked delete. These shadow tables are small, so
            # the quadratic re-scan is not a concern (#79324 keeps the high-water path for the big
            # single-key tables).
            cur = conn.execute(
                f"DELETE FROM {tbl} WHERE ({key}) IN (SELECT {key} FROM {tbl} LIMIT {self._FTS_REBUILD_CHUNK_ROWS})"
            )
            return _drop(conn) if cur.rowcount == 0 else True  # True: more trash tables / chunks may remain

        def _drop(conn, marker_key: Optional[str] = None) -> bool:
            """Drained — the DROP is cheap now. True: re-check for more trash."""
            conn.execute(f"DROP TABLE IF EXISTS {tbl}")
            if marker_key is not None:
                _delete_meta(conn, marker_key)
            logger.info("Old FTS shadow table %s torn down.", tbl)
            return True

        try:
            return bool(self._execute_write(_do))
        except sqlite3.OperationalError as exc:
            logger.debug("FTS trash teardown chunk failed (will retry): %s", exc)
            return True

    def _fts_cjk_reset_if_stale(self) -> None:
        """From-scratch rebuild of a stale cjk index (triggers were dropped, gap extent unknown):
        drop table + triggers, clear the breadcrumb, recreate (fresh backfill markers)."""
        if not self._fts_cjk_loaded:
            return

        def _do(conn):
            if _meta_row(conn, FTS_CJK_STALE_KEY) is None:
                return False
            for trig in _FTS_CJK_TRIGGERS:
                conn.execute(f"DROP TRIGGER IF EXISTS {trig}")
            conn.execute("DROP TABLE IF EXISTS messages_fts_cjk")
            conn.execute("DROP VIEW IF EXISTS messages_fts_cjk_src")
            _delete_meta(conn, FTS_CJK_STALE_KEY, "fts_cjk_rebuild_high_water", "fts_cjk_rebuild_progress")
            return True
        if self._execute_write(_do):
            # Recreate OUTSIDE the write transaction: executescript() implicitly commits.
            self._ensure_cjk_schema_committed()

    def _ensure_cjk_schema_committed(self) -> None:
        with self._lock:
            self._ensure_fts_cjk_schema(self._conn)
            self._conn.commit()

    def _fts_external_index_empty_with_messages(self, conn) -> bool:
        """True when the base FTS table indexes nothing while ``messages`` has rows (post-demote
        empty-index shape). Caller holds ``self._lock``. docsize is the authoritative "is this
        rowid indexed" surface; EXISTS not COUNT(*) because this runs on every writable open."""
        try:
            if not conn.execute("SELECT EXISTS(SELECT 1 FROM messages)").fetchone()[0]:
                return False
            return not conn.execute("SELECT EXISTS(SELECT 1 FROM messages_fts_docsize)").fetchone()[0]
        except sqlite3.OperationalError:
            return False  # table absent / FTS disabled mid-init — not this failure class

    def _reseed_missing_progress(self, conn) -> None:
        """high_water without progress: fts_rebuild_step reads missing progress as "done by
        another process" and optimize would no-op then stamp. Reset to known-empty, re-seed.
        Truncation goes through FTS5 ``'delete-all'`` (a plain DELETE is O(rows) and corrupts
        the index when indexed rows diverged from ``messages``); the backfill worker replays
        without an anti-join, so it needs a known-empty index. A missing docsize table counts
        as empty."""
        if _meta_row(conn, "fts_rebuild_progress") is None:
            if not self._fts_index_known_empty(conn):
                self._reset_fts_index_to_empty(conn)
            self.set_meta("fts_rebuild_progress", "0", cursor=conn)

    @staticmethod
    def _fts_index_known_empty(conn) -> bool:
        """True when the base external-content index holds no rows (a missing table counts as empty)."""
        try:
            return int(conn.execute("SELECT COUNT(*) FROM messages_fts_docsize").fetchone()[0]) == 0
        except sqlite3.OperationalError:
            return True

    @staticmethod
    def _reset_fts_index_to_empty(conn) -> None:
        """Truncate the v23 external-content tables via FTS5 ``'delete-all'`` (O(1); a plain DELETE is
        O(rows) and corrupts the index when indexed rows diverged from ``messages``)."""
        for tbl in ("messages_fts", "messages_fts_trigram"):
            with contextlib.suppress(sqlite3.OperationalError):  # table absent — already an empty surface
                conn.execute(f"INSERT INTO {tbl}({tbl}) VALUES('delete-all')")

    def _seed_fts_rebuild_markers(self, conn, *, force: bool = False) -> int:
        """Write ``fts_rebuild_high_water`` / ``fts_rebuild_progress`` for a full backfill; returns
        the high-water id. Without ``force`` an existing high_water only gets a missing
        progress key repaired. Caller holds the write transaction."""
        existing_hw = _meta_row(conn, "fts_rebuild_high_water")
        if existing_hw is not None and not force:
            self._reseed_missing_progress(conn)
            return int(existing_hw[0])
        hw = conn.execute("SELECT COALESCE(MAX(id), 0) FROM messages").fetchone()[0]
        self.set_meta("fts_rebuild_high_water", str(hw), cursor=conn)
        self.set_meta("fts_rebuild_progress", "0", cursor=conn)
        return int(hw)

    def _repair_optimize_bookkeeping(self) -> None:
        """Heal interrupted demote/backfill bookkeeping before optimize runs: orphan high_water
        gets progress re-seeded; an empty external index with messages and no markers gets
        a full backfill claim. Never invents markers on a still-legacy inline DB — optimize
        would then skip demote and INSERT against the inline table forever.

        Covers two post-#65798 failure classes:
        """
        def _do(conn):
            if _meta_row(conn, "fts_rebuild_high_water") is not None:
                self._reseed_missing_progress(conn)
                return
            if self._db_has_legacy_inline_fts(conn):
                return  # demote owns marker creation
            if self._fts_external_index_empty_with_messages(conn):
                _delete_meta(conn, "fts_storage_version")
                self._seed_fts_rebuild_markers(conn, force=True)
        self._execute_write(_do)

    def fts_optimize_available(self) -> bool:
        """True when `optimize_fts_storage()` has work: legacy inline FTS or a v23 trigram still
        carrying ``tool_calls`` (``_db_needs_fts_storage_upgrade``), an interrupted optimize
        (markers/trash), a CJK backfill on this tokenizer-capable host, or an empty external
        index without markers. False when FTS5 is unavailable."""
        if not self._fts_enabled or self.read_only:
            return False
        with self._read_ctx() as conn:
            return (
                self._db_needs_fts_storage_upgrade(conn)
                or _meta_row(conn, "fts_rebuild_high_water") is not None  # interrupted optimize
                # CJK work is only offerable when THIS process can tokenize.
                or (self._fts_cjk_loaded and (
                    _meta_row(conn, "fts_cjk_rebuild_high_water") is not None
                    or _meta_row(conn, FTS_CJK_STALE_KEY) is not None
                ))
                or self._has_fts_trash(conn)
                or self._fts_external_index_empty_with_messages(conn)
            )

    def _demote_legacy_fts_to_trash(self) -> int:
        """Demote upgrade-eligible FTS vtables and stage their shadow tables for chunked
        teardown; returns MAX(messages.id) as the rebuild high water. O(1) schema surgery
        — the heavy delete is deferred. Markers land in the same BEGIN IMMEDIATE, BEFORE
        the empty v23 schema is created (``executescript`` implicitly COMMITs), closing
        the crash window where trash + empty v23 tables exist with no backfill claim."""
        def _stage(conn):
            self._drop_fts_triggers(conn)
            conn.execute("DROP VIEW IF EXISTS messages_fts_trigram_src")
            if conn.execute(
                "SELECT 1 FROM sqlite_master WHERE type = 'table' AND name IN ('messages_fts', 'messages_fts_trigram') "
                "AND sql LIKE 'CREATE VIRTUAL TABLE%' LIMIT 1"
            ).fetchone():
                conn.execute("PRAGMA writable_schema=ON")
                conn.execute(
                    "DELETE FROM sqlite_master WHERE type = 'table' "
                    "AND name IN ('messages_fts', 'messages_fts_trigram') AND sql LIKE 'CREATE VIRTUAL TABLE%'"
                )
                conn.execute("PRAGMA writable_schema=RESET")
                for row in conn.execute(
                    "SELECT name FROM sqlite_master WHERE type = 'table' "
                    "AND (name LIKE 'messages_fts_%' ESCAPE '\\' "
                    "OR name LIKE 'messages_fts_trigram_%' ESCAPE '\\') "
                    # messages_fts_cjk* is an independent v23+ index, not part of the
                    # demoted legacy layout: fts5's xRename renames the entire shadow
                    # family in one step, so sweeping the cjk vtable here aborts the
                    # loop on the next cjk shadow entry and drags _config — needed by
                    # the vtable constructor — into the trash family (#103647).
                    "AND name NOT LIKE 'messages\\_fts\\_cjk%' ESCAPE '\\'"
                ).fetchall():
                    conn.execute(f"ALTER TABLE {row[0]} RENAME TO fts_v22_trash_{row[0]}")
            # Claim the backfill BEFORE the empty v23 tables exist so a crash before
            # schema ensure resumes instead of stamping an empty index.
            hw = self._seed_fts_rebuild_markers(conn, force=True)
            _delete_meta(conn, "fts_optimize_available")
            return hw

        hw = int(self._execute_write(_stage))
        # Outside the write transaction (executescript commits); markers are durable.
        self._ensure_v23_fts_tables("failed to create v23 messages_fts during optimize-storage demote")
        return hw

    def _ensure_v23_fts_tables(self, failure_message: str) -> None:
        """Ensure the v23 base + trigram tables under the lock (IF NOT EXISTS, cheap); raise
        *failure_message* without the base table (the backfill loop would retry forever)."""
        with self._lock:
            base_ok = self._ensure_fts_schema(self._conn, "messages_fts", FTS_SQL)
            trigram_ok = self._ensure_fts_schema(self._conn, "messages_fts_trigram", FTS_TRIGRAM_SQL)
            self._trigram_available = bool(trigram_ok)
            if not base_ok:
                raise sqlite3.OperationalError(failure_message)
            self._conn.commit()

    def _optimize_vacuum(self) -> bool:
        """Phase 3: reclaim freed pages to the OS. False when VACUUM failed (usually no free disk
        for its temp copy; a later VACUUM reclaims)."""
        try:
            with self._lock:
                self._conn.execute("VACUUM")
            vacuum_ok = True
        except sqlite3.OperationalError as exc:
            logger.warning("VACUUM after FTS optimize failed: %s", exc)
            vacuum_ok = False
        # Best-effort WAL fold-back, REFUSED (SQLITE_BUSY) while another connection holds a
        # read-mark (callers size via logical_size_bytes, not stat()). PASSIVE, never TRUNCATE:
        # a TRUNCATE reset from a transient CLI would race a live writer.
        try:
            with self._lock:
                # Best-effort: fold the WAL back into the main file so the on-disk size settles now rather
                # than at close(). Callers must therefore NOT size the result by stat()ing the file; use
                # :meth:`logical_size_bytes`, which is truthful immediately regardless of readers. See
                # #45383.
                self._conn.execute("PRAGMA wal_checkpoint(PASSIVE)")
        except Exception as exc:
            logger.debug("WAL checkpoint (PASSIVE) after optimize VACUUM failed: %s", exc)
        return vacuum_ok

    def _optimize_settle(self, conn) -> Optional[str]:
        """Phase 4 (inside the write transaction, so a concurrent writer cannot race a stamp past
        incomplete work): stamp the FTS layout (source of truth for "optimized"), clear the
        "available" flag, advance a lagging schema_version. Returns a refusal reason or None.
        Refuses while optimize work remains; an empty base index against non-empty messages
        also refuses (settling there meant permanent search-index loss)."""
        if _meta_row(conn, "fts_rebuild_high_water") is not None:
            return "backfill_incomplete"
        if self._has_fts_trash(conn):
            return "teardown_incomplete"
        if self._fts_external_index_empty_with_messages(conn):
            return "backfill_incomplete"
        self.set_meta("fts_storage_version", str(FTS_STORAGE_VERSION), cursor=conn)
        _delete_meta(conn, "fts_optimize_available")
        conn.execute("UPDATE schema_version SET version = ? WHERE version < ?", (SCHEMA_VERSION, SCHEMA_VERSION))
        return None

    def optimize_fts_storage(
        self, *, progress_cb: Optional[Callable[[dict[str, Any]], None]] = None, vacuum: bool = True
    ) -> dict[str, Any]:
        """Repair an older FTS layout into the current v23 shape, foreground and to completion:
        legacy-v22 inline -> external-content, or a v23 ``messages_fts_trigram`` that still stores
        ``tool_calls``. Re-running resumes. ``progress_cb`` receives {"phase", "percent",
        "indexed", "total"}. A missing trigram tokenizer is not fatal (CJK falls back to LIKE)."""
        if not self._fts_enabled:
            return {"ok": False, "reason": "fts5_unavailable"}
        if self.read_only:
            return {"ok": False, "reason": "read_only"}

        # Heal bookkeeping BEFORE deciding whether to demote again.
        self._repair_optimize_bookkeeping()
        with self._lock:
            needs_storage_upgrade = self._db_needs_fts_storage_upgrade(self._conn)
        pending = self.get_meta("fts_rebuild_high_water") is not None
        if needs_storage_upgrade and not pending:
            self._demote_legacy_fts_to_trash()
        elif pending and not needs_storage_upgrade:
            # Resume mid-demote: the process may have died between the staged demote
            # commit and schema ensure.
            self._ensure_v23_fts_tables("failed to re-create v23 messages_fts on optimize-storage resume")

        # A stale CJK index can only be recovered from scratch; then ensure table +
        # markers exist (a v23 DB gaining the cjk index for the first time).
        self._fts_cjk_reset_if_stale()
        if self._fts_cjk_loaded:
            self._ensure_cjk_schema_committed()

        def _emit(phase: str) -> None:
            if progress_cb is None:
                return
            st = self.fts_rebuild_status() or self.fts_cjk_rebuild_status()
            progress_cb({"phase": phase, "percent": st["percent"] if st else 100,
                         "indexed": st["indexed"] if st else 0, "total": st["total"] if st else 0})

        def _drive(phase: str, step) -> None:
            """Run *step* to completion; the inter-chunk sleep is the single place the duty
            cycle is enforced — back-to-back BEGIN IMMEDIATE chunks starve a live
            gateway/CLI out of its lock retries."""
            while True:
                _t0 = time.monotonic()
                if not step():
                    break
                _emit(phase)
                time.sleep(max(self._FTS_REBUILD_MIN_PAUSE, (time.monotonic() - _t0) * self._FTS_REBUILD_DUTY_FACTOR))

        # Phase 1: base backfill; 1b: CJK-bigram backfill (own marker pair).
        _emit("backfill")
        _drive("backfill", self.fts_rebuild_step)
        _emit("backfill")
        _drive("backfill", self.fts_cjk_rebuild_step)
        # Phase 2: tear down the demoted legacy shadow tables in chunks.
        _emit("teardown")
        _drive("teardown", self._fts_teardown_trash_step)
        with self._read_ctx() as conn:
            still_pending = _meta_row(conn, "fts_rebuild_high_water") is not None
            still_trash = self._has_fts_trash(conn)
            empty_index = self._fts_external_index_empty_with_messages(conn)
        if still_pending or still_trash or empty_index:
            reason = "backfill_incomplete" if still_pending or empty_index else "teardown_incomplete"
            logger.warning("FTS storage optimization did not settle (%s): pending=%s trash=%s empty_index=%s",
                           reason, still_pending, still_trash, empty_index)
            return {"ok": False, "reason": reason, "vacuumed": None}

        vacuum_ok = None
        if vacuum:
            _emit("vacuum")
            vacuum_ok = self._optimize_vacuum()
        refusal = self._execute_write(self._optimize_settle)
        if refusal is not None:
            # A concurrent process changed state since the pre-vacuum check; a re-run can still settle.
            logger.warning("FTS storage optimization settle refused (%s)", refusal)
            return {"ok": False, "reason": refusal, "vacuumed": vacuum_ok}
        _emit("done")
        logger.info("FTS storage optimization complete (layout v%d).", FTS_STORAGE_VERSION)
        return {"ok": True, "vacuumed": vacuum_ok}

    # ── Read views ─────────────────────────────────────────────────────────

    def get_anchored_view(
        self, session_id: str, around_message_id: int, window: int = 5, bookend: int = 3,
        keep_roles: Optional[tuple[str, ...]] = ("user", "assistant")) -> dict[str, Any]:
        """Anchored window (``get_messages_around``) plus session bookends, so one call yields the
        goal and the resolution of a long session. ``window`` is filtered to ``keep_roles``
        (None disables) EXCEPT the anchor; ``bookend_start`` / ``bookend_end`` are the
        first/last ``bookend`` non-empty-content messages with ids strictly outside the
        window (empty when it overlaps the head/tail). Empty result when the anchor isn't
        in the session."""
        bookend = max(bookend, 0)
        primitive = self.get_messages_around(session_id, around_message_id, window=window)
        window_rows = primitive["window"]
        if not window_rows:
            return {"window": [], "messages_before": 0, "messages_after": 0, "bookend_start": [], "bookend_end": []}

        filtered_window = window_rows
        if keep_roles is not None:
            keep_set = set(keep_roles)
            filtered_window = [m for m in window_rows if m.get("id") == around_message_id or m.get("role") in keep_set]
        bookend_start_rows: list[Any] = []
        bookend_end_rows: list[Any] = []
        if bookend > 0:
            role_clause = "" if keep_roles is None else f" AND role IN ({','.join('?' for _ in keep_roles)})"
            role_params = [] if keep_roles is None else list(keep_roles)
            with self._read_ctx() as conn:
                def _bookend(op: str, boundary_id: int, order: str):
                    return conn.execute(
                        f"SELECT * FROM messages "
                        f"WHERE session_id = ? AND id {op} ?{role_clause} "
                        f"AND length(content) > 0 "
                        f"ORDER BY id {order} LIMIT ?",
                        (session_id, boundary_id, *role_params, bookend),
                    ).fetchall()
                bookend_start_rows = _bookend("<", window_rows[0]["id"], "ASC")
                # End rows come back DESC for the LIMIT cap; flip to ASC.
                bookend_end_rows = list(reversed(_bookend(">", window_rows[-1]["id"], "DESC")))

        def _hydrate(row) -> dict[str, Any]:
            return self._row_to_message_dict(row, warn_context="get_anchored_view", summary_flag=False)
        return {
            "window": filtered_window, "messages_before": primitive["messages_before"],
            "messages_after": primitive["messages_after"],
            "bookend_start": [_hydrate(r) for r in bookend_start_rows],
            "bookend_end": [_hydrate(r) for r in bookend_end_rows],
        }

    def list_recent_user_messages(
        self, session_id: str, limit: int = 20, include_inactive: bool = False) -> list[dict[str, Any]]:
        """The *limit* most-recent real user turns, newest first, as ``{id, timestamp, preview}``
        (80 chars, whitespace collapsed); used by /rewind and ``/undo [N]``. Bookkeeping rows
        (``display_kind`` set) are excluded. Legacy compaction handoffs are role='user' rows
        with NO display_kind — invisible to SQL — so fetch with headroom and drop them in the
        decode loop; otherwise ``/undo N`` pairs an in-memory count that excludes handoffs
        with a DB pick that includes them."""
        active_clause = "" if include_inactive else " AND active = 1"
        # A /steer row is typed for the renderer but is human input: keep it so the DB pick agrees
        # with the in-memory user_originated_turn_view count.
        display_clause = " AND (display_kind IS NULL OR display_kind = '' OR display_kind = 'steer')"
        with self._read_ctx() as conn:
            rows = conn.execute(
                "SELECT id, timestamp, content FROM messages WHERE session_id = ? AND role = 'user'"
                f"{active_clause}{display_clause} "
                "ORDER BY id DESC LIMIT ?",
                (session_id, int(limit) * 2 + 5),
            ).fetchall()
        from agent.context_compressor import ContextCompressor
        result: list[dict[str, Any]] = []
        for row in rows:
            if len(result) >= int(limit):
                break
            decoded = self._decode_content(row["content"])
            if ContextCompressor._is_context_summary_content(decoded):
                continue  # compaction handoff — never a user-originated turn
            if isinstance(decoded, str):  # a /skill turn embeds the whole skill body; show what was typed
                preview = describe_skill_invocation(decoded) or decoded
            else:
                preview = _flatten_text(decoded)
            preview = " ".join(preview.split())
            if len(preview) > 80:
                preview = preview[:77] + "..."
            result.append({"id": row["id"], "timestamp": row["timestamp"], "preview": preview})
        return result

    # ── Query analysis ─────────────────────────────────────────────────────

    @staticmethod
    def _sanitize_fts5_query(query: str) -> str:
        """Sanitize user input for FTS5 MATCH (raw special characters raise): preserve paired
        quoted phrases, strip unmatched special characters, and quote hyphenated/dotted
        terms so FTS5 matches them as phrases (``chat-send``, ``P2.2``, ``my-app.config.ts``)."""
        # Cap before any regex processing so adversarial input stays bounded.
        query = query[:MAX_FTS5_QUERY_CHARS]

        # 1. Protect balanced quoted phrases via numbered placeholders (``"[^"]*"`` pairs
        # left-to-right without backtracking); a leftover unmatched quote becomes whitespace.
        _quoted_parts: list = []

        def _hold(m: re.Match[str]) -> str:
            _quoted_parts.append(m.group(0))
            return f"\x00Q{len(_quoted_parts) - 1}\x00"

        sanitized = _QUOTED_PHRASE_RE.sub(_hold, query).replace('"', " ")

        # 2. Strip FTS5-special characters (an unquoted ``TODO: fix`` parses as
        # ``column:term``). ``%`` is only spared for the CJK LIKE fallback.
        sanitized = _FTS5_SPECIAL_RE.sub(" ", sanitized)
        if "%" in sanitized and not SessionSearchMixin._contains_cjk(sanitized):
            sanitized = sanitized.replace("%", " ")
        # 3. Collapse repeated * and drop leading * (prefix needs a char).
        sanitized = re.sub(r"\*+", "*", sanitized)
        sanitized = re.sub(r"(^|\s)\*", r"\1", sanitized)
        # 4. Drop dangling boolean operators at start/end (syntax errors).
        sanitized = re.sub(r"(?i)^(AND|OR|NOT)\b\s*", "", sanitized.strip())
        sanitized = re.sub(r"(?i)\s+(AND|OR|NOT)\s*$", "", sanitized.strip())
        # 5. Quote dotted/hyphenated/underscored terms in ONE pass (sequential passes
        # double-quote ``my-app.config``).
        sanitized = re.sub(r"\b(\w+(?:[._-]\w+)+)\b", r'"\1"', sanitized)
        # 6. Restore preserved quoted phrases.
        for i, quoted in enumerate(_quoted_parts):
            sanitized = sanitized.replace(f"\x00Q{i}\x00", quoted)
        return sanitized.strip()

    @staticmethod
    def _contains_cjk(text: str) -> bool:
        return any(_is_cjk(ord(ch)) for ch in text)

    @staticmethod
    def _count_cjk(text: str) -> int:
        return sum(1 for ch in text if _is_cjk(ord(ch)))

    @staticmethod
    def _has_lone_cjk_run(query: str) -> bool:
        """True when any maximal CJK run is a single char: the cjk-bigram index stores unigrams
        only for isolated chars, so such a term can't match inside longer runs — keep LIKE."""
        run = 0
        for ch in query:
            if _is_cjk(ord(ch)):
                run += 1
            else:
                if run == 1:
                    return True
                run = 0
        return run == 1

    @staticmethod
    def _or_relaxed_query(query: str) -> Optional[str]:
        """The sanitized implicit-AND query rewritten as an any-term OR query, or ``None`` when
        relaxation does not apply: fewer than two searchable units (a single term cannot relax)
        or explicit ``OR``/``NOT`` (the caller expressed exact semantics). Quoted phrases stay
        whole units: ``"docker networking" tls`` -> ``"docker networking" OR tls``."""
        units: list[str] = []
        for raw_token in _LIKE_TOKEN_RE.findall(query):
            upper = raw_token.upper()
            if upper in {"OR", "NOT"}:
                return None
            if upper != "AND":
                units.append(raw_token)
        return " OR ".join(units) if len(units) >= 2 else None

    @staticmethod
    def _trigram_eligible_tokens(query: str) -> bool:
        """True when every non-operator token is >=3 chars: a shorter token produces no
        trigrams, and with FTS5's implicit AND one such token empties the whole MATCH."""
        tokens = _non_operator_tokens(query.strip('"').strip())
        return bool(tokens) and all(len(t) >= 3 for t in tokens)

    @classmethod
    def _has_short_cjk_token(cls, raw_query: str) -> bool:
        """True when any non-operator CJK token has fewer than 3 CJK chars — the trigram
        tokenizer needs >=3 per token, so such a query must take the LIKE route."""
        return any(cls._count_cjk(t) < 3 for t in _non_operator_tokens(raw_query) if cls._contains_cjk(t))

    def _trigram_route_ok(self, raw_query: str) -> bool:
        """Per-token CJK length gate for the trigram index: ``广西 OR 桂林 OR 漓江`` has 6
        CJK chars total but 2 per token, so trigram returns 0."""
        return (self._count_cjk(raw_query) >= 3 and not self._has_short_cjk_token(raw_query)
                and self._trigram_available)

    def _describe_search_path(self, query: str) -> str:
        """Best-effort name of the routing path a query takes (log-only)."""
        try:
            if self._fts_stale:
                return "like_scan_fts_stale"
            sanitized = self._sanitize_fts5_query(query or "")
            if not sanitized:
                return "empty"
            if not self._contains_cjk(sanitized):
                return "fts5"
            raw = sanitized.strip('"').strip()
            if self._fts_cjk_available and not self._has_lone_cjk_run(raw):
                return "fts_cjk"
            return "trigram" if self._trigram_route_ok(raw) else "like_scan"
        except Exception:
            return "unknown"

    # ── Query builders / runners ───────────────────────────────────────────

    @staticmethod
    def _fts_match_sql(table: str, match_query: str, order_by_sql: str, *, limit: int, offset: int,
                       **filters) -> tuple[str, list]:
        """MATCH query + params against one FTS5 index joined to messages/sessions."""
        where = [f"{table} MATCH ?"]
        params: list = [match_query]
        _search_filter_clauses(where, params, **filters)
        params.extend([limit, offset])
        sql = _search_select_sql(
            f"snippet({table}, -1, '>>>', '<<<', '...', 40) AS snippet",
            f"{table}\n            JOIN messages m ON m.id = {table}.rowid", where, order_by_sql, "LIMIT ? OFFSET ?",
        )
        return sql, params

    def _match_rows(self, table: str, match_query: str, order_by_sql: str, *, fail_open: Optional[str] = None,
                    operational_debug: Optional[str] = None, **kwargs) -> Optional[list[dict[str, Any]]]:
        """Run one MATCH against *table*; ``None`` when the query cannot execute (tokenizer /
        syntax) so the caller falls back. *fail_open* names the index for the
        substring-capable routes: a corruption-class ``DatabaseError`` there detaches the
        derived indexes (``_enter_fts_fail_open``) and answers from canonical rows — a live
        search never runs the unbounded rebuild. Other ``DatabaseError``s propagate."""
        sql, params = self._fts_match_sql(table, match_query, order_by_sql, **kwargs)
        try:
            return [dict(row) for row in self._read_all(sql, params)]
        except sqlite3.OperationalError:
            if operational_debug:
                logger.debug(operational_debug, exc_info=True)
            return None
        except sqlite3.DatabaseError as exc:
            if fail_open is None or not self._enter_fts_fail_open(exc):
                raise
            logger.warning(
                "%s FTS search hit a corruption error (%s); detached FTS and falling back to canonical LIKE.",
                fail_open, exc)
            return None

    def _like_rows(self, where: list[str], params: list, *, order_by: str, limit_sql: str) -> list[dict[str, Any]]:
        """Canonical-table LIKE scan; ``params[0]`` is the snippet anchor term."""
        sql = _search_select_sql(_LIKE_SNIPPET_SQL, "messages m", where, order_by, limit_sql)
        return [dict(row) for row in self._read_all(sql, params)]

    @staticmethod
    def _compile_like_boolean_query(query: str) -> tuple[str, list[Any], Optional[str]]:
        """Compile the supported FTS boolean subset into LIKE predicates: terms within an OR
        group are ANDed (FTS5's implicit conjunction) and ``NOT`` negates the next term."""
        groups: list[list[tuple[str, bool]]] = [[]]
        negate_next = False
        for raw_token in _LIKE_TOKEN_RE.findall(query):
            operator = raw_token.upper()
            if operator == "OR":
                if groups[-1]:
                    groups.append([])
                negate_next = False
                continue
            if operator in {"AND", "NEAR"}:
                continue
            if operator == "NOT":
                negate_next = True
                continue
            term = raw_token.strip('"').strip("*").strip()
            if term:
                groups[-1].append((term, negate_next))
                negate_next = False

        compiled_groups: list[str] = []
        params: list[Any] = []
        snippet_term: Optional[str] = None
        for group in groups:
            if not group or not any(not negated for _, negated in group):
                continue
            clauses: list[str] = []
            for term, negated in group:
                clauses.append(f"NOT {_LIKE_COALESCED_COLUMN_SQL}" if negated else _LIKE_COALESCED_COLUMN_SQL)
                params.extend(_like_params(term))
                if snippet_term is None and not negated:
                    snippet_term = term
            compiled_groups.append(f"({' AND '.join(clauses)})")
        return " OR ".join(compiled_groups), params, snippet_term

    def _search_messages_like_fallback(
        self, query: str, *, limit: int, offset: int, sort: Optional[str], **filters) -> list[dict[str, Any]]:
        """Search canonical messages while derived FTS state is stale."""
        predicate, params, snippet_term = self._compile_like_boolean_query(query)
        if not predicate or snippet_term is None:
            return []
        where = [f"({predicate})"]
        _search_filter_clauses(where, params, **filters)
        order = "ASC" if isinstance(sort, str) and sort.strip().lower() == "oldest" else "DESC"
        return self._like_rows(where, [snippet_term, *params, limit, offset],
                               order_by=f"ORDER BY m.timestamp {order}, m.id {order}", limit_sql="LIMIT ? OFFSET ?")

    def _refresh_fts_stale_state(self) -> None:
        """Observe fail-open initiated by another process sharing state.db."""
        if self._fts_stale or not self._fts_enabled:
            return
        try:
            stale = self._read_one("SELECT 1 FROM state_meta WHERE key = ? LIMIT 1", (FTS_STALE_KEY,))
        except sqlite3.Error:
            return
        if stale is not None:
            self._fts_stale = True
            self._fts_enabled = self._trigram_available = self._fts_cjk_available = False

    def _finalize_search_matches(
        self, matches: list[dict[str, Any]], result_fields: Optional[Collection[str]] = None) -> list[dict[str, Any]]:
        """Attach neighboring messages in bounded batches, only when context is requested."""
        if result_fields is None or "context" in result_fields:
            for start in range(0, len(matches), 500):
                batch = matches[start:start + 500]
                contexts = {match["id"]: [] for match in batch}
                try:
                    sql = _CONTEXT_WINDOW_SQL.format(ids=",".join("?" for _ in contexts))
                    with self._read_ctx() as conn:
                        rows = conn.execute(sql, list(contexts)).fetchall()
                    for row in rows:
                        contexts[row["match_id"]].append(row)
                except Exception:
                    contexts = {}
                for match in batch:
                    try:
                        match["context"] = [
                            {"role": row["role"], "content": _flatten_text(self._decode_content(row["content"]))[:200]}
                            for row in contexts.get(match["id"], [])]
                    except Exception:
                        match["context"] = []
        # No route selects full content; the pop guards any future one that does.
        for match in matches:
            match.pop("content", None)
        if result_fields is not None:
            matches = [{field: match[field] for field in result_fields if field in match} for match in matches]
        return matches

    # ── search_messages ────────────────────────────────────────────────────

    def search_messages(
        self, query: str, source_filter: list[str] | None = None, exclude_sources: list[str] | None = None,
        role_filter: list[str] | None = None, limit: int = 20, offset: int = 0, sort: str | None = None,
        include_inactive: bool = False, fields: Optional[Collection[str]] = None,
        after_ts: Optional[int] = None, before_ts: Optional[int] = None,
    ) -> list[dict[str, Any]]:
        """:meth:`_search_messages_impl` plus one log line per slow search with the routing
        path taken. Threshold HERMES_SEARCH_SLOW_MS (default 1000; 0 logs every call)."""
        started = time.time()
        rows = None
        try:
            rows = self._search_messages_impl(
                query, source_filter=source_filter, exclude_sources=exclude_sources, role_filter=role_filter,
                limit=limit, offset=offset, sort=sort, include_inactive=include_inactive, fields=fields,
                after_ts=after_ts, before_ts=before_ts)
            return rows
        finally:
            elapsed_ms = (time.time() - started) * 1000.0
            if elapsed_ms >= _search_slow_ms():
                logger.info("slow session search: path=%s elapsed=%.0fms rows=%s query=%r",
                            self._describe_search_path(query), elapsed_ms, len(rows) if rows is not None else "err",
                            query[: 200])

    def _search_messages_impl(
        self, query: str, source_filter: list[str] | None = None, exclude_sources: list[str] | None = None,
        role_filter: list[str] | None = None, limit: int = 20, offset: int = 0, sort: str | None = None,
        include_inactive: bool = False, fields: Optional[Collection[str]] = None,
        after_ts: Optional[int] = None, before_ts: Optional[int] = None,
    ) -> list[dict[str, Any]]:
        """FTS5 search across session messages (keywords, ``"phrases"``, AND/OR/NOT, ``prefix*``).
        Returns snippet + session metadata + 1-message context per hit; ``fields`` selects a
        projection. ``sort``: None = BM25 rank; "newest"/"oldest" = timestamp then rank (the
        CJK LIKE fallback ignores it). Rewound rows (``active=0, compacted=0``) are excluded
        by default; compaction-archived rows ARE included; ``include_inactive`` = every row.
        ``after_ts``/``before_ts`` bound ``sessions.started_at`` on every route (FTS5, CJK,
        trigram, LIKE fallback, unindexed-gap supplement)."""
        result_fields = self._search_message_fields(fields)
        if not query or not query.strip():
            return []
        query = self._sanitize_fts5_query(query)
        if not query:
            return []
        filters = dict(include_inactive=include_inactive, source_filter=source_filter,
                       exclude_sources=exclude_sources, role_filter=role_filter,
                       after_ts=after_ts, before_ts=before_ts)
        # New oversized tool results index only a bounded prefix; an explicit tool-role search is the
        # opt-in full-body path and scans canonical rows via LIKE.
        if role_filter and "tool" in role_filter:
            matches = self._search_messages_like_fallback(query, limit=limit, offset=offset, sort=sort, **filters)
            return self._finalize_search_matches(matches, result_fields=result_fields)
        self._refresh_fts_stale_state()
        if self._fts_stale:
            matches = self._search_messages_like_fallback(query, limit=limit, offset=offset, sort=sort, **filters)
            return self._finalize_search_matches(matches, result_fields=result_fields)
        if not self._fts_enabled:
            return []

        order_by_sql = _FTS_ORDER_BY.get(sort.strip().lower() if isinstance(sort, str) else None, "ORDER BY rank")
        route = dict(order_by_sql=order_by_sql, limit=limit, offset=offset, **filters)
        # Tool rows and FTS_TRIGRAM_EXCLUDED_SOURCES sessions are excluded from the trigram/cjk
        # indexes (see FTS_TRIGRAM_SQL); an explicit filter for them must scan the base table.
        wants_unindexed_rows = (bool(role_filter) and "tool" in role_filter) or (
            bool(source_filter) and any(src in FTS_TRIGRAM_EXCLUDED_SOURCES for src in source_filter))
        is_cjk = self._contains_cjk(query)
        if is_cjk:
            matches = self._search_cjk(query, wants_unindexed_rows, route)
        else:
            sql, params = self._fts_match_sql("messages_fts", query, **route)
            try:
                matches = [dict(row) for row in self._read_all(sql, params)]
            except sqlite3.OperationalError:
                return []  # FTS5 syntax error despite sanitization
            except sqlite3.DatabaseError as exc:
                # Corruption parent class: detach the derived indexes and answer from
                # canonical rows; repair paths own the rebuild.
                # A corrupt FTS index raises the malformed / "fts5: corrupt structure record" class on the
                # MATCH read, the same class the write path handles (#66296). OperationalError (query
                # syntax) is a subclass caught above; this arm is the corruption parent. The existing
                # stale-open/repair paths retain rebuild ownership.
                if not self._enter_fts_fail_open(exc):
                    raise
                matches = self._search_messages_like_fallback(query, limit=limit, offset=offset, sort=sort, **filters)

        # Deferred-rebuild supplement: while the backfill is pending the FTS indexes miss
        # the (progress, high_water] gap; top up with a bounded LIKE scan so old messages
        # never vanish mid-rebuild. Cost decays to zero as the backfill advances.
        if self.fts_rebuild_status() is not None and len(matches) < limit:
            try:
                gap_matches = self._search_unindexed_gap(query, limit - len(matches), **filters)
                seen_ids = {m["id"] for m in matches}
                matches.extend(m for m in gap_matches if m["id"] not in seen_ids)
            except sqlite3.OperationalError as exc:
                logger.debug("Unindexed-gap supplement skipped: %s", exc)

        # unicode61 puts no boundary between Latin and adjacent CJK ("修改youer服务端" is
        # one token, so MATCH "youer" misses). On a zero-result Latin miss retry the
        # substring-capable indexes: cjk first (exact ranked match), then trigram (>=3-char
        # tokens). Gated on a miss so hits keep their ranking ("cat" may then match
        # "concatenate"). Skipped for role='tool' (both indexes exclude tool rows).
        if not matches and not is_cjk and not (bool(role_filter) and "tool" in role_filter):
            fb_query = _quote_fts_tokens(query.strip('"').strip())
            # ── CJK-bigram route (messages_fts_cjk, cjk_unicode61) ────── When the bigram index is
            # available it serves EVERY CJK query shape the legacy code split between trigram (>=3
            # chars/token) and LIKE full scans (1-2 char tokens) — the whole point of the index (PR #65544).
            # Exceptions stay on the legacy routes: - role_filter=['tool'] queries (tool rows aren't in the
            # cjk index, same exclusion as trigram), - queries containing a LONE 1-char CJK run: the index
            # stores bigrams for runs >=2, so a single-char term can only match isolated chars — LIKE
            # substring semantics are broader.
            if self._fts_cjk_available:
                matches = self._match_rows("messages_fts_cjk", fb_query, **route) or matches
            if not matches and self._trigram_available and self._trigram_eligible_tokens(query):
                matches = self._match_rows("messages_fts_trigram", fb_query, **route) or matches

        # OR-relaxed retry: the implicit AND between terms means a paraphrased multi-word query
        # misses a stored sentence that lacks even ONE word ("when does Sarah like her standup
        # scheduled" vs "Sarah prefers the standup meeting scheduled ... Thursday mornings"). Once
        # the exact query and the substring fallbacks all miss, retry the unicode61 index matching
        # ANY term. The caller's ``sort`` still applies (``route`` carries order_by_sql): rank order
        # puts rows covering more terms first, newest/oldest keep their timestamp order. Gated on a
        # zero-result miss so hits keep exact-match semantics; explicit OR/NOT, single-term and
        # CJK-routed queries are left alone.
        if not matches and not is_cjk and not self._fts_stale:
            relaxed = self._or_relaxed_query(query)
            if relaxed is not None:
                matches = self._match_rows("messages_fts", relaxed, fail_open="OR-relaxed",
                                           operational_debug="OR-relaxed FTS retry failed; keeping empty result",
                                           **route) or matches
        return self._finalize_search_matches(matches, result_fields=result_fields)

    def _search_cjk(self, query: str, wants_unindexed_rows: bool, route: dict[str, Any]) -> list[dict[str, Any]]:
        """CJK routing: the unicode61 table splits CJK into single characters (false positives,
        missed phrases). cjk-bigram serves every shape except queries wanting rows the
        substring indexes exclude (role='tool', cron/subagent sources) and LONE
        1-char CJK runs (bigrams only exist for runs >=2 — LIKE is broader); then trigram
        (>=3 CJK chars per token); then a LIKE substring scan with one clause per
        non-operator token so "广西 OR 桂林 OR 漓江" matches each term."""
        raw_query = _strip_cjk_wildcards(query).strip('"').strip()
        match_query = _quote_fts_tokens(raw_query)
        if self._fts_cjk_available and not wants_unindexed_rows and not self._has_lone_cjk_run(raw_query):
            matches = self._match_rows(
                "messages_fts_cjk", match_query, fail_open="CJK-bigram",
                operational_debug="messages_fts_cjk query failed; falling back to trigram/LIKE", **route)
            if matches is not None:
                return matches
        if self._trigram_route_ok(raw_query) and not wants_unindexed_rows:
            matches = self._match_rows("messages_fts_trigram", match_query, fail_open="Trigram", **route)
            if matches is not None:
                return matches
        non_op_tokens = _non_operator_tokens(raw_query) or [raw_query]
        like_params: list = [p for tok in non_op_tokens for p in _like_params(tok)]
        like_where = [f"({' OR '.join([_LIKE_ANY_COLUMN_SQL] * len(non_op_tokens))})"]
        filters = {k: route[k] for k in ("include_inactive", "source_filter", "exclude_sources", "role_filter",
                                         "after_ts", "before_ts")}
        _search_filter_clauses(like_where, like_params, **filters)
        # instr() for the snippet uses the first search token.
        return self._like_rows(like_where, [non_op_tokens[0], *like_params, route["limit"], route["offset"]],
                               order_by="ORDER BY m.timestamp DESC", limit_sql="LIMIT ? OFFSET ?")

    def _search_unindexed_gap(self, fts_query: str, limit: int, **filters) -> list[dict[str, Any]]:
        """LIKE-scan ids in (fts_rebuild_progress, fts_rebuild_high_water] — rows the deferred
        rebuild hasn't indexed yet. The FTS query degrades to AND-joined substring terms
        (quoted phrases kept whole): recall-over-precision mid-rebuild."""
        status = self.fts_rebuild_status()
        if status is None or limit <= 0:
            return []
        terms = [tok for tok in (t.strip('"').strip("*").strip() for t in _LIKE_TOKEN_RE.findall(fts_query))
                 if tok and tok.upper() not in _LIKE_SKIP_TOKENS]
        if not terms:
            return []
        where = ["m.id > ? AND m.id <= ?", *([_LIKE_ANY_COLUMN_SQL] * len(terms))]
        params: list = [status["indexed"], status["total"], *(p for term in terms for p in _like_params(term))]
        _search_filter_clauses(where, params, **filters)
        return self._like_rows(where, [terms[0], *params, limit], order_by="ORDER BY m.timestamp DESC",
                               limit_sql="LIMIT ?")

    def search_sessions_by_id(
        self, query: str, limit: int = 20, include_archived: bool = True, source: str | None = None,
        sources: list[str] | None = None, exclude_sources: list[str] | None = None) -> list[dict[str, Any]]:
        """Search surfaced sessions by exact/prefix/substring session id. Also matches
        ``_lineage_root_id`` so an old compression root id resolves to the live continuation."""
        needle = (query or "").strip().lower()
        if not needle or limit <= 0:
            return []
        # list_sessions_rich pushes the id LIKE filter (own id + forward compression
        # chain) into SQL; over-fetch so the in-Python ranking has candidates.
        candidates = self.list_sessions_rich(
            source=source, sources=sources, exclude_sources=exclude_sources, limit=max(limit * 4, limit),
            offset=0, include_archived=include_archived, order_by_last_active=True, id_query=needle)

        def score(row: dict[str, Any]) -> int:
            normalized = [v.lower() for v in (str(row.get("id") or ""), str(row.get("_lineage_root_id") or "")) if v]
            if any(value == needle for value in normalized):
                return 0
            return 1 if any(value.startswith(needle) for value in normalized) else 2
        ranked = sorted(enumerate(candidates), key=lambda item: (score(item[1]), item[0]))
        return [row for _, row in ranked[:limit]]

    # ── FTS maintenance commands ───────────────────────────────────────────

    def _fts_table_exists(self, name: str) -> bool:
        """True if an FTS5 virtual table is queryable ("no such table" and "vtable
        constructor failed" — missing tokenizer / mid-teardown — both count as not)."""
        try:
            self._conn.execute(f"SELECT 1 FROM {name} LIMIT 0")
            return True
        except sqlite3.DatabaseError:
            return False

    def _present_fts_tables(self) -> list[str]:
        """Queryable FTS tables (caller holds ``self._lock``)."""
        return [tbl for tbl in self._FTS_TABLES if self._fts_table_exists(tbl)]

    def optimize_fts(self) -> int:
        """Merge fragmented FTS5 segments into one per index (``'optimize'``). Pure
        maintenance: changes neither results nor ``snippet()`` output, only layout and
        speed; VACUUM then returns the freed pages. Returns the number optimized. A quarantined
        handle never issues ``'optimize'``: it rewrites index segments in place and would compound
        structural damage (or a split WAL generation) instead of leaving it diagnosable."""
        self._raise_if_db_corrupt()
        optimized = 0
        with self._lock:
            self._raise_if_db_replaced()
            if self._conn is None:
                self._reopen_after_close_locked(context="write")
            for tbl in self._present_fts_tables():
                try:
                    self._conn.execute(f"INSERT INTO {tbl}({tbl}) VALUES('optimize')")
                    optimized += 1
                except sqlite3.DatabaseError as exc:  # SQLITE_CORRUPT is not an OperationalError
                    logger.warning("FTS optimize failed for %s: %s", tbl, exc)
        return optimized

    def rebuild_fts(self) -> int:
        """Rebuild FTS5 indexes from ``messages`` (``'rebuild'``) — the recovery for a corrupt index
        that rejects writes while reads succeed. Two processes rebuilding one state.db
        concurrently corrupted production DBs, so this admits through
        ``fts_rebuild_admission`` and FAILS CLOSED, returning 0 on deferral (callers treat 0
        as "no progress" and use the stale-FTS breadcrumb path). Returns indexes rebuilt.

        Uses the FTS5 ``'rebuild'`` command, which rewrites the internal b-tree segments from the content
        rows. Unlike ``optimize_fts`` (which merges existing segments), ``rebuild`` discards and recreates
        the index data entirely — the more destructive of the two, so it is quarantined the same way. See
        #50502.
        A full structural rebuild must never run concurrently in two processes sharing one state.db — that
        interleaving has structurally corrupted the database in production (PR #93200) — so this admits
        through the cross-process ``fts_rebuild_admission`` authority and FAILS CLOSED: if another process
        holds the rebuild lock beyond the bounded wait, this call defers (returns 0) rather than racing it.
        Callers already treat 0 as "rebuild made no progress" and fall back to the stale-FTS breadcrumb
        path, which retries in-process from the gateway housekeeping tick (``retry_deferred_fts_recovery``)
        and at next startup.
        """
        self._raise_if_db_corrupt()
        rebuilt = 0
        with fts_rebuild_admission(self.db_path) as admitted:
            if not admitted:
                logger.warning(
                    "Deferred in-place FTS rebuild: another process holds the rebuild authority for this state.db.")
                return 0
            with self._lock:
                self._raise_if_db_replaced()
                if self._conn is None:
                    self._reopen_after_close_locked(context="write")
                for tbl in self._present_fts_tables():
                    try:
                        self._conn.execute(f"INSERT INTO {tbl}({tbl}) VALUES('rebuild')")
                        self._conn.commit()
                        rebuilt += 1
                    except sqlite3.DatabaseError as exc:  # SQLITE_CORRUPT is not an OperationalError (#133375)
                        self._conn.rollback()
                        if is_malformed_db_error(exc):
                            logger.error(
                                "FTS rebuild failed with a corruption-class error for %s: %s; "
                                "run 'hermes sessions repair' (or 'hermes doctor --fix') to rebuild the index offline",
                                tbl, exc)
                        else:
                            logger.warning("FTS rebuild failed for %s: %s", tbl, exc)
        return rebuilt

    def _merge_fts_incrementally(self, *, max_pages: int, max_commands: Optional[int] = None) -> int:
        """Run bounded FTS5 ``'merge'`` commands against each present index. A positive merge rank
        stops after ~that many output pages, so each command holds the write lock for
        milliseconds regardless of index size (``'optimize'`` takes 9-18 s per index on a
        10 GB DB). ``usermerge`` is lowered to its minimum of 2 (persisted in ``%_config``,
        once per instance) so a merge acts on ANY level with >= 2 segments; at the default
        4 a fragmented index cannot converge. Up to *max_commands* per index, stopping on
        the no-progress signal ``total_changes`` delta < 2 (the INSERT itself is 1). Each
        command is its own implicit transaction, so processes interleave mid-pass. Missing
        tables are skipped (optimize_fts_storage drops + backfills them live); other SQLite
        errors propagate. Returns commands executed."""
        _positive_int("max_pages", max_pages)
        if max_commands is None:
            max_commands = self._FTS_MERGE_COMMANDS_PER_PASS
        _positive_int("max_commands", max_commands)
        executed = 0
        with self._lock:
            for tbl in self._present_fts_tables():
                if not self._fts_usermerge_floor_applied:
                    self._conn.execute(f"INSERT INTO {tbl}({tbl}, rank) VALUES('usermerge', 2)")
                for _ in range(max_commands):
                    before = self._conn.total_changes
                    self._conn.execute(f"INSERT INTO {tbl}({tbl}, rank) VALUES('merge', ?)", (max_pages,))
                    executed += 1
                    if self._conn.total_changes - before < 2:
                        break
            self._fts_usermerge_floor_applied = True
        return executed