mirror of
https://github.com/NousResearch/hermes-agent.git
synced 2026-11-07 19:26:26 +03:00
alt-glitch
ac61edf748
Open no longer deletes the retired messages_fts_cjk index inline. It drops the triggers and the source view, deletes the fts_cjk_* markers, removes the vtable's schema row, and renames each remaining shadow table to fts_v22_trash_messages_fts_cjk_<suffix>. `hermes sessions optimize-storage` then empties and drops those tables in chunks, as it already does for the v22 index. A large index was otherwise deleted under the write lock at open, past the 20 s open and 60 s write budgets of other processes. The same path runs on every SQLite version, so the separate pre-3.47 fallback is gone. The removal runs whenever any messages_fts_cjk object is left, not only when DROP TABLE fails. A run that stopped after removing the schema row left shadow tables that no later open cleaned up, and every open repeated the warning. The warning now comes from the open path and fires only when something was removed. The repair health probe (`hermes sessions repair --check-only`, the console repair check, doctor) runs the removal and its checks in one BEGIN IMMEDIATE transaction that it always rolls back. The probe judges the store as the next open will leave it, and the file stays unchanged. The rebuild helpers _rebuild_status, _rebuild_step and _rebuild_finish are folded into fts_rebuild_status, fts_rebuild_step and _fts_rebuild_finish. Their prefix, finish and finish_when_empty parameters had one value each since the CJK backfill was removed. Behaviour is unchanged.
1442 lines
80 KiB
Python
1442 lines
80 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 contextvars
|
|
import logging
|
|
import re
|
|
import sqlite3
|
|
import threading
|
|
import time
|
|
from dataclasses import dataclass, field
|
|
from functools import partial
|
|
from typing import Any, Callable, Collection, Dict, List, Optional, Tuple
|
|
|
|
from agent.skill_commands import describe_skill_invocation
|
|
from hermes_state_errors import is_malformed_db_error
|
|
from hermes_state_common import (
|
|
FTS_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,
|
|
escape_like as _escape_like, fts_rebuild_admission, fts_trigram_session_sql, routed_sessions_setting,
|
|
)
|
|
from hermes_state_fts import fts_layout_current, fts_layout_drift
|
|
|
|
# Pre-split logger identity so log filtering/capture is unchanged.
|
|
logger = logging.getLogger("hermes_state")
|
|
|
|
|
|
@dataclass
|
|
class _BackfillProgress:
|
|
"""How many messages a backfill pass (``backfill`` = base + trigram) has
|
|
indexed out of the messages at or below its high water. Counts rows, never ids: a pruned
|
|
store's id gaps would otherwise read 0% then jump to 99%. The step engine adds the rows each
|
|
window covered (:meth:`SessionSearchMixin.fts_rebuild_step`)."""
|
|
|
|
phase: str
|
|
total: int = 0
|
|
indexed: int = 0
|
|
|
|
def info(self) -> dict[str, Any]:
|
|
done, total = self.indexed, self.total
|
|
return {"phase": self.phase, "percent": min(100, 100 * done // total) if total else 100,
|
|
"indexed": min(done, total), "total": total}
|
|
|
|
|
|
@dataclass
|
|
class _TeardownProgress:
|
|
"""Which demoted shadow table optimize-storage is draining and how far (``progress_cb``
|
|
payload). Owned and kept current by :meth:`SessionSearchMixin._fts_teardown_trash_step`.
|
|
``counted=False`` (no progress callback) skips the per-table ``COUNT(*)``."""
|
|
|
|
table: str = ""
|
|
table_index: int = 0
|
|
table_count: int = 0
|
|
total: int = 0
|
|
drained: int = 0
|
|
counted: bool = True
|
|
|
|
def info(self) -> dict[str, Any]:
|
|
done, total = self.drained, self.total
|
|
return {"phase": "teardown", "percent": min(100, 100 * done // total) if total else 100,
|
|
"indexed": done, "total": total, "table": self.table,
|
|
"table_index": self.table_index, "table_count": self.table_count}
|
|
|
|
|
|
@dataclass
|
|
class _VacuumProgress:
|
|
"""VACUUM is one statement: liveness is elapsed time against the size before and the estimated
|
|
size after (live pages only), never a percent."""
|
|
|
|
size_bytes: int
|
|
estimated_bytes: int
|
|
started: float = field(default_factory=time.monotonic)
|
|
|
|
def info(self) -> dict[str, Any]:
|
|
return {"phase": "vacuum", "percent": 0, "indexed": 0, "total": 0,
|
|
"elapsed": time.monotonic() - self.started, "size_bytes": self.size_bytes,
|
|
"estimated_bytes": self.estimated_bytes}
|
|
|
|
|
|
class _HolderWatch:
|
|
"""Re-runs the foreign-holder scan (the one the held-store refusal and ``--force`` use) every
|
|
*interval* seconds on a side thread while optimize-storage rebuilds and tears down. The chunk
|
|
loop only reads :attr:`held`, so a quiet store pays nothing per chunk, while a writer that opens
|
|
state.db mid-run (Desktop launched, gateway restarted) moves the loop onto the paced cycle within
|
|
one interval instead of waiting behind back-to-back chunks."""
|
|
|
|
def __init__(self, scan: Callable[[], list], interval: float):
|
|
self._scan, self._interval = scan, interval
|
|
self._held, self._stop = threading.Event(), threading.Event()
|
|
self._thread: Optional[threading.Thread] = None
|
|
|
|
@property
|
|
def held(self) -> bool:
|
|
return self._held.is_set()
|
|
|
|
def _refresh(self) -> None:
|
|
(self._held.set if self._scan() else self._held.clear)()
|
|
|
|
def _run(self) -> None:
|
|
while not self._stop.wait(self._interval):
|
|
self._refresh()
|
|
|
|
def __enter__(self) -> "_HolderWatch":
|
|
self._refresh() # a holder already there paces from the first chunk
|
|
# copy_context().run: the watcher keeps the caller's profile scope (HX012).
|
|
self._thread = threading.Thread(target=contextvars.copy_context().run, args=(self._run,),
|
|
name="optimize-holder-watch", daemon=True)
|
|
self._thread.start()
|
|
return self
|
|
|
|
def __exit__(self, *_exc) -> None:
|
|
self._stop.set()
|
|
if self._thread is not None:
|
|
self._thread.join()
|
|
|
|
|
|
|
|
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 trigram route
|
|
quotes every token before MATCH (so ``*`` matches a literal asterisk) and
|
|
LIKE has no ``*`` wildcard at all (only ``%``/``_``). Left in place, every
|
|
CJK search arriving from the web/desktop search box — which appends the
|
|
star to each unquoted token so partial English words match — searches for
|
|
a term ending in a literal ``*`` and returns nothing (#90636). Only
|
|
TRAILING stars go: a star written inside a quoted phrase is the user's
|
|
own text, and a token that is ALL stars keeps its original form so it
|
|
cannot degrade to a match-everything empty term.
|
|
"""
|
|
if "*" not in raw_query:
|
|
return raw_query
|
|
stripped: list[str] = []
|
|
for token in raw_query.split():
|
|
if token.upper() in _FTS_OPERATORS:
|
|
stripped.append(token)
|
|
else:
|
|
stripped.append(token.rstrip("*") or token)
|
|
return " ".join(stripped) or raw_query
|
|
|
|
|
|
def _flatten_text(decoded: Any) -> str:
|
|
"""Multimodal part list -> joined text (or the placeholder); str passes through; else ''."""
|
|
if isinstance(decoded, list):
|
|
parts = [p.get("text", "") for p in decoded if isinstance(p, dict) and p.get("type") == "text"]
|
|
return " ".join(t for t in parts if t).strip() or "[multimodal content]"
|
|
return decoded if isinstance(decoded, str) else ""
|
|
|
|
|
|
def _positive_int(name: str, value: Any) -> None:
|
|
if isinstance(value, bool) or not isinstance(value, int):
|
|
raise TypeError(f"{name} must be an integer")
|
|
if value <= 0:
|
|
raise ValueError(f"{name} must be greater than zero")
|
|
|
|
|
|
def _search_select_sql(snippet_sql: str, from_sql: str, where: list[str], order_by: str, limit_sql: str) -> str:
|
|
"""Result-row SELECT shared by the FTS and LIKE routes (SQL text is pinned)."""
|
|
return f"""
|
|
SELECT m.id, m.session_id, m.role,
|
|
{snippet_sql},
|
|
{_SEARCH_SELECT_TAIL}
|
|
FROM {from_sql}
|
|
JOIN sessions s ON s.id = m.session_id
|
|
WHERE {' AND '.join(where)}
|
|
{order_by}
|
|
{limit_sql}
|
|
"""
|
|
|
|
|
|
def _search_filter_clauses(
|
|
where: list[str], params: list, *, include_inactive: bool, source_filter: Optional[list[str]],
|
|
exclude_sources: Optional[list[str]], role_filter: Optional[list[str]],
|
|
after_ts: Optional[int] = None, before_ts: Optional[int] = None) -> None:
|
|
"""Append the visibility/source/role/session-start predicates every search route shares. Live
|
|
rows (active=1) AND compaction-archived rows (compacted=1) are discoverable; only
|
|
rewind/undo rows (active=0, compacted=0) are hidden. ``after_ts``/``before_ts`` bound
|
|
``sessions.started_at`` (inclusive / exclusive) inside the query so LIMIT cannot be
|
|
filled by out-of-window hits."""
|
|
if not include_inactive:
|
|
where.append("(m.active = 1 OR m.compacted = 1)")
|
|
# display_kind="hidden" rows are model-facing scaffolding the person never saw; a hit would confuse.
|
|
where.append("COALESCE(m.display_kind, '') <> 'hidden'")
|
|
if source_filter is not None:
|
|
where.append(f"s.source IN ({','.join('?' for _ in source_filter)})")
|
|
params.extend(source_filter)
|
|
if exclude_sources is not None:
|
|
where.append(f"s.source NOT IN ({','.join('?' for _ in exclude_sources)})")
|
|
params.extend(exclude_sources)
|
|
if role_filter:
|
|
where.append(f"m.role IN ({','.join('?' for _ in role_filter)})")
|
|
params.extend(role_filter)
|
|
if after_ts is not None:
|
|
where.append("s.started_at >= ?")
|
|
params.append(int(after_ts))
|
|
if before_ts is not None:
|
|
where.append("s.started_at < ?")
|
|
params.append(int(before_ts))
|
|
|
|
|
|
class SessionSearchMixin:
|
|
"""See module docstring — mixin for SessionDB (Search cluster)."""
|
|
|
|
_SEARCH_MESSAGE_RESULT_FIELDS = (
|
|
"id", "session_id", "role", "snippet", "timestamp", "tool_name", "source", "model", "session_started", "context"
|
|
)
|
|
|
|
@classmethod
|
|
def _search_message_fields(cls, fields: Optional[Collection[str]]) -> Optional[tuple[str, ...]]:
|
|
"""Validate and canonically order an optional result projection."""
|
|
if fields is None:
|
|
return None
|
|
if isinstance(fields, str):
|
|
raise TypeError("search fields must be a collection of field names, not a string")
|
|
requested = set(fields)
|
|
unknown = requested.difference(cls._SEARCH_MESSAGE_RESULT_FIELDS)
|
|
if unknown:
|
|
raise ValueError(f"unknown search result field(s): {', '.join(sorted(unknown))}")
|
|
return tuple(field for field in cls._SEARCH_MESSAGE_RESULT_FIELDS if field in requested)
|
|
|
|
def _try_incremental_merge_fts(self) -> None:
|
|
"""One bounded FTS5 merge pass that never fails the already-committed write (a caller
|
|
must never replay an ambiguous, possibly-durable write — even on the bare
|
|
SystemError CPython's sqlite3 layer can raise under cross-thread errmsg scrambling)."""
|
|
if not self._fts_enabled:
|
|
return
|
|
try:
|
|
self._merge_fts_incrementally(max_pages=self._FTS_MERGE_MAX_PAGES_PER_INDEX)
|
|
except Exception as exc:
|
|
# The canonical write is already committed before this cadence runs. No maintenance failure —
|
|
# including the bare SystemError the CPython sqlite3 layer can raise under cross-thread errmsg
|
|
# scrambling — may escape and make the caller replay an ambiguous, possibly-durable write
|
|
# (#90734, #85079).
|
|
logger.warning("FTS incremental merge failed after commit: %s", exc)
|
|
|
|
# ── Deferred rebuild engine ─────────────────────
|
|
|
|
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."""
|
|
rows = self._read_all("SELECT key, value FROM state_meta WHERE key IN (?, ?)",
|
|
("fts_rebuild_high_water", "fts_rebuild_progress"))
|
|
meta = {r["key"]: r["value"] for r in rows}
|
|
high_water = meta.get("fts_rebuild_high_water")
|
|
if high_water is None or int(high_water) <= 0:
|
|
return None
|
|
total, progress = int(high_water), int(meta.get("fts_rebuild_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.
|
|
_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 messages_fts(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 <= ?"
|
|
)
|
|
_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."""
|
|
def _do(conn):
|
|
hw_row = _meta_row(conn, "fts_rebuild_high_water")
|
|
if hw_row is not None:
|
|
hw = int(hw_row[0])
|
|
conn.execute(self._BASE_BOUNDARY_SWEEP_SQL, (FTS_TOOL_CONTENT_PREFIX_CHARS, hw - 1000, hw + 1000))
|
|
if self._trigram_available:
|
|
conn.execute(self._TRIGRAM_BOUNDARY_SWEEP_SQL, (hw - 1000, hw + 1000))
|
|
_delete_meta(conn, "fts_rebuild_high_water", "fts_rebuild_progress")
|
|
self._execute_write(_do)
|
|
logger.info("Deferred FTS rebuild complete — all messages indexed.")
|
|
|
|
def fts_rebuild_step(self, progress: Optional[_BackfillProgress] = None) -> bool:
|
|
"""Backfill one chunk (``_fts_chunk_rows`` messages) 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. A failed chunk raises (nothing committed;
|
|
the caller decides whether to retry). *progress* gains the messages the chunk covered."""
|
|
if not self._fts_enabled:
|
|
return False
|
|
inserts = [self._CHUNK_INSERT_SQL]
|
|
if self._trigram_available:
|
|
inserts.append(self._TRIGRAM_CHUNK_INSERT_SQL)
|
|
high_water_raw = self.get_meta("fts_rebuild_high_water")
|
|
if high_water_raw is None:
|
|
return False
|
|
high_water = int(high_water_raw)
|
|
chunk = self._fts_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, "fts_rebuild_progress")
|
|
if row is None:
|
|
return False, 0 # finished (or cleared) by another process
|
|
done = int(row[0])
|
|
if done >= high_water:
|
|
return False, 0
|
|
# The window is the next `chunk` messages, not `chunk` ids: an id gap (pruned
|
|
# sessions) neither shrinks a chunk nor lets the next one span a million dense ids.
|
|
nth = conn.execute("SELECT id FROM messages WHERE id > ? AND id <= ? ORDER BY id LIMIT 1 OFFSET ?",
|
|
(done, high_water, chunk - 1)).fetchone()
|
|
upper = int(nth[0]) if nth is not None else high_water
|
|
covered = chunk if nth is not None else 0
|
|
if nth is None and progress is not None: # the last, short window (< chunk rows)
|
|
covered = conn.execute("SELECT COUNT(*) FROM messages WHERE id > ? AND id <= ?",
|
|
(done, upper)).fetchone()[0]
|
|
for sql in inserts:
|
|
conn.execute(sql, (done, upper))
|
|
# Progress lands in the same transaction as its rows (crash-atomic).
|
|
conn.execute("UPDATE state_meta SET value = ? WHERE key = ?", (str(upper), "fts_rebuild_progress"))
|
|
return upper < high_water, covered
|
|
|
|
more, covered = self._execute_write(_do)
|
|
if progress is not None:
|
|
# Counted after the commit: _execute_write may replay _do on contention.
|
|
progress.indexed += covered
|
|
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_teardown_trash_step(self, progress: Optional[_TeardownProgress] = None) -> bool:
|
|
"""Tear down one chunk (``_fts_chunk_rows`` rows) of a demoted v22 FTS shadow table (a PLAIN
|
|
table now); True while work remains. A failed chunk raises (nothing committed).
|
|
|
|
With *progress* (optimize-storage), the first step on each trash table deletes nothing: it
|
|
points *progress* at the table and restarts the chunk size (rows cost ~1000x more in one
|
|
shadow table than in another); later steps add the rows they deleted.
|
|
|
|
Single-column INTEGER-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 tables
|
|
keep the chunked ``LIMIT`` delete — they are small by construction.
|
|
"""
|
|
limit = self._fts_chunk_rows
|
|
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 trash and progress is not None and progress.table != trash[0]:
|
|
self._fts_teardown_start_table(conn, trash, progress)
|
|
return True
|
|
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_row = conn.execute(
|
|
f"SELECT {key} FROM {tbl} WHERE {key} > ? ORDER BY {key} LIMIT 1 OFFSET ?",
|
|
(high_water, limit - 1),
|
|
).fetchone() or conn.execute(f"SELECT MAX({key}) FROM {tbl} WHERE {key} > ?", (high_water,)).fetchone()
|
|
if upper_row is None or upper_row[0] is None:
|
|
return _drop(conn, marker_key), 0
|
|
upper = upper_row[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, cur.rowcount
|
|
# 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 {int(limit)})")
|
|
return (_drop(conn) if cur.rowcount == 0 else True), cur.rowcount
|
|
|
|
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
|
|
|
|
more, deleted = self._execute_write(_do)
|
|
if progress is not None:
|
|
# Counted after the commit: _execute_write may replay _do on contention.
|
|
progress.drained += deleted
|
|
return bool(more)
|
|
|
|
def _fts_teardown_start_table(self, conn, trash: list[str], progress: _TeardownProgress) -> None:
|
|
"""Point *progress* at ``trash[0]`` (its place in the run, the rows left past its resume
|
|
marker) and restart the chunk size. One COUNT per table, on the pooled reader *conn*."""
|
|
tbl = trash[0]
|
|
# Only INTEGER-pk tables (rowid aliases) carry a resume marker; count what is left past it.
|
|
marker = _meta_row(conn, f"fts_teardown_{tbl}_progress")
|
|
where, params = ("WHERE rowid > ?", (int(marker[0]),)) if marker is not None else ("", ())
|
|
table_count = progress.table_count or len(trash)
|
|
# COUNT before publishing the table: a failed COUNT must leave progress on the previous
|
|
# table so the retry runs this start step again instead of inheriting its totals.
|
|
total = int(conn.execute(f"SELECT COUNT(*) FROM {tbl} {where}", params).fetchone()[0]) if progress.counted else 0
|
|
progress.table_count, progress.total, progress.drained = table_count, total, 0
|
|
progress.table_index = max(1, table_count - len(trash) + 1)
|
|
self._fts_chunk_rows = self._FTS_REBUILD_CHUNK_ROWS
|
|
progress.table = tbl
|
|
|
|
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), 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
|
|
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 '\\')"
|
|
).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 _backfill_progress(self, phase: str) -> Optional[_BackfillProgress]:
|
|
"""Messages at or below the rebuild high water, and how many of them the resume marker already
|
|
covers (one scan per pass); None when the pass has nothing pending."""
|
|
st = self.fts_rebuild_status()
|
|
if st is None:
|
|
return None
|
|
with self._read_ctx() as conn:
|
|
# ``+id`` keeps the planner off the rowid range: counting from a covering index reads
|
|
# far fewer pages (1.97M messages, cold: 73 s on the table b-tree, 1.9 s here).
|
|
total, done = conn.execute("SELECT COUNT(*), COALESCE(SUM(id <= ?), 0) FROM messages WHERE +id <= ?",
|
|
(st["indexed"], st["total"])).fetchone()
|
|
return _BackfillProgress(phase=phase, total=int(total), indexed=int(done))
|
|
|
|
# A chunk that fails this many times in a row ends the run (each lock failure already waited out
|
|
# _execute_write's patience); a full disk or I/O error never heals by retrying.
|
|
_OPTIMIZE_CHUNK_ATTEMPTS = 3
|
|
_VACUUM_TICK_S = 1.0
|
|
# How often the side thread re-scans for another process holding state.db (~15 ms on Linux).
|
|
_HOLDER_SCAN_INTERVAL_S = 2.0
|
|
|
|
def _optimize_drive(self, step: Callable[[], bool], info: Optional[Callable[[], dict[str, Any]]],
|
|
emit: Callable[[dict[str, Any]], None], watch: _HolderWatch) -> None:
|
|
"""Run *step* to completion, emitting ``info()`` (plus ``paced``) after every chunk. Back to
|
|
back while no other process holds state.db (see the chunk engine note in hermes_state_fts);
|
|
while *watch* sees a holder, the duty cycle (short chunks + a pause) keeps its writes moving.
|
|
Chunk size adapts to its measured cost; a step may restart it (teardown on each new table)."""
|
|
self._fts_chunk_rows = self._FTS_REBUILD_CHUNK_ROWS
|
|
failures = 0
|
|
while True:
|
|
t0 = time.monotonic()
|
|
try:
|
|
more = step()
|
|
except sqlite3.OperationalError as exc:
|
|
failures += 1
|
|
if failures >= self._OPTIMIZE_CHUNK_ATTEMPTS:
|
|
raise
|
|
logger.warning("FTS optimize chunk failed (attempt %d, retrying): %s", failures, exc)
|
|
time.sleep(0.25 * failures)
|
|
continue
|
|
failures = 0
|
|
if not more:
|
|
return
|
|
cost, paced = time.monotonic() - t0, watch.held
|
|
if info is not None:
|
|
emit({**info(), "paced": paced})
|
|
target = self._FTS_REBUILD_PACED_TARGET_S if paced else self._FTS_REBUILD_CHUNK_TARGET_S
|
|
self._fts_chunk_rows = self._next_chunk_rows(self._fts_chunk_rows, cost, target)
|
|
if paced:
|
|
time.sleep(max(self._FTS_REBUILD_MIN_PAUSE, cost * self._FTS_REBUILD_DUTY_FACTOR))
|
|
|
|
def _optimize_vacuum(self, tick: Optional[Callable[[dict[str, Any]], None]] = None) -> 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). VACUUM is one statement that runs for minutes
|
|
on a big store, so *tick* gets a :class:`_VacuumProgress` payload at once and every
|
|
``_VACUUM_TICK_S`` from a side thread, which is stopped and joined before this returns."""
|
|
ticker, stop = None, threading.Event()
|
|
if tick is not None:
|
|
with self._read_ctx() as conn:
|
|
page_size, pages, free = (conn.execute(f"PRAGMA {p}").fetchone()[0]
|
|
for p in ("page_size", "page_count", "freelist_count"))
|
|
state = _VacuumProgress(size_bytes=pages * page_size, estimated_bytes=(pages - free) * page_size)
|
|
|
|
def _tick() -> None:
|
|
tick(state.info())
|
|
while not stop.wait(self._VACUUM_TICK_S):
|
|
tick(state.info())
|
|
|
|
# copy_context().run: the ticker keeps the caller's profile scope (HX012).
|
|
ticker = threading.Thread(target=contextvars.copy_context().run, args=(_tick,),
|
|
name="optimize-vacuum-ticker", daemon=True)
|
|
ticker.start()
|
|
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
|
|
finally:
|
|
stop.set()
|
|
if ticker is not None:
|
|
ticker.join()
|
|
# 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"
|
|
current = fts_layout_current(conn)
|
|
if not current and fts_layout_drift(conn):
|
|
return "layout_drift"
|
|
# Not current with no older layout left means a definition this code does not recognise (a newer
|
|
# release's): the work is done, but the stamp is not this release's to write.
|
|
if current:
|
|
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; a chunk that keeps failing raises (nothing is lost).
|
|
``progress_cb`` receives {"phase", "percent", "indexed", "total", "paced"} after every chunk
|
|
(``backfill`` counts messages; plus :meth:`_TeardownProgress.info` fields)
|
|
and every second of the VACUUM (:meth:`_VacuumProgress.info`). Whenever another process holds
|
|
state.db (re-scanned every ``_HOLDER_SCAN_INTERVAL_S``) chunks are paced so its writes get
|
|
through. 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")
|
|
|
|
emit = progress_cb or (lambda _info: None)
|
|
teardown = _TeardownProgress(counted=progress_cb is not None)
|
|
with _HolderWatch(self._foreign_state_db_holders, self._HOLDER_SCAN_INTERVAL_S) as watch:
|
|
# Phase 1: base backfill.
|
|
bf = self._backfill_progress("backfill") if progress_cb else None
|
|
if bf is not None:
|
|
emit({**bf.info(), "paced": watch.held})
|
|
step = partial(self.fts_rebuild_step, bf) if bf else self.fts_rebuild_step
|
|
self._optimize_drive(step, bf.info if bf else None, emit, watch)
|
|
# Phase 2: tear down the demoted legacy shadow tables in chunks.
|
|
self._optimize_drive(partial(self._fts_teardown_trash_step, teardown), teardown.info, emit, watch)
|
|
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, vacuum_skipped = None, False
|
|
if vacuum and self._foreign_state_db_holders():
|
|
# VACUUM holds the write lock for minutes; a live writer would time out mid-append.
|
|
logger.info("Skipping VACUUM after FTS optimize: another process is using %s", self.db_path)
|
|
vacuum_ok, vacuum_skipped = False, True
|
|
elif vacuum:
|
|
vacuum_ok = self._optimize_vacuum(progress_cb)
|
|
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({"phase": "done", "percent": 100, "indexed": 0, "total": 0})
|
|
logger.info("FTS storage optimization complete (layout v%d).", FTS_STORAGE_VERSION)
|
|
return {"ok": True, "vacuumed": vacuum_ok, **({"vacuum_skipped": "in_use"} if vacuum_skipped else {})}
|
|
|
|
# ── 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 _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()
|
|
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 = False
|
|
|
|
def _finalize_search_matches(
|
|
self, matches: list[dict[str, Any]], result_fields: Optional[Collection[str]] = None) -> list[dict[str, Any]]:
|
|
"""Attach neighboring messages in bounded batches, only when context is requested."""
|
|
if result_fields is None or "context" in result_fields:
|
|
for start in range(0, len(matches), 500):
|
|
batch = matches[start:start + 500]
|
|
contexts = {match["id"]: [] for match in batch}
|
|
try:
|
|
sql = _CONTEXT_WINDOW_SQL.format(ids=",".join("?" for _ in contexts))
|
|
with self._read_ctx() as conn:
|
|
rows = conn.execute(sql, list(contexts)).fetchall()
|
|
for row in rows:
|
|
contexts[row["match_id"]].append(row)
|
|
except Exception:
|
|
contexts = {}
|
|
for match in batch:
|
|
try:
|
|
match["context"] = [
|
|
{"role": row["role"], "content": _flatten_text(self._decode_content(row["content"]))[:200]}
|
|
for row in contexts.get(match["id"], [])]
|
|
except Exception:
|
|
match["context"] = []
|
|
# No route selects full content; the pop guards any future one that does.
|
|
for match in matches:
|
|
match.pop("content", None)
|
|
if result_fields is not None:
|
|
matches = [{field: match[field] for field in result_fields if field in match} for match in matches]
|
|
return matches
|
|
|
|
# ── search_messages ────────────────────────────────────────────────────
|
|
|
|
def search_messages(
|
|
self, query: str, source_filter: list[str] | None = None, exclude_sources: list[str] | None = None,
|
|
role_filter: list[str] | None = None, limit: int = 20, offset: int = 0, sort: str | None = None,
|
|
include_inactive: bool = False, fields: Optional[Collection[str]] = None,
|
|
after_ts: Optional[int] = None, before_ts: Optional[int] = None,
|
|
) -> list[dict[str, Any]]:
|
|
""":meth:`_search_messages_impl` plus one log line per slow search with the routing
|
|
path taken. Threshold HERMES_SEARCH_SLOW_MS (default 1000; 0 logs every call)."""
|
|
started = time.time()
|
|
rows = None
|
|
try:
|
|
rows = self._search_messages_impl(
|
|
query, source_filter=source_filter, exclude_sources=exclude_sources, role_filter=role_filter,
|
|
limit=limit, offset=offset, sort=sort, include_inactive=include_inactive, fields=fields,
|
|
after_ts=after_ts, before_ts=before_ts)
|
|
return rows
|
|
finally:
|
|
elapsed_ms = (time.time() - started) * 1000.0
|
|
if elapsed_ms >= _search_slow_ms():
|
|
logger.info("slow session search: path=%s elapsed=%.0fms rows=%s query=%r",
|
|
self._describe_search_path(query), elapsed_ms, len(rows) if rows is not None else "err",
|
|
query[: 200])
|
|
|
|
def _search_messages_impl(
|
|
self, query: str, source_filter: list[str] | None = None, exclude_sources: list[str] | None = None,
|
|
role_filter: list[str] | None = None, limit: int = 20, offset: int = 0, sort: str | None = None,
|
|
include_inactive: bool = False, fields: Optional[Collection[str]] = None,
|
|
after_ts: Optional[int] = None, before_ts: Optional[int] = None,
|
|
) -> list[dict[str, Any]]:
|
|
"""FTS5 search across session messages (keywords, ``"phrases"``, AND/OR/NOT, ``prefix*``).
|
|
Returns snippet + session metadata + 1-message context per hit; ``fields`` selects a
|
|
projection. ``sort``: None = BM25 rank; "newest"/"oldest" = timestamp then rank (the
|
|
CJK LIKE fallback ignores it). Rewound rows (``active=0, compacted=0``) are excluded
|
|
by default; compaction-archived rows ARE included; ``include_inactive`` = every row.
|
|
``after_ts``/``before_ts`` bound ``sessions.started_at`` on every route (FTS5, CJK,
|
|
trigram, LIKE fallback, unindexed-gap supplement)."""
|
|
result_fields = self._search_message_fields(fields)
|
|
if not query or not query.strip():
|
|
return []
|
|
query = self._sanitize_fts5_query(query)
|
|
if not query:
|
|
return []
|
|
filters = dict(include_inactive=include_inactive, source_filter=source_filter,
|
|
exclude_sources=exclude_sources, role_filter=role_filter,
|
|
after_ts=after_ts, before_ts=before_ts)
|
|
# New oversized tool results index only a bounded prefix; an explicit tool-role search is the
|
|
# opt-in full-body path and scans canonical rows via LIKE.
|
|
if role_filter and "tool" in role_filter:
|
|
matches = self._search_messages_like_fallback(query, limit=limit, offset=offset, sort=sort, **filters)
|
|
return self._finalize_search_matches(matches, result_fields=result_fields)
|
|
self._refresh_fts_stale_state()
|
|
if self._fts_stale:
|
|
matches = self._search_messages_like_fallback(query, limit=limit, offset=offset, sort=sort, **filters)
|
|
return self._finalize_search_matches(matches, result_fields=result_fields)
|
|
if not self._fts_enabled:
|
|
return []
|
|
|
|
order_by_sql = _FTS_ORDER_BY.get(sort.strip().lower() if isinstance(sort, str) else None, "ORDER BY rank")
|
|
route = dict(order_by_sql=order_by_sql, limit=limit, offset=offset, **filters)
|
|
# Tool rows and FTS_TRIGRAM_EXCLUDED_SOURCES sessions are excluded from the trigram
|
|
# index (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
|
|
# trigram index (>=3-char tokens). Gated on a miss so hits keep their ranking ("cat"
|
|
# may then match "concatenate"). Skipped for role='tool' (trigram excludes 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())
|
|
if 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). Trigram serves >=3 CJK chars per token unless the query wants rows it
|
|
excludes (role='tool', cron/subagent sources); everything else is 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._trigram_route_ok(raw_query) and not wants_unindexed_rows:
|
|
matches = self._match_rows("messages_fts_trigram", match_query, fail_open="Trigram", **route)
|
|
if matches is not None:
|
|
return matches
|
|
non_op_tokens = _non_operator_tokens(raw_query) or [raw_query]
|
|
like_params: list = [p for tok in non_op_tokens for p in _like_params(tok)]
|
|
like_where = [f"({' OR '.join([_LIKE_ANY_COLUMN_SQL] * len(non_op_tokens))})"]
|
|
filters = {k: route[k] for k in ("include_inactive", "source_filter", "exclude_sources", "role_filter",
|
|
"after_ts", "before_ts")}
|
|
_search_filter_clauses(like_where, like_params, **filters)
|
|
# instr() for the snippet uses the first search token.
|
|
return self._like_rows(like_where, [non_op_tokens[0], *like_params, route["limit"], route["offset"]],
|
|
order_by="ORDER BY m.timestamp DESC", limit_sql="LIMIT ? OFFSET ?")
|
|
|
|
def _search_unindexed_gap(self, fts_query: str, limit: int, **filters) -> list[dict[str, Any]]:
|
|
"""LIKE-scan ids in (fts_rebuild_progress, fts_rebuild_high_water] — rows the deferred
|
|
rebuild hasn't indexed yet. The FTS query degrades to AND-joined substring terms
|
|
(quoted phrases kept whole): recall-over-precision mid-rebuild."""
|
|
status = self.fts_rebuild_status()
|
|
if status is None or limit <= 0:
|
|
return []
|
|
terms = [tok for tok in (t.strip('"').strip("*").strip() for t in _LIKE_TOKEN_RE.findall(fts_query))
|
|
if tok and tok.upper() not in _LIKE_SKIP_TOKENS]
|
|
if not terms:
|
|
return []
|
|
where = ["m.id > ? AND m.id <= ?", *([_LIKE_ANY_COLUMN_SQL] * len(terms))]
|
|
params: list = [status["indexed"], status["total"], *(p for term in terms for p in _like_params(term))]
|
|
_search_filter_clauses(where, params, **filters)
|
|
return self._like_rows(where, [terms[0], *params, limit], order_by="ORDER BY m.timestamp DESC",
|
|
limit_sql="LIMIT ?")
|
|
|
|
def search_sessions_by_id(
|
|
self, query: str, limit: int = 20, include_archived: bool = True, source: str | None = None,
|
|
sources: list[str] | None = None, exclude_sources: list[str] | None = None) -> list[dict[str, Any]]:
|
|
"""Search surfaced sessions by exact/prefix/substring session id. Also matches
|
|
``_lineage_root_id`` so an old compression root id resolves to the live continuation."""
|
|
needle = (query or "").strip().lower()
|
|
if not needle or limit <= 0:
|
|
return []
|
|
# list_sessions_rich pushes the id LIKE filter (own id + forward compression
|
|
# chain) into SQL; over-fetch so the in-Python ranking has candidates.
|
|
candidates = self.list_sessions_rich(
|
|
source=source, sources=sources, exclude_sources=exclude_sources, limit=max(limit * 4, limit),
|
|
offset=0, include_archived=include_archived, order_by_last_active=True, id_query=needle)
|
|
|
|
def score(row: dict[str, Any]) -> int:
|
|
normalized = [v.lower() for v in (str(row.get("id") or ""), str(row.get("_lineage_root_id") or "")) if v]
|
|
if any(value == needle for value in normalized):
|
|
return 0
|
|
return 1 if any(value.startswith(needle) for value in normalized) else 2
|
|
ranked = sorted(enumerate(candidates), key=lambda item: (score(item[1]), item[0]))
|
|
return [row for _, row in ranked[:limit]]
|
|
|
|
# ── FTS maintenance commands ───────────────────────────────────────────
|
|
|
|
def _fts_table_exists(self, name: str) -> bool:
|
|
"""True if an FTS5 virtual table is queryable ("no such table" and "vtable
|
|
constructor failed" — missing tokenizer / mid-teardown — both count as not)."""
|
|
try:
|
|
self._conn.execute(f"SELECT 1 FROM {name} LIMIT 0")
|
|
return True
|
|
except sqlite3.DatabaseError:
|
|
return False
|
|
|
|
def _present_fts_tables(self) -> list[str]:
|
|
"""Queryable FTS tables (caller holds ``self._lock``)."""
|
|
return [tbl for tbl in self._FTS_TABLES if self._fts_table_exists(tbl)]
|
|
|
|
def optimize_fts(self) -> int:
|
|
"""Merge fragmented FTS5 segments into one per index (``'optimize'``). Pure
|
|
maintenance: changes neither results nor ``snippet()`` output, only layout and
|
|
speed; VACUUM then returns the freed pages. Returns the number optimized. A quarantined
|
|
handle never issues ``'optimize'``: it rewrites index segments in place and would compound
|
|
structural damage (or a split WAL generation) instead of leaving it diagnosable."""
|
|
self._raise_if_db_corrupt()
|
|
optimized = 0
|
|
with self._lock:
|
|
self._raise_if_db_replaced()
|
|
if self._conn is None:
|
|
self._reopen_after_close_locked(context="write")
|
|
for tbl in self._present_fts_tables():
|
|
try:
|
|
self._conn.execute(f"INSERT INTO {tbl}({tbl}) VALUES('optimize')")
|
|
optimized += 1
|
|
except sqlite3.DatabaseError as exc: # SQLITE_CORRUPT is not an OperationalError
|
|
logger.warning("FTS optimize failed for %s: %s", tbl, exc)
|
|
return optimized
|
|
|
|
def rebuild_fts(self) -> int:
|
|
"""Rebuild FTS5 indexes from ``messages`` (``'rebuild'``) — the recovery for a corrupt index
|
|
that rejects writes while reads succeed. Two processes rebuilding one state.db
|
|
concurrently corrupted production DBs, so this admits through
|
|
``fts_rebuild_admission`` and FAILS CLOSED, returning 0 on deferral (callers treat 0
|
|
as "no progress" and use the stale-FTS breadcrumb path). Returns indexes rebuilt.
|
|
|
|
Uses the FTS5 ``'rebuild'`` command, which rewrites the internal b-tree segments from the content
|
|
rows. Unlike ``optimize_fts`` (which merges existing segments), ``rebuild`` discards and recreates
|
|
the index data entirely — the more destructive of the two, so it is quarantined the same way. See
|
|
#50502.
|
|
A full structural rebuild must never run concurrently in two processes sharing one state.db — that
|
|
interleaving has structurally corrupted the database in production (PR #93200) — so this admits
|
|
through the cross-process ``fts_rebuild_admission`` authority and FAILS CLOSED: if another process
|
|
holds the rebuild lock beyond the bounded wait, this call defers (returns 0) rather than racing it.
|
|
Callers already treat 0 as "rebuild made no progress" and fall back to the stale-FTS breadcrumb
|
|
path, which retries in-process from the gateway housekeeping tick (``retry_deferred_fts_recovery``)
|
|
and at next startup.
|
|
"""
|
|
self._raise_if_db_corrupt()
|
|
rebuilt = 0
|
|
with fts_rebuild_admission(self.db_path) as admitted:
|
|
if not admitted:
|
|
logger.warning(
|
|
"Deferred in-place FTS rebuild: another process holds the rebuild authority for this state.db.")
|
|
return 0
|
|
with self._lock:
|
|
self._raise_if_db_replaced()
|
|
if self._conn is None:
|
|
self._reopen_after_close_locked(context="write")
|
|
for tbl in self._present_fts_tables():
|
|
try:
|
|
self._conn.execute(f"INSERT INTO {tbl}({tbl}) VALUES('rebuild')")
|
|
self._conn.commit()
|
|
rebuilt += 1
|
|
except sqlite3.DatabaseError as exc: # SQLITE_CORRUPT is not an OperationalError (#133375)
|
|
self._conn.rollback()
|
|
if is_malformed_db_error(exc):
|
|
logger.error(
|
|
"FTS rebuild failed with a corruption-class error for %s: %s; "
|
|
"run 'hermes sessions repair' (or 'hermes doctor --fix') to rebuild the index offline",
|
|
tbl, exc)
|
|
else:
|
|
logger.warning("FTS rebuild failed for %s: %s", tbl, exc)
|
|
return rebuilt
|
|
|
|
def _merge_fts_incrementally(self, *, max_pages: int, max_commands: Optional[int] = None) -> int:
|
|
"""Run bounded FTS5 ``'merge'`` commands against each present index. A positive merge rank
|
|
stops after ~that many output pages, so each command holds the write lock for
|
|
milliseconds regardless of index size (``'optimize'`` takes 9-18 s per index on a
|
|
10 GB DB). ``usermerge`` is lowered to its minimum of 2 (persisted in ``%_config``,
|
|
once per instance) so a merge acts on ANY level with >= 2 segments; at the default
|
|
4 a fragmented index cannot converge. Up to *max_commands* per index, stopping on
|
|
the no-progress signal ``total_changes`` delta < 2 (the INSERT itself is 1). Each
|
|
command is its own implicit transaction, so processes interleave mid-pass. Missing
|
|
tables are skipped (optimize_fts_storage drops + backfills them live); other SQLite
|
|
errors propagate. Returns commands executed."""
|
|
_positive_int("max_pages", max_pages)
|
|
if max_commands is None:
|
|
max_commands = self._FTS_MERGE_COMMANDS_PER_PASS
|
|
_positive_int("max_commands", max_commands)
|
|
executed = 0
|
|
with self._lock:
|
|
for tbl in self._present_fts_tables():
|
|
if not self._fts_usermerge_floor_applied:
|
|
self._conn.execute(f"INSERT INTO {tbl}({tbl}, rank) VALUES('usermerge', 2)")
|
|
for _ in range(max_commands):
|
|
before = self._conn.total_changes
|
|
self._conn.execute(f"INSERT INTO {tbl}({tbl}, rank) VALUES('merge', ?)", (max_pages,))
|
|
executed += 1
|
|
if self._conn.total_changes - before < 2:
|
|
break
|
|
self._fts_usermerge_floor_applied = True
|
|
return executed
|