fix(gateway): require FTS provenance before transcript rebuild-and-retry

Widen #96038's fail-closed classifier to the gateway transcript retry
path: SessionStore._is_fts_corruption_error no longer treats a generic
'database disk image is malformed' as FTS-only damage. It now delegates
to SessionDB._is_fts_write_corruption_error (SQLITE_CORRUPT_VTAB result
code or explicit fts5 corrupt-structure text) and only keeps the
messages_fts-named cases. Structural corruption falls through to the
bounded retry/backoff path instead of rebuilding FTS and retrying writes
against a damaged database.

Sibling site spotted in PR #98090 by @fangliquanflq.

contributors/emails/diatche@gmail.com

@@ -0,0 +1 @@
+diatche

gateway/session.py

@@ -3930,22 +3930,28 @@ class SessionStore:
 
     @staticmethod
     def _is_fts_corruption_error(exc: Exception) -> bool:
-        """True if *exc* looks like an FTS index corruption error.
+        """True only when the failure is provably scoped to the FTS index.
 
-        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"``.
+        A generic ``database disk image is malformed`` (bare SQLITE_CORRUPT)
+        can mean structural damage to canonical B-trees, not just the FTS
+        shadow tables — treating it as FTS-only here made the store rebuild
+        the index and retry transcript writes against a structurally corrupt
+        database (#97940). Only errors that name ``messages_fts`` or carry
+        FTS provenance per ``SessionDB._is_fts_write_corruption_error``
+        (``SQLITE_CORRUPT_VTAB`` result code, or explicit ``fts5:`` corrupt
+        structure text) may authorize the one-shot rebuild-and-retry.
+        Everything else falls through to the bounded retry/backoff path.
         """
         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",
-            )
-        )
+        if "messages_fts" in text:
+            return True
+        import sqlite3
+
+        from hermes_state import SessionDB
+
+        if isinstance(exc, sqlite3.DatabaseError):
+            return SessionDB._is_fts_write_corruption_error(exc)
+        return False
 
     def _rebuild_fts_once(self) -> bool:
         """Attempt FTS5 ``rebuild`` command once per store lifetime.

tests/gateway/test_session.py

@@ -1541,15 +1541,29 @@ class TestGatewaySessionDbRecovery:
         assert "child" 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
+    def test_fts_corruption_error_requires_fts_provenance(self):
+        """_is_fts_corruption_error must not treat a generic malformed-image
+        error as FTS-scoped (#97940): bare SQLITE_CORRUPT can mean canonical
+        B-tree damage. It must also not match unrelated error strings
         containing 'fts' as a substring (e.g. 'shifts', 'gifts')."""
-        assert SessionStore._is_fts_corruption_error(
+        import sqlite3
+
+        # Generic structural corruption: no FTS provenance -> fail closed.
+        assert not SessionStore._is_fts_corruption_error(
             RuntimeError("database disk image is malformed")
         )
+        assert not SessionStore._is_fts_corruption_error(
+            sqlite3.DatabaseError("database disk image is malformed")
+        )
+        # FTS-scoped errors remain eligible for the one-shot rebuild.
         assert SessionStore._is_fts_corruption_error(
             RuntimeError("no such table: messages_fts")
         )
+        assert SessionStore._is_fts_corruption_error(
+            sqlite3.DatabaseError(
+                'fts5: corrupt structure record for table "messages_fts"'
+            )
+        )
         assert not SessionStore._is_fts_corruption_error(
             RuntimeError("shifts were applied")
         )