50873d2154
Six fixes for the wave-7 picker/input cluster:
DeepSeek -> "Deepseek") and left the gemini- branch's words lowercase
("Gemini 2.5 pro"). Vendor casing + parameter counts now applied after
title-case (GLM, DeepSeek, MiniMax, OpenAI, ERNIE, MiMo, BGE, VL, IT,
FP8, AI; 8b -> 8B, a3b -> A3B), and the gemini branch title-cases like
every other branch.
and was unreachable by keyboard (rows are highlighted, never DOM-focused,
so Radix's own ArrowRight never fires). The chevron is now visible on
every model row and ArrowRight (caret parked at query end) hands focus
to the highlighted trigger and opens its sub; ArrowLeft returns focus
to the search field. Consolidates #86968 + #104532.
-fast/-thinking/-preview ids to the base label. The tag now rides the
display name on every surface, and formatModelPillLabel no longer
doubles Fast for a -fast variant id.
all MoA presets were disabled: manual picks are sticky by design
(d595e636c8), but the virtual moa provider's catalog row disappears
entirely once no preset is enabled, so that one absence is
authoritative (moaPickRemoved) and the pick reseeds from the profile
default. Narrow moa-only exception — no general catalog diff.
token (nimb -> nimb*); none of the CJK routes can honour it (bigram and
trigram routes quote tokens so the star matches literally; LIKE has no
star wildcard at all), so CJK searches returned zero results. The star
is now stripped per token on the CJK path only.
measure() effect deps (stale measurements after toggling Inbox style)
and the card estimate undershot the four-line/wrapped-title worst case
(74px), painting rows over their neighbours on cold start. Card
estimate raised to the worst-case-covering 96px and the deps fixed.
1371 lines
75 KiB
Python
1371 lines
75 KiB
Python
"""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_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: # 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)
|
|
|
|
# ── 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, 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,
|
|
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, 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,
|
|
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,
|
|
sources: List[str] = None, exclude_sources: List[str] = 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.OperationalError as exc:
|
|
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.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
|
|
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
|
|
|
|
|
|
# ---- BEGIN PLUGIN-COMPAT (revert-scheduled; see COMPAT_MANIFEST.md) ----
|
|
# Names external plugins imported from this module before the Sep 2026 decomposition.
|
|
# Internal code MUST NOT use these (scripts/check_compat_pointers.py fails CI if it does).
|
|
# The whole block is removed by reverting the commit that added it.
|
|
import json # noqa: F401,E402
|
|
import os # noqa: F401,E402
|
|
# ---- END PLUGIN-COMPAT ----
|