fix(state): bound routine FTS merge work

hermes_state.py

@@ -1976,16 +1976,13 @@ class SessionDB:
     _WRITE_RETRY_MAX_S = 0.150   # 150ms
     # Attempt a WAL checkpoint every N successful writes (PASSIVE mode).
     _CHECKPOINT_EVERY_N_WRITES = 50
-    # Merge fragmented FTS5 segments every N successful writes. The message
-    # triggers append one segment per insert; left unmaintained these grow
-    # into tens of thousands of segments, so every MATCH must scan them all
-    # and every insert pays a growing automerge cost — which lengthens the
-    # write-lock hold time and starves competing writers (gateway + cron
-    # processes share one state.db), surfacing as "database is locked".
-    # 'optimize' is a no-op once the index is already merged, so an idle DB
-    # pays almost nothing; the cadence is deliberately coarse so the one-off
-    # merge cost is amortised far below the checkpoint cadence.
-    _OPTIMIZE_EVERY_N_WRITES = 1000
+    # Retain the existing coarse 1000-write maintenance cadence, but give each
+    # FTS5 index only one positive-rank merge command per pass. SQLite treats
+    # that rank as an approximate output-page budget. A 500-page budget keeps
+    # routine work small while automerge handles normal background compaction;
+    # unlike ``optimize``, it cannot chase every remaining segment in one write.
+    _FTS_MERGE_EVERY_N_WRITES = 1000
+    _FTS_MERGE_MAX_PAGES_PER_INDEX = 500
     # Session imports intentionally use a lower cap than exports: import holds
     # one BEGIN IMMEDIATE transaction, so bounded batches avoid starving live
     # gateway/CLI writers. The dashboard accepts one exported JSON/JSONL file
@@ -2611,8 +2608,8 @@ class SessionDB:
                 self._write_count += 1
                 if self._write_count % self._CHECKPOINT_EVERY_N_WRITES == 0:
                     self._try_wal_checkpoint()
-                if self._write_count % self._OPTIMIZE_EVERY_N_WRITES == 0:
-                    self._try_optimize_fts()
+                if self._write_count % self._FTS_MERGE_EVERY_N_WRITES == 0:
+                    self._try_incremental_merge_fts()
                 return result
             except sqlite3.OperationalError as exc:
                 err_msg = str(exc).lower()
@@ -2740,21 +2737,19 @@ class SessionDB:
         except Exception as exc:
             logger.warning("WAL checkpoint (PASSIVE) failed: %s", exc)
 
-    def _try_optimize_fts(self) -> None:
-        """Best-effort FTS5 segment merge. Never raises.
-
-        Runs on the ``_OPTIMIZE_EVERY_N_WRITES`` cadence from the write hot
-        path (off the lock — ``optimize_fts`` re-acquires ``self._lock``
-        itself, mirroring ``_try_wal_checkpoint``). ``read_only`` connections
-        never reach the write path, so this is implicitly skipped for them.
-        Once the index is merged the 'optimize' command is close to free, so
-        the steady-state cost is negligible; the expensive case is only the
-        first merge of a long-neglected index.
-        """
+    def _try_incremental_merge_fts(self) -> None:
+        """Run one bounded FTS5 merge pass without failing the completed write."""
+        if not self._fts_enabled:
+            return
         try:
-            self.optimize_fts()
-        except Exception:
-            pass  # Best effort — never fatal.
+            self._merge_fts_incrementally(
+                max_pages=self._FTS_MERGE_MAX_PAGES_PER_INDEX
+            )
+        except sqlite3.Error as exc:
+            # Routine maintenance is best effort, but unexpected SQLite errors
+            # must remain visible instead of being silently mistaken for an
+            # optional missing index.
+            logger.warning("FTS incremental merge failed: %s", exc)
 
     def close(self):
         """Close the database connection.
@@ -11300,6 +11295,42 @@ class SessionDB:
             logger.debug("Could not read logical DB size: %s", exc)
             return None
 
+    def _merge_fts_incrementally(self, *, max_pages: int) -> int:
+        """Issue one positive-rank FTS5 ``merge`` per present index.
+
+        A positive rank tells SQLite to stop after approximately that many
+        output pages. This method deliberately does not loop: one invocation
+        executes at most one merge command for each table in ``_FTS_TABLES``.
+        Missing tables are valid schema variants; all other SQLite errors
+        propagate to the caller.
+        """
+        if isinstance(max_pages, bool) or not isinstance(max_pages, int):
+            raise TypeError("max_pages must be an integer")
+        if max_pages <= 0:
+            raise ValueError("max_pages must be greater than zero")
+
+        merged = 0
+        with self._lock:
+            rows = self._conn.execute(
+                "SELECT name FROM sqlite_master "
+                "WHERE type = 'table' AND name IN (?, ?)",
+                self._FTS_TABLES,
+            ).fetchall()
+            present = {row[0] for row in rows}
+            if "messages_fts" not in present:
+                raise sqlite3.OperationalError(
+                    "required FTS5 index is missing: messages_fts"
+                )
+            for tbl in self._FTS_TABLES:
+                if tbl not in present:
+                    continue
+                self._conn.execute(
+                    f"INSERT INTO {tbl}({tbl}, rank) VALUES('merge', ?)",
+                    (max_pages,),
+                )
+                merged += 1
+        return merged
+
     def vacuum(self) -> int:
         """Run VACUUM to reclaim disk space after large deletes.
 

tests/test_hermes_state.py

@@ -5414,7 +5414,15 @@ class TestOptimizeFts:
         """A fresh DB has both FTS indexes; optimize merges both."""
         db.create_session(session_id="s1", source="cli")
         db.append_message(session_id="s1", role="user", content="hello world")
-        assert db.optimize_fts() == 2
+        statements = []
+        db._conn.set_trace_callback(statements.append)
+        try:
+            assert db.optimize_fts() == 2
+        finally:
+            db._conn.set_trace_callback(None)
+        optimize_sql = [sql for sql in statements if "'optimize'" in sql]
+        assert len(optimize_sql) == 2
+        assert not any("'merge'" in sql for sql in optimize_sql)
 
     def test_optimize_preserves_search_and_snippet(self, db):
         """Optimize is layout-only: MATCH results + snippets are unchanged."""
@@ -5466,39 +5474,103 @@ class TestOptimizeFts:
         # Search still works after repeated optimization.
         assert len(db.search_messages("repeat")) == 1
 
-    def test_write_path_optimizes_fts_on_cadence(self, db, monkeypatch):
-        """Writes periodically merge FTS segments so they never accumulate
-        into the tens-of-thousands that lengthen the write-lock hold and
-        starve competing writers ("database is locked")."""
-        db._OPTIMIZE_EVERY_N_WRITES = 5
-        calls = {"n": 0}
-        real_optimize = db.optimize_fts
-
-        def _counting_optimize():
-            calls["n"] += 1
-            return real_optimize()
-
-        monkeypatch.setattr(db, "optimize_fts", _counting_optimize)
-        # create_session is write #1; appends are #2.. -> #5 and #10 trigger.
+    def test_incremental_merge_is_one_bounded_command_per_present_index(self, db):
         db.create_session(session_id="s1", source="cli")
-        for i in range(9):
+        db.append_message(session_id="s1", role="user", content="bounded merge")
+        statements = []
+        db._conn.set_trace_callback(statements.append)
+        try:
+            assert db._merge_fts_incrementally(max_pages=37) == 2
+        finally:
+            db._conn.set_trace_callback(None)
+
+        merge_sql = [sql for sql in statements if "VALUES('merge', 37)" in sql]
+        assert len(merge_sql) == 2
+        assert sum(
+            "messages_fts(messages_fts, rank)" in sql for sql in merge_sql
+        ) == 1
+        assert sum(
+            "messages_fts_trigram(messages_fts_trigram, rank)" in sql
+            for sql in merge_sql
+        ) == 1
+        assert not any("'optimize'" in sql for sql in statements)
+
+    def test_incremental_merge_skips_absent_optional_index(self, db):
+        with db._lock:
+            for trigger in (
+                "messages_fts_trigram_insert",
+                "messages_fts_trigram_delete",
+                "messages_fts_trigram_update",
+            ):
+                db._conn.execute(f"DROP TRIGGER {trigger}")
+            db._conn.execute("DROP TABLE messages_fts_trigram")
+
+        statements = []
+        db._conn.set_trace_callback(statements.append)
+        try:
+            assert db._merge_fts_incrementally(max_pages=19) == 1
+        finally:
+            db._conn.set_trace_callback(None)
+        merge_sql = [sql for sql in statements if "VALUES('merge', 19)" in sql]
+        assert len(merge_sql) == 1
+        assert "messages_fts(messages_fts, rank)" in merge_sql[0]
+
+    def test_incremental_merge_does_not_hide_missing_required_index(self, db):
+        with db._lock:
+            for trigger in (
+                "messages_fts_insert",
+                "messages_fts_delete",
+                "messages_fts_update",
+            ):
+                db._conn.execute(f"DROP TRIGGER {trigger}")
+            db._conn.execute("DROP TABLE messages_fts")
+
+        with pytest.raises(
+            sqlite3.OperationalError,
+            match="required FTS5 index is missing: messages_fts",
+        ):
+            db._merge_fts_incrementally(max_pages=19)
+
+    def test_write_path_merges_fts_only_at_cadence_boundary(self, db, monkeypatch):
+        """Routine writes use bounded merge and never full optimize."""
+        db._FTS_MERGE_EVERY_N_WRITES = 5
+        calls = []
+
+        def _counting_merge(*, max_pages):
+            calls.append(max_pages)
+            return 0
+
+        def _unexpected_optimize():
+            raise AssertionError("routine cadence must not call optimize")
+
+        monkeypatch.setattr(db, "_merge_fts_incrementally", _counting_merge)
+        monkeypatch.setattr(db, "optimize_fts", _unexpected_optimize)
+        db.create_session(session_id="s1", source="cli")
+        for i in range(3):
             db.append_message(session_id="s1", role="user", content=f"needle {i}")
-        assert calls["n"] == 2
-        # The auto-merge is layout-only: search is unaffected.
+        assert calls == []  # Four successful writes are below the boundary.
+        db.append_message(session_id="s1", role="user", content="needle 3")
+        assert calls == [500]  # The fifth write gets the production page budget.
+        for i in range(4, 8):
+            db.append_message(session_id="s1", role="user", content=f"needle {i}")
+        assert calls == [500]
+        db.append_message(session_id="s1", role="user", content="needle 8")
+        assert calls == [500, 500]  # The tenth write is the next boundary.
         assert len(db.search_messages("needle")) == 9
 
-    def test_write_path_optimize_failure_never_breaks_write(self, db, monkeypatch):
-        """A failing periodic optimize must not fail the surrounding write."""
-        db._OPTIMIZE_EVERY_N_WRITES = 2
+    def test_write_path_merge_failure_is_logged_without_breaking_write(
+        self, db, monkeypatch, caplog
+    ):
+        db._FTS_MERGE_EVERY_N_WRITES = 2
 
-        def _boom():
-            raise sqlite3.OperationalError("simulated optimize failure")
+        def _boom(*, max_pages):
+            raise sqlite3.OperationalError("simulated merge failure")
 
-        monkeypatch.setattr(db, "optimize_fts", _boom)
+        monkeypatch.setattr(db, "_merge_fts_incrementally", _boom)
         db.create_session(session_id="s1", source="cli")  # write #1
-        # write #2 trips the cadence; the swallowed failure must not propagate.
         db.append_message(session_id="s1", role="user", content="still persists")
         assert len(db.get_messages("s1")) == 1
+        assert "FTS incremental merge failed: simulated merge failure" in caplog.text
 
 
 class TestAutoMaintenance: