"""Full-text / trigram / CJK message search and FTS maintenance for SessionDB. Mixin contract: this is a plain mixin class consumed by ``hermes_state.SessionDB``. It defines no ``__init__`` and no state of its own; methods access the host's attributes (``self._conn``, ``self.db_path``, ``self._execute_write`` and other SessionDB methods) established by ``SessionDB.__init__``. It must never import hermes_state (cycle) — shared module-level constants live in hermes_state_common. """ import logging import json import os 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_common import ( FTS_CJK_STALE_KEY, FTS_SQL, FTS_STALE_KEY, FTS_STORAGE_VERSION, FTS_TOOL_CONTENT_PREFIX_CHARS, FTS_TOOL_FULL_CONTENT_HIGH_WATER_KEY, FTS_TRIGRAM_EXCLUDED_SOURCES, FTS_TRIGRAM_SQL, fts_trigram_session_sql, MAX_FTS5_QUERY_CHARS, SCHEMA_VERSION, _FTS_CJK_TRIGGERS, escape_like as _escape_like, fts_rebuild_admission, ) # Moved methods logged under the "hermes_state" logger before the split; # keep that logger identity so log filtering/capture behavior is unchanged. logger = logging.getLogger("hermes_state") # Characters FTS5's query grammar rejects outside a quoted phrase. Anything # missing from this set reaches MATCH raw and raises, which the execute site # swallows into zero results — the failure this strip step exists to prevent. # Assembled through re.escape so the backslash cannot be eaten as a regex # escape inside the class (it was, while the set was written as a literal). # # ``%`` is deliberately excluded: a CJK query falls back to a LIKE search that # needs it preserved as a literal (that path escapes wildcards itself), so # stripping it here widened those queries onto unrelated rows. _FTS5_SPECIAL_CHARS = '+{}():"^@/#&|~[]<>,;!?$=\\\'' _FTS5_SPECIAL_RE = re.compile(f"[{re.escape(_FTS5_SPECIAL_CHARS)}]") 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: """Run one bounded FTS5 merge pass without failing the completed write.""" if not self._fts_enabled: return try: self._merge_fts_incrementally( max_pages=self._FTS_MERGE_MAX_PAGES_PER_INDEX ) except Exception as exc: # noqa: BLE001 - post-commit maintenance # 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) def fts_rebuild_status(self) -> Optional[Dict[str, Any]]: """Return deferred-rebuild progress, or None when no rebuild pending. Shape: {"pending": True, "total": , "indexed": , "percent": <0-100 int>}. Consumed by search_messages() notes and by status surfaces (dashboard/desktop can poll this to render a progress indicator). Reads state_meta directly via _read_ctx instead of calling get_meta() (which takes self._lock) so search_messages doesn't block on the writer lock when checking rebuild status. """ with self._read_ctx() as conn: row = conn.execute( "SELECT key, value FROM state_meta WHERE key IN (?, ?)", ("fts_rebuild_high_water", "fts_rebuild_progress"), ).fetchall() meta = {r["key"]: r["value"] for r in row} high_water = meta.get("fts_rebuild_high_water") if high_water is None: return None progress = int(meta.get("fts_rebuild_progress") or 0) total = int(high_water) if total <= 0: return None pct = min(100, int(100 * progress / total)) return {"pending": True, "total": total, "indexed": progress, "percent": pct} def _fts_rebuild_finish(self) -> None: """Finalize the deferred rebuild: boundary sweep + clear markers. The sweep is cheap insurance against any write that slipped through the migration-boundary instant (between high_water capture and trigger activation): re-index any row near the boundary that the index is missing. docsize has one row per indexed doc, so the anti-join is exact and runs on a narrow id range. The trigram half of the sweep is gated on ``self._trigram_available`` for the same reason ``fts_rebuild_step()`` gates its backfill INSERT: when the SQLite build has no trigram tokenizer (or the table was never created), an unconditional INSERT raises ``no such table`` and aborts the whole rebuild — taking ``optimize_fts_storage()`` down with it. """ include_trigram = self._trigram_available def _do(conn): hw_row = conn.execute( "SELECT value FROM state_meta WHERE key = 'fts_rebuild_high_water'" ).fetchone() if hw_row is not None: hw = int(hw_row[0]) # Sweep a generous window around the boundary. lo, hi = hw - 1000, hw + 1000 conn.execute( "INSERT INTO messages_fts(rowid, content, tool_name, tool_calls) " "SELECT m.id, " "CASE WHEN m.role = 'tool' AND m.id > ? " "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)", (hw, FTS_TOOL_CONTENT_PREFIX_CHARS, lo, hi), ) if include_trigram: conn.execute( "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 " "WHERE m.id > ? AND m.id <= ? AND m.role <> 'tool' " f"AND {fts_trigram_session_sql('s')} " "AND NOT EXISTS (SELECT 1 FROM messages_fts_trigram_docsize d WHERE d.id = m.id)", (lo, hi), ) conn.execute( "DELETE FROM state_meta WHERE key IN " "('fts_rebuild_high_water', 'fts_rebuild_progress')" ) self._execute_write(_do) logger.info("Deferred FTS rebuild complete — all messages indexed.") def _fts_teardown_trash_step(self) -> bool: """Tear down one chunk of a demoted v22 FTS shadow table. The trash tables are PLAIN tables (their vtable parent was demoted away during the migration), so chunked DELETE + final DROP involve no FTS5 machinery at all. Returns True while teardown work remains. 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). Compound-key trash tables (multi-column PK) cannot use a scalar high-water comparison, so they keep the legacy chunked ``LIMIT`` delete — those shadow tables are small by construction. """ 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 ] pk_cols = [name for name, _typ in pk_info] key = ", ".join(pk_cols) if pk_cols else "rowid" if len(pk_cols) == 1 and (not pk_info or pk_info[0][1] == "INTEGER"): # High-water drain: delete only rows past the marker key. # The marker is read/written inside the same BEGIN IMMEDIATE # transaction as the DELETE, so concurrent callers claim # disjoint key ranges instead of re-deleting. Only integer # PKs can anchor a numeric high-water comparison — the FTS # config shadow table (TEXT pk like 'version') falls back to # the legacy chunked delete below. marker_key = f"fts_teardown_{tbl}_progress" row = conn.execute( "SELECT value FROM state_meta WHERE key = ?", (marker_key,), ).fetchone() high_water = int(row[0]) if row is not None else 0 # Claim the chunk's upper bound: the LAST row of the # LIMIT window, so a full chunk is deleted per step. upper_rows = conn.execute( f"SELECT {key} FROM {tbl} WHERE {key} > ? " f"ORDER BY {key} LIMIT {self._FTS_REBUILD_CHUNK_ROWS}", (high_water,), ).fetchall() if not upper_rows: # Drained — the DROP is cheap now. conn.execute(f"DROP TABLE IF EXISTS {tbl}") conn.execute( "DELETE FROM state_meta WHERE key = ?", (marker_key,) ) logger.info("Old FTS shadow table %s torn down.", tbl) return True upper = upper_rows[-1][0] cur = conn.execute( f"DELETE FROM {tbl} WHERE {key} > ? AND {key} <= ?", (high_water, upper), ) if cur.rowcount > 0: conn.execute( "INSERT INTO state_meta (key, value) VALUES (?, ?) " "ON CONFLICT(key) DO UPDATE SET value = excluded.value", (marker_key, str(upper)), ) 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 " f"(SELECT {key} FROM {tbl} LIMIT {self._FTS_REBUILD_CHUNK_ROWS})" ) if cur.rowcount == 0: # Empty — the DROP is cheap now. conn.execute(f"DROP TABLE IF EXISTS {tbl}") logger.info("Old FTS shadow table %s torn down.", tbl) return True # re-check: more trash tables / chunks may remain 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_rebuild_step(self) -> bool: """Backfill one chunk of the deferred FTS rebuild. Returns True when more work remains, False when the rebuild is complete (or none is pending). Safe to call from any process at any time; chunks are claimed atomically inside the write transaction, so concurrent callers interleave instead of duplicating rows. """ if not self._fts_enabled: return False high_water_raw = self.get_meta("fts_rebuild_high_water") if high_water_raw is None: return False high_water = int(high_water_raw) include_trigram = self._trigram_available chunk = self._FTS_REBUILD_CHUNK_ROWS def _do(conn): # Re-read progress inside the write transaction (BEGIN IMMEDIATE # is already held by _execute_write) — this is the claim: two # workers can't read the same progress value concurrently. row = conn.execute( "SELECT value FROM state_meta WHERE key = 'fts_rebuild_progress'" ).fetchone() if row is None: return False # finished (or cleared) by another process progress = int(row[0]) if progress >= high_water: return False # The chunk upper bound is an id, not a row count, so gaps from # deleted rows don't shrink chunks below the claimed range. upper = min(progress + chunk, high_water) conn.execute( "INSERT INTO messages_fts(rowid, content, tool_name, tool_calls) " "SELECT id, content, tool_name, tool_calls FROM messages " "WHERE id > ? AND id <= ?", (progress, upper), ) if include_trigram: conn.execute( "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 " "WHERE m.id > ? AND m.id <= ? AND m.role <> 'tool' " f"AND {fts_trigram_session_sql('s')}", (progress, upper), ) # Publish progress in the same transaction as the rows it # covers — crash-atomic: either both land or neither does. conn.execute( "UPDATE state_meta SET value = ? " "WHERE key = 'fts_rebuild_progress'", (str(upper),), ) return upper < high_water try: more = self._execute_write(_do) except sqlite3.OperationalError as exc: logger.debug("FTS rebuild chunk failed (will retry): %s", exc) return True # transient (lock contention) — caller retries if more is False: status = self.fts_rebuild_status() if high_water <= 0 or ( status is not None and status["indexed"] >= status["total"] ): self._fts_rebuild_finish() return False return bool(more) def fts_cjk_rebuild_status(self) -> Optional[Dict[str, Any]]: """CJK-index backfill progress, or None when none is pending.""" with self._read_ctx() as conn: row = conn.execute( "SELECT key, value FROM state_meta WHERE key IN (?, ?)", ("fts_cjk_rebuild_high_water", "fts_cjk_rebuild_progress"), ).fetchall() meta = {r["key"]: r["value"] for r in row} high_water = meta.get("fts_cjk_rebuild_high_water") if high_water is None: return None progress = int(meta.get("fts_cjk_rebuild_progress") or 0) total = int(high_water) if total <= 0: return None pct = min(100, int(100 * progress / total)) return {"pending": True, "total": total, "indexed": progress, "percent": pct} 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 high_water_raw = self.get_meta("fts_cjk_rebuild_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): row = conn.execute( "SELECT value FROM state_meta " "WHERE key = 'fts_cjk_rebuild_progress'" ).fetchone() if row is None: return False # finished (or cleared) by another process progress = int(row[0]) if progress >= high_water: return False upper = min(progress + chunk, high_water) conn.execute( "INSERT INTO messages_fts_cjk(rowid, content, tool_name, tool_calls) " "SELECT id, content, tool_name, tool_calls FROM messages " "WHERE id > ? AND id <= ? AND role <> 'tool'", (progress, upper), ) conn.execute( "UPDATE state_meta SET value = ? " "WHERE key = 'fts_cjk_rebuild_progress'", (str(upper),), ) return upper < high_water try: more = self._execute_write(_do) except sqlite3.OperationalError as exc: logger.debug("CJK FTS rebuild chunk failed (will retry): %s", exc) return True if more is False: status = self.fts_cjk_rebuild_status() if status is not None and status["indexed"] >= status["total"]: self._fts_cjk_rebuild_finish() return False return bool(more) def _fts_cjk_rebuild_finish(self) -> None: """Boundary sweep + clear the cjk markers; index becomes servable.""" def _do(conn): hw_row = conn.execute( "SELECT value FROM state_meta " "WHERE key = 'fts_cjk_rebuild_high_water'" ).fetchone() if hw_row is not None: hw = int(hw_row[0]) lo, hi = hw - 1000, hw + 1000 conn.execute( "INSERT INTO messages_fts_cjk(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 <= ? AND m.role <> 'tool' " "AND NOT EXISTS (SELECT 1 FROM messages_fts_cjk_docsize d WHERE d.id = m.id)", (lo, hi), ) conn.execute( "DELETE FROM state_meta WHERE key IN " "('fts_cjk_rebuild_high_water', 'fts_cjk_rebuild_progress')" ) self._execute_write(_do) self._fts_cjk_available = True logger.info("CJK FTS index backfill complete — serving CJK search.") def _fts_cjk_reset_if_stale(self) -> None: """Rebuild path for a stale cjk index (triggers were dropped). The gap's extent is unknown, so the only safe recovery is a from- scratch rebuild: drop the table + triggers, clear the breadcrumb, recreate via ``_ensure_fts_cjk_schema`` (which sets fresh backfill markers on a populated DB). Called from ``optimize_fts_storage`` on a tokenizer-capable host; no-op when not stale. """ if not self._fts_cjk_loaded: return def _do(conn): stale = conn.execute( "SELECT 1 FROM state_meta WHERE key = ?", (FTS_CJK_STALE_KEY,), ).fetchone() if not stale: 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") conn.execute( "DELETE FROM state_meta WHERE key IN " f"('{FTS_CJK_STALE_KEY}', 'fts_cjk_rebuild_high_water', " "'fts_cjk_rebuild_progress')" ) return True was_stale = self._execute_write(_do) if was_stale: # Recreate outside the write transaction — _ensure_fts_cjk_schema # uses executescript(), which implicitly commits any pending # transaction and must not run inside _execute_write's BEGIN # IMMEDIATE. Sets fresh backfill markers on a populated DB. 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 exists but indexes nothing while ``messages`` has rows. Caller must hold ``self._lock``. This is the post-demote empty-index shape: external-content FTS with zero ``messages_fts_docsize`` rows against a non-empty messages table. Healthy installs (and mid-backfill installs that still hold markers) never match. """ try: has_msg = conn.execute( "SELECT EXISTS(SELECT 1 FROM messages)" ).fetchone()[0] if not has_msg: return False # docsize is the authoritative "is this rowid indexed" surface for # external-content FTS5; probing the virtual table itself is # not reliable across SQLite builds. EXISTS instead of COUNT(*): # this runs on every writable open via the _init_schema stamp # condition, and COUNT(*) is a full b-tree scan (~100ms on a # 2M-row table) while EXISTS is O(1). has_fts = conn.execute( "SELECT EXISTS(SELECT 1 FROM messages_fts_docsize)" ).fetchone()[0] return not has_fts except sqlite3.OperationalError: # Table absent / FTS disabled mid-init — not this failure class. return False def _fts_index_known_empty(self, conn) -> bool: """True when the base external-content index holds no rows. A missing table counts as empty: the schema ensure that follows creates it fresh. """ try: n = conn.execute( "SELECT COUNT(*) FROM messages_fts_docsize" ).fetchone()[0] return int(n) == 0 except sqlite3.OperationalError: return True def _reset_fts_index_to_empty(self, conn) -> None: """Delete every indexed row from the v23 external-content tables. Uses the FTS5 ``'delete-all'`` special command — the documented O(1) truncate for external-content tables. A plain no-WHERE ``DELETE`` is O(rows) on external-content FTS5 (each row's delete tokens are regenerated from the content table; measured ~12µs/row, minutes on a large index, while holding the write lock) and corrupts the index if indexed rows have diverged from ``messages`` — precisely the broken- bookkeeping shape this repair path handles. The backfill chunk worker replays its whole selected id range with no anti-join, so a replay from zero is only safe once the index is known empty — this is how a partially indexed DB gets there. """ for tbl in ("messages_fts", "messages_fts_trigram"): try: conn.execute(f"INSERT INTO {tbl}({tbl}) VALUES('delete-all')") except sqlite3.OperationalError: pass # table absent — already an empty surface 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. When ``force`` is False and high_water is already set, only repairs a missing progress key (stuck no-op when high_water exists alone), and only after the index is known empty: the chunk worker replays its whole selected id range without an anti-join, so a partially indexed DB is first reset to a known-empty surface rather than rebuilt from zero on top of surviving rows. Caller must hold the write transaction / lock as appropriate. """ existing_hw = conn.execute( "SELECT value FROM state_meta WHERE key = 'fts_rebuild_high_water'" ).fetchone() if existing_hw is not None and not force: hw = int(existing_hw[0]) progress = conn.execute( "SELECT value FROM state_meta WHERE key = 'fts_rebuild_progress'" ).fetchone() if progress is None: # high_water without progress: fts_rebuild_step treats missing # progress as "done by another process" and optimize would # no-op then stamp. Re-seed progress so the chunk loop runs. if not self._fts_index_known_empty(conn): self._reset_fts_index_to_empty(conn) conn.execute( "INSERT INTO state_meta (key, value) VALUES " "('fts_rebuild_progress', '0') " "ON CONFLICT(key) DO UPDATE SET value = excluded.value" ) conn.execute( "INSERT INTO state_meta (key, value) VALUES (?, ?) " "ON CONFLICT(key) DO UPDATE SET value = excluded.value", (FTS_TOOL_FULL_CONTENT_HIGH_WATER_KEY, str(hw)), ) return hw hw = conn.execute( "SELECT COALESCE(MAX(id), 0) FROM messages" ).fetchone()[0] for k, v in ( ("fts_rebuild_high_water", str(hw)), ("fts_rebuild_progress", "0"), (FTS_TOOL_FULL_CONTENT_HIGH_WATER_KEY, str(hw)), ): conn.execute( "INSERT INTO state_meta (key, value) VALUES (?, ?) " "ON CONFLICT(key) DO UPDATE SET value = excluded.value", (k, v), ) return int(hw) def _repair_optimize_bookkeeping(self) -> None: """Heal interrupted demote/backfill bookkeeping before optimize runs. Covers two post-#65798 failure classes: 1. Empty external-content index with messages present and no rebuild markers (demote crash window after empty v23 tables landed but before markers, or settle that stamped without backfill). Seed a full backfill. 2. ``fts_rebuild_high_water`` present without ``fts_rebuild_progress`` (partial meta) — seed progress so the chunk loop is not a no-op, resetting a partially populated index to a known-empty surface first so the anti-join-free chunk replay cannot duplicate rows. Must not invent markers on a still-legacy inline DB: that would make ``optimize_fts_storage`` skip demote (``legacy and not pending``) and attempt v23-shaped INSERTs against the inline table forever. """ def _do(conn): existing_hw = conn.execute( "SELECT value FROM state_meta " "WHERE key = 'fts_rebuild_high_water'" ).fetchone() if existing_hw is not None: # Repair orphan high_water-without-progress only. Never # invent a fresh claim on a healthy complete index. progress = conn.execute( "SELECT 1 FROM state_meta " "WHERE key = 'fts_rebuild_progress'" ).fetchone() if progress is None: if not self._fts_index_known_empty(conn): self._reset_fts_index_to_empty(conn) conn.execute( "INSERT INTO state_meta (key, value) VALUES " "('fts_rebuild_progress', '0') " "ON CONFLICT(key) DO UPDATE SET value = '0'" ) return # No markers. On a still-legacy DB demote owns marker creation. if self._db_has_legacy_inline_fts(conn): return # Non-legacy empty external index (demote crash window / premature # stamp): seed a full backfill claim. if self._fts_external_index_empty_with_messages(conn): conn.execute( "DELETE FROM state_meta WHERE key = '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 to do: either this DB is a legacy inline-FTS install that can be optimized to the v23 external-content schema, or a previous optimize run was interrupted (legacy vtables already demoted, but backfill markers and/or trash tables remain) and re-running would resume it, or this DB is v23 with the old tool-calls-inclusive trigram projection (repairable via this same migration flow), or the CJK-bigram index needs a backfill/rebuild on this tokenizer-capable host, or a prior demote left an empty external-content index without markers (healable on re-run). False for fresh and fully-optimized installs (and when FTS5 is unavailable).""" if not self._fts_enabled or self.read_only: return False with self._read_ctx() as conn: if self._db_has_legacy_inline_fts(conn): return True if self._db_has_trigram_tool_calls_projection(self._conn): return True # Interrupted optimize: demotion already removed the legacy # vtables (so the check above is False), but the transition is # unfinished until the backfill markers are cleared and the # demoted trash tables are torn down. Search stays complete # through the gap supplement meanwhile; re-running resumes. if conn.execute( "SELECT 1 FROM state_meta " "WHERE key = 'fts_rebuild_high_water' LIMIT 1" ).fetchone(): return True # CJK-bigram index work — only offerable when THIS process can # tokenize: a pending backfill (markers set at creation on a # populated DB) or a stale index awaiting a from-scratch rebuild. if self._fts_cjk_loaded and conn.execute( "SELECT 1 FROM state_meta WHERE key IN " f"('fts_cjk_rebuild_high_water', '{FTS_CJK_STALE_KEY}') LIMIT 1" ).fetchone(): return True if self._has_fts_trash(conn): return True # Pre-fix crash window: empty external-content index with # messages still present, no markers, no trash (teardown already # finished or never needed). Re-run seeds markers and backfills. return 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 to the chunked teardown, exactly as the validated auto path did. Markers are written in the same BEGIN IMMEDIATE as the demote, *before* the empty v23 schema is created. Schema creation uses ``executescript`` and therefore cannot run inside that transaction (it issues an implicit COMMIT — see the CJK recreate path). Creating the empty schema only after markers are durable closes 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") had = bool(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()) if had: 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") shadows = [ r[0] for r in conn.execute( "SELECT name FROM sqlite_master WHERE type = 'table' " "AND (name LIKE 'messages_fts_%' ESCAPE '\\' " "OR name LIKE 'messages_fts_trigram_%' ESCAPE '\\')" ).fetchall() ] for sh in shadows: conn.execute(f"ALTER TABLE {sh} RENAME TO fts_v22_trash_{sh}") # Claim the backfill *before* empty v23 tables exist. A crash # between this commit and schema ensure still leaves markers, so # optimize-storage resumes instead of tearing down trash and # stamping an empty index as complete. hw = self._seed_fts_rebuild_markers(conn, force=True) conn.execute( "DELETE FROM state_meta WHERE key = 'fts_optimize_available'" ) return hw hw = int(self._execute_write(_stage)) # Create the empty v23 schema outside the write transaction — # ``_ensure_fts_schema`` uses executescript(), which implicitly # commits any pending transaction and must not run inside # ``_execute_write``'s BEGIN IMMEDIATE (same rule as the CJK recreate # path above). Markers are already durable. 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( "failed to create v23 messages_fts during optimize-storage demote" ) self._conn.commit() return hw 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-compatible shape, foreground and to completion. Supports two paths: - legacy-v22 inline -> demote to v23 external-content - v23 installs where ``messages_fts_trigram`` still stores ``tool_calls`` payloads Safe to re-run: if a previous attempt was interrupted it resumes from the progress marker. ``progress_cb`` receives {"phase", "percent", "indexed", "total"} dicts for a CLI progress bar. Returns a summary dict. The trigram tokenizer being unavailable is not fatal — the base index is still rebuilt (CJK falls back to LIKE), mirroring normal startup. """ if not self._fts_enabled: return {"ok": False, "reason": "fts5_unavailable"} if self.read_only: return {"ok": False, "reason": "read_only"} # Heal empty-index / orphan-marker bookkeeping from an interrupted # demote *before* deciding whether to demote again. This re-seeds # markers when trash was already staged (or torn down) without a # backfill claim so the phases below actually run. self._repair_optimize_bookkeeping() # Only demote if we're actually still on the legacy shape. If a prior # run already demoted (markers/trash present), skip straight to # finishing the backfill + teardown — this is what makes re-running # after an interruption safe. 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: markers exist, empty v23 tables may still be # missing if the process died between the staged demote commit and # schema ensure. Re-ensure is IF NOT EXISTS and cheap. 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: # Fail fast: without the base table the backfill loop # below would retry "no such table" errors forever. raise sqlite3.OperationalError( "failed to re-create v23 messages_fts " "on optimize-storage resume" ) self._conn.commit() # A stale CJK index (triggers dropped by a tokenizer-less process) # can only be recovered from scratch — reset it now so the cjk # backfill phase below rebuilds it. No-op without the tokenizer. self._fts_cjk_reset_if_stale() # An optimized v23 DB gaining the cjk index for the first time (no # legacy work left, tokenizer newly installed): ensure the table + # markers exist so the backfill phase has work to claim. if self._fts_cjk_loaded: with self._lock: self._ensure_fts_cjk_schema(self._conn) self._conn.commit() def _emit(phase: str) -> None: if progress_cb is None: return st = self.fts_rebuild_status() if st is None: st = 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 _pause(chunk_seconds: float) -> None: """Inter-chunk throttle (see the chunk-engine note above). The chunk methods themselves never sleep, so this loop is the single place the duty cycle is enforced: without it, back-to-back BEGIN IMMEDIATE chunks starve any live gateway/CLI process sharing the DB out of its lock retries (the measured ~85% write-lock ownership that froze concurrent sessions). """ time.sleep(max( self._FTS_REBUILD_MIN_PAUSE, chunk_seconds * self._FTS_REBUILD_DUTY_FACTOR, )) # Phase 1: backfill (foreground, throttled between chunks so a live # gateway sharing the DB stays responsive). _emit("backfill") while True: _t0 = time.monotonic() if not self.fts_rebuild_step(): break _emit("backfill") _pause(time.monotonic() - _t0) _emit("backfill") # Phase 1b: backfill the CJK-bigram index (its own marker pair; a # no-op when the tokenizer isn't loadable or nothing is pending). while True: _t0 = time.monotonic() if not self.fts_cjk_rebuild_step(): break _emit("backfill") _pause(time.monotonic() - _t0) # Phase 2: tear down the demoted legacy shadow tables in chunks. _emit("teardown") while True: _t0 = time.monotonic() if not self._fts_teardown_trash_step(): break _emit("teardown") _pause(time.monotonic() - _t0) # Refuse to stamp "optimized" while work remains or the base index is # still empty against a non-empty messages table. Pre-fix code could # tear down trash and settle after a no-op backfill when markers were # missing — permanent search-index loss for historical rows. with self._read_ctx() as conn: still_pending = conn.execute( "SELECT 1 FROM state_meta " "WHERE key = 'fts_rebuild_high_water' LIMIT 1" ).fetchone() 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} # Phase 3: reclaim freed pages to the OS. vacuum_ok = None if vacuum: _emit("vacuum") try: with self._lock: self._conn.execute("VACUUM") vacuum_ok = True except sqlite3.OperationalError as exc: # Most common cause: not enough free disk for VACUUM's temp # copy. The optimization still succeeded; space just isn't # reclaimed until a later VACUUM. Non-fatal. logger.warning("VACUUM after FTS optimize failed: %s", exc) vacuum_ok = False # Best-effort: fold the WAL back into the main file so the on-disk # size settles now rather than at close(). NOTE this is REFUSED # (SQLITE_BUSY) while any other connection holds a WAL read-mark — # e.g. a live gateway sharing the DB — so it is not sufficient on # its own. Callers must therefore NOT size the result by stat()ing # the file; use :meth:`logical_size_bytes`, which is truthful # immediately regardless of readers. # PASSIVE, not TRUNCATE: optimize-storage runs from a transient CLI # process; a TRUNCATE reset here would race a live gateway writer # and tear B-tree pages (#45383). (The TRUNCATE was already refused # SQLITE_BUSY while the gateway holds a read-mark, per the note # above; PASSIVE removes the reset attempt entirely.) try: with self._lock: self._conn.execute("PRAGMA wal_checkpoint(PASSIVE)") except Exception as exc: logger.debug( "WAL checkpoint (PASSIVE) after optimize VACUUM failed: %s", exc, ) # Phase 4: stamp the FTS storage layout as current, clear the "available" # flag, and advance schema_version if it was somehow still behind (the # main version normally advances on open now, but bump defensively so a # DB opened only by pre-decoupling code still settles). The FTS-layout # marker is the source of truth for "is this DB optimized". def _settle(conn): # Re-check inside the write transaction so a concurrent writer # cannot race a stamp past incomplete work. Returns a refusal # reason (stamping nothing) or None once the stamp is written. if conn.execute( "SELECT 1 FROM state_meta " "WHERE key = 'fts_rebuild_high_water' LIMIT 1" ).fetchone() 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" conn.execute( "INSERT INTO state_meta (key, value) VALUES ('fts_storage_version', ?) " "ON CONFLICT(key) DO UPDATE SET value = excluded.value", (str(FTS_STORAGE_VERSION),), ) conn.execute("DELETE FROM state_meta WHERE key = 'fts_optimize_available'") conn.execute( "UPDATE schema_version SET version = ? WHERE version < ?", (SCHEMA_VERSION, SCHEMA_VERSION), ) return None refusal = self._execute_write(_settle) if refusal is not None: # A concurrent process re-seeded markers, left trash, or emptied # the index between the pre-vacuum check above and this write # transaction. Nothing was stamped. Report the failure instead of # crashing the CLI with a traceback; 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} 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]: """Return an anchored window plus session bookends. Built on top of ``get_messages_around``. Three slices: - ``window``: messages immediately surrounding the anchor. Filtered to ``keep_roles`` (tool-response noise dropped by default), EXCEPT the anchor itself is always preserved regardless of role. - ``bookend_start``: first ``bookend`` user/assistant messages of the session — but only those whose id is strictly before the window's first message id. Empty when the window already overlaps the session head. Empty-content messages (tool-call-only assistant turns) are skipped so they don't crowd out actual prose openings. - ``bookend_end``: last ``bookend`` user/assistant messages of the session, same non-overlap rule at the tail. Bookends let an FTS5 hit anywhere in a long session yield the goal (opening) and the resolution (closing) on a single call — without loading the whole transcript. Returns ``{"window": [], "messages_before": 0, "messages_after": 0, "bookend_start": [], "bookend_end": []}`` when the anchor isn't in the session. ``keep_roles=None`` disables role filtering (raw window + raw bookends). """ if bookend < 0: bookend = 0 # Reuse the primitive — handles anchor-existence, content decoding, # tool_calls deserialisation, and boundary counts. 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": [], } # Apply role filter to the window, but never drop the anchor itself. 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 ] else: filtered_window = window_rows window_min_id = window_rows[0]["id"] window_max_id = window_rows[-1]["id"] # Fetch bookends only when there's room outside the window. SQL filters # by id range, role, and non-empty content — tool-call-only assistant # turns (content='' with tool_calls populated) are excluded so they # don't crowd out actual prose openings/closings. bookend_start_rows: List[Any] = [] bookend_end_rows: List[Any] = [] if bookend > 0: with self._read_ctx() as conn: role_clause = "" role_params: list = [] if keep_roles is not None: role_placeholders = ",".join("?" for _ in keep_roles) role_clause = f" AND role IN ({role_placeholders})" role_params = list(keep_roles) bookend_start_rows = conn.execute( f"SELECT * FROM messages " f"WHERE session_id = ? AND id < ?{role_clause} " f"AND length(content) > 0 " f"ORDER BY id ASC LIMIT ?", (session_id, window_min_id, *role_params, bookend), ).fetchall() bookend_end_rows = conn.execute( f"SELECT * FROM messages " f"WHERE session_id = ? AND id > ?{role_clause} " f"AND length(content) > 0 " f"ORDER BY id DESC LIMIT ?", (session_id, window_max_id, *role_params, bookend), ).fetchall() # End rows came back DESC for the LIMIT cap; flip to ASC. bookend_end_rows = list(reversed(bookend_end_rows)) def _hydrate(row) -> Dict[str, Any]: msg = dict(row) if "content" in msg: msg["content"] = self._decode_content(msg["content"]) if msg.get("tool_calls"): try: msg["tool_calls"] = json.loads(msg["tool_calls"]) except (json.JSONDecodeError, TypeError): logger.warning( "Failed to deserialize tool_calls in get_anchored_view, falling back to []" ) msg["tool_calls"] = [] if msg.get("display_metadata") is not None: msg["display_metadata"] = self._decode_display_metadata(msg["display_metadata"]) return msg 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]]: """Return the *limit* most-recent user messages, newest first. Each entry is a dict with keys ``id``, ``timestamp``, ``preview``. ``preview`` is the first 80 characters of the message content (with line breaks collapsed to spaces). Used by the /rewind slash command picker, CLI/TUI/gateway ``/undo [N]``, and any other caller that needs real user-turn targets. Bookkeeping timeline rows (``display_kind`` set — e.g. model_switch, async_delegation_complete, auto_continue, hidden) are excluded. They are durable ``role='user'`` rows for the API transcript, but no client counts them as user turns (desktop demotes them to system / drops them; the CLI already uses ``not m.get("display_kind")``). Including them here made ``/undo`` soft-delete from a marker instead of the last real turn — same class of index skew as the prompt.submit ordinal bug. By default only active messages are returned. """ active_clause = "" if include_inactive else " AND active = 1" # Match CLI/desktop: only real user turns, not timeline bookkeeping. display_clause = " AND (display_kind IS NULL OR display_kind = '')" # Legacy standalone compaction handoffs (persisted pre-#80622) are # durable role='user' rows with NO display_kind — SQL can't see them, # so fetch with headroom and drop them in the decode loop below. # Without this, /undo N and rewind pair an in-memory count that # excludes handoffs with a DB pick that includes them, soft-deleting # the wrong turn. fetch_limit = int(limit) * 2 + 5 with self._read_ctx() as conn: cursor = 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, fetch_limit), ) rows = cursor.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): # Compaction handoff — never a user-originated turn (#80622). continue if isinstance(decoded, list): # Multimodal — flatten text parts. text_parts = [ p.get("text", "") for p in decoded if isinstance(p, dict) and p.get("type") == "text" ] preview = " ".join(t for t in text_parts if t).strip() if not preview: preview = "[multimodal content]" elif isinstance(decoded, str): # A /skill turn embeds the whole skill body; show what the user # typed instead of the skill's opening prose. preview = describe_skill_invocation(decoded) or decoded else: preview = "" preview = " ".join(preview.split()) # collapse whitespace if len(preview) > 80: preview = preview[:77] + "..." result.append( { "id": row["id"], "timestamp": row["timestamp"], "preview": preview, } ) return result @staticmethod def _sanitize_fts5_query(query: str) -> str: """Sanitize user input for safe use in FTS5 MATCH queries. FTS5 has its own query syntax where characters like ``"``, ``(``, ``)``, ``+``, ``*``, ``{``, ``}``, the column-filter operator ``:`` and bare boolean operators (``AND``, ``OR``, ``NOT``) have special meaning. Passing raw user input directly to MATCH can cause ``sqlite3.OperationalError``. Strategy: - Preserve properly paired quoted phrases (``"exact phrase"``) - Strip unmatched FTS5-special characters that would cause errors - Wrap unquoted hyphenated and dotted terms in quotes so FTS5 matches them as exact phrases instead of splitting on the hyphen/dot (e.g. ``chat-send``, ``P2.2``, ``my-app.config.ts``) """ # Cap user-controlled FTS input before any regex processing. Search # queries do not need to be arbitrarily large, and bounding them keeps # sanitizer/runtime behavior predictable under adversarial input. query = query[:MAX_FTS5_QUERY_CHARS] # Step 1: Extract balanced double-quoted phrases and protect them # from further processing via numbered placeholders. Do this with a # single linear scan rather than a regex so pathological quote runs # cannot induce backtracking. _quoted_parts: list = [] pieces: list[str] = [] i = 0 while i < len(query): ch = query[i] if ch != '"': pieces.append(ch) i += 1 continue end = query.find('"', i + 1) if end == -1: # Unmatched quote: replace with whitespace like the old # sanitizer's special-char stripping step. pieces.append(" ") i += 1 continue _quoted_parts.append(query[i:end + 1]) pieces.append(f"\x00Q{len(_quoted_parts) - 1}\x00") i = end + 1 sanitized = "".join(pieces) # Step 2: Strip remaining (unmatched) FTS5-special characters. ``:`` is # FTS5's column-filter operator (``col:term``); since the FTS table has a # single ``content`` column, an unquoted colon query like ``TODO: fix`` # parses as ``column:term`` and raises "no such column" — swallowed at # the execute site into zero results. Strip it like the others. # The class below is every character FTS5's query grammar rejects # outside a quoted phrase. Anything omitted here reaches MATCH raw and # raises, which the execute site swallows into zero results — the # failure mode this step exists to prevent. Measured against a real # FTS5 table: ``it's``, ``gateway/run.py``, ``user@host``, ``a,b`` and # ``50%`` all raised before the class was completed. sanitized = _FTS5_SPECIAL_RE.sub(" ", sanitized) # Step 2b: ``%`` is excluded from the class above only to protect the # CJK LIKE-fallback path (LIKE treats % as a wildcard the fallback # builds itself). A non-CJK query never reaches that fallback # (``is_cjk`` gates it), so ``50%`` would sail into MATCH raw and # raise like the rest. Strip it whenever the query has no CJK. if "%" in sanitized and not SessionSearchMixin._contains_cjk(sanitized): sanitized = sanitized.replace("%", " ") # Step 3: Collapse repeated * (e.g. "***") into a single one, # and remove leading * (prefix-only needs at least one char before *) sanitized = re.sub(r"\*+", "*", sanitized) sanitized = re.sub(r"(^|\s)\*", r"\1", sanitized) # Step 4: Remove dangling boolean operators at start/end that would # cause syntax errors (e.g. "hello AND" or "OR world") sanitized = re.sub(r"(?i)^(AND|OR|NOT)\b\s*", "", sanitized.strip()) sanitized = re.sub(r"(?i)\s+(AND|OR|NOT)\s*$", "", sanitized.strip()) # Step 5: Wrap unquoted dotted and/or hyphenated terms in double # quotes. FTS5's tokenizer splits on dots and hyphens, turning # ``chat-send`` into ``chat AND send`` and ``P2.2`` into ``p2 AND 2``. # Quoting preserves phrase semantics. A single pass avoids the # double-quoting bug that would occur if dotted, hyphenated and underscored # patterns were applied sequentially (e.g. ``my-app.config``). sanitized = re.sub(r"\b(\w+(?:[._-]\w+)+)\b", r'"\1"', sanitized) # Step 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 _is_cjk_codepoint(cp: int) -> bool: return (0x4E00 <= cp <= 0x9FFF or # CJK Unified Ideographs 0x3400 <= cp <= 0x4DBF or # CJK Extension A 0x20000 <= cp <= 0x2A6DF or # CJK Extension B 0x3000 <= cp <= 0x303F or # CJK Symbols 0x3040 <= cp <= 0x309F or # Hiragana 0x30A0 <= cp <= 0x30FF or # Katakana 0xAC00 <= cp <= 0xD7AF) # Hangul Syllables @staticmethod def _contains_cjk(text: str) -> bool: """Check if text contains CJK (Chinese, Japanese, Korean) characters.""" for ch in text: cp = ord(ch) if (0x4E00 <= cp <= 0x9FFF or # CJK Unified Ideographs 0x3400 <= cp <= 0x4DBF or # CJK Extension A 0x20000 <= cp <= 0x2A6DF or # CJK Extension B 0x3000 <= cp <= 0x303F or # CJK Symbols 0x3040 <= cp <= 0x309F or # Hiragana 0x30A0 <= cp <= 0x30FF or # Katakana 0xAC00 <= cp <= 0xD7AF): # Hangul Syllables return True return False @classmethod def _count_cjk(cls, text: str) -> int: """Count CJK characters in text.""" return sum(1 for ch in text if cls._is_cjk_codepoint(ord(ch))) @classmethod def _has_lone_cjk_run(cls, query: str) -> bool: """True when any maximal CJK run in the query is a single char. The cjk-bigram index stores bigrams for runs >=2 chars and unigrams only for isolated chars, so a 1-char CJK term can't match inside longer runs there — those queries keep the LIKE substring route. """ run = 0 for ch in query: if cls._is_cjk_codepoint(ord(ch)): run += 1 else: if run == 1: return True run = 0 return run == 1 @staticmethod def _trigram_eligible_tokens(query: str) -> bool: """True when every non-operator token is long enough for the trigram tokenizer to match (>=3 chars). The trigram tokenizer indexes overlapping 3-character sequences, so a token shorter than 3 chars produces no trigrams and can never match. With FTS5's implicit-AND between tokens, a single short token makes the whole MATCH return nothing, so the trigram path is only worth taking when every searchable token qualifies. """ tokens = [ t for t in query.strip('"').strip().split() if t.upper() not in {"AND", "OR", "NOT"} ] return bool(tokens) and all(len(t) >= 3 for t in tokens) def _run_trigram_search( self, raw_query: str, *, table: str = "messages_fts_trigram", order_by_sql: str, include_inactive: bool, source_filter: List[str] = None, exclude_sources: List[str] = None, role_filter: List[str] = None, limit: int = 20, offset: int = 0, ) -> Optional[List[Dict[str, Any]]]: """Run a search against a substring-capable FTS index. ``table`` is ``messages_fts_trigram`` (default) or ``messages_fts_cjk``. The trigram tokenizer indexes overlapping 3-byte sequences, so it matches substrings regardless of word boundaries — both CJK phrases the unicode61 tokenizer splits into single characters and Latin runs the unicode61 tokenizer fuses onto adjacent CJK (e.g. ``修改youer服务端``). The cjk-bigram tokenizer splits Latin runs off adjacent CJK, giving the same recovery as an exact ranked token match. Each non-operator token is quoted to neutralise FTS5 special characters while boolean operators (AND/OR/NOT) are preserved. Returns the matching rows, or ``None`` when the query cannot be executed (e.g. the tokenizer is unavailable at runtime) so the caller can fall back to another strategy. """ tokens = raw_query.split() parts = [] for tok in tokens: if tok.upper() in {"AND", "OR", "NOT"}: parts.append(tok) else: parts.append('"' + tok.replace('"', '""') + '"') trigram_query = " ".join(parts) tri_where = [f"{table} MATCH ?"] tri_params: list = [trigram_query] if not include_inactive: tri_where.append("(m.active = 1 OR m.compacted = 1)") if source_filter is not None: tri_where.append(f"s.source IN ({','.join('?' for _ in source_filter)})") tri_params.extend(source_filter) if exclude_sources is not None: tri_where.append(f"s.source NOT IN ({','.join('?' for _ in exclude_sources)})") tri_params.extend(exclude_sources) if role_filter: tri_where.append(f"m.role IN ({','.join('?' for _ in role_filter)})") tri_params.extend(role_filter) tri_sql = f""" SELECT m.id, m.session_id, m.role, snippet({table}, -1, '>>>', '<<<', '...', 40) AS snippet, m.timestamp, m.tool_name, s.source, s.model, s.started_at AS session_started FROM {table} JOIN messages m ON m.id = {table}.rowid JOIN sessions s ON s.id = m.session_id WHERE {' AND '.join(tri_where)} {order_by_sql} LIMIT ? OFFSET ? """ tri_params.extend([limit, offset]) with self._read_ctx() as conn: try: tri_cursor = conn.execute(tri_sql, tri_params) except sqlite3.OperationalError: # Query failed at runtime — let the caller fall back. return None return [dict(row) for row in tri_cursor.fetchall()] def search_messages( self, query: str, source_filter: List[str] = None, exclude_sources: List[str] = None, role_filter: List[str] = None, limit: int = 20, offset: int = 0, sort: str = None, include_inactive: bool = False, fields: Optional[Collection[str]] = None, ) -> List[Dict[str, Any]]: """Instrumented wrapper around :meth:`_search_messages_impl`. Logs one line per slow search with the routing path taken, so production latency stays attributable per query shape (the 2026-07 session_search investigation needed trace archaeology to discover the LIKE full scans; this makes the next regression a grep). 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, ) return rows finally: try: threshold = float(os.getenv("HERMES_SEARCH_SLOW_MS", "1000")) except (TypeError, ValueError): threshold = 1000.0 elapsed_ms = (time.time() - started) * 1000.0 if elapsed_ms >= threshold: 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 _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" tokens = [ t for t in raw.split() if t.upper() not in {"AND", "OR", "NOT"} and self._contains_cjk(t) ] short = any(self._count_cjk(t) < 3 for t in tokens) if self._count_cjk(raw) >= 3 and not short and self._trigram_available: return "trigram" return "like_scan" except Exception: return "unknown" @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 by default, matching FTS5's implicit conjunction. ``NOT`` negates the following term inside that group instead of being discarded, so ``python NOT java`` becomes a positive Python match plus a Java exclusion. """ groups: List[List[Tuple[str, bool]]] = [[]] negate_next = False for raw_token in re.findall(r'"[^"]+"|\S+', 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: escaped = ( term.replace("\\", "\\\\") .replace("%", "\\%") .replace("_", "\\_") ) clause = ( "(COALESCE(m.content, '') LIKE ? ESCAPE '\\' OR " "COALESCE(m.tool_name, '') LIKE ? ESCAPE '\\' OR " "COALESCE(m.tool_calls, '') LIKE ? ESCAPE '\\')" ) clauses.append(f"NOT {clause}" if negated else clause) params.extend([f"%{escaped}%"] * 3) 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, *, source_filter: Optional[List[str]], exclude_sources: Optional[List[str]], role_filter: Optional[List[str]], limit: int, offset: int, sort: Optional[str], include_inactive: bool, ) -> 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})"] if not include_inactive: where.append("(m.active = 1 OR m.compacted = 1)") 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) order = ( "ASC" if isinstance(sort, str) and sort.strip().lower() == "oldest" else "DESC" ) sql = f""" SELECT m.id, m.session_id, m.role, substr(m.content, max(1, instr(m.content, ?) - 40), 120) AS snippet, m.timestamp, m.tool_name, s.source, s.model, s.started_at AS session_started FROM messages m JOIN sessions s ON s.id = m.session_id WHERE {' AND '.join(where)} ORDER BY m.timestamp {order}, m.id {order} LIMIT ? OFFSET ? """ with self._read_ctx() as conn: rows = conn.execute( sql, [snippet_term, *params, limit, offset] ).fetchall() return [dict(row) for row in rows] 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: with self._read_ctx() as conn: stale = conn.execute( "SELECT 1 FROM state_meta WHERE key = ? LIMIT 1", (FTS_STALE_KEY,), ).fetchone() except sqlite3.Error: return if stale is not None: self._fts_stale = True self._fts_enabled = False self._trigram_available = False 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 and trim full content from results. Context (1 message before + after each match) is only loaded when the selected result projection consumes it. Each query takes its own fresh read transaction via _read_ctx, so we never hold a lock across N sequential queries. """ context_matches = ( matches if result_fields is None or "context" in result_fields else () ) for match in context_matches: try: with self._read_ctx() as conn: ctx_cursor = conn.execute( """WITH target AS ( SELECT session_id, timestamp, id FROM messages WHERE id = ? ) SELECT role, content FROM ( SELECT m.id, m.timestamp, m.role, m.content FROM messages m JOIN target t ON t.session_id = m.session_id WHERE (m.timestamp < t.timestamp) OR (m.timestamp = t.timestamp AND m.id < t.id) ORDER BY m.timestamp DESC, m.id DESC LIMIT 1 ) UNION ALL SELECT role, content FROM messages WHERE id = ? UNION ALL SELECT role, content FROM ( SELECT m.id, m.timestamp, m.role, m.content FROM messages m JOIN target t ON t.session_id = m.session_id WHERE (m.timestamp > t.timestamp) OR (m.timestamp = t.timestamp AND m.id > t.id) ORDER BY m.timestamp ASC, m.id ASC LIMIT 1 )""", (match["id"], match["id"]), ) context_msgs = [] for row in ctx_cursor.fetchall(): decoded = self._decode_content(row["content"]) if isinstance(decoded, list): text_parts = [ part.get("text", "") for part in decoded if isinstance(part, dict) and part.get("type") == "text" ] text = " ".join(t for t in text_parts if t).strip() preview = text or "[multimodal content]" elif isinstance(decoded, str): preview = decoded else: preview = "" context_msgs.append( {"role": row["role"], "content": preview[:200]} ) match["context"] = context_msgs except Exception: match["context"] = [] # Full message content is never selected by any search route: every # SELECT returns snippet + metadata only (saves I/O on multi-MB tool # rows and the tokens a content column would cost downstream). The # context query above re-fetches its 3-message window by id, so # nothing reads content from the match rows themselves. The pop stays # as a belt-and-braces guard for any future route that selects it. 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 def _search_messages_impl( self, query: str, source_filter: List[str] = None, exclude_sources: List[str] = None, role_filter: List[str] = None, limit: int = 20, offset: int = 0, sort: str = None, include_inactive: bool = False, fields: Optional[Collection[str]] = None, ) -> List[Dict[str, Any]]: """ Full-text search across session messages using FTS5. Supports FTS5 query syntax: - Simple keywords: "docker deployment" - Phrases: '"exact phrase"' - Boolean: "docker OR kubernetes", "python NOT java" - Prefix: "deploy*" Returns matching messages with session metadata, content snippet, and surrounding context (1 message before and after the match). ``fields`` selects a result projection; omitting it preserves the complete legacy result. Context is only loaded when that projection consumes it. ``sort`` controls temporal ordering: - ``None`` (default): FTS5 BM25 relevance only. Time-neutral. - ``"newest"``: order by message timestamp DESC, then by rank. - ``"oldest"``: order by message timestamp ASC, then by rank. The short-CJK LIKE fallback already orders by timestamp DESC and ignores ``sort``. The trigram CJK path honours ``sort`` like the main FTS5 path. Rewound (``active=0``, ``compacted=0``) rows are excluded by default — the user took those back. Compaction-archived rows (``active=0``, ``compacted=1``) ARE included by default: they were summarized away from the live context but remain part of the conversation's record, so the pre-compaction transcript stays discoverable after in-place compaction (#38763). Pass ``include_inactive=True`` to search every row regardless. """ 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 [] # New oversized tool results only index a bounded prefix to keep the # foreground write transaction short. 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, source_filter=source_filter, exclude_sources=exclude_sources, role_filter=role_filter, limit=limit, offset=offset, sort=sort, include_inactive=include_inactive, ) 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, source_filter=source_filter, exclude_sources=exclude_sources, role_filter=role_filter, limit=limit, offset=offset, sort=sort, include_inactive=include_inactive, ) return self._finalize_search_matches( matches, result_fields=result_fields ) if not self._fts_enabled: return [] # Normalise sort. Anything not in the allowed set falls back to None # (FTS5 rank-only) so callers can pass through user input without # validation. if isinstance(sort, str): sort_norm = sort.strip().lower() if sort_norm not in ("newest", "oldest"): sort_norm = None else: sort_norm = None # ORDER BY shared across the main FTS5 path and trigram CJK path. # With sort set, timestamp is primary and rank is the tiebreaker. if sort_norm == "newest": order_by_sql = "ORDER BY m.timestamp DESC, rank" elif sort_norm == "oldest": order_by_sql = "ORDER BY m.timestamp ASC, rank" else: order_by_sql = "ORDER BY rank" # Build WHERE clauses dynamically where_clauses = ["messages_fts MATCH ?"] params: list = [query] if not include_inactive: # Live rows (active=1) AND compaction-archived rows (compacted=1) # are discoverable; only rewind/undo rows (active=0, compacted=0) # are hidden. See archive_and_compact() / #38763. where_clauses.append("(m.active = 1 OR m.compacted = 1)") if source_filter is not None: source_placeholders = ",".join("?" for _ in source_filter) where_clauses.append(f"s.source IN ({source_placeholders})") params.extend(source_filter) if exclude_sources is not None: exclude_placeholders = ",".join("?" for _ in exclude_sources) where_clauses.append(f"s.source NOT IN ({exclude_placeholders})") params.extend(exclude_sources) if role_filter: role_placeholders = ",".join("?" for _ in role_filter) where_clauses.append(f"m.role IN ({role_placeholders})") params.extend(role_filter) where_sql = " AND ".join(where_clauses) params.extend([limit, offset]) sql = f""" SELECT m.id, m.session_id, m.role, snippet(messages_fts, -1, '>>>', '<<<', '...', 40) AS snippet, m.timestamp, m.tool_name, s.source, s.model, s.started_at AS session_started FROM messages_fts JOIN messages m ON m.id = messages_fts.rowid JOIN sessions s ON s.id = m.session_id WHERE {where_sql} {order_by_sql} LIMIT ? OFFSET ? """ # CJK queries bypass the unicode61 FTS5 table. The default tokenizer # splits CJK characters into individual tokens, so "大别山项目" becomes # "大 AND 别 AND 山 AND 项 AND 目" — producing false positives and # missing exact phrase matches. # # For queries with 3+ CJK characters, we use the trigram FTS5 table # (indexed substring matching with ranking and snippets). For shorter # CJK queries (1-2 chars), trigram can't match (it needs ≥9 UTF-8 # bytes = 3 CJK chars), so we fall back to LIKE. matches: List[Dict[str, Any]] = [] is_cjk = self._contains_cjk(query) if is_cjk: raw_query = query.strip('"').strip() cjk_count = self._count_cjk(raw_query) # Per-token CJK length check (#20494): trigram needs >=3 CJK chars # per token. A query like "广西 OR 桂林 OR 漓江" has cjk_count=6 # (>=3) but each individual token is only 2 chars — trigram returns 0. # Route to LIKE when any non-operator CJK token is <3 CJK chars. _tokens_for_check = [ t for t in raw_query.split() if t.upper() not in {"AND", "OR", "NOT"} and self._contains_cjk(t) ] _any_short_cjk = any( self._count_cjk(t) < 3 for t in _tokens_for_check ) _trigram_succeeded = False # Tool rows are excluded from the trigram index (they're ~90% of # message bytes and machine noise — see FTS_TRIGRAM_SQL). A CJK # query explicitly filtering on role='tool' must therefore use # the LIKE fallback, which scans the base table directly. _wants_tool_rows = bool(role_filter) and "tool" in role_filter # Cron and subagent transcripts are excluded too (see # FTS_TRIGRAM_EXCLUDED_SOURCES); an explicit filter for them # must likewise scan the base table. _wants_cron_rows = bool(source_filter) and any( src in FTS_TRIGRAM_EXCLUDED_SOURCES for src in source_filter ) # ── 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 and not _wants_tool_rows and not _wants_cron_rows and not self._has_lone_cjk_run(raw_query) ): tokens = raw_query.split() parts = [] for tok in tokens: if tok.upper() in {"AND", "OR", "NOT"}: parts.append(tok) else: parts.append('"' + tok.replace('"', '""') + '"') cjk_query = " ".join(parts) cjk_where = ["messages_fts_cjk MATCH ?"] cjk_params: list = [cjk_query] if not include_inactive: cjk_where.append("(m.active = 1 OR m.compacted = 1)") if source_filter is not None: cjk_where.append(f"s.source IN ({','.join('?' for _ in source_filter)})") cjk_params.extend(source_filter) if exclude_sources is not None: cjk_where.append(f"s.source NOT IN ({','.join('?' for _ in exclude_sources)})") cjk_params.extend(exclude_sources) if role_filter: cjk_where.append(f"m.role IN ({','.join('?' for _ in role_filter)})") cjk_params.extend(role_filter) cjk_sql = f""" SELECT m.id, m.session_id, m.role, snippet(messages_fts_cjk, -1, '>>>', '<<<', '...', 40) AS snippet, m.timestamp, m.tool_name, s.source, s.model, s.started_at AS session_started FROM messages_fts_cjk JOIN messages m ON m.id = messages_fts_cjk.rowid JOIN sessions s ON s.id = m.session_id WHERE {' AND '.join(cjk_where)} {order_by_sql} LIMIT ? OFFSET ? """ cjk_params.extend([limit, offset]) try: with self._read_ctx() as conn: cjk_cursor = conn.execute(cjk_sql, cjk_params) matches = [dict(row) for row in cjk_cursor.fetchall()] _trigram_succeeded = True except sqlite3.OperationalError: # Tokenizer missing on this connection / query syntax — # the trigram + LIKE routes below still answer. logger.debug( "messages_fts_cjk query failed; falling back to " "trigram/LIKE", exc_info=True, ) except sqlite3.DatabaseError as exc: # A full-message rebuild is unbounded and holds the writer # lock, so a live search never performs one. Detach the # derived indexes and answer from canonical rows instead. # Non-FTS corruption is not safe to reinterpret here. if not self._enter_fts_fail_open(exc): raise logger.warning( "CJK-bigram FTS search hit a corruption error (%s); " "detached FTS and falling back to canonical LIKE.", exc, ) if ( not _trigram_succeeded and cjk_count >= 3 and not _any_short_cjk and self._trigram_available and not _wants_tool_rows and not _wants_cron_rows ): # Trigram FTS5 path — quote each non-operator token to handle # FTS5 special chars (%, *, etc.) while preserving boolean # operators (AND, OR, NOT) for multi-term queries. tokens = raw_query.split() parts = [] for tok in tokens: if tok.upper() in {"AND", "OR", "NOT"}: parts.append(tok) else: parts.append('"' + tok.replace('"', '""') + '"') trigram_query = " ".join(parts) tri_where = ["messages_fts_trigram MATCH ?"] tri_params: list = [trigram_query] if not include_inactive: tri_where.append("(m.active = 1 OR m.compacted = 1)") if source_filter is not None: tri_where.append(f"s.source IN ({','.join('?' for _ in source_filter)})") tri_params.extend(source_filter) if exclude_sources is not None: tri_where.append(f"s.source NOT IN ({','.join('?' for _ in exclude_sources)})") tri_params.extend(exclude_sources) if role_filter: tri_where.append(f"m.role IN ({','.join('?' for _ in role_filter)})") tri_params.extend(role_filter) tri_sql = f""" SELECT m.id, m.session_id, m.role, snippet(messages_fts_trigram, -1, '>>>', '<<<', '...', 40) AS snippet, m.timestamp, m.tool_name, s.source, s.model, s.started_at AS session_started FROM messages_fts_trigram JOIN messages m ON m.id = messages_fts_trigram.rowid JOIN sessions s ON s.id = m.session_id WHERE {' AND '.join(tri_where)} {order_by_sql} LIMIT ? OFFSET ? """ tri_params.extend([limit, offset]) try: with self._read_ctx() as conn: tri_cursor = conn.execute(tri_sql, tri_params) matches = [dict(row) for row in tri_cursor.fetchall()] _trigram_succeeded = True except sqlite3.OperationalError: # Trigram query failed at runtime — fall through to LIKE. pass except sqlite3.DatabaseError as exc: # Preserve the same bounded recovery contract as the CJK # and main FTS paths: detach derived indexes, then fall # through to the canonical LIKE query. A non-FTS storage # error remains fatal rather than being hidden as a miss. if not self._enter_fts_fail_open(exc): raise logger.warning( "Trigram FTS search hit a corruption error (%s); " "detached FTS and falling back to canonical LIKE.", exc, ) if not _trigram_succeeded: # Short / mixed CJK query, trigram unavailable, or trigram # <3 CJK chars. Fall back to LIKE substring search. # For multi-token OR queries (e.g. "广西 OR 桂林 OR 漓江"), # build one LIKE condition per non-operator token so each term # is matched independently (#20494). non_op_tokens = [ t for t in raw_query.split() if t.upper() not in {"AND", "OR", "NOT"} ] or [raw_query] token_clauses = [] like_params: list = [] for tok in non_op_tokens: esc = _escape_like(tok) token_clauses.append( "(m.content LIKE ? ESCAPE '\\' OR m.tool_name LIKE ? ESCAPE '\\' OR m.tool_calls LIKE ? ESCAPE '\\')" ) like_params += [f"%{esc}%", f"%{esc}%", f"%{esc}%"] like_where = [f"({' OR '.join(token_clauses)})"] if not include_inactive: # Same visibility rule as the FTS5 paths: live rows and # compaction-archived rows are discoverable; rewind/undo # rows (active=0, compacted=0) are hidden (#38763). like_where.append("(m.active = 1 OR m.compacted = 1)") if source_filter is not None: like_where.append(f"s.source IN ({','.join('?' for _ in source_filter)})") like_params.extend(source_filter) if exclude_sources is not None: like_where.append(f"s.source NOT IN ({','.join('?' for _ in exclude_sources)})") like_params.extend(exclude_sources) if role_filter: like_where.append(f"m.role IN ({','.join('?' for _ in role_filter)})") like_params.extend(role_filter) like_sql = f""" SELECT m.id, m.session_id, m.role, substr(m.content, max(1, instr(m.content, ?) - 40), 120) AS snippet, m.timestamp, m.tool_name, s.source, s.model, s.started_at AS session_started FROM messages m JOIN sessions s ON s.id = m.session_id WHERE {' AND '.join(like_where)} ORDER BY m.timestamp DESC LIMIT ? OFFSET ? """ like_params.extend([limit, offset]) # instr() for snippet uses first search token like_params = [non_op_tokens[0]] + like_params with self._read_ctx() as conn: like_cursor = conn.execute(like_sql, like_params) matches = [dict(row) for row in like_cursor.fetchall()] else: try: with self._read_ctx() as conn: cursor = conn.execute(sql, params) matches = [dict(row) for row in cursor.fetchall()] except sqlite3.OperationalError: # FTS5 query syntax error despite sanitization — return empty return [] except sqlite3.DatabaseError as exc: # 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. Live search must remain bounded, so detach the # derived indexes and answer from canonical message rows. 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, source_filter=source_filter, exclude_sources=exclude_sources, role_filter=role_filter, limit=limit, offset=offset, sort=sort, include_inactive=include_inactive, ) # Deferred-rebuild supplement (schema v23): while the background # backfill is pending, the FTS indexes only cover rows outside the # (progress, high_water] gap. Top the results up with a bounded LIKE # scan over just that id range so search never silently loses old # messages mid-rebuild. The range shrinks as the backfill advances, # so this cost decays to zero. The CJK LIKE-fallback path above # already scans the whole base table and needs no supplement. rebuild_status = self.fts_rebuild_status() if rebuild_status is not None and len(matches) < limit: try: gap_matches = self._search_unindexed_gap( query, limit - len(matches), include_inactive=include_inactive, source_filter=source_filter, exclude_sources=exclude_sources, role_filter=role_filter, ) 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) # Pure-Latin queries run against the unicode61 ``messages_fts`` table, # whose tokenizer does not insert a boundary between Latin letters and # adjacent CJK characters: "修改youer服务端" is indexed as one token, # so MATCH "youer" finds nothing even though the substring is present # (#54242). When the exact-token search returns nothing, retry on the # substring-capable indexes. Preference order: # 1. messages_fts_cjk (when built): its tokenizer splits Latin runs # off adjacent CJK, so "youer" is an exact ranked token match. # 2. messages_fts_trigram: substring matching, needs >=3-char # tokens (shorter tokens produce no trigrams). # Gated on a zero-result miss so successful Latin searches keep their # unicode61 ranking — strictly additive, never reorders existing # hits. Trade-off on the trigram leg: any zero-result Latin query # gains substring semantics (e.g. "cat" can then match # "concatenate"). Genuinely absent terms still return []. Skipped for # role_filter=['tool'] queries — both fallback indexes exclude tool # rows (v23), so a retry could never add hits. if ( not matches and not is_cjk and not (bool(role_filter) and "tool" in role_filter) ): _fb_query = query.strip('"').strip() if self._fts_cjk_available: cjk_fb = self._run_trigram_search( _fb_query, table="messages_fts_cjk", order_by_sql=order_by_sql, include_inactive=include_inactive, source_filter=source_filter, exclude_sources=exclude_sources, role_filter=role_filter, limit=limit, offset=offset, ) if cjk_fb: matches = cjk_fb if ( not matches and self._trigram_available and self._trigram_eligible_tokens(query) ): tri_matches = self._run_trigram_search( _fb_query, order_by_sql=order_by_sql, include_inactive=include_inactive, source_filter=source_filter, exclude_sources=exclude_sources, role_filter=role_filter, limit=limit, offset=offset, ) if tri_matches: matches = tri_matches return self._finalize_search_matches(matches, result_fields=result_fields) def _search_unindexed_gap( self, fts_query: str, limit: int, *, include_inactive: bool = False, source_filter: Optional[List[str]] = None, exclude_sources: Optional[List[str]] = None, role_filter: Optional[List[str]] = None, ) -> List[Dict[str, Any]]: """LIKE-scan the rows the deferred rebuild hasn't indexed yet. Only touches ids in (fts_rebuild_progress, fts_rebuild_high_water] — a range that shrinks to nothing as the backfill advances. The FTS query is degraded to per-token substring terms (AND-joined; quoted phrases kept whole), which is deliberately recall-over-precision: temporary results beat silently missing ones mid-rebuild. """ status = self.fts_rebuild_status() if status is None or limit <= 0: return [] progress, high_water = status["indexed"], status["total"] # Degrade the FTS query to LIKE terms: strip operators/wildcards, # keep quoted phrases intact, AND the rest. terms: List[str] = [] for raw_tok in re.findall(r'"[^"]+"|\S+', fts_query): tok = raw_tok.strip('"').strip("*").strip() if not tok or tok.upper() in {"AND", "OR", "NOT", "NEAR"}: continue terms.append(tok) if not terms: return [] where = ["m.id > ? AND m.id <= ?"] params: list = [progress, high_water] for term in terms: esc = _escape_like(term) where.append( "(m.content LIKE ? ESCAPE '\\' OR m.tool_name LIKE ? ESCAPE '\\' " "OR m.tool_calls LIKE ? ESCAPE '\\')" ) params += [f"%{esc}%"] * 3 if not include_inactive: where.append("(m.active = 1 OR m.compacted = 1)") 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) sql = f""" SELECT m.id, m.session_id, m.role, substr(m.content, max(1, instr(m.content, ?) - 40), 120) AS snippet, m.timestamp, m.tool_name, s.source, s.model, s.started_at AS session_started FROM messages m JOIN sessions s ON s.id = m.session_id WHERE {' AND '.join(where)} ORDER BY m.timestamp DESC LIMIT ? """ params = [terms[0]] + params + [limit] with self._read_ctx() as conn: rows = conn.execute(sql, params).fetchall() return [dict(r) for r in rows] def search_sessions_by_id( self, query: str, limit: int = 20, include_archived: bool = True, source: str = None, sources: List[str] = None, exclude_sources: List[str] = None, ) -> List[Dict[str, Any]]: """Search surfaced sessions by exact/prefix/substring session id. Desktop search uses this alongside FTS message search so users can paste a session id from logs, CLI output, or another Hermes surface and jump straight to that conversation. Matching also checks ``_lineage_root_id`` for projected compression-chain tips, so an old root id still resolves to the live continuation row. """ needle = (query or "").strip().lower() if not needle or limit <= 0: return [] # SQL-bounded: list_sessions_rich pushes the id LIKE filter into the # query (matching the row's own id AND any id in its forward # compression chain), so we only materialize matching rows instead of # scanning every session. Fetch a small multiple of `limit` so the # in-Python exact/prefix/substring ranking below has enough candidates # to order, then truncate. 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: ids = [str(row.get("id") or ""), str(row.get("_lineage_root_id") or "")] normalized = [value.lower() for value in ids if value] if any(value == needle for value in normalized): return 0 if any(value.startswith(needle) for value in normalized): return 1 return 2 ranked = sorted( enumerate(candidates), key=lambda item: (score(item[1]), item[0]), ) return [row for _, row in ranked[:limit]] def _fts_table_exists(self, name: str) -> bool: """True if an FTS5 virtual table is queryable in this DB.""" try: self._conn.execute(f"SELECT 1 FROM {name} LIMIT 0") return True except sqlite3.DatabaseError: # OperationalError ("no such table") or the broader # DatabaseError class ("vtable constructor failed", raised when # e.g. a required tokenizer is missing or the table is mid- # teardown) — in every case the table is not queryable. return False def optimize_fts(self) -> int: """Merge fragmented FTS5 b-tree segments into one per index. FTS5 indexes grow as a series of incremental segments — one per ``INSERT`` batch driven by the message triggers. Over tens of thousands of messages these segments accumulate, which both bloats the ``*_data`` shadow tables and slows ``MATCH`` queries that must scan every segment. The special ``'optimize'`` command rewrites each index as a single merged segment. This is purely a maintenance operation — it changes neither search results nor ``snippet()`` output, only on-disk layout and query speed. It is complementary to VACUUM: ``optimize`` compacts the FTS index internally, then VACUUM returns the freed pages to the OS. Skips any FTS table that does not exist (e.g. the trigram index when disabled via ``HERMES_DISABLE_FTS_TRIGRAM`` or not yet created), so it is safe to call unconditionally. Returns the number of FTS indexes that were optimized. """ optimized = 0 with self._lock: for tbl in self._FTS_TABLES: if not self._fts_table_exists(tbl): continue try: # The column name in the INSERT must match the table name # for FTS5 special commands. self._conn.execute( f"INSERT INTO {tbl}({tbl}) VALUES('optimize')" ) optimized += 1 except sqlite3.OperationalError as exc: logger.warning( "FTS optimize failed for %s: %s", tbl, exc ) return optimized def rebuild_fts(self) -> int: """Rebuild FTS5 indexes from the canonical ``messages`` table. Uses the FTS5 ``'rebuild'`` command, which rewrites the internal b-tree segments from the content rows. This is the documented recovery for a corrupt FTS index that rejects message writes while reads still succeed (issue #50502). Unlike ``optimize_fts`` (which merges existing segments), ``rebuild`` discards and recreates the index data entirely. 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. Safe to call when FTS tables don't exist (skips them). Returns the number of FTS indexes that were rebuilt. """ rebuilt = 0 with fts_rebuild_admission(getattr(self, "db_path", None)) 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: high_water = self._conn.execute( "SELECT COALESCE(MAX(id), 0) FROM messages" ).fetchone()[0] self._conn.execute( "INSERT INTO state_meta (key, value) VALUES (?, ?) " "ON CONFLICT(key) DO UPDATE SET value = excluded.value", (FTS_TOOL_FULL_CONTENT_HIGH_WATER_KEY, str(high_water)), ) for tbl in self._FTS_TABLES: if not self._fts_table_exists(tbl): continue try: self._conn.execute( f"INSERT INTO {tbl}({tbl}) VALUES('rebuild')" ) self._conn.commit() rebuilt += 1 except sqlite3.OperationalError as exc: self._conn.rollback() 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 tells SQLite to stop after approximately that many output pages, so each command holds the write lock for milliseconds regardless of index size — unlike ``'optimize'``, which rewrites the whole index in one transaction (measured 9-18 s per index on a 10 GB production DB, long enough to exhaust a competing writer's entire lock-retry patience). Protocol (SQLite FTS5 §6.8-6.9): - ``usermerge`` is lowered to its minimum of 2 (persisted in the ``%_config`` shadow table, applied once per instance) so a positive merge acts on ANY level holding >= 2 segments. With the default of 4, levels below that threshold are never merged by a positive-rank command and a fragmented index cannot converge. - Up to *max_commands* merge commands run per index, stopping early on the documented no-progress signal: the delta in ``total_changes`` is < 2 (the command's own INSERT accounts for 1 change; >= 2 means real merge work happened). Each command is its own implicit transaction (the connection runs with ``isolation_level=None``), so the SQLite write lock is released between commands and competing processes can interleave writes mid-pass. Missing tables are valid schema variants (FTS variants are optional, and ``optimize_fts_storage`` legitimately drops + backfills these tables while writers keep running) and are skipped, mirroring ``optimize_fts``. Other SQLite errors propagate to the caller. Returns the number of merge commands executed. """ if isinstance(max_pages, bool) or not isinstance(max_pages, int): raise TypeError("max_pages must be an integer") if max_pages <= 0: raise ValueError("max_pages must be greater than zero") if max_commands is None: max_commands = self._FTS_MERGE_COMMANDS_PER_PASS if isinstance(max_commands, bool) or not isinstance(max_commands, int): raise TypeError("max_commands must be an integer") if max_commands <= 0: raise ValueError("max_commands must be greater than zero") executed = 0 with self._lock: for tbl in self._FTS_TABLES: if not self._fts_table_exists(tbl): continue # One-time (per instance) usermerge floor; the value is # persisted in the index's config shadow table so future # connections inherit it. Setting config is a metadata-only # write — it never touches segment data. if not getattr(self, "_fts_usermerge_floor_applied", False): 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