fix(state): complete the bounded-merge protocol — usermerge floor, progress-bounded continuation, tolerate mid-rebuild missing index

Follow-up on top of #65554 (@the3asic):

- Lower FTS5 'usermerge' to its minimum of 2 (persisted in the config
  shadow table, applied once per SessionDB instance). Without this a
  positive-rank 'merge' skips any level holding fewer than 4 segments
  (SQLite FTS5 §6.8), so the fragmented-index case the cadence targets
  never converges.
- Run up to _FTS_MERGE_COMMANDS_PER_PASS (4) bounded merge commands per
  index per cadence, stopping early on the documented no-progress
  signal (total_changes delta < 2). Each command is its own implicit
  transaction, so the write lock is released between commands.
- Skip a missing messages_fts instead of raising: the chunked
  optimize-storage rebuild legitimately drops + backfills FTS tables
  while writers keep running; warning every 1000 writes for the whole
  backfill window would be noise, and optimize_fts() has always
  treated missing tables as skippable.
- Replace traced-SQL shape assertions with behavioral tests: real
  fragmented-index convergence (automerge suppressed, 60 segments) and
  a 3-segment below-default-usermerge compaction test that fails
  against a bare positive-rank merge (sabotage-verified).

Validation on a sqlite3.backup() copy of a real 10.7 GB production
state.db (1.49M messages, 1.3 GB + 2.9 GB FTS shadow tables):
worst per-command write-lock hold 41.8 ms (was 9.2 s / 18.1 s per
index with 'optimize'), search results byte-identical, fts5
integrity-check and PRAGMA integrity_check clean, steady-state pass
0.0 ms.

hermes_state.py

@@ -1976,13 +1976,21 @@ class SessionDB:
     _WRITE_RETRY_MAX_S = 0.150   # 150ms
     # Attempt a WAL checkpoint every N successful writes (PASSIVE mode).
     _CHECKPOINT_EVERY_N_WRITES = 50
-    # 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.
+    # Retain the existing coarse 1000-write maintenance cadence, but replace
+    # the unbounded FTS5 ``'optimize'`` (measured holding the write lock for
+    # 9-18 s per index on a 10 GB production DB — longer than a competing
+    # writer's full retry patience, surfacing as "database is locked" /
+    # session_persistence_failed) with bounded ``'merge'`` commands. A
+    # positive merge rank is an approximate output-page budget, so each
+    # command holds the write lock for milliseconds; up to
+    # ``_FTS_MERGE_COMMANDS_PER_PASS`` commands run per index per cadence,
+    # stopping early on the documented no-progress signal. ``usermerge`` is
+    # lowered to 2 so positive merges act on any level with >= 2 segments —
+    # without that, levels below the default threshold of 4 are skipped and
+    # a fragmented index never converges (SQLite FTS5 §6.8-6.9).
     _FTS_MERGE_EVERY_N_WRITES = 1000
     _FTS_MERGE_MAX_PAGES_PER_INDEX = 500
+    _FTS_MERGE_COMMANDS_PER_PASS = 4
     # 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
@@ -2020,6 +2028,9 @@ class SessionDB:
         # in place at most once per SessionDB instance so a genuinely
         # unrecoverable database can't put writers into a rebuild loop.
         self._fts_runtime_rebuild_attempted = False
+        # One-shot guard for the usermerge-floor config write on the
+        # incremental FTS merge cadence (see _merge_fts_incrementally).
+        self._fts_usermerge_floor_applied = False
         self._fts_enabled = False
         self._trigram_available = False
         # CJK-bigram index (cjk_unicode61 loadable tokenizer). _fts_cjk_loaded:
@@ -11295,41 +11306,76 @@ 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.
+    def _merge_fts_incrementally(
+        self, *, max_pages: int, max_commands: Optional[int] = None
+    ) -> int:
+        """Run bounded FTS5 ``'merge'`` commands against each present index.
 
-        A positive 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.
+        A positive merge rank tells SQLite to stop after approximately that
+        many output pages, so each command holds the write lock for
+        milliseconds regardless of index size — unlike ``'optimize'``, which
+        rewrites the whole index in one transaction (measured 9-18 s per
+        index on a 10 GB production DB, long enough to exhaust a competing
+        writer's entire lock-retry patience).
+
+        Protocol (SQLite FTS5 §6.8-6.9):
+
+        - ``usermerge`` is lowered to its minimum of 2 (persisted in the
+          ``%_config`` shadow table, applied once per instance) so a
+          positive merge acts on ANY level holding >= 2 segments. With the
+          default of 4, levels below that threshold are never merged by a
+          positive-rank command and a fragmented index cannot converge.
+        - Up to *max_commands* merge commands run per index, stopping early
+          on the documented no-progress signal: the delta in
+          ``total_changes`` is < 2 (the command's own INSERT accounts
+          for 1 change; >= 2 means real merge work happened).
+
+        Each command is its own implicit transaction (the connection runs
+        with ``isolation_level=None``), so the SQLite write lock is released
+        between commands and competing processes can interleave writes
+        mid-pass. Missing tables are valid schema variants (FTS variants are
+        optional, and ``optimize_fts_storage`` legitimately drops + backfills
+        these tables while writers keep running) and are skipped, mirroring
+        ``optimize_fts``. Other SQLite errors propagate to the caller.
+
+        Returns the number of merge commands executed.
         """
         if isinstance(max_pages, bool) or not isinstance(max_pages, int):
             raise TypeError("max_pages must be an integer")
         if max_pages <= 0:
             raise ValueError("max_pages must be greater than zero")
+        if max_commands is None:
+            max_commands = self._FTS_MERGE_COMMANDS_PER_PASS
+        if isinstance(max_commands, bool) or not isinstance(max_commands, int):
+            raise TypeError("max_commands must be an integer")
+        if max_commands <= 0:
+            raise ValueError("max_commands must be greater than zero")
 
-        merged = 0
+        executed = 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:
+                if not self._fts_table_exists(tbl):
                     continue
-                self._conn.execute(
-                    f"INSERT INTO {tbl}({tbl}, rank) VALUES('merge', ?)",
-                    (max_pages,),
-                )
-                merged += 1
-        return merged
+                # One-time (per instance) usermerge floor; the value is
+                # persisted in the index's config shadow table so future
+                # connections inherit it. Setting config is a metadata-only
+                # write — it never touches segment data.
+                if not getattr(self, "_fts_usermerge_floor_applied", False):
+                    self._conn.execute(
+                        f"INSERT INTO {tbl}({tbl}, rank) "
+                        "VALUES('usermerge', 2)"
+                    )
+                for _ in range(max_commands):
+                    before = self._conn.total_changes
+                    self._conn.execute(
+                        f"INSERT INTO {tbl}({tbl}, rank) VALUES('merge', ?)",
+                        (max_pages,),
+                    )
+                    executed += 1
+                    if self._conn.total_changes - before < 2:
+                        break
+            self._fts_usermerge_floor_applied = True
+        return executed
 
     def vacuum(self) -> int:
         """Run VACUUM to reclaim disk space after large deletes.

tests/test_hermes_state.py

@@ -5474,27 +5474,115 @@ class TestOptimizeFts:
         # Search still works after repeated optimization.
         assert len(db.search_messages("repeat")) == 1
 
-    def test_incremental_merge_is_one_bounded_command_per_present_index(self, db):
+    def test_incremental_merge_bounded_commands_per_present_index(self, db):
+        """Each pass issues bounded 'merge' commands, never 'optimize'."""
         db.create_session(session_id="s1", source="cli")
         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
+            executed = db._merge_fts_incrementally(max_pages=37)
         finally:
             db._conn.set_trace_callback(None)
 
+        # At least one merge command per present FTS index, and never more
+        # than the per-pass command cap per index.
+        present = [t for t in db._FTS_TABLES if db._fts_table_exists(t)]
+        assert len(present) >= 2  # messages_fts + trigram on a fresh DB
         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 len(merge_sql) == executed
+        assert len(present) <= executed <= (
+            len(present) * db._FTS_MERGE_COMMANDS_PER_PASS
+        )
+        for tbl in present:
+            n = sum(f"{tbl}({tbl}, rank)" in sql for sql in merge_sql)
+            assert 1 <= n <= db._FTS_MERGE_COMMANDS_PER_PASS
+        # The usermerge floor is applied so positive merges can make
+        # progress on levels with >= 2 segments (SQLite FTS5 §6.8).
+        assert any("VALUES('usermerge', 2)" in sql for sql in statements)
         assert not any("'optimize'" in sql for sql in statements)
 
+    def test_incremental_merge_converges_on_fragmented_index(self, db):
+        """Bounded passes make real progress on a fragmented index and
+        reach a no-more-work steady state — the failure mode of a bare
+        positive-rank merge (usermerge default 4) is that it never merges
+        anything and the index stays fragmented forever."""
+        db.create_session(session_id="s1", source="cli")
+        # Suppress automerge so every insert leaves its own level-0 segment
+        # — a deliberately fragmented index that only explicit merge
+        # commands can compact. Config is scoped to this test's temp DB.
+        with db._lock:
+            for tbl in ("messages_fts", "messages_fts_trigram"):
+                db._conn.execute(
+                    f"INSERT INTO {tbl}({tbl}, rank) VALUES('automerge', 0)"
+                )
+        for i in range(60):
+            db.append_message(
+                session_id="s1", role="user",
+                content=f"fragment needle {i} lorem ipsum dolor",
+            )
+
+        # First pass must do real merge work (shadow-table rows change).
+        before = db._conn.total_changes
+        executed_first = db._merge_fts_incrementally(max_pages=500)
+        assert executed_first >= 2
+        assert db._conn.total_changes - before > executed_first  # real work
+
+        # Repeated passes converge: eventually a pass issues exactly one
+        # no-progress command per present index and stops early.
+        present = sum(1 for t in db._FTS_TABLES if db._fts_table_exists(t))
+        for _ in range(50):
+            if db._merge_fts_incrementally(max_pages=500) == present:
+                break
+        else:
+            pytest.fail("bounded merge passes never converged")
+
+        # Merging is layout-only: every row is still searchable.
+        assert len(db.search_messages("fragment", limit=100)) == 60
+
+    def test_incremental_merge_compacts_below_default_usermerge(self, db):
+        """A level with only 2-3 segments — below FTS5's default usermerge
+        threshold of 4 — must still get merged. A bare positive-rank
+        'merge' skips such levels entirely (SQLite FTS5 §6.8), which is
+        why the cadence lowers usermerge to 2 first; without the floor,
+        light fragmentation persists forever and every MATCH pays for it."""
+
+        def _segment_count(tbl: str) -> int:
+            # FTS5 structure record: id=10 in the %_data shadow table.
+            # Format: 4-byte cookie, then varint nlevel, varint nsegment...
+            # nsegment fits in one varint byte for small indexes; parse the
+            # second varint (single-byte values < 128 read directly).
+            blob = db._conn.execute(
+                f"SELECT block FROM {tbl}_data WHERE id=10"
+            ).fetchone()[0]
+            pos = 4
+            for _ in range(1):  # skip nlevel varint (single byte here)
+                assert blob[pos] < 0x80
+                pos += 1
+            assert blob[pos] < 0x80
+            return blob[pos]
+
+        db.create_session(session_id="s1", source="cli")
+        with db._lock:
+            db._conn.execute(
+                "INSERT INTO messages_fts(messages_fts, rank) "
+                "VALUES('automerge', 0)"
+            )
+        # Exactly 3 level-0 segments: below the default usermerge of 4.
+        for i in range(3):
+            db.append_message(
+                session_id="s1", role="user", content=f"sparse token{i}"
+            )
+        assert _segment_count("messages_fts") == 3
+
+        for _ in range(10):
+            db._merge_fts_incrementally(max_pages=500)
+
+        assert _segment_count("messages_fts") == 1, (
+            "3-segment level was not compacted — usermerge floor missing?"
+        )
+        assert len(db.search_messages("sparse")) == 3
+
     def test_incremental_merge_skips_absent_optional_index(self, db):
         with db._lock:
             for trigger in (
@@ -5508,14 +5596,18 @@ class TestOptimizeFts:
         statements = []
         db._conn.set_trace_callback(statements.append)
         try:
-            assert db._merge_fts_incrementally(max_pages=19) == 1
+            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]
+        assert merge_sql
+        assert all("messages_fts(messages_fts, rank)" in sql for sql in merge_sql)
 
-    def test_incremental_merge_does_not_hide_missing_required_index(self, db):
+    def test_incremental_merge_skips_missing_primary_index(self, db):
+        """A missing messages_fts is a valid transient state (chunked
+        optimize-storage rebuild drops + backfills it while writers keep
+        running) — the cadence must skip it, not raise a warning every
+        1000 writes for the whole backfill window."""
         with db._lock:
             for trigger in (
                 "messages_fts_insert",
@@ -5525,11 +5617,16 @@ class TestOptimizeFts:
                 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)
+        statements = []
+        db._conn.set_trace_callback(statements.append)
+        try:
+            executed = db._merge_fts_incrementally(max_pages=19)
+        finally:
+            db._conn.set_trace_callback(None)
+        assert executed >= 1  # trigram index still present and merged
+        assert not any(
+            "messages_fts(messages_fts, rank)" in sql for sql in statements
+        )
 
     def test_write_path_merges_fts_only_at_cadence_boundary(self, db, monkeypatch):
         """Routine writes use bounded merge and never full optimize."""