fix(gateway): trust SQLite codes for FTS recovery

gateway/session_transcript.py

@@ -6,6 +6,7 @@
 
 import contextlib
 import logging
+import re
 import threading
 from agent.turn_context import extract_api_content_sidecar
 from typing import TYPE_CHECKING, Any, Dict, List, Optional
@@ -15,6 +16,9 @@
 
 # Log-record parity with the origin module.
 logger = logging.getLogger("gateway.session")
+_MISSING_HERMES_FTS_TABLE_RE = re.compile(
+    r"no such table:\s+(?:main\.)?messages_fts(?:_trigram|_cjk)?$"
+)
 
 
 class TranscriptReadError(RuntimeError):
@@ -366,10 +370,18 @@ def _is_fts_corruption_error(exc: Exception) -> bool:
         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).
         """
-        if "messages_fts" in str(exc).lower():
-            return True
         import sqlite3
         from hermes_state import SessionDB
+        text = str(exc).lower()
+        if _MISSING_HERMES_FTS_TABLE_RE.fullmatch(text.strip()):
+            return True
+        # A live SQLite result code is stronger provenance than prose. In
+        # particular, a constraint failure can include an FTS table name
+        # without proving that the index itself is corrupt.
+        if getattr(exc, "sqlite_errorcode", None) is not None:
+            return isinstance(exc, sqlite3.DatabaseError) and SessionDB._is_fts_write_corruption_error(exc)
+        if "messages_fts" in text:
+            return True
         return isinstance(exc, sqlite3.DatabaseError) and SessionDB._is_fts_write_corruption_error(exc)
 
     def _rebuild_fts_once(self) -> bool:

tests/gateway/test_session.py

@@ -1560,11 +1560,41 @@ def test_fts_corruption_error_requires_fts_provenance(self):
         assert SessionStore._is_fts_corruption_error(
             RuntimeError("no such table: messages_fts")
         )
+        missing_fts = sqlite3.connect(":memory:")
+        try:
+            missing_fts.executescript(
+                "CREATE TABLE messages(content TEXT);"
+                "CREATE TRIGGER messages_fts_insert AFTER INSERT ON messages BEGIN "
+                "INSERT INTO messages_fts(content) VALUES (new.content); END;"
+            )
+            with pytest.raises(sqlite3.OperationalError) as missing:
+                missing_fts.execute("INSERT INTO messages(content) VALUES ('hello')")
+        finally:
+            missing_fts.close()
+        assert "no such table: main.messages_fts" in str(missing.value).lower()
+        assert SessionStore._is_fts_corruption_error(missing.value)
+        lookalike = sqlite3.connect(":memory:")
+        try:
+            lookalike.executescript(
+                "CREATE TABLE messages(content TEXT);"
+                "CREATE TRIGGER messages_fts_archive_insert AFTER INSERT ON messages BEGIN "
+                "INSERT INTO messages_fts_archive(content) VALUES (new.content); END;"
+            )
+            with pytest.raises(sqlite3.OperationalError) as missing_archive:
+                lookalike.execute("INSERT INTO messages(content) VALUES ('hello')")
+        finally:
+            lookalike.close()
+        assert not SessionStore._is_fts_corruption_error(missing_archive.value)
         assert SessionStore._is_fts_corruption_error(
             sqlite3.DatabaseError(
                 'fts5: corrupt structure record for table "messages_fts"'
             )
         )
+        contradictory = sqlite3.IntegrityError(
+            'fts5: corrupt structure record for table "messages_fts"'
+        )
+        contradictory.sqlite_errorcode = sqlite3.SQLITE_CONSTRAINT_TRIGGER
+        assert not SessionStore._is_fts_corruption_error(contradictory)
         assert not SessionStore._is_fts_corruption_error(
             RuntimeError("shifts were applied")
         )