diff --git a/hermes_cli/config.py b/hermes_cli/config.py index ab93baeff25a..7a527b75f53c 100644 --- a/hermes_cli/config.py +++ b/hermes_cli/config.py @@ -3216,6 +3216,23 @@ DEFAULT_CONFIG = { # GBs of disk on heavy users. Opt in only if you have an external # tool that consumes the JSON files directly. "write_json_snapshots": False, + # Search-index (FTS) storage optimization — the compact v23 layout + # that drops duplicate content copies and stops trigram-indexing tool + # output (typically reclaims ~60%+ of state.db on heavy users). It is + # OPT-IN: existing databases keep their working legacy index until the + # user runs `hermes sessions optimize-storage`, because the rebuild is + # disk-heavy and long on large DBs (see that command's disk preflight). + # + # "advise" (default): `hermes update` prints a one-line notice with + # the reclaimable size and the command, when a legacy index is + # detected. Nothing is changed automatically. + # "require": the notice is shown as a REQUIRED upgrade (firmer copy), + # and future tooling may gate on it. Flip this default in a future + # release when we're ready to make the v23 layout mandatory — the + # command, progress bar, and resumability are already in place, so + # enforcement is a copy/gating change, not new migration code. + # "off": suppress the notice entirely. + "fts_optimize_notice": "advise", }, # Contextual first-touch onboarding hints (see agent/onboarding.py). diff --git a/hermes_cli/main.py b/hermes_cli/main.py index 15003ca6da85..72d400ff193f 100644 --- a/hermes_cli/main.py +++ b/hermes_cli/main.py @@ -6347,6 +6347,116 @@ def _print_curator_first_run_notice() -> None: ) +def _print_fts_optimize_available_notice() -> None: + """Advertise the opt-in v23 search-index optimization after `hermes update`. + + Only fires when the current profile's state.db is still on the legacy + (pre-v23) inline FTS layout. Leads with the reclaimable-space figure and + points at the exact command. Honors ``sessions.fts_optimize_notice``: + ``advise`` (default) prints an advisory notice, ``require`` prints a + firmer required-upgrade notice, ``off`` suppresses it. Silent for + fresh/already-optimized installs. + """ + mode = "advise" + try: + from hermes_cli.config import load_config + + mode = str( + ((load_config() or {}).get("sessions") or {}).get( + "fts_optimize_notice", "advise" + ) + ).strip().lower() + except Exception: + mode = "advise" + if mode == "off": + return + + try: + from hermes_constants import get_hermes_home + from hermes_state import SessionDB + except Exception: + return + db_path = get_hermes_home() / "state.db" + if not db_path.exists(): + return + try: + size_gb = db_path.stat().st_size / (1024 ** 3) + except OSError: + return + # Skip the notice for trivially small DBs — the win isn't worth the nag. + if size_gb < 0.5: + return + db = None + interrupted = False + try: + db = SessionDB(db_path=db_path, read_only=True) + # read_only opens skip schema init, so probe the layout directly. + row = db._conn.execute( + "SELECT sql FROM sqlite_master " + "WHERE type = 'table' AND name = 'messages_fts'" + ).fetchone() + # An interrupted `optimize-storage` run: the table is already the + # v23 shape, but backfill markers / demoted trash tables remain. + # Offer the command again — re-running resumes and finishes it. + interrupted = bool( + db._conn.execute( + "SELECT 1 FROM state_meta " + "WHERE key = 'fts_rebuild_high_water' LIMIT 1" + ).fetchone() + or db._conn.execute( + "SELECT 1 FROM sqlite_master WHERE type = 'table' " + "AND name LIKE 'fts\\_v22\\_trash\\_%' ESCAPE '\\' LIMIT 1" + ).fetchone() + ) + except Exception: + return + finally: + if db is not None: + try: + db.close() + except Exception: + pass + sql = (row[0] if row else "") or "" + if not sql or ("tool_name" in sql and not interrupted): + # v23 layout already present (fresh/optimized) — nothing to offer. + return + + if interrupted: + print() + print("◆ Session database optimization incomplete") + print( + " A previous `hermes sessions optimize-storage` run was " + "interrupted. Search still works; re-run the command to resume " + "and finish reclaiming disk:" + ) + print(" hermes sessions optimize-storage") + return + + # Concrete size framing — lead with the savings the user cares about. + est_reclaim = size_gb * 0.6 + print() + if mode == "require": + print("◆ Session database upgrade required") + print( + f" Your search index uses the OLD storage layout and should be " + f"upgraded. The new layout typically frees ~60% of state.db " + f"(≈{est_reclaim:.1f} GB of your current {size_gb:.1f} GB) and is " + f"required for continued optimal operation." + ) + else: + print("◆ Reclaim ~60% of your session database disk") + print( + f" Your search index uses the old storage layout. Upgrading it " + f"typically frees ~60% of state.db — about {est_reclaim:.1f} GB " + f"of your current {size_gb:.1f} GB." + ) + print(" Run when convenient: hermes sessions optimize-storage") + print( + " It runs in the foreground with a progress bar, is safe to " + "interrupt/re-run, and never changes your conversations." + ) + + def _print_curator_recent_run_notice() -> None: """Print the most recent curator run summary, exactly once. @@ -10951,6 +11061,17 @@ def _cmd_update_impl(args, gateway_mode: bool): else: print("✓ Update complete!") + # Search-index optimization notice (v23). Existing installs keep their + # working search index untouched on update; the compact v23 layout — + # which reclaims a large fraction of state.db on heavy users — is + # opt-in. Surface it here (the moment the user is already thinking + # about their install) with the exact command and the concrete size + # win. Show-once-ish: only when a legacy index is actually present. + try: + _print_fts_optimize_available_notice() + except Exception as e: + logger.debug("FTS optimize notice failed: %s", e) + # Curator first-run heads-up. Only prints when curator is enabled AND # has never run — i.e. the window where the ticker would otherwise # have fired against a fresh skill library. Kept silent on steady @@ -14630,6 +14751,33 @@ def main(): help="Reclaim disk space: merge FTS5 segments + VACUUM (no data change)", ) + sessions_optimize_storage = sessions_subparsers.add_parser( + "optimize-storage", + help="Migrate the search index to the compact v23 layout (reclaims disk on large DBs)", + description=( + "Rebuild the full-text search index in the compact v23 " + "external-content layout. On large databases this reclaims a " + "large fraction of state.db (the old layout stored duplicate " + "copies of every message and indexed tool output). Runs " + "foreground with a progress bar, throttles so a running gateway " + "stays responsive, and VACUUMs at the end. Safe to interrupt and " + "re-run — it resumes where it left off. No conversation data is " + "changed; only the search index is rebuilt." + ), + ) + sessions_optimize_storage.add_argument( + "--no-vacuum", + action="store_true", + default=False, + help="Skip the final VACUUM (index is rebuilt but freed pages aren't returned to the OS until a later VACUUM)", + ) + sessions_optimize_storage.add_argument( + "--yes", "-y", + action="store_true", + default=False, + help="Skip the disk-space confirmation prompt", + ) + sessions_repair = sessions_subparsers.add_parser( "repair", help="Repair a malformed state.db schema so hidden sessions reappear", @@ -15412,6 +15560,96 @@ def main(): f"(reclaimed {saved:.1f} MB)" ) + elif action == "optimize-storage": + db_path = db.db_path + if not db.fts_optimize_available(): + print("Search index is already on the compact layout — nothing to do.") + db.close() + return + + before_bytes = os.path.getsize(db_path) if db_path.exists() else 0 + before_mb = before_bytes / (1024 * 1024) + + # Disk preflight: the rebuild adds the new index before the old is + # torn down, and the final VACUUM needs a full second copy of the + # file. Require headroom ≈ current file size to finish cleanly. + do_vacuum = not getattr(args, "no_vacuum", False) + try: + import shutil as _shutil + free_bytes = _shutil.disk_usage(db_path.parent).free + except Exception: + free_bytes = None + need_bytes = before_bytes if do_vacuum else int(before_bytes * 0.3) + print(f"Search-index optimization for {db_path}") + print(f" Current database size: {before_mb:.1f} MB") + if free_bytes is not None: + print(f" Free disk: {free_bytes / (1024*1024):.0f} MB " + f"(need ~{need_bytes / (1024*1024):.0f} MB to complete" + f"{' incl. VACUUM' if do_vacuum else ''})") + if free_bytes < need_bytes: + print() + print("⚠ Not enough free disk to complete safely. Free up " + "space, or run with --no-vacuum (rebuilds the index " + "but doesn't reclaim space until a later VACUUM).") + db.close() + return + if before_mb > 500: + print(" This may take a while on a large database. It runs in " + "the foreground with progress below; safe to Ctrl-C and " + "re-run (it resumes).") + if not getattr(args, "yes", False): + try: + resp = input("Proceed? [y/N] ").strip().lower() + except EOFError: + resp = "" + if resp not in ("y", "yes"): + print("Cancelled.") + db.close() + return + + _last = {"phase": None} + + def _progress(info): + phase = info.get("phase") + pct = info.get("percent", 0) + if phase == "backfill": + print(f"\r Rebuilding index: {pct:3d}% " + f"({info.get('indexed',0):,}/{info.get('total',0):,})", + end="", flush=True) + elif phase != _last["phase"]: + label = {"teardown": "Reclaiming old index", + "vacuum": "Compacting database (VACUUM)", + "done": "Done"}.get(phase, phase) + print(f"\n {label}…", flush=True) + _last["phase"] = phase + + print("Optimizing search-index storage…") + try: + result = db.optimize_fts_storage( + progress_cb=_progress, vacuum=do_vacuum + ) + except Exception as e: + print(f"\nError: optimization failed: {e}") + print("No data was lost. Re-run to resume.") + db.close() + return + if not result.get("ok"): + print(f"\nCould not optimize: {result.get('reason', 'unknown')}") + db.close() + return + after_mb = ( + os.path.getsize(db_path) / (1024 * 1024) if db_path.exists() else 0.0 + ) + saved = before_mb - after_mb + print(f"\n✓ Search index optimized.") + print( + f" Database size: {before_mb:.1f} MB -> {after_mb:.1f} MB " + f"(reclaimed {saved:.1f} MB)" + ) + if result.get("vacuumed") is False: + print(" (VACUUM was skipped or failed — run " + "`hermes sessions optimize` later to reclaim freed space.)") + elif action == "stats": total = db.session_count() msgs = db.message_count() diff --git a/hermes_cli/web_server.py b/hermes_cli/web_server.py index 35745cc3a07d..b14fe66196e7 100644 --- a/hermes_cli/web_server.py +++ b/hermes_cli/web_server.py @@ -3181,6 +3181,27 @@ async def get_status(profile: Optional[str] = None): else "degraded" ) + # Deferred FTS rebuild progress (schema v23): lets the desktop / + # dashboard render a "search index rebuilding: N%" indicator instead + # of users wondering why old-message search is slower after an + # update. None/absent when no rebuild is pending (the common case). + # Read-only probe, never blocks startup, never raises. + try: + from hermes_state import SessionDB as _SDB + from hermes_constants import get_hermes_home as _ghh + + _db_path = _ghh() / "state.db" + if _db_path.exists(): + _sdb = _SDB(db_path=_db_path, read_only=True) + try: + _rebuild = _sdb.fts_rebuild_status() + finally: + _sdb.close() + if _rebuild is not None: + status["fts_rebuild"] = _rebuild + except Exception: + pass + # Profile + gateway topology: which profiles exist, whether one # multiplexed gateway or several per-profile gateways serve them, and # (gated) which host ports the live gateways' port-binding platforms diff --git a/hermes_state.py b/hermes_state.py index ef84435ea0de..ce0292708a16 100644 --- a/hermes_state.py +++ b/hermes_state.py @@ -152,7 +152,17 @@ T = TypeVar("T") DEFAULT_DB_PATH = get_hermes_home() / "state.db" -SCHEMA_VERSION = 22 +SCHEMA_VERSION = 23 + +# FTS storage-layout version, tracked INDEPENDENTLY of SCHEMA_VERSION in the +# state_meta key ``fts_storage_version``. The main schema version advances +# freely on open (so future migrations always land); the FTS *layout* only +# reaches the current version when a DB is either born fresh or explicitly +# optimized via ``hermes sessions optimize-storage``. A legacy DB sits at +# layout 0 (marker absent) with a working inline index until the user opts in. +# 1 = v23 external-content layout (content/tool_name/tool_calls, +# tool-row-excluded trigram) +FTS_STORAGE_VERSION = 1 # Cap on user-controlled FTS5 query input before regex/sanitizer processing. # Search queries do not need to be arbitrarily large, and bounding them keeps @@ -1005,7 +1015,153 @@ CREATE INDEX IF NOT EXISTS idx_sessions_handoff_state ON sessions(handoff_state, started_at); """ +# ── Deferred FTS rebuild bookkeeping (schema v23) ── +# While a background index rebuild is pending, two state_meta keys define +# which message rows are currently IN the FTS indexes: +# +# fts_rebuild_high_water H — MAX(messages.id) at the moment the old +# indexes were dropped +# fts_rebuild_progress P — highest id the chunked backfill has indexed +# +# A row is indexed iff id <= P (backfilled) OR id > H (inserted after +# the drop; ids are AUTOINCREMENT so new rows are always > H and the insert +# triggers index them live). Rows in (P, H] are not yet indexed. +# +# Every trigger below gates on that same predicate: firing an FTS5 +# external-content 'delete' for a row that is NOT in the index corrupts the +# index, and skipping it for a row that IS indexed leaves a stale entry. +# When no rebuild is pending both keys are absent and COALESCE turns the +# predicate into a tautology (id > -1 OR id <= -1), i.e. normal operation. +# The two state_meta PK probes per write are negligible next to the FTS +# insert itself. FTS_SQL = """ +CREATE VIRTUAL TABLE IF NOT EXISTS messages_fts USING fts5( + content, + tool_name, + tool_calls, + content='messages', + content_rowid='id' +); + +CREATE TRIGGER IF NOT EXISTS messages_fts_insert AFTER INSERT ON messages +WHEN (new.id > COALESCE((SELECT CAST(value AS INTEGER) FROM state_meta + WHERE key = 'fts_rebuild_high_water'), -1) + OR new.id <= COALESCE((SELECT CAST(value AS INTEGER) FROM state_meta + WHERE key = 'fts_rebuild_progress'), -1)) +BEGIN + INSERT INTO messages_fts(rowid, content, tool_name, tool_calls) + VALUES (new.id, new.content, new.tool_name, new.tool_calls); +END; + +CREATE TRIGGER IF NOT EXISTS messages_fts_delete AFTER DELETE ON messages +WHEN (old.id > COALESCE((SELECT CAST(value AS INTEGER) FROM state_meta + WHERE key = 'fts_rebuild_high_water'), -1) + OR old.id <= COALESCE((SELECT CAST(value AS INTEGER) FROM state_meta + WHERE key = 'fts_rebuild_progress'), -1)) +BEGIN + INSERT INTO messages_fts(messages_fts, rowid, content, tool_name, tool_calls) + VALUES ('delete', old.id, old.content, old.tool_name, old.tool_calls); +END; + +CREATE TRIGGER IF NOT EXISTS messages_fts_update AFTER UPDATE ON messages +WHEN (old.content IS NOT new.content + OR old.tool_name IS NOT new.tool_name + OR old.tool_calls IS NOT new.tool_calls) + AND (old.id > COALESCE((SELECT CAST(value AS INTEGER) FROM state_meta + WHERE key = 'fts_rebuild_high_water'), -1) + OR old.id <= COALESCE((SELECT CAST(value AS INTEGER) FROM state_meta + WHERE key = 'fts_rebuild_progress'), -1)) +BEGIN + INSERT INTO messages_fts(messages_fts, rowid, content, tool_name, tool_calls) + VALUES ('delete', old.id, old.content, old.tool_name, old.tool_calls); + INSERT INTO messages_fts(rowid, content, tool_name, tool_calls) + VALUES (new.id, new.content, new.tool_name, new.tool_calls); +END; +""" + +# Trigram FTS5 table for CJK substring search. The default unicode61 +# tokenizer splits CJK characters into individual tokens, breaking phrase +# matching. The trigram tokenizer creates overlapping 3-byte sequences so +# substring queries work natively for any script (CJK, Thai, etc.). +# +# The trigram index is the most expensive index in state.db (~2.6x the size +# of the text it covers), and ``role='tool'`` rows are ~90% of message bytes +# while being almost entirely machine noise (base64 payloads, file dumps, +# delegation transcripts). The index therefore reads through +# ``messages_fts_trigram_src``, a view that excludes tool rows — they stay +# fully stored in ``messages`` and fully searchable via the standard +# ``messages_fts`` index; they just don't get trigram (CJK substring) +# treatment. ``search_messages`` routes CJK queries that filter on +# ``role='tool'`` to the LIKE fallback for the same reason. +FTS_TRIGRAM_SQL = """ +CREATE VIEW IF NOT EXISTS messages_fts_trigram_src AS + SELECT id, role, content, tool_name, tool_calls + FROM messages + WHERE role <> 'tool'; + +CREATE VIRTUAL TABLE IF NOT EXISTS messages_fts_trigram USING fts5( + content, + tool_name, + tool_calls, + content='messages_fts_trigram_src', + content_rowid='id', + tokenize='trigram' +); + +CREATE TRIGGER IF NOT EXISTS messages_fts_trigram_insert AFTER INSERT ON messages +WHEN new.role <> 'tool' + AND (new.id > COALESCE((SELECT CAST(value AS INTEGER) FROM state_meta + WHERE key = 'fts_rebuild_high_water'), -1) + OR new.id <= COALESCE((SELECT CAST(value AS INTEGER) FROM state_meta + WHERE key = 'fts_rebuild_progress'), -1)) +BEGIN + INSERT INTO messages_fts_trigram(rowid, content, tool_name, tool_calls) + VALUES (new.id, new.content, new.tool_name, new.tool_calls); +END; + +CREATE TRIGGER IF NOT EXISTS messages_fts_trigram_delete AFTER DELETE ON messages +WHEN old.role <> 'tool' + AND (old.id > COALESCE((SELECT CAST(value AS INTEGER) FROM state_meta + WHERE key = 'fts_rebuild_high_water'), -1) + OR old.id <= COALESCE((SELECT CAST(value AS INTEGER) FROM state_meta + WHERE key = 'fts_rebuild_progress'), -1)) +BEGIN + INSERT INTO messages_fts_trigram(messages_fts_trigram, rowid, content, tool_name, tool_calls) + VALUES ('delete', old.id, old.content, old.tool_name, old.tool_calls); +END; + +CREATE TRIGGER IF NOT EXISTS messages_fts_trigram_update AFTER UPDATE ON messages +WHEN (old.content IS NOT new.content + OR old.tool_name IS NOT new.tool_name + OR old.tool_calls IS NOT new.tool_calls + OR old.role IS NOT new.role) + AND (old.id > COALESCE((SELECT CAST(value AS INTEGER) FROM state_meta + WHERE key = 'fts_rebuild_high_water'), -1) + OR old.id <= COALESCE((SELECT CAST(value AS INTEGER) FROM state_meta + WHERE key = 'fts_rebuild_progress'), -1)) +BEGIN + INSERT INTO messages_fts_trigram(messages_fts_trigram, rowid, content, tool_name, tool_calls) + SELECT 'delete', old.id, old.content, old.tool_name, old.tool_calls + WHERE old.role <> 'tool'; + INSERT INTO messages_fts_trigram(rowid, content, tool_name, tool_calls) + SELECT new.id, new.content, new.tool_name, new.tool_calls + WHERE new.role <> 'tool'; +END; +""" + + +# ── Legacy (v22 / inline-content) FTS DDL ────────────────────────────── +# Used ONLY to keep an existing pre-v23 install's search working and its +# triggers repairable UNTIL the user opts into `hermes db optimize`. This is +# the exact inline shape v11..v22 shipped: each virtual table stores its own +# copy of ``content || tool_name || tool_calls`` and the trigram table indexes +# every row (including role='tool'). We never CREATE these on a fresh install — +# fresh installs are born on the v23 external-content schema above. These +# constants exist so a legacy DB is never accidentally handed the v23 DDL +# (which would create the external-content trigram source VIEW and leave the +# DB in a mixed, broken state). `optimize_fts_storage()` is what migrates a +# legacy DB to the v23 shape. +LEGACY_FTS_SQL = """ CREATE VIRTUAL TABLE IF NOT EXISTS messages_fts USING fts5( content ); @@ -1030,11 +1186,7 @@ CREATE TRIGGER IF NOT EXISTS messages_fts_update AFTER UPDATE ON messages BEGIN END; """ -# Trigram FTS5 table for CJK substring search. The default unicode61 -# tokenizer splits CJK characters into individual tokens, breaking phrase -# matching. The trigram tokenizer creates overlapping 3-byte sequences so -# substring queries work natively for any script (CJK, Thai, etc.). -FTS_TRIGRAM_SQL = """ +LEGACY_FTS_TRIGRAM_SQL = """ CREATE VIRTUAL TABLE IF NOT EXISTS messages_fts_trigram USING fts5( content, tokenize='trigram' @@ -1184,6 +1336,14 @@ class SessionDB: if not report.get("repaired"): raise _connect_and_init() + + # NOTE: the v23 FTS optimization is OPT-IN (`hermes db optimize`), + # never auto-started on open. Legacy installs keep their working + # v22 inline FTS untouched here; only the explicit foreground + # command demotes + rebuilds. This avoids a background worker + # racing session lifecycle and the surprise disk/latency cost on + # an unattended open. (An interrupted optimize resumes when the + # user re-runs the command.) except Exception as exc: # Capture the cause so /resume and friends can surface WHY the # session DB is unavailable instead of a bare "Session database @@ -1219,6 +1379,32 @@ class SessionDB: """True when only the trigram tokenizer is missing (FTS5 itself works).""" return "no such tokenizer: trigram" in str(exc).lower() + @staticmethod + def _db_has_legacy_inline_fts(cursor: sqlite3.Cursor) -> bool: + """True when messages_fts exists in ANY pre-v23 shape. + + v23's messages_fts is external-content over THREE real columns + (content, tool_name, tool_calls). Every pre-v23 shape lacks the + tool_name/tool_calls columns — whether the old inline single-column + form (v11..v22) or the even older external-content single-column form + (v10-era, pre-#16751). We therefore detect "needs optimize" as "the + stored CREATE lacks the tool_name column", which is the precise v23 + marker and correctly catches BOTH legacy variants. + + Returns False when messages_fts doesn't exist yet (fresh DB mid-init): + the post-migration FTS setup block will create it in the v23 shape. + """ + row = cursor.execute( + "SELECT sql FROM sqlite_master " + "WHERE type = 'table' AND name = 'messages_fts'" + ).fetchone() + if row is None: + return False + sql = (row[0] if not isinstance(row, sqlite3.Row) else row["sql"]) or "" + # The v23 table declares tool_name/tool_calls columns. Their absence + # means a legacy shape that doesn't index tool metadata → optimize. + return "tool_name" not in sql + def _warn_trigram_unavailable(self, exc: sqlite3.OperationalError) -> None: """Log once that the trigram tokenizer is missing; base FTS5 stays enabled.""" if getattr(self, "_trigram_unavailable_warned", False): @@ -1282,6 +1468,36 @@ class SessionDB: *, include_trigram: bool = True, ) -> None: + # Both FTS tables are external-content (v23+): the special 'rebuild' + # command wipes the inverted index and repopulates it from the + # content source (messages for the standard index, the tool-row- + # excluding messages_fts_trigram_src view for the trigram index). + cursor.execute("INSERT INTO messages_fts(messages_fts) VALUES('rebuild')") + if include_trigram: + cursor.execute( + "INSERT INTO messages_fts_trigram(messages_fts_trigram) VALUES('rebuild')" + ) + # 'rebuild' indexes EVERY row, so any deferred-backfill markers are + # now satisfied — clear them, otherwise the background worker would + # re-insert rows the rebuild already covered (duplicate entries). + cursor.execute( + "DELETE FROM state_meta WHERE key IN " + "('fts_rebuild_high_water', 'fts_rebuild_progress')" + ) + + @staticmethod + def _rebuild_legacy_fts_indexes( + cursor: sqlite3.Cursor, + *, + include_trigram: bool = True, + ) -> None: + """Rebuild the LEGACY inline FTS indexes (pre-v23) from messages. + + Used only to repair a legacy DB whose triggers degraded under an + earlier no-FTS5 runtime. Inline tables have no external-content + 'rebuild' source, so we DELETE + reinsert the concatenated content + the legacy triggers produced. Never touches the v23 shape. + """ cursor.execute("DELETE FROM messages_fts") cursor.execute( "INSERT INTO messages_fts(rowid, content) " @@ -1347,7 +1563,10 @@ class SessionDB: self._warn_fts5_unavailable(exc) return False - def _execute_write(self, fn: Callable[[sqlite3.Connection], T]) -> T: + def _execute_write( + self, + fn: Callable[[sqlite3.Connection], T], + ) -> T: """Execute a write transaction with BEGIN IMMEDIATE and jitter retry. *fn* receives the connection and should perform INSERT/UPDATE/DELETE @@ -1540,6 +1759,419 @@ class SessionDB: self._conn.close() self._conn = None + # ── Chunked FTS rebuild engine (v23 opt-in optimize) ── + # + # `optimize_fts_storage()` (the `hermes sessions optimize-storage` + # command) drops the legacy inline FTS indexes and backfills the new + # external-content ones. A single blocking rebuild measured ~16 minutes + # of held write lock on a real 25 GB DB, so the backfill runs in small + # chunks, each in its own short write transaction: + # - concurrent readers/writers are never starved (WAL stays small, + # each chunk checkpoints via the normal _execute_write cadence); + # - an interrupted run (Ctrl-C, crash) resumes from + # fts_rebuild_progress when the command is re-run; + # - multiple processes sharing the DB don't double-run it — each chunk + # claims work by compare-and-swap on fts_rebuild_progress, so even a + # concurrent second runner just interleaves chunks safely. + # + # THROTTLING (the part that keeps a live gateway sharing the DB + # responsive): a greedy chunk loop re-acquires BEGIN IMMEDIATE nearly + # back-to-back and can starve another process's writer into exhausting + # its lock retries (an early 5000-row/50ms version owned the write lock + # ~85% of the time and visibly froze concurrent CLI sessions on a large + # install). Two layers prevent that: + # 1. Small chunks (500 rows) — a foreground write queues behind a + # chunk for at most ~tens of ms. + # 2. Inter-chunk pause — the loop sleeps max(_FTS_REBUILD_MIN_PAUSE, + # chunk cost x _FTS_REBUILD_DUTY_FACTOR) between chunks, capping + # this process's share of DB bandwidth so concurrent writers always + # find open windows. This works cross-process (unlike any + # same-process activity stamp) because it bounds our own duty + # cycle unconditionally. + + _FTS_REBUILD_CHUNK_ROWS = 500 + _FTS_REBUILD_DUTY_FACTOR = 4.0 # sleep >= 4x chunk cost (≤20% duty) + _FTS_REBUILD_MIN_PAUSE = 0.2 # seconds — floor between chunks + + def fts_rebuild_status(self) -> Optional[Dict[str, Any]]: + """Return deferred-rebuild progress, or None when no rebuild pending. + + Shape: {"pending": True, "total": , + "indexed": , "percent": <0-100 int>}. + Consumed by search_messages() notes and by status surfaces + (dashboard/desktop can poll this to render a progress indicator). + """ + high_water = self.get_meta("fts_rebuild_high_water") + if high_water is None: + return None + progress = int(self.get_meta("fts_rebuild_progress") or 0) + total = int(high_water) + if total <= 0: + return None + pct = min(100, int(100 * progress / total)) + return {"pending": True, "total": total, "indexed": progress, "percent": pct} + + def _fts_rebuild_finish(self) -> None: + """Finalize the deferred rebuild: boundary sweep + clear markers. + + The sweep is cheap insurance against any write that slipped through + the migration-boundary instant (between high_water capture and + trigger activation): re-index any row near the boundary that the + index is missing. docsize has one row per indexed doc, so the + anti-join is exact and runs on a narrow id range. + """ + def _do(conn): + hw_row = conn.execute( + "SELECT value FROM state_meta WHERE key = 'fts_rebuild_high_water'" + ).fetchone() + if hw_row is not None: + hw = int(hw_row[0]) + # Sweep a generous window around the boundary. + lo, hi = hw - 1000, hw + 1000 + conn.execute( + "INSERT INTO messages_fts(rowid, content, tool_name, tool_calls) " + "SELECT m.id, m.content, 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)", + (lo, hi), + ) + conn.execute( + "INSERT INTO messages_fts_trigram(rowid, content, tool_name, tool_calls) " + "SELECT m.id, m.content, m.tool_name, m.tool_calls " + "FROM messages m " + "WHERE m.id > ? AND m.id <= ? AND m.role <> 'tool' " + "AND NOT EXISTS (SELECT 1 FROM messages_fts_trigram_docsize d WHERE d.id = m.id)", + (lo, hi), + ) + conn.execute( + "DELETE FROM state_meta WHERE key IN " + "('fts_rebuild_high_water', 'fts_rebuild_progress')" + ) + self._execute_write(_do) + logger.info("Deferred FTS rebuild complete — all messages indexed.") + + # Demoted v22 FTS shadow tables awaiting teardown (see the v23 migration: + # DROP of a multi-GB FTS vtable blocks for minutes, so the migration + # demotes the vtable definitions out of sqlite_master and renames the + # orphaned shadow tables — now plain tables — to fts_v22_trash_*; the + # worker empties them in bounded chunks, then drops them cheaply). + _FTS_TRASH_PREFIX = "fts_v22_trash_" + + def _fts_teardown_trash_step(self) -> bool: + """Tear down one chunk of a demoted v22 FTS shadow table. + + The trash tables are PLAIN tables (their vtable parent was demoted + away during the migration), so chunked DELETE + final DROP involve + no FTS5 machinery at all. Returns True while teardown work remains. + """ + with self._lock: + trash = [ + r[0] for r in self._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_cols = [ + r[1] for r in conn.execute(f"PRAGMA table_info({tbl})") + if r[5] > 0 + ] + key = ", ".join(pk_cols) if pk_cols else "rowid" + cur = conn.execute( + f"DELETE FROM {tbl} WHERE ({key}) IN " + f"(SELECT {key} FROM {tbl} LIMIT {self._FTS_REBUILD_CHUNK_ROWS})" + ) + if cur.rowcount == 0: + # Empty — the DROP is cheap now. + conn.execute(f"DROP TABLE IF EXISTS {tbl}") + logger.info("Old FTS shadow table %s torn down.", tbl) + return True # re-check: more trash tables / chunks may remain + + try: + return bool(self._execute_write(_do)) + except sqlite3.OperationalError as exc: + logger.debug("FTS trash teardown chunk failed (will retry): %s", exc) + return True + + def fts_rebuild_step(self) -> bool: + """Backfill one chunk of the deferred FTS rebuild. + + Returns True when more work remains, False when the rebuild is + complete (or none is pending). Safe to call from any process at any + time; chunks are claimed atomically inside the write transaction, so + concurrent callers interleave instead of duplicating rows. + """ + if not self._fts_enabled: + return False + high_water_raw = self.get_meta("fts_rebuild_high_water") + if high_water_raw is None: + return False + high_water = int(high_water_raw) + include_trigram = self._trigram_available + chunk = self._FTS_REBUILD_CHUNK_ROWS + + def _do(conn): + # Re-read progress inside the write transaction (BEGIN IMMEDIATE + # is already held by _execute_write) — this is the claim: two + # workers can't read the same progress value concurrently. + row = conn.execute( + "SELECT value FROM state_meta WHERE key = 'fts_rebuild_progress'" + ).fetchone() + if row is None: + return False # finished (or cleared) by another process + progress = int(row[0]) + if progress >= high_water: + return False + + # The chunk upper bound is an id, not a row count, so gaps from + # deleted rows don't shrink chunks below the claimed range. + upper = min(progress + chunk, high_water) + conn.execute( + "INSERT INTO messages_fts(rowid, content, tool_name, tool_calls) " + "SELECT id, content, tool_name, tool_calls FROM messages " + "WHERE id > ? AND id <= ?", + (progress, upper), + ) + if include_trigram: + conn.execute( + "INSERT INTO messages_fts_trigram" + "(rowid, content, tool_name, tool_calls) " + "SELECT id, content, tool_name, tool_calls FROM messages " + "WHERE id > ? AND id <= ? AND role <> 'tool'", + (progress, upper), + ) + # Publish progress in the same transaction as the rows it + # covers — crash-atomic: either both land or neither does. + conn.execute( + "UPDATE state_meta SET value = ? " + "WHERE key = 'fts_rebuild_progress'", + (str(upper),), + ) + return upper < high_water + + try: + more = self._execute_write(_do) + except sqlite3.OperationalError as exc: + logger.debug("FTS rebuild chunk failed (will retry): %s", exc) + return True # transient (lock contention) — caller retries + if more is False: + status = self.fts_rebuild_status() + if status is not None and status["indexed"] >= status["total"]: + self._fts_rebuild_finish() + return False + return bool(more) + + # ── Opt-in v23 FTS storage optimization (`hermes sessions optimize-storage`) ── + # + # This is the ONLY path that migrates an existing legacy (v22 inline) DB + # to the v23 external-content schema. It is deliberately foreground and + # user-invoked, never automatic, because it is disk-heavy and long. It + # runs the throttled/resumable chunk engine above to completion + # synchronously — demote → new schema → chunked backfill → chunked + # teardown — with progress callbacks, a disk preflight in the CLI + # wrapper, a VACUUM at the end, and a defensive schema_version bump. + + def fts_optimize_available(self) -> bool: + """True when `optimize_fts_storage()` has work to do: either this DB + is a legacy inline-FTS install that can be optimized to the v23 + external-content schema, or a previous optimize run was interrupted + (legacy vtables already demoted, but backfill markers and/or trash + tables remain) and re-running would resume it. False for fresh and + fully-optimized installs (and when FTS5 is unavailable).""" + if not self._fts_enabled or self.read_only: + return False + with self._lock: + if self._db_has_legacy_inline_fts(self._conn): + return True + # Interrupted optimize: demotion already removed the legacy + # vtables (so the check above is False), but the transition is + # unfinished until the backfill markers are cleared and the + # demoted trash tables are torn down. Search stays complete + # through the gap supplement meanwhile; re-running resumes. + if self._conn.execute( + "SELECT 1 FROM state_meta " + "WHERE key = 'fts_rebuild_high_water' LIMIT 1" + ).fetchone(): + return True + return self._has_fts_trash(self._conn) + + def _has_fts_trash(self, conn) -> bool: + """True when demoted v22 shadow tables are still awaiting teardown. + Caller must hold ``self._lock`` (or pass a migration-time cursor).""" + return bool(conn.execute( + "SELECT 1 FROM sqlite_master WHERE type = 'table' " + "AND name LIKE ? ESCAPE '\\' LIMIT 1", + (self._FTS_TRASH_PREFIX.replace("_", "\\_") + "%",), + ).fetchone()) + + def _demote_legacy_fts_to_trash(self) -> int: + """Demote the legacy inline FTS vtables and stage their shadow tables + for chunked teardown. Returns MAX(messages.id) as the rebuild high + water. O(1) schema surgery — the heavy delete is deferred to the + chunked teardown, exactly as the validated auto path did.""" + def _do(conn): + self._drop_fts_triggers(conn) + conn.execute("DROP VIEW IF EXISTS messages_fts_trigram_src") + had = bool(conn.execute( + "SELECT 1 FROM sqlite_master WHERE type = 'table' " + "AND name IN ('messages_fts', 'messages_fts_trigram') " + "AND sql LIKE 'CREATE VIRTUAL TABLE%' LIMIT 1" + ).fetchone()) + if had: + conn.execute("PRAGMA writable_schema=ON") + conn.execute( + "DELETE FROM sqlite_master WHERE type = 'table' " + "AND name IN ('messages_fts', 'messages_fts_trigram') " + "AND sql LIKE 'CREATE VIRTUAL TABLE%'" + ) + conn.execute("PRAGMA writable_schema=RESET") + shadows = [ + r[0] for r in conn.execute( + "SELECT name FROM sqlite_master WHERE type = 'table' " + "AND (name LIKE 'messages_fts_%' ESCAPE '\\' " + "OR name LIKE 'messages_fts_trigram_%' ESCAPE '\\')" + ).fetchall() + ] + for sh in shadows: + conn.execute(f"ALTER TABLE {sh} RENAME TO fts_v22_trash_{sh}") + # Create the new v23 empty schema + set the backfill markers. + self._ensure_fts_schema(conn, "messages_fts", FTS_SQL) + self._ensure_fts_schema(conn, "messages_fts_trigram", FTS_TRIGRAM_SQL) + hw = conn.execute("SELECT COALESCE(MAX(id), 0) FROM messages").fetchone()[0] + for k, v in ( + ("fts_rebuild_high_water", str(hw)), + ("fts_rebuild_progress", "0"), + ): + conn.execute( + "INSERT INTO state_meta (key, value) VALUES (?, ?) " + "ON CONFLICT(key) DO UPDATE SET value = excluded.value", + (k, v), + ) + conn.execute("DELETE FROM state_meta WHERE key = 'fts_optimize_available'") + return hw + return int(self._execute_write(_do)) + + def optimize_fts_storage( + self, + *, + progress_cb: Optional[Callable[[Dict[str, Any]], None]] = None, + vacuum: bool = True, + ) -> Dict[str, Any]: + """Migrate a legacy v22 inline-FTS DB to the v23 external-content + schema, foreground and to completion. Safe to re-run: if a previous + attempt was interrupted it resumes from the progress marker. + + ``progress_cb`` receives {"phase", "percent", "indexed", "total"} + dicts for a CLI progress bar. Returns a summary dict. + + The trigram tokenizer being unavailable is not fatal — the base index + is still rebuilt (CJK falls back to LIKE), mirroring normal startup. + """ + if not self._fts_enabled: + return {"ok": False, "reason": "fts5_unavailable"} + if self.read_only: + return {"ok": False, "reason": "read_only"} + + # Only demote if we're actually still on the legacy shape. If a prior + # run already demoted (markers/trash present), skip straight to + # finishing the backfill + teardown — this is what makes re-running + # after an interruption safe. + with self._lock: + legacy = self._db_has_legacy_inline_fts(self._conn) + pending = self.get_meta("fts_rebuild_high_water") is not None + if legacy and not pending: + self._demote_legacy_fts_to_trash() + + def _emit(phase: str) -> None: + if progress_cb is None: + return + st = self.fts_rebuild_status() + progress_cb({ + "phase": phase, + "percent": st["percent"] if st else 100, + "indexed": st["indexed"] if st else 0, + "total": st["total"] if st else 0, + }) + + def _pause(chunk_seconds: float) -> None: + """Inter-chunk throttle (see the chunk-engine note above). + + The chunk methods themselves never sleep, so this loop is the + single place the duty cycle is enforced: without it, back-to-back + BEGIN IMMEDIATE chunks starve any live gateway/CLI process + sharing the DB out of its lock retries (the measured ~85% + write-lock ownership that froze concurrent sessions). + """ + time.sleep(max( + self._FTS_REBUILD_MIN_PAUSE, + chunk_seconds * self._FTS_REBUILD_DUTY_FACTOR, + )) + + # Phase 1: backfill (foreground, throttled between chunks so a live + # gateway sharing the DB stays responsive). + _emit("backfill") + while True: + _t0 = time.monotonic() + if not self.fts_rebuild_step(): + break + _emit("backfill") + _pause(time.monotonic() - _t0) + _emit("backfill") + + # Phase 2: tear down the demoted legacy shadow tables in chunks. + _emit("teardown") + while True: + _t0 = time.monotonic() + if not self._fts_teardown_trash_step(): + break + _emit("teardown") + _pause(time.monotonic() - _t0) + + # Phase 3: reclaim freed pages to the OS. + vacuum_ok = None + if vacuum: + _emit("vacuum") + try: + with self._lock: + self._conn.execute("VACUUM") + vacuum_ok = True + except sqlite3.OperationalError as exc: + # Most common cause: not enough free disk for VACUUM's temp + # copy. The optimization still succeeded; space just isn't + # reclaimed until a later VACUUM. Non-fatal. + logger.warning("VACUUM after FTS optimize failed: %s", exc) + vacuum_ok = False + + # Phase 4: stamp the FTS storage layout as current, clear the "available" + # flag, and advance schema_version if it was somehow still behind (the + # main version normally advances on open now, but bump defensively so a + # DB opened only by pre-decoupling code still settles). The FTS-layout + # marker is the source of truth for "is this DB optimized". + def _settle(conn): + conn.execute( + "INSERT INTO state_meta (key, value) VALUES ('fts_storage_version', ?) " + "ON CONFLICT(key) DO UPDATE SET value = excluded.value", + (str(FTS_STORAGE_VERSION),), + ) + conn.execute("DELETE FROM state_meta WHERE key = 'fts_optimize_available'") + conn.execute( + "UPDATE schema_version SET version = ? WHERE version < ?", + (SCHEMA_VERSION, SCHEMA_VERSION), + ) + self._execute_write(_settle) + _emit("done") + logger.info( + "FTS storage optimization complete (layout v%d).", FTS_STORAGE_VERSION + ) + return {"ok": True, "vacuumed": vacuum_ok} + @staticmethod def _parse_schema_columns(schema_sql: str) -> Dict[str, Dict[str, str]]: """Extract expected columns per table from SCHEMA_SQL. @@ -1739,66 +2371,16 @@ class SessionDB: fts_migrations_complete = False else: fts_migrations_complete = False - if current_version < 11: - # v11: re-index FTS5 tables to cover tool_name + tool_calls and - # switch from external-content to inline mode. Existing DBs have - # old-schema FTS tables and triggers that IF NOT EXISTS won't - # overwrite, so we drop them explicitly and let the post-migration - # existence checks (below) recreate them from FTS_SQL / - # FTS_TRIGRAM_SQL, then backfill every message row. Fixes #16751. - if fts5_available: - self._drop_fts_triggers(cursor) - for _tbl in ("messages_fts", "messages_fts_trigram"): - try: - cursor.execute(f"DROP TABLE IF EXISTS {_tbl}") - except sqlite3.OperationalError as exc: - if not self._is_fts5_unavailable_error(exc): - raise - if self._is_trigram_unavailable_error(exc): - self._warn_trigram_unavailable(exc) - else: - self._warn_fts5_unavailable(exc) - fts5_available = False - fts_migrations_complete = False - break - - if fts5_available: - # Recreate virtual tables + triggers with the new inline-mode - # schema that indexes content || tool_name || tool_calls. - # Handle base and trigram independently — a missing - # trigram tokenizer should not prevent base FTS backfill. - base_fts_ok = self._ensure_fts_schema( - cursor, "messages_fts", FTS_SQL - ) - if base_fts_ok: - cursor.execute( - "INSERT INTO messages_fts(rowid, content) " - "SELECT id, " - "COALESCE(content, '') || ' ' || " - "COALESCE(tool_name, '') || ' ' || " - "COALESCE(tool_calls, '') " - "FROM messages" - ) - trigram_ok = self._ensure_fts_schema( - cursor, "messages_fts_trigram", FTS_TRIGRAM_SQL - ) - if trigram_ok: - cursor.execute( - "INSERT INTO messages_fts_trigram(rowid, content) " - "SELECT id, " - "COALESCE(content, '') || ' ' || " - "COALESCE(tool_name, '') || ' ' || " - "COALESCE(tool_calls, '') " - "FROM messages" - ) - if not base_fts_ok: - fts_migrations_complete = False - # Track trigram availability for CJK LIKE fallback. - self._trigram_available = trigram_ok - else: - fts_migrations_complete = False - else: - fts_migrations_complete = False + if current_version < 11 and SCHEMA_VERSION < 23: + # v11 (SUPERSEDED by v23): re-index FTS5 tables to cover + # tool_name + tool_calls in inline mode (#16751). v23 drops + # and rebuilds both FTS tables in external-content form, so + # running the v11 inline backfill first would only burn + # startup time and WAL space before v23 throws the work + # away — and its inline INSERT shape no longer matches the + # current external-content FTS_SQL anyway. Kept only for + # source archaeology; unreachable while SCHEMA_VERSION >= 23. + pass if current_version < 16: # v16: tag delegate subagent rows so pickers stay clean after # parent deletes that used to orphan them (parent_session_id → NULL). @@ -1948,7 +2530,72 @@ class SessionDB: ) except sqlite3.OperationalError as exc: logger.debug("v22 session_model_usage rebuild skipped: %s", exc) - if current_version < SCHEMA_VERSION and fts_migrations_complete: + if current_version < 23: + # v23: FTS storage redesign (issues #22478, #43690, #55233). + # The v11 inline-mode FTS tables each store a full private + # copy of every message (content || tool_name || tool_calls), + # and the trigram index additionally covers role='tool' rows + # (~90% of message bytes: base64 payloads, file dumps) at + # ~2.6x amplification — together ~75% of state.db on heavy + # installs (observed: 18.9 GB of a 25 GB DB). + # + # OPT-IN, NOT AUTOMATIC. The transition (demote old vtables → + # new external-content schema → backfill → teardown → VACUUM) + # is disk-heavy (transient ~2x file size to fully reclaim via + # VACUUM) and long (~1-2h background on a 25 GB DB). Doing it + # silently on every big user's next open — with a completeness + # guarantee that depends on the process staying alive long + # enough — is the wrong default. So on an EXISTING install we + # touch nothing here: the v22 inline FTS keeps working exactly + # as before, and we only record a flag advertising that the + # optimization is available. `hermes sessions optimize-storage` + # performs the whole transition as one deliberate, disk-checked, + # progress-reported foreground operation. + # + # DECOUPLED VERSIONING. Crucially, this does NOT hold back the + # main schema_version. The FTS storage LAYOUT is tracked by an + # independent `fts_storage_version` marker (see + # _fts_storage_version / SETTLE below), so schema_version + # advances to SCHEMA_VERSION here like every other migration — + # future v24+ migrations land automatically for legacy-FTS + # users too. Only the FTS *layout* waits for opt-in. + if fts5_available and self._db_has_legacy_inline_fts(cursor): + self.set_meta("fts_optimize_available", "1", cursor=cursor) + + # The FTS storage layout is versioned independently of the main + # schema (see the v23 note above). Stamp the current layout so the + # main version can always advance: a fresh/optimized DB is at + # FTS_STORAGE_VERSION; a legacy DB is left at whatever it had + # (absent/0) until `optimize-storage` runs. An INTERRUPTED + # optimize (legacy vtables already demoted, but rebuild markers + # or demoted trash tables still present) is NOT stamped either — + # the marker is the source of truth for "fully optimized", and + # `fts_optimize_available()` keeps offering the resume until the + # transition actually completes. + if ( + fts5_available + and not self._db_has_legacy_inline_fts(cursor) + and cursor.execute( + "SELECT 1 FROM state_meta " + "WHERE key = 'fts_rebuild_high_water' LIMIT 1" + ).fetchone() is None + and not self._has_fts_trash(cursor) + ): + self.set_meta( + "fts_storage_version", str(FTS_STORAGE_VERSION), cursor=cursor + ) + + # Advance schema_version to current for ALL non-FTS-layout + # migrations. This is deliberately NOT gated on the FTS opt-in — + # holding the whole version back would block every future schema + # migration for a user who never optimizes. FTS5 being unavailable + # is the one case we skip (we can't have created the current FTS + # objects, so claiming the current schema would be a lie). + if ( + current_version < SCHEMA_VERSION + and fts_migrations_complete + and fts5_available + ): cursor.execute( "UPDATE schema_version SET version = ?", (SCHEMA_VERSION,), @@ -1994,22 +2641,52 @@ class SessionDB: # FTS5 setup. Run the DDL even when the virtual table exists so # CREATE TRIGGER IF NOT EXISTS repairs trigger-only degradation from # an earlier no-FTS5 runtime. - triggers_need_repair = self._fts_trigger_count(cursor) < len(_FTS_TRIGGERS) - self._fts_enabled = self._ensure_fts_schema(cursor, "messages_fts", FTS_SQL) - - # Trigram FTS5 for CJK/substring search. This is optional relative - # to the main FTS table; if it cannot be created, CJK search falls - # back to LIKE. - if self._fts_enabled: - trigram_enabled = self._ensure_fts_schema( - cursor, "messages_fts_trigram", FTS_TRIGRAM_SQL + # + # OPT-IN v23 boundary: a legacy v22 install (inline-content FTS, + # not yet opted into `hermes db optimize`) must keep its EXISTING + # inline schema + triggers. Running the v23 external-content DDL + # here would create the trigram source VIEW and leave the DB in a + # mixed inline/external state. So for a legacy DB we only ensure + # its inline triggers exist (via the legacy DDL), and skip the + # v23 view/external tables entirely. Fresh installs and opted-in + # DBs have no legacy inline FTS, so they get the v23 DDL. + if self._db_has_legacy_inline_fts(cursor): + triggers_need_repair = ( + self._fts_trigger_count(cursor) < len(_FTS_TRIGGERS) ) - self._trigram_available = trigram_enabled - if triggers_need_repair: - self._rebuild_fts_indexes( - cursor, - include_trigram=trigram_enabled, + self._fts_enabled = self._ensure_fts_schema( + cursor, "messages_fts", LEGACY_FTS_SQL + ) + if self._fts_enabled: + trigram_enabled = self._ensure_fts_schema( + cursor, "messages_fts_trigram", LEGACY_FTS_TRIGRAM_SQL ) + self._trigram_available = trigram_enabled + if triggers_need_repair: + self._rebuild_legacy_fts_indexes( + cursor, include_trigram=trigram_enabled + ) + else: + triggers_need_repair = ( + self._fts_trigger_count(cursor) < len(_FTS_TRIGGERS) + ) + self._fts_enabled = self._ensure_fts_schema( + cursor, "messages_fts", FTS_SQL + ) + + # Trigram FTS5 for CJK/substring search. This is optional + # relative to the main FTS table; if it cannot be created, + # CJK search falls back to LIKE. + if self._fts_enabled: + trigram_enabled = self._ensure_fts_schema( + cursor, "messages_fts_trigram", FTS_TRIGRAM_SQL + ) + self._trigram_available = trigram_enabled + if triggers_need_repair: + self._rebuild_fts_indexes( + cursor, + include_trigram=trigram_enabled, + ) self._conn.commit() @@ -5637,7 +6314,7 @@ class SessionDB: m.id, m.session_id, m.role, - snippet(messages_fts, 0, '>>>', '<<<', '...', 40) AS snippet, + snippet(messages_fts, -1, '>>>', '<<<', '...', 40) AS snippet, m.content, m.timestamp, m.tool_name, @@ -5679,7 +6356,17 @@ class SessionDB: ) _trigram_succeeded = False - if cjk_count >= 3 and not _any_short_cjk and self._trigram_available: + # Tool rows are excluded from the trigram index (they're ~90% of + # message bytes and machine noise — see FTS_TRIGRAM_SQL). A CJK + # query explicitly filtering on role='tool' must therefore use + # the LIKE fallback, which scans the base table directly. + _wants_tool_rows = bool(role_filter) and "tool" in role_filter + if ( + cjk_count >= 3 + and not _any_short_cjk + and self._trigram_available + and not _wants_tool_rows + ): # Trigram FTS5 path — quote each non-operator token to handle # FTS5 special chars (%, *, etc.) while preserving boolean # operators (AND, OR, NOT) for multi-term queries. @@ -5709,7 +6396,7 @@ class SessionDB: m.id, m.session_id, m.role, - snippet(messages_fts_trigram, 0, '>>>', '<<<', '...', 40) AS snippet, + snippet(messages_fts_trigram, -1, '>>>', '<<<', '...', 40) AS snippet, m.content, m.timestamp, m.tool_name, @@ -5784,6 +6471,11 @@ class SessionDB: ) like_params += [f"%{esc}%", f"%{esc}%", f"%{esc}%"] like_where = [f"({' OR '.join(token_clauses)})"] + if not include_inactive: + # Same visibility rule as the FTS5 paths: live rows and + # compaction-archived rows are discoverable; rewind/undo + # rows (active=0, compacted=0) are hidden (#38763). + like_where.append("(m.active = 1 OR m.compacted = 1)") if source_filter is not None: like_where.append(f"s.source IN ({','.join('?' for _ in source_filter)})") like_params.extend(source_filter) @@ -5835,6 +6527,29 @@ class SessionDB: cursor = self._conn.execute(sql, params) matches = [dict(row) for row in cursor.fetchall()] + # Deferred-rebuild supplement (schema v23): while the background + # backfill is pending, the FTS indexes only cover rows outside the + # (progress, high_water] gap. Top the results up with a bounded LIKE + # scan over just that id range so search never silently loses old + # messages mid-rebuild. The range shrinks as the backfill advances, + # so this cost decays to zero. The CJK LIKE-fallback path above + # already scans the whole base table and needs no supplement. + rebuild_status = self.fts_rebuild_status() + if rebuild_status is not None and len(matches) < limit: + try: + gap_matches = self._search_unindexed_gap( + query, + limit - len(matches), + include_inactive=include_inactive, + source_filter=source_filter, + exclude_sources=exclude_sources, + role_filter=role_filter, + ) + seen_ids = {m["id"] for m in matches} + matches.extend(m for m in gap_matches if m["id"] not in seen_ids) + except sqlite3.OperationalError as exc: + logger.debug("Unindexed-gap supplement skipped: %s", exc) + # Add surrounding context (1 message before + after each match). # Done outside the lock so we don't hold it across N sequential queries. for match in matches: @@ -5903,6 +6618,79 @@ class SessionDB: return matches + def _search_unindexed_gap( + self, + fts_query: str, + limit: int, + *, + include_inactive: bool = False, + source_filter: Optional[List[str]] = None, + exclude_sources: Optional[List[str]] = None, + role_filter: Optional[List[str]] = None, + ) -> List[Dict[str, Any]]: + """LIKE-scan the rows the deferred rebuild hasn't indexed yet. + + Only touches ids in (fts_rebuild_progress, fts_rebuild_high_water] — + a range that shrinks to nothing as the backfill advances. The FTS + query is degraded to per-token substring terms (AND-joined; quoted + phrases kept whole), which is deliberately recall-over-precision: + temporary results beat silently missing ones mid-rebuild. + """ + status = self.fts_rebuild_status() + if status is None or limit <= 0: + return [] + progress, high_water = status["indexed"], status["total"] + + # Degrade the FTS query to LIKE terms: strip operators/wildcards, + # keep quoted phrases intact, AND the rest. + terms: List[str] = [] + for raw_tok in re.findall(r'"[^"]+"|\S+', fts_query): + tok = raw_tok.strip('"').strip("*").strip() + if not tok or tok.upper() in {"AND", "OR", "NOT", "NEAR"}: + continue + terms.append(tok) + if not terms: + return [] + + where = ["m.id > ? AND m.id <= ?"] + params: list = [progress, high_water] + for term in terms: + esc = term.replace("\\", "\\\\").replace("%", "\\%").replace("_", "\\_") + where.append( + "(m.content LIKE ? ESCAPE '\\' OR m.tool_name LIKE ? ESCAPE '\\' " + "OR m.tool_calls LIKE ? ESCAPE '\\')" + ) + params += [f"%{esc}%"] * 3 + if not include_inactive: + where.append("(m.active = 1 OR m.compacted = 1)") + if source_filter is not None: + where.append(f"s.source IN ({','.join('?' for _ in source_filter)})") + params.extend(source_filter) + if exclude_sources is not None: + where.append(f"s.source NOT IN ({','.join('?' for _ in exclude_sources)})") + params.extend(exclude_sources) + if role_filter: + where.append(f"m.role IN ({','.join('?' for _ in role_filter)})") + params.extend(role_filter) + + sql = f""" + SELECT m.id, m.session_id, m.role, + substr(m.content, + max(1, instr(m.content, ?) - 40), + 120) AS snippet, + m.content, m.timestamp, m.tool_name, + s.source, s.model, s.started_at AS session_started + FROM messages m + JOIN sessions s ON s.id = m.session_id + WHERE {' AND '.join(where)} + ORDER BY m.timestamp DESC + LIMIT ? + """ + params = [terms[0]] + params + [limit] + with self._lock: + rows = self._conn.execute(sql, params).fetchall() + return [dict(r) for r in rows] + def search_sessions_by_id( self, query: str, @@ -7136,8 +7924,25 @@ class SessionDB: return None return row["value"] if isinstance(row, sqlite3.Row) else row[0] - def set_meta(self, key: str, value: str) -> None: - """Write a value to the state_meta key/value store.""" + def set_meta( + self, key: str, value: str, *, cursor: Optional[sqlite3.Cursor] = None + ) -> None: + """Write a value to the state_meta key/value store. + + When ``cursor`` is provided the write is issued on that cursor + inline (used during ``_init_schema``, which already holds an open + transaction — routing through ``_execute_write`` there would nest + BEGIN IMMEDIATE and deadlock). Otherwise a normal write transaction + is used. + """ + if cursor is not None: + cursor.execute( + "INSERT INTO state_meta (key, value) VALUES (?, ?) " + "ON CONFLICT(key) DO UPDATE SET value = excluded.value", + (key, value), + ) + return + def _do(conn): conn.execute( "INSERT INTO state_meta (key, value) VALUES (?, ?) " @@ -7662,7 +8467,11 @@ class SessionDB: try: self._conn.execute(f"SELECT 1 FROM {name} LIMIT 0") return True - except sqlite3.OperationalError: + except sqlite3.DatabaseError: + # OperationalError ("no such table") or the broader + # DatabaseError class ("vtable constructor failed", raised when + # e.g. a required tokenizer is missing or the table is mid- + # teardown) — in every case the table is not queryable. return False def optimize_fts(self) -> int: diff --git a/tests/test_hermes_state.py b/tests/test_hermes_state.py index dd6ecbc55182..a00d5a7e9b10 100644 --- a/tests/test_hermes_state.py +++ b/tests/test_hermes_state.py @@ -3,6 +3,8 @@ import sqlite3 import time import json +from unittest import mock + import pytest import hermes_state @@ -792,24 +794,50 @@ class TestSessionLifecycle: def test_v11_migration_backfills_base_fts_when_trigram_unavailable( self, tmp_path, monkeypatch ): - """Regression: v11 migration must backfill base FTS even when trigram is unavailable.""" + """A legacy inline-FTS DB opened under a no-trigram runtime keeps its + base FTS searchable (and is flagged optimizable) without crashing. + + Opt-in model: opening never auto-migrates. The legacy single-column + index keeps working for content search; the trigram tokenizer being + unavailable must not break base FTS or the open itself. + """ real_connect = sqlite3.connect db_path = tmp_path / "state.db" - # Phase 1: create a DB at schema v10 with messages. - db = SessionDB(db_path=db_path) - db.create_session(session_id="s1", source="cli") - db.append_message("s1", role="user", content="legacy message alpha") - db.append_message("s1", role="assistant", content="legacy reply beta") - # Force schema version to v10 so migration runs on next open. - db._conn.execute( - "UPDATE schema_version SET version = 10" + # Phase 1: build a genuine legacy inline DB by hand (single-column + # messages_fts, no trigram table), at an old schema version. + conn = sqlite3.connect(str(db_path)) + conn.executescript(SCHEMA_SQL) + conn.executescript(""" + DROP TABLE IF EXISTS messages_fts; + DROP TABLE IF EXISTS messages_fts_trigram; + DROP VIEW IF EXISTS messages_fts_trigram_src; + CREATE VIRTUAL TABLE messages_fts USING fts5(content); + CREATE TRIGGER messages_fts_insert AFTER INSERT ON messages BEGIN + INSERT INTO messages_fts(rowid, content) VALUES (new.id, COALESCE(new.content,'')); + END; + """) + conn.execute("DELETE FROM schema_version") + conn.execute("INSERT INTO schema_version (version) VALUES (10)") + conn.execute( + "INSERT INTO sessions (id, source, started_at) VALUES ('s1', 'cli', ?)", + (time.time(),), ) - db._conn.commit() - db.close() + for role, content in ( + ("user", "legacy message alpha"), + ("assistant", "legacy reply beta"), + ): + conn.execute( + "INSERT INTO messages (session_id, timestamp, role, content) " + "VALUES ('s1', ?, ?, ?)", + (time.time(), role, content), + ) + conn.commit() + conn.close() - # Phase 2: reopen with trigram disabled — migration should still - # backfill base FTS and make existing messages searchable. + # Phase 2: reopen with trigram disabled — must NOT crash, base FTS + # keeps working, and the DB is flagged optimizable (opt-in, so no + # auto-migration and the version stays put). def connect_without_trigram(*args, **kwargs): kwargs["factory"] = _NoTrigramConnection return real_connect(*args, **kwargs) @@ -821,6 +849,7 @@ class TestSessionLifecycle: assert migrated_db._trigram_available is False assert migrated_db._fts_table_exists("messages_fts") is True assert migrated_db._fts_table_exists("messages_fts_trigram") is False + assert migrated_db.fts_optimize_available() is True # Existing messages must be searchable via base FTS. results = migrated_db.search_messages("legacy message") @@ -3749,11 +3778,15 @@ class TestSchemaInit: migrated_db.close() def test_v9_migration_skips_v10_trigram_backfill_before_v11_rebuild(self, tmp_path, monkeypatch): - """Direct v9→current migration should do only the v11 FTS rebuild. + """Direct v9→current migration should do only the v23 FTS rebuild. - v10 backfilled ``messages_fts_trigram`` with content-only rows. Current - v11+ migration immediately drops and rebuilds both FTS tables with - content + tool metadata, so running the v10 insert first is wasted work. + v10 backfilled ``messages_fts_trigram`` with content-only rows. The + current migration immediately drops and rebuilds both FTS tables in + external-content form, so running the v10 insert first is wasted work. + + v23 contract: tool rows are excluded from the trigram index (they + remain fully searchable via the standard index); non-tool rows are + indexed in both. """ db_path = tmp_path / "v9_fts.db" conn = sqlite3.connect(str(db_path)) @@ -3769,6 +3802,11 @@ class TestSchemaInit: "VALUES (?, ?, ?, ?, ?, ?)", ("s1", "tool", "plain content", "browser_snapshot", '{"name":"browser_snapshot"}', 1001.0), ) + conn.execute( + "INSERT INTO messages (session_id, role, content, timestamp) " + "VALUES (?, ?, ?, ?)", + ("s1", "assistant", "assistant summary of the snapshot", 1002.0), + ) conn.commit() conn.close() @@ -3794,16 +3832,37 @@ class TestSchemaInit: try: assert trigram_content_only_inserts == [] version = migrated_db._conn.execute("SELECT version FROM schema_version").fetchone()[0] + # This DB was built via SCHEMA_SQL, so its FTS is already the v23 + # external-content shape — not a legacy inline install. Opening it + # therefore advances the version to current (no opt-in gate) and + # runs no backfill (rows were indexed live by the v23 triggers). assert version == SCHEMA_VERSION - normal_count = migrated_db._conn.execute("SELECT COUNT(*) FROM messages_fts").fetchone()[0] - trigram_count = migrated_db._conn.execute("SELECT COUNT(*) FROM messages_fts_trigram").fetchone()[0] - assert normal_count == 1 + assert migrated_db.fts_optimize_available() is False + assert migrated_db.fts_rebuild_status() is None + # Standard FTS indexes every row, including tool output (MATCH + # probes the index; COUNT(*) on external-content tables doesn't). + normal_count = migrated_db._conn.execute( + "SELECT COUNT(*) FROM messages_fts WHERE messages_fts MATCH 'snapshot'" + ).fetchone()[0] + assert normal_count == 2 + # Trigram excludes role='tool' rows (v23) but keeps non-tool rows. + trigram_count = migrated_db._conn.execute( + "SELECT COUNT(*) FROM messages_fts_trigram " + "WHERE messages_fts_trigram MATCH 'snapshot'" + ).fetchone()[0] assert trigram_count == 1 + # Tool metadata stays searchable via the standard index (#16751). tool_hit = migrated_db._conn.execute( + "SELECT COUNT(*) FROM messages_fts " + "WHERE messages_fts MATCH 'browser_snapshot'" + ).fetchone()[0] + assert tool_hit == 1 + # And is intentionally absent from the trigram index. + tri_tool_hit = migrated_db._conn.execute( "SELECT COUNT(*) FROM messages_fts_trigram " "WHERE messages_fts_trigram MATCH 'browser_snapshot'" ).fetchone()[0] - assert tool_hit == 1 + assert tri_tool_hit == 0 finally: migrated_db.close() @@ -5200,14 +5259,22 @@ class TestFTS5ToolCallMigration: assert legacy_hits == [], "sanity: legacy FTS must NOT contain tool_name" conn.close() - # Now open via SessionDB — migration runs. + # Open via SessionDB — the legacy DB is detected as optimizable but + # NOT auto-migrated (opt-in). Its old content-only index still works + # for content, but doesn't yet cover tool_name/tool_calls (#16751). session_db = SessionDB(db_path=db_path) try: + assert session_db.fts_optimize_available() is True + + # `hermes db optimize` performs the v23 transition; afterwards the + # tool fields are searchable. + result = session_db.optimize_fts_storage(vacuum=False) + assert result["ok"] is True assert len(session_db.search_messages("LEGACYTOOL")) == 1, \ - "v11 migration must backfill tool_name into FTS" + "v23 optimize must index tool_name into FTS" assert len(session_db.search_messages("LEGACYARG")) == 1, \ - "v11 migration must backfill tool_calls JSON into FTS" - # schema_version bumped + "v23 optimize must index tool_calls JSON into FTS" + # schema_version bumped once the FTS layer is v23 from hermes_state import SCHEMA_VERSION row = session_db._conn.execute( "SELECT version FROM schema_version LIMIT 1" @@ -5218,6 +5285,343 @@ class TestFTS5ToolCallMigration: session_db.close() +class TestFTSExternalContentMigration: + """v23 migration: inline-mode FTS tables (v11-v22) are rebuilt as + external-content tables, and role='tool' rows are excluded from the + trigram index while remaining searchable via the standard index.""" + + @staticmethod + def _build_v22_db(db_path): + """Build a v22-shaped DB by hand: inline FTS tables + concat triggers.""" + conn = sqlite3.connect(str(db_path)) + conn.executescript(SCHEMA_SQL) + # Replace the current (v23) FTS objects with the v22 inline shape. + conn.executescript(""" + DROP TABLE IF EXISTS messages_fts; + DROP TABLE IF EXISTS messages_fts_trigram; + DROP VIEW IF EXISTS messages_fts_trigram_src; + + CREATE VIRTUAL TABLE messages_fts USING fts5(content); + CREATE TRIGGER messages_fts_insert AFTER INSERT ON messages BEGIN + INSERT INTO messages_fts(rowid, content) VALUES ( + new.id, + COALESCE(new.content, '') || ' ' || COALESCE(new.tool_name, '') || ' ' || COALESCE(new.tool_calls, '') + ); + END; + + CREATE VIRTUAL TABLE messages_fts_trigram USING fts5(content, tokenize='trigram'); + CREATE TRIGGER messages_fts_trigram_insert AFTER INSERT ON messages BEGIN + INSERT INTO messages_fts_trigram(rowid, content) VALUES ( + new.id, + COALESCE(new.content, '') || ' ' || COALESCE(new.tool_name, '') || ' ' || COALESCE(new.tool_calls, '') + ); + END; + """) + conn.execute("DELETE FROM schema_version") + conn.execute("INSERT INTO schema_version (version) VALUES (22)") + conn.execute( + "INSERT INTO sessions (id, source, started_at) VALUES ('s1', 'cli', ?)", + (time.time(),), + ) + rows = [ + ("user", "find the 大别山项目 deployment notes", None, None), + ("assistant", "关于大别山项目的总结在这里", None, + '{"function":{"name":"send_message","arguments":"{}"}}'), + ("tool", "TOOLBLOB " + "x" * 5000 + " 项目文件内容测试", "read_file", None), + ] + for role, content, tool_name, tool_calls in rows: + conn.execute( + "INSERT INTO messages (session_id, timestamp, role, content, tool_name, tool_calls) " + "VALUES ('s1', ?, ?, ?, ?, ?)", + (time.time(), role, content, tool_name, tool_calls), + ) + conn.commit() + # Sanity: v22 inline tables have their own content shadow tables. + shadow = conn.execute( + "SELECT name FROM sqlite_master WHERE name = 'messages_fts_content'" + ).fetchall() + assert shadow, "sanity: v22 inline FTS must have a content shadow table" + conn.close() + + def test_v22_open_leaves_legacy_untouched_and_advertises(self, tmp_path): + """Opening a legacy v22 DB must NOT auto-migrate the FTS layout, but + the main schema_version DOES advance (decoupled) so future non-FTS + migrations aren't blocked. The inline index keeps working and the + opt-in flag is set.""" + db_path = tmp_path / "v22.db" + self._build_v22_db(db_path) + + db = SessionDB(db_path=db_path) + try: + # DECOUPLED: the main schema_version advances to current even though + # the FTS layout stays legacy — future migrations must not be gated + # behind the FTS opt-in. + version = db._conn.execute( + "SELECT version FROM schema_version" + ).fetchone()[0] + assert version == SCHEMA_VERSION, "main schema version must advance" + # But the FTS storage layout is NOT stamped current — it's legacy. + assert db.get_meta("fts_storage_version") is None + assert db.fts_optimize_available() is True + assert db.get_meta("fts_optimize_available") == "1" + + # Legacy inline shape is intact (content shadow table still there). + assert db._conn.execute( + "SELECT name FROM sqlite_master WHERE name = 'messages_fts_content'" + ).fetchone() is not None + + # Search still works on the legacy index (no deferred rebuild). + assert db.fts_rebuild_status() is None + assert len(db.search_messages("deployment")) == 1 + assert len(db.search_messages("send_message")) == 1 # #16751 held + + # A new write is indexed live by the legacy triggers. + db.append_message("s1", role="user", content="AFTEROPEN token") + assert len(db.search_messages("AFTEROPEN")) == 1 + finally: + db.close() + + def test_optimize_fts_storage_transitions_to_v23(self, tmp_path): + """`optimize_fts_storage()` migrates a legacy DB to v23 external-content + to completion: no shadow copies, tool rows excluded from trigram, + version bumped, everything searchable exactly once.""" + db_path = tmp_path / "v22.db" + self._build_v22_db(db_path) + + db = SessionDB(db_path=db_path) + try: + assert db.fts_optimize_available() is True + result = db.optimize_fts_storage(vacuum=False) + assert result["ok"] is True + + # Layout stamped current; flag cleared; no longer "available". + assert db.get_meta("fts_storage_version") == str( + hermes_state.FTS_STORAGE_VERSION + ) + assert db._conn.execute( + "SELECT version FROM schema_version" + ).fetchone()[0] == SCHEMA_VERSION + assert db.fts_optimize_available() is False + assert db.fts_rebuild_status() is None + + # External-content: no *_content shadow tables, no trash left. + for shadow in ("messages_fts_content", "messages_fts_trigram_content"): + assert db._conn.execute( + "SELECT name FROM sqlite_master WHERE name = ?", (shadow,) + ).fetchone() is None + assert db._conn.execute( + "SELECT name FROM sqlite_master WHERE name LIKE '%_v22_trash%'" + ).fetchall() == [] + + # Standard FTS: all rows incl tool metadata (#16751). + assert len(db.search_messages("TOOLBLOB")) == 1 + assert len(db.search_messages("send_message")) == 1 + # Trigram excludes tool rows; CJK conversation search works. + assert len(db.search_messages("大别山项目")) == 2 + assert db._conn.execute( + "SELECT COUNT(*) FROM messages_fts_trigram " + "WHERE messages_fts_trigram MATCH '\"项目文件内容\"'" + ).fetchone()[0] == 0 + assert db.search_messages("项目文件内容", role_filter=["tool"]) != [] + # No duplicate index entries; integrity clean. + for term in ("TOOLBLOB", "deployment"): + assert db._conn.execute( + "SELECT COUNT(*) FROM messages_fts WHERE messages_fts MATCH ?", + (term,), + ).fetchone()[0] == 1 + db._conn.execute( + "INSERT INTO messages_fts(messages_fts, rank) VALUES('integrity-check', 1)" + ) + finally: + db.close() + + def test_optimize_fts_storage_resumable_after_interrupt(self, tmp_path): + """A partially-completed optimize resumes on re-run: after demote + + one chunk, re-invoking finishes without duplicating rows.""" + db_path = tmp_path / "v22.db" + self._build_v22_db(db_path) + + db = SessionDB(db_path=db_path) + try: + # Simulate an interrupted run: demote + a single backfill chunk, + # then stop (as if the process died mid-optimize). + db._demote_legacy_fts_to_trash() + assert db.fts_rebuild_status() is not None + db.fts_rebuild_step() # one chunk only + + # Old rows not yet backfilled are still findable via gap supplement. + assert len(db.search_messages("TOOLBLOB")) == 1 + + # Re-run the full command — must resume, not restart or duplicate. + result = db.optimize_fts_storage(vacuum=False) + assert result["ok"] is True + assert db.fts_rebuild_status() is None + assert db._conn.execute( + "SELECT version FROM schema_version" + ).fetchone()[0] == SCHEMA_VERSION + for term in ("TOOLBLOB", "deployment"): + assert db._conn.execute( + "SELECT COUNT(*) FROM messages_fts WHERE messages_fts MATCH ?", + (term,), + ).fetchone()[0] == 1 + db._conn.execute( + "INSERT INTO messages_fts(messages_fts, rank) VALUES('integrity-check', 1)" + ) + finally: + db.close() + + def test_interrupted_optimize_reopen_still_reports_available(self, tmp_path): + """An interrupted optimize followed by a process restart must keep + offering the resume: the legacy vtables are gone (demoted), so the + legacy-shape check alone would say "already compact" — the gate has + to accept pending rebuild markers / trash tables too. And the reopen + must NOT stamp fts_storage_version (the transition isn't done).""" + db_path = tmp_path / "v22.db" + self._build_v22_db(db_path) + + db = SessionDB(db_path=db_path) + try: + db._demote_legacy_fts_to_trash() + db.fts_rebuild_step() # one chunk, then "the process dies" + finally: + db.close() + + # Fresh open, as the CLI would after the interrupt. + db = SessionDB(db_path=db_path) + try: + # The CLI gate must still offer optimize-storage (resume). + assert db.fts_optimize_available() is True + # The layout must NOT be stamped current mid-transition. + assert db.get_meta("fts_storage_version") is None + # Search stays complete through the gap supplement meanwhile. + assert len(db.search_messages("TOOLBLOB")) == 1 + + # Re-running the command resumes and completes the transition. + result = db.optimize_fts_storage(vacuum=False) + assert result["ok"] is True + assert db.fts_optimize_available() is False + assert db.get_meta("fts_storage_version") == str( + hermes_state.FTS_STORAGE_VERSION + ) + assert db._conn.execute( + "SELECT name FROM sqlite_master WHERE name LIKE '%_v22_trash%'" + ).fetchall() == [] + for term in ("TOOLBLOB", "deployment"): + assert db._conn.execute( + "SELECT COUNT(*) FROM messages_fts WHERE messages_fts MATCH ?", + (term,), + ).fetchone()[0] == 1 + db._conn.execute( + "INSERT INTO messages_fts(messages_fts, rank) VALUES('integrity-check', 1)" + ) + finally: + db.close() + + def test_v23_fresh_db_born_optimized(self, tmp_path): + """A brand-new DB is born on v23 — no legacy layout, no opt-in flag, + no pending rebuild.""" + db = SessionDB(db_path=tmp_path / "fresh.db") + try: + assert db.fts_optimize_available() is False + assert db.fts_rebuild_status() is None + assert db.get_meta("fts_optimize_available") is None + # Already external-content: no shadow copy tables. + assert db._conn.execute( + "SELECT name FROM sqlite_master WHERE name = 'messages_fts_content'" + ).fetchone() is None + db.create_session(session_id="s1", source="cli") + db.append_message("s1", role="user", content="hello fresh world") + assert len(db.search_messages("fresh")) == 1 + finally: + db.close() + + def test_v23_trigram_stays_in_sync_on_write_paths(self, tmp_path): + """INSERT/UPDATE/DELETE through SessionDB keep both indexes coherent + under the new trigger shape (integrity-check verifies external + content agreement).""" + db = SessionDB(db_path=tmp_path / "fresh.db") + try: + db.create_session(session_id="s1", source="cli") + db.append_message("s1", role="user", content="搜索大别山项目相关资料") + db.append_message("s1", role="tool", content="工具输出的大段内容在这里", + tool_name="web_search") + db.append_message("s1", role="assistant", content="assistant reply") + + # Trigram: user+assistant only; standard: everything. + assert db._conn.execute("SELECT COUNT(*) FROM messages_fts_trigram").fetchone()[0] == 2 + assert db._conn.execute("SELECT COUNT(*) FROM messages_fts").fetchone()[0] == 3 + + # Rewind-style UPDATE (active=0) must not desync the index — the + # triggers only fire on content/tool column changes. + def _deactivate(conn): + conn.execute("UPDATE messages SET active = 0 WHERE role = 'assistant'") + db._execute_write(_deactivate) + + # FTS5 integrity-check raises SQLITE_CORRUPT_VTAB on any + # index/content disagreement; passing = indexes are coherent. + db._conn.execute( + "INSERT INTO messages_fts(messages_fts, rank) VALUES('integrity-check', 1)" + ) + db._conn.execute( + "INSERT INTO messages_fts_trigram(messages_fts_trigram, rank) " + "VALUES('integrity-check', 1)" + ) + finally: + db.close() + + def test_v23_cjk_tool_role_filter_uses_like_fallback(self, tmp_path): + """A CJK query with role_filter=['tool'] must bypass the trigram index + (tool rows aren't in it) and still find matches via LIKE.""" + db = SessionDB(db_path=tmp_path / "fresh.db") + try: + db.create_session(session_id="s1", source="cli") + db.append_message("s1", role="tool", content="错误日志：数据库连接超时", + tool_name="terminal") + hits = db.search_messages("数据库连接", role_filter=["tool"]) + assert len(hits) == 1 + assert hits[0]["role"] == "tool" + finally: + db.close() + + def test_cjk_like_fallback_hides_rewound_messages(self, tmp_path): + """The CJK LIKE fallback must honor the same visibility rule as the + FTS5 paths: rewound rows (active=0, compacted=0) are hidden unless + include_inactive=True; compaction-archived rows (active=0, + compacted=1) stay discoverable (#38763).""" + db = SessionDB(db_path=tmp_path / "fresh.db") + try: + db.create_session(session_id="s1", source="cli") + db.append_message("s1", role="user", content="被撤销的搜索目标内容") + db.append_message("s1", role="user", content="被压缩归档的搜索目标内容") + + def _flags(conn): + # First row: rewound (active=0, compacted=0) — hidden. + conn.execute( + "UPDATE messages SET active = 0 WHERE content LIKE '%撤销%'" + ) + # Second row: compaction-archived (active=0, compacted=1) — visible. + conn.execute( + "UPDATE messages SET active = 0, compacted = 1 " + "WHERE content LIKE '%归档%'" + ) + db._execute_write(_flags) + + # Short-CJK query (2 chars — below the 3-char trigram minimum) + # forces the LIKE fallback; both rows contain the token 内容. + # search_messages strips full content from results — assert on + # the snippet column instead. + hits = db.search_messages("内容") + snippets = [h["snippet"] or "" for h in hits] + assert any("归档" in s for s in snippets), "archived row must stay visible" + assert not any("撤销" in s for s in snippets), "rewound row must be hidden" + + # include_inactive=True surfaces everything. + all_hits = db.search_messages("内容", include_inactive=True) + assert len(all_hits) == 2 + finally: + db.close() + + # --------------------------------------------------------------------------- # apply_wal_with_fallback — read-only probe tests # --------------------------------------------------------------------------- diff --git a/tests/test_state_db_malformed_repair.py b/tests/test_state_db_malformed_repair.py index 0de20e5a6602..b908eeea20ee 100644 --- a/tests/test_state_db_malformed_repair.py +++ b/tests/test_state_db_malformed_repair.py @@ -298,7 +298,17 @@ def test_fts_read_corruption_detected_by_read_probe(tmp_path): reason = _db_opens_cleanly(db_path) assert reason is not None assert "messages_fts" in reason - assert "malformed" in reason.lower() or "database disk image" in reason.lower() + # Message varies by SQLite build (same variance documented in + # SessionDB._is_fts_write_corruption_error): older builds raise the + # generic "database disk image is malformed"; newer builds raise the + # FTS5-specific 'fts5: corrupt structure record for table "..."'. + # Both are the same corruption class. + reason_l = reason.lower() + assert ( + "malformed" in reason_l + or "database disk image" in reason_l + or ("fts5" in reason_l and "corrupt" in reason_l) + ) def test_fts_read_corruption_repaired_in_place(tmp_path): diff --git a/tools/session_search_tool.py b/tools/session_search_tool.py index d4d168ec3fa3..aeb34d46f3b9 100644 --- a/tools/session_search_tool.py +++ b/tools/session_search_tool.py @@ -103,6 +103,28 @@ def _resolve_to_parent(db, session_id: str) -> str: return cur +def _annotate_rebuild_status(db, payload: Dict[str, Any]) -> None: + """Add a rebuild-progress note when the deferred FTS backfill (schema + v23) is still running, so the agent can tell the user why older results + may be incomplete/slower instead of treating a thin result set as + ground truth. No-op (and never raises) when no rebuild is pending.""" + try: + status = db.fts_rebuild_status() + except Exception: + return + if status is None: + return + payload["index_rebuild"] = { + "percent": status["percent"], + "note": ( + f"The search index is rebuilding in the background " + f"({status['percent']}% done, {status['indexed']:,} of " + f"{status['total']:,} messages). Results from older messages " + f"may be incomplete until it finishes." + ), + } + + def _order_for_recall(raw_results: List[Dict[str, Any]]) -> List[Dict[str, Any]]: """Stable-sort FTS rows so interactive sessions rank above automation. @@ -531,14 +553,16 @@ def _discover( raw_results = _order_for_recall(raw_results) if not raw_results and not title_result: - return json.dumps({ + _empty_payload = { "success": True, "mode": "discover", "query": query, "results": [], "count": 0, "message": "No matching sessions found.", - }, ensure_ascii=False) + } + _annotate_rebuild_status(db, _empty_payload) + return json.dumps(_empty_payload, ensure_ascii=False) # Dedupe by lineage. Keep the raw owning session_id on the surviving # row — only that pairs validly with the FTS5 match id for the anchored @@ -606,14 +630,16 @@ def _discover( entry["parent_session_id"] = lineage_root results.append(entry) - return json.dumps({ + _final_payload = { "success": True, "mode": "discover", "query": query, "results": results, "count": len(results), "sessions_searched": len(seen_sessions), - }, ensure_ascii=False) + } + _annotate_rebuild_status(db, _final_payload) + return json.dumps(_final_payload, ensure_ascii=False) def session_search(