fix(gateway): retry failed transcript appends and rebuild corrupted FTS

gateway/session.py

@@ -1028,6 +1028,10 @@ def __init__(self, sessions_dir: Path, config: GatewayConfig,
         self._persisted_routing_generation = 0
         self._inflight_lock = threading.Lock()
         self._inflight_sessions: Dict[str, _SessionFlight] = {}
+        self._transcript_retry_lock = threading.Lock()
+        self._dirty_transcripts: Dict[str, List[Dict[str, Any]]] = {}
+        self._transcript_append_failures: Dict[str, int] = {}
+        self._fts_rebuild_attempted = False
         self._has_active_processes_fn = has_active_processes_fn
         # Whether to keep writing the legacy sessions.json mirror alongside
         # the primary gateway_routing table in state.db. Default True for
@@ -2505,31 +2509,140 @@ def append_to_transcript(self, session_id: str, message: Dict[str, Any], skip_db
                      _flush_messages_to_session_db(), preventing the
                      duplicate-write bug (#860).
         """
-        if self._db and not skip_db:
+        if not self._db or skip_db:
+            return
+        with self._transcript_retry_lock:
+            pending = self._dirty_transcripts.setdefault(session_id, [])
+            pending.append(dict(message))
+            # 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._db.append_message(
-                    session_id=session_id,
-                    role=message.get("role", "unknown"),
-                    content=message.get("content"),
-                    tool_name=message.get("tool_name"),
-                    tool_calls=message.get("tool_calls"),
-                    tool_call_id=message.get("tool_call_id"),
-                    reasoning=message.get("reasoning") if message.get("role") == "assistant" else None,
-                    reasoning_content=message.get("reasoning_content") if message.get("role") == "assistant" else None,
-                    reasoning_details=message.get("reasoning_details") if message.get("role") == "assistant" else None,
-                    codex_reasoning_items=message.get("codex_reasoning_items") if message.get("role") == "assistant" else None,
-                    codex_message_items=message.get("codex_message_items") if message.get("role") == "assistant" else None,
-                    # Platform-side message id (yuanbao msg_id, telegram update_id, …).
-                    # Accept either explicit ``platform_message_id`` or the legacy
-                    # ``message_id`` key the JSONL transcript used.
-                    platform_message_id=(
-                        message.get("platform_message_id") or message.get("message_id")
-                    ),
-                    observed=bool(message.get("observed")),
-                    timestamp=message.get("timestamp"),
+                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,
                 )
-            except Exception as e:
-                logger.debug("Session DB operation failed: %s", e)
+                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 handles retry queuing."""
+        self._db.append_message(
+            session_id=session_id,
+            role=message.get("role", "unknown"),
+            content=message.get("content"),
+            tool_name=message.get("tool_name"),
+            tool_calls=message.get("tool_calls"),
+            tool_call_id=message.get("tool_call_id"),
+            reasoning=message.get("reasoning") if message.get("role") == "assistant" else None,
+            reasoning_content=message.get("reasoning_content") if message.get("role") == "assistant" else None,
+            reasoning_details=message.get("reasoning_details") if message.get("role") == "assistant" else None,
+            codex_reasoning_items=message.get("codex_reasoning_items") if message.get("role") == "assistant" else None,
+            codex_message_items=message.get("codex_message_items") if message.get("role") == "assistant" else None,
+            platform_message_id=(message.get("platform_message_id") or message.get("message_id")),
+            observed=bool(message.get("observed")),
+            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 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 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
+        if db is None or not hasattr(db, "rebuild_fts"):
+            return False
+        try:
+            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,
+            )
+        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
@@ -2565,6 +2678,7 @@ def rewrite_transcript(self, session_id: str, messages: List[Dict[str, Any]]) ->
         """
         if not self._db:
             return True
+        self._clear_dirty_transcript(session_id)
         try:
             self._db.replace_messages(session_id, messages)
             return True
@@ -2607,6 +2721,7 @@ def rewind_session(self, session_id: str, n: int = 1) -> Optional[Dict[str, Any]
         """
         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 @@ def optimize_fts(self) -> int:
                     )
         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

@@ -1660,6 +1660,157 @@ def flaky_encode(cls, content):
 
 
 class TestGatewaySessionDbRecovery:
+    def test_transcript_append_rebuilds_fts_and_retries_dirty_rows_in_order(self):
+        import threading
+
+        class FakeDb:
+            def __init__(self):
+                self.attempts = []
+                self.persisted = []
+                self.rebuild_calls = 0
+
+            def rebuild_fts(self):
+                self.rebuild_calls += 1
+                return 1
+
+            def append_message(self, **kwargs):
+                content = kwargs["content"]
+                self.attempts.append(content)
+                if len(self.attempts) <= 2:
+                    raise RuntimeError("database disk image is malformed")
+                self.persisted.append(content)
+
+        store = object.__new__(SessionStore)
+        store._db = FakeDb()
+        store._transcript_retry_lock = threading.Lock()
+        store._dirty_transcripts = {}
+        store._transcript_append_failures = {}
+        store._fts_rebuild_attempted = False
+
+        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
+
+    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())
         source = SessionSource(