fix: harden transcript append retry — lock, matcher, encapsulation, cap

Follow-up fixes for salvaged PR #65637:

1. Clear _dirty_transcripts in rewrite_transcript + rewind_session —
   stale pending messages were re-inserted after /retry, /undo, /compress.

2. Narrow _is_fts_corruption_error to specific SQLite error strings —
   bare 'fts' substring matched 'shifts', 'gifts', etc.

3. Move DB write outside _transcript_retry_lock — holding the lock
   during writes serialized all sessions' transcript appends and blocked
   during FTS rebuild. Now the lock guards only the pending queue.

4. Push rebuild_fts() into SessionDB — SessionStore was reaching into
   _conn/_lock private attrs. SessionDB.rebuild_fts() follows the same
   pattern as optimize_fts().

5. Cap pending per session at 200 — prevents unbounded memory growth
   when DB is persistently broken. Oldest messages dropped with warning.

Added 4 new tests: dirty-clear on rewrite/rewind, FTS matcher false
positives, pending cap enforcement.

gateway/session.py

@@ -2514,34 +2514,59 @@ class SessionStore:
         with self._transcript_retry_lock:
             pending = self._dirty_transcripts.setdefault(session_id, [])
             pending.append(dict(message))
-            while pending:
-                try:
-                    self._append_transcript_message(session_id, pending[0])
-                except Exception as exc:
-                    if self._is_fts_corruption_error(exc) and self._rebuild_fts_once():
-                        try:
-                            self._append_transcript_message(session_id, pending[0])
-                        except Exception as retry_exc:
-                            exc = retry_exc
-                        else:
-                            pending.pop(0)
-                            continue
+            # Cap pending messages per session to avoid unbounded memory
+            # growth when the DB is persistently broken. Drop the oldest.
+            if len(pending) > self._MAX_PENDING_PER_SESSION:
+                dropped = pending.pop(0)
+                logger.warning(
+                    "Session DB transcript pending queue full for %s "
+                    "(cap=%d); dropping oldest message to make room",
+                    session_id, self._MAX_PENDING_PER_SESSION,
+                )
+            # Snapshot the first pending message, then release the lock
+            # before the DB write so other sessions are not blocked.
+            msg = pending[0]
+        # DB write outside the retry lock — other sessions can append
+        # concurrently. We re-acquire the lock only to update the queue.
+        while True:
+            try:
+                self._append_transcript_message(session_id, msg)
+            except Exception as exc:
+                if self._is_fts_corruption_error(exc) and self._rebuild_fts_once():
+                    try:
+                        self._append_transcript_message(session_id, msg)
+                    except Exception as retry_exc:
+                        exc = retry_exc
+                    else:
+                        with self._transcript_retry_lock:
+                            if pending and pending[0] is msg:
+                                pending.pop(0)
+                            if not pending:
+                                self._dirty_transcripts.pop(session_id, None)
+                                self._transcript_append_failures.pop(session_id, None)
+                        continue
+                with self._transcript_retry_lock:
                     failures = self._transcript_append_failures.get(session_id, 0) + 1
                     self._transcript_append_failures[session_id] = failures
-                    logger.warning(
-                        "Session DB transcript append failed for %s "
-                        "(failure_count=%d, pending=%d); will retry: %s",
-                        session_id, failures, len(pending), exc,
-                    )
-                    break
-                else:
-                    pending.pop(0)
-            if not pending:
-                self._dirty_transcripts.pop(session_id, None)
-                self._transcript_append_failures.pop(session_id, None)
+                logger.warning(
+                    "Session DB transcript append failed for %s "
+                    "(failure_count=%d, pending=%d); will retry: %s",
+                    session_id, failures, len(pending), exc,
+                )
+                return
+            else:
+                with self._transcript_retry_lock:
+                    if pending and pending[0] is msg:
+                        pending.pop(0)
+                    if not pending:
+                        self._dirty_transcripts.pop(session_id, None)
+                        self._transcript_append_failures.pop(session_id, None)
+                        return
+                    msg = pending[0]
+                continue
 
     def _append_transcript_message(self, session_id: str, message: Dict[str, Any]) -> None:
-        """Write one transcript row. Caller serializes retries per session store."""
+        """Write one transcript row. Caller handles retry queuing."""
         self._db.append_message(
             session_id=session_id,
             role=message.get("role", "unknown"),
@@ -2559,40 +2584,65 @@ class SessionStore:
             timestamp=message.get("timestamp"),
         )
 
+    # Maximum in-memory pending messages per session before dropping the
+    # oldest. Prevents unbounded growth when the DB is persistently broken.
+    _MAX_PENDING_PER_SESSION = 200
+
     @staticmethod
     def _is_fts_corruption_error(exc: Exception) -> bool:
+        """True if *exc* looks like an FTS index corruption error.
+
+        Matches the specific SQLite error strings for malformed disk images
+        and FTS table corruption — not bare ``"fts"`` substrings which match
+        unrelated words like ``"shifts"`` or ``"gifts"``.
+        """
         text = str(exc).lower()
-        return "database disk image is malformed" in text or "fts" in text
+        return any(
+            marker in text
+            for marker in (
+                "database disk image is malformed",
+                "malformed database schema",
+                "messages_fts",
+                "no such table: messages_fts",
+            )
+        )
 
     def _rebuild_fts_once(self) -> bool:
-        """Attempt SQLite's documented FTS5 rebuild command once per store."""
+        """Attempt FTS5 ``rebuild`` command once per store lifetime.
+
+        Delegates to ``SessionDB.rebuild_fts()`` which handles locking and
+        table-existence checks internally. Returns ``True`` when at least
+        one index was rebuilt.
+        """
         if self._fts_rebuild_attempted:
             return False
         self._fts_rebuild_attempted = True
         db = self._db
-        conn = getattr(db, "_conn", None)
-        lock = getattr(db, "_lock", None)
-        if conn is None or lock is None:
+        if db is None or not hasattr(db, "rebuild_fts"):
             return False
-        rebuilt = 0
         try:
-            with lock:
-                for table in getattr(db, "_FTS_TABLES", ("messages_fts", "messages_fts_trigram")):
-                    try:
-                        if hasattr(db, "_fts_table_exists") and not db._fts_table_exists(table):
-                            continue
-                        conn.execute(f"INSERT INTO {table}({table}) VALUES('rebuild')")
-                        conn.commit()
-                        rebuilt += 1
-                    except Exception as exc:
-                        conn.rollback()
-                        logger.warning("Session DB FTS rebuild failed for %s: %s", table, exc)
+            rebuilt = db.rebuild_fts()
         except Exception as exc:
             logger.warning("Session DB FTS rebuild failed: %s", exc)
             return False
         if rebuilt:
-            logger.warning("Rebuilt %d Session DB FTS index(es) after append corruption", rebuilt)
+            logger.warning(
+                "Rebuilt %d Session DB FTS index(es) after append corruption",
+                rebuilt,
+            )
         return rebuilt > 0
+
+    def _clear_dirty_transcript(self, session_id: str) -> None:
+        """Drop queued pending messages for a session.
+
+        Called by ``rewrite_transcript`` and ``rewind_session`` so that
+        /retry, /undo, /compress — which replace or truncate the transcript —
+        don't leave stale messages that would be re-inserted on the next
+        append.
+        """
+        with self._transcript_retry_lock:
+            self._dirty_transcripts.pop(session_id, None)
+            self._transcript_append_failures.pop(session_id, None)
     
     def has_platform_message_id(
         self, session_id: str, platform_message_id: str
@@ -2628,6 +2678,7 @@ class SessionStore:
         """
         if not self._db:
             return True
+        self._clear_dirty_transcript(session_id)
         try:
             self._db.replace_messages(session_id, messages)
             return True
@@ -2670,6 +2721,7 @@ class SessionStore:
         """
         if not self._db:
             return None
+        self._clear_dirty_transcript(session_id)
         if n < 1:
             n = 1
         try:

hermes_state.py

@@ -7273,6 +7273,37 @@ class SessionDB:
                     )
         return optimized
 
+    def rebuild_fts(self) -> int:
+        """Rebuild FTS5 indexes from the canonical ``messages`` table.
+
+        Uses the FTS5 ``'rebuild'`` command, which rewrites the internal
+        b-tree segments from the content rows. This is the documented
+        recovery for a corrupt FTS index that rejects message writes while
+        reads still succeed (issue #50502). Unlike ``optimize_fts`` (which
+        merges existing segments), ``rebuild`` discards and recreates the
+        index data entirely.
+
+        Safe to call when FTS tables don't exist (skips them).
+        Returns the number of FTS indexes that were rebuilt.
+        """
+        rebuilt = 0
+        with self._lock:
+            for tbl in self._FTS_TABLES:
+                if not self._fts_table_exists(tbl):
+                    continue
+                try:
+                    self._conn.execute(
+                        f"INSERT INTO {tbl}({tbl}) VALUES('rebuild')"
+                    )
+                    self._conn.commit()
+                    rebuilt += 1
+                except sqlite3.OperationalError as exc:
+                    self._conn.rollback()
+                    logger.warning(
+                        "FTS rebuild failed for %s: %s", tbl, exc
+                    )
+        return rebuilt
+
     def vacuum(self) -> int:
         """Run VACUUM to reclaim disk space after large deletes.
 

tests/gateway/test_session.py

@@ -1663,30 +1663,15 @@ class TestGatewaySessionDbRecovery:
     def test_transcript_append_rebuilds_fts_and_retries_dirty_rows_in_order(self):
         import threading
 
-        class FakeConnection:
-            def __init__(self):
-                self.sql = []
-
-            def execute(self, sql):
-                self.sql.append(sql)
-
-            def commit(self):
-                pass
-
-            def rollback(self):
-                pass
-
         class FakeDb:
-            _FTS_TABLES = ("messages_fts",)
-
             def __init__(self):
-                self._lock = threading.Lock()
-                self._conn = FakeConnection()
                 self.attempts = []
                 self.persisted = []
+                self.rebuild_calls = 0
 
-            def _fts_table_exists(self, _table):
-                return True
+            def rebuild_fts(self):
+                self.rebuild_calls += 1
+                return 1
 
             def append_message(self, **kwargs):
                 content = kwargs["content"]
@@ -1704,14 +1689,127 @@ class TestGatewaySessionDbRecovery:
 
         store.append_to_transcript("s1", {"role": "user", "content": "first"})
         assert [m["content"] for m in store._dirty_transcripts["s1"]] == ["first"]
+        assert store._db.rebuild_calls == 1
 
         store.append_to_transcript("s1", {"role": "assistant", "content": "second"})
 
         assert store._db.persisted == ["first", "second"]
         assert "s1" not in store._dirty_transcripts
-        assert store._db._conn.sql == [
-            "INSERT INTO messages_fts(messages_fts) VALUES('rebuild')"
-        ]
+
+    def test_transcript_append_clears_dirty_on_rewrite(self):
+        """rewrite_transcript must clear pending dirty messages so /retry
+        and /compress don't re-insert replaced rows."""
+        import threading
+
+        class FakeDb:
+            def __init__(self):
+                self.persisted = []
+                self.replaced = []
+
+            def rebuild_fts(self):
+                return 0
+
+            def append_message(self, **kwargs):
+                raise RuntimeError("database disk image is malformed")
+
+            def replace_messages(self, session_id, messages):
+                self.replaced.append((session_id, messages))
+
+        store = object.__new__(SessionStore)
+        store._db = FakeDb()
+        store._transcript_retry_lock = threading.Lock()
+        store._dirty_transcripts = {}
+        store._transcript_append_failures = {}
+        store._fts_rebuild_attempted = True  # prevent rebuild attempt
+
+        # Queue a failed message
+        store.append_to_transcript("s1", {"role": "user", "content": "stale"})
+        assert "s1" in store._dirty_transcripts
+
+        # rewrite_transcript should clear the dirty queue
+        store.rewrite_transcript("s1", [{"role": "user", "content": "fresh"}])
+        assert "s1" not in store._dirty_transcripts
+        assert len(store._db.replaced) == 1
+
+    def test_transcript_append_clears_dirty_on_rewind(self):
+        """rewind_session must clear pending dirty messages so /undo
+        doesn't re-insert rewound rows."""
+        import threading
+
+        class FakeDb:
+            def __init__(self):
+                self.persisted = []
+
+            def rebuild_fts(self):
+                return 0
+
+            def append_message(self, **kwargs):
+                raise RuntimeError("database disk image is malformed")
+
+            def list_recent_user_messages(self, session_id, limit=10):
+                return [{"id": 1, "content": "old"}]
+
+            def rewind_to_message(self, session_id, target_id):
+                return {"target_message": {"id": target_id, "content": "old"}}
+
+        store = object.__new__(SessionStore)
+        store._db = FakeDb()
+        store._transcript_retry_lock = threading.Lock()
+        store._dirty_transcripts = {}
+        store._transcript_append_failures = {}
+        store._fts_rebuild_attempted = True
+
+        store.append_to_transcript("s1", {"role": "user", "content": "stale"})
+        assert "s1" in store._dirty_transcripts
+
+        store.rewind_session("s1", 1)
+        assert "s1" not in store._dirty_transcripts
+
+    def test_fts_corruption_error_does_not_match_false_positives(self):
+        """_is_fts_corruption_error must not match unrelated error strings
+        containing 'fts' as a substring (e.g. 'shifts', 'gifts')."""
+        assert SessionStore._is_fts_corruption_error(
+            RuntimeError("database disk image is malformed")
+        )
+        assert SessionStore._is_fts_corruption_error(
+            RuntimeError("no such table: messages_fts")
+        )
+        assert not SessionStore._is_fts_corruption_error(
+            RuntimeError("shifts were applied")
+        )
+        assert not SessionStore._is_fts_corruption_error(
+            RuntimeError("gifts received")
+        )
+
+    def test_pending_queue_caps_at_max(self):
+        """Pending queue should drop oldest messages when exceeding the cap
+        to prevent unbounded memory growth on persistent DB failure."""
+        import threading
+
+        class FakeDb:
+            def __init__(self):
+                self.count = 0
+
+            def rebuild_fts(self):
+                return 0
+
+            def append_message(self, **kwargs):
+                self.count += 1
+                raise RuntimeError("database disk image is malformed")
+
+        store = object.__new__(SessionStore)
+        store._db = FakeDb()
+        store._transcript_retry_lock = threading.Lock()
+        store._dirty_transcripts = {}
+        store._transcript_append_failures = {}
+        store._fts_rebuild_attempted = True
+
+        # Fill beyond the cap
+        for i in range(store._MAX_PENDING_PER_SESSION + 10):
+            store.append_to_transcript("s1", {"role": "user", "content": f"msg{i}"})
+
+        pending = store._dirty_transcripts.get("s1", [])
+        assert len(pending) <= store._MAX_PENDING_PER_SESSION
 
     def test_new_session_records_gateway_peer_fields(self, tmp_path):
         store = SessionStore(sessions_dir=tmp_path, config=GatewayConfig())