fix(state): classify FTS-scoped corruption as fts_index, never whole-file damage

classify_persistence_error bucketed every _DB_CORRUPTION_MARKERS hit as "corrupt",
so an error SQLite itself scoped to the FTS5 index layer (SQLITE_CORRUPT_VTAB, or an
`fts5: corrupt structure record for table "messages_fts"` report) that escaped the
write path — the detach in _enter_fts_fail_open refused (generation/lock check),
or a read/search path with no fail-open at all — reached the turn boundary and the
gateway startup notice as structural corruption: the turn ended with `.recover` /
restore-backup advice on a file whose canonical tables were provably healthy.

One provenance rule, hermes_state_errors.is_fts_scoped_corruption_error, now feeds
both the write-repair gate (SessionDB._is_fts_write_corruption_error delegates to it,
so the gateway transcript retry inherits it) and the classifier: a known result code
outranks prose (only SQLITE_CORRUPT_VTAB is FTS-scoped; bare SQLITE_CORRUPT/NOTADB
and any contradictory code fail closed), and without a code the text must both carry
a corruption marker and name a messages_fts* object. The new "fts_index" cause
renders index-scoped guidance (doctor --fix / restart, do not run recovery) in the
turn explainer and the home-channel notice. The structural fail-close is untouched:
bare malformed / not-a-database still quarantine and still classify "corrupt".

Salvaged from PR #97843 (SulthanZahran1), trimmed: the quick_check-backed
"corrupt_unconfirmed" tier is dropped — on a live handle that just observed an
unscoped SQLITE_CORRUPT, PRAGMA quick_check on a damaged shadow b-tree raises rather
than reports on 3.53.1, so the probe could never downgrade the exact shape it was
built for, and a verdict that softens quarantine guidance on prose alone weakens the
fail-close. #97841 (Finn763) reached the same fts_index cause via text markers
only; its LIKE-degradation intent already lives in _search_messages_impl (_fts_stale).

Fixes #97794
Co-authored-by: finn763 <165816600+finn763@users.noreply.github.com>

agent/turn_explainers.py

@@ -151,6 +151,16 @@ _PERSISTENCE_CAUSE_EXPLANATIONS: Dict[str, str] = {
         "3. Restore from a backup in {backups_dir}/\n"
         "Then send your message again."
     ),
+    # SQLite scoped the corruption to the FTS index and the derived indexes could not be
+    # detached, so this write did not land; the message store itself is intact (#97794).
+    "fts_index": (
+        "the turn was stopped because the session search index (FTS5) "
+        "is corrupt and could not be detached, so this message was not "
+        "saved. The message store itself is not damaged: do not run "
+        "recovery tools or restore a backup. Run `hermes doctor --fix` "
+        "(or restart Hermes, which repairs the index on open), then "
+        "send your message again."
+    ),
     "disk": (
         "the turn was stopped because session storage could not "
         "be written (the transcript would have been lost on "

contributors/emails/zsulthan9@gmail.com

@@ -0,0 +1,2 @@
+SulthanZahran1
+# PR #97843 salvage

gateway/run_notifications.py

@@ -835,7 +835,8 @@ class GatewayNotificationsMixin:
                 return
         from hermes_constants import get_default_hermes_root
         from hermes_state import _default_db_path, classify_persistence_error, format_session_db_unavailable
-        if classify_persistence_error(error) == "corrupt":
+        cause = classify_persistence_error(error)
+        if cause == "corrupt":
             # Copy-pasteable, so name the real store (profiles / HERMES_HOME do not live under ~/.hermes).
             db_path = _default_db_path()
             backups_dir = get_default_hermes_root() / "backups"
@@ -854,6 +855,15 @@ class GatewayNotificationsMixin:
                 f"3. Restore from a backup in {backups_dir}/\n"
                 "Run `hermes doctor` for sanitized diagnostics."
             )
+        elif cause == "fts_index":
+            # Index-scoped corruption: the message tables are not damaged, so the recover /
+            # restore advice above would be destructive on a healthy file (#97794).
+            message = (
+                "⚠️ Session database reported a corruption error confined to the search index "
+                "(FTS5); the message tables are not damaged. Messages may not be persisted until "
+                "it is repaired: run `hermes doctor --fix`, then restart the gateway. Do not run "
+                "recovery tools or restore a backup unless `hermes doctor` confirms damage."
+            )
         else:
             message = (
                 f"⚠️ Session database unavailable — messages may not be persisted. "

hermes_state_errors.py

@@ -3,6 +3,7 @@ Shared by hermes_state and its mixins; string predicates match wrapped RPC
 strings as well as live sqlite3 exceptions."""
 
 import errno
+import re
 import sqlite3
 
 # Malformed schema: ``sqlite_master`` itself is inconsistent (typically a DUPLICATE
@@ -74,8 +75,8 @@ def is_disk_full_error(exc: BaseException | str | None) -> bool:
 
 # Every classify_persistence_error bucket; consumers enumerate this tuple.
 PERSISTENCE_ERROR_CAUSES = (
-    "locked", "compression", "compression_closed", "turn_lease", "corrupt", "replaced",
-    "deleted_wal", "disk", "unknown",
+    "locked", "compression", "compression_closed", "turn_lease", "corrupt", "fts_index",
+    "replaced", "deleted_wal", "disk", "unknown",
 )
 
 
@@ -88,6 +89,38 @@ _DB_CORRUPTION_MARKERS = (
     "malformed", "file is not a database", "not a database", "database corruption",
 )
 
+# The module constant exists on Python 3.11+; the numeric value is stable across SQLite releases.
+SQLITE_CORRUPT_VTAB = getattr(sqlite3, "SQLITE_CORRUPT_VTAB", 267)
+
+# Every FTS object hangs off this prefix: the virtual tables and their _data/_idx/_content/
+# _docsize/_config shadow b-trees. FTS5 names the table in its own corruption reports.
+_FTS_OBJECT_RE = re.compile(r"\bmessages_fts\w*")
+
+
+def is_fts_scoped_corruption_error(exc_or_str) -> bool:
+    """Corruption SQLite itself attributes to the FTS index layer: the ONE provenance rule
+    shared by the write-repair gate (``SessionDB._is_fts_write_corruption_error``), the
+    gateway transcript retry and :func:`classify_persistence_error` (#96038, #97794).
+
+    A known result code outranks prose: ``SQLITE_CORRUPT_VTAB`` is FTS-scoped even with
+    the generic malformed-image text older SQLite builds emit, while bare ``SQLITE_CORRUPT``
+    / ``SQLITE_NOTADB`` carry no object scope and any other known code contradicts
+    FTS-looking prose, so both fail closed. Only without a code (Python < 3.11, RPC-wrapped
+    strings) does the text decide, and then only an ``fts5:`` corruption report or a
+    corruption marker that names a ``messages_fts*`` object counts.
+    """
+    if exc_or_str is None:
+        return False
+    code = getattr(exc_or_str, "sqlite_errorcode", None)
+    if code is not None:
+        return code == SQLITE_CORRUPT_VTAB
+    text = (exc_or_str if isinstance(exc_or_str, str) else str(exc_or_str)).lower()
+    if not _FTS_OBJECT_RE.search(text):
+        return False
+    if text.startswith("fts5:") and "corrupt" in text:
+        return True
+    return any(marker in text for marker in _DB_CORRUPTION_MARKERS)
+
 
 class CompressionSessionClosedError(RuntimeError):
     """A durable write targeted a parent already closed by compression."""
@@ -205,8 +238,10 @@ def classify_persistence_error(exc_or_str) -> str:
     matches: "locked" = busy, retry; "disk" = full/read-only/permissions;
     "compression" = a live lease refused the write; "compression_closed" = adopt
     the rotated session id; "turn_lease" = fencing, not storage; "corrupt" =
-    file damage (repair path, not disk space); "replaced" = main-file replacement;
-    "deleted_wal" = a retired sidecar generation requiring capture inspection."""
+    file damage (repair path, not disk space); "fts_index" = SQLite scoped the
+    corruption to the FTS index (the transcript store is not damaged); "replaced" =
+    main-file replacement; "deleted_wal" = a retired sidecar generation requiring
+    capture inspection."""
     if exc_or_str is None:
         return "unknown"
     # Lease refusals contain neither "locked" nor "busy": match by type first,
@@ -216,6 +251,10 @@ def classify_persistence_error(exc_or_str) -> str:
     for exc_type, cause in _PERSISTENCE_CAUSE_BY_TYPE:
         if isinstance(exc_or_str, exc_type):
             return cause
+    # Provenance before prose: an FTS-scoped result code (or, without one, an fts5 report
+    # naming messages_fts*) is index damage, never whole-file corruption (#97794).
+    if is_fts_scoped_corruption_error(exc_or_str):
+        return "fts_index"
     text = str(exc_or_str).lower()
     for markers, cause in _PERSISTENCE_CAUSE_BY_PHRASE:
         if any(marker in text for marker in markers):

hermes_state_fts.py

@@ -10,6 +10,7 @@ from typing import Sequence
 
 from hermes_constants import get_hermes_home
 from hermes_state_common import FTS_CJK_STALE_KEY, FTS_STALE_KEY, _FTS_CJK_TRIGGERS, _FTS_TRIGGERS
+from hermes_state_errors import is_fts_scoped_corruption_error
 
 # caplog tests pin the "hermes_state" logger name.
 logger = logging.getLogger("hermes_state")
@@ -341,13 +342,11 @@ class SessionFtsSetupMixin:
 
     @staticmethod
     def _is_fts_write_corruption_error(exc: sqlite3.DatabaseError) -> bool:
-        """Corruption SQLite identifies as FTS-scoped (SQLITE_CORRUPT_VTAB, or an
-        ``fts5:`` message on older builds); a bare malformed image is structural."""
-        error_code = getattr(exc, "sqlite_errorcode", None)
-        if error_code is not None:
-            return error_code == getattr(sqlite3, "SQLITE_CORRUPT_VTAB", 267)
-        msg = str(exc).lower()
-        return msg.startswith("fts5:") and "corrupt structure" in msg
+        """Corruption SQLite identifies as FTS-scoped (SQLITE_CORRUPT_VTAB, or an ``fts5:``
+        report naming ``messages_fts*`` on builds without result codes); a bare malformed
+        image is structural. One rule, shared with ``classify_persistence_error`` and the
+        gateway transcript retry: see :func:`hermes_state_errors.is_fts_scoped_corruption_error`."""
+        return is_fts_scoped_corruption_error(exc)
 
     def _enter_fts_fail_open(self, exc: sqlite3.DatabaseError) -> bool:
         """Detach corrupt FTS indexes so canonical writes can continue. Breadcrumb +

tests/run_agent/test_turn_completion_explainer.py

@@ -173,6 +173,26 @@ def test_explanation_persistence_corrupt_backups_dir_follows_hermes_home(monkeyp
     assert "{backups_dir}" not in out
 
 
+def test_explanation_persistence_fts_index_never_advises_recovery():
+    """#97794: an FTS-scoped failure must never send the user down the recover /
+    restore-backup path on a healthy file, and must not claim the transcript was lost."""
+    out = AIAgent._format_turn_completion_explanation(
+        "session_persistence_failed", "fts_index"
+    )
+    lower = out.lower()
+    assert out.strip() != ""
+    assert "sessions recover" not in lower
+    assert ".recover" not in lower
+    # Negative advice ("do not ... restore a backup") is fine; instructions are not.
+    assert "recovery options" not in lower
+    assert "restore from a backup" not in lower and "backups/" not in lower
+    assert "would have been lost" not in lower
+    assert "free" not in lower  # never disk-space advice
+    assert "hermes doctor" in lower
+    assert "search index" in lower and "not damaged" in lower
+    assert "send your message again" in lower  # the handle stays live
+
+
 def test_explanation_persistence_replaced_cause_forbids_inplace_repair():
     out = AIAgent._format_turn_completion_explanation(
         "session_persistence_failed", "replaced"
@@ -350,6 +370,7 @@ def test_persistence_error_causes_tuple_matches_classifier():
         "Session 'abc' is being compressed by another writer",
         "Session turn lease lost; refusing transcript write for 'abc'",
         "database disk image is malformed",
+        'fts5: corrupt structure record for table "messages_fts"',
         "FATAL: state.db was replaced underneath the gateway",
         "FATAL: a live process holds a deleted state.db-wal or state.db-shm inode.",
         "database or disk is full",
@@ -360,6 +381,63 @@ def test_persistence_error_causes_tuple_matches_classifier():
         assert classify_persistence_error(probe) in PERSISTENCE_ERROR_CAUSES
 
 
+def test_classify_persistence_error_fts_provenance_order():
+    """Result code first, prose only without one — the #96038 rule the write-repair gate
+    already enforces, now shared with the classifier so there is one definition of
+    "provably FTS-only" (#97794 review)."""
+    import sqlite3
+
+    from hermes_state import SessionDB, classify_persistence_error
+    from hermes_state_errors import SQLITE_CORRUPT_VTAB, is_fts_scoped_corruption_error
+
+    def _err(text, code=None, cls=sqlite3.DatabaseError):
+        exc = cls(text)
+        if code is not None:
+            exc.sqlite_errorcode = code
+        return exc
+
+    # Tier 1 — a known result code decides. SQLITE_CORRUPT_VTAB is FTS-scoped even with the
+    # generic text older SQLite builds emit; bare SQLITE_CORRUPT / SQLITE_NOTADB are unscoped.
+    vtab = _err("database disk image is malformed", SQLITE_CORRUPT_VTAB)
+    assert classify_persistence_error(vtab) == "fts_index"
+    assert SessionDB._is_fts_write_corruption_error(vtab)  # same verdict as the write gate
+    assert classify_persistence_error(
+        _err("database disk image is malformed", sqlite3.SQLITE_CORRUPT)
+    ) == "corrupt"
+    assert classify_persistence_error(
+        _err("file is not a database", sqlite3.SQLITE_NOTADB)
+    ) == "corrupt"
+    # A contradictory known code outranks FTS-looking prose (the #96038 regression shape).
+    contradictory = _err(
+        'fts5: corrupt structure record for table "messages_fts"',
+        sqlite3.SQLITE_CONSTRAINT_TRIGGER,
+        sqlite3.IntegrityError,
+    )
+    assert not is_fts_scoped_corruption_error(contradictory)
+    assert classify_persistence_error(contradictory) != "fts_index"
+    assert not SessionDB._is_fts_write_corruption_error(contradictory)
+
+    # Tier 2 — no code (Python < 3.11, RPC-wrapped strings): the report must name a
+    # messages_fts* object. Both shapes from #97794's evidence logs qualify.
+    assert classify_persistence_error(
+        _err('fts5: corrupt structure record for table "messages_fts"')
+    ) == "fts_index"
+    assert classify_persistence_error(
+        'fts5: corruption found reading blob 2061584302081 from table "messages_fts"'
+    ) == "fts_index"
+    assert classify_persistence_error(
+        'fts5: corrupt structure record for table "messages_fts_trigram"'
+    ) == "fts_index"
+    assert classify_persistence_error(
+        "malformed inverted index for FTS5 table main.messages_fts"
+    ) == "fts_index"
+    # Generic markers without provenance stay conservative; fts5 text without corruption,
+    # or an FTS name without a corruption marker, is not corruption at all.
+    assert classify_persistence_error("database disk image is malformed") == "corrupt"
+    assert classify_persistence_error('fts5: syntax error near "x"') == "unknown"
+    assert classify_persistence_error("no such table: messages_fts") == "unknown"
+
+
 # --------------------------------------------------------------------------
 # 2. Enable/disable seam
 # --------------------------------------------------------------------------

tests/state/test_fts_index_fail_open.py

@@ -0,0 +1,133 @@
+"""Regression tests for #97794: an FTS5-index-only failure must not kill the turn, and an
+FTS-scoped error that escapes must not be rendered as whole-file damage.
+
+The write path already fails open on provenance-proven FTS corruption (detach the derived
+indexes, retry the canonical write) and quarantines the handle on unscoped corruption
+(#97940 / #90837). These tests pin the contract at the boundaries the issue was filed
+against — the agent flush whose failure ends the turn, and the cause that drives the
+user-facing guidance:
+
+* the flush succeeds after an FTS-only stomp and the exact user message is durable in
+  ``messages`` (the turn proceeds);
+* an FTS-scoped error that still escapes (detach refused) classifies as ``fts_index`` and
+  never quarantines the handle;
+"""
+
+import sqlite3
+from types import SimpleNamespace
+
+import pytest
+
+from hermes_state import SessionDB
+from run_agent import AIAgent
+
+
+def _flush_agent(db, session_id):
+    """Bind the real flush methods onto a stand-in over a live SessionDB."""
+    agent = SimpleNamespace(
+        _session_db=db,
+        _session_db_created=True,
+        _persist_disabled=False,
+        session_id=session_id,
+        _session_persist_lock=None,
+        _flushed_db_message_ids=set(),
+        _flushed_db_message_session_id=None,
+        _last_flushed_db_idx=0,
+        _db_flush_scan_prefix=None,
+        _persist_user_message_idx=None,
+        _persist_user_message_override=None,
+        _persist_user_message_timestamp=None,
+        _pending_cli_user_message=None,
+        _active_session_turn_lease_holder=None,
+        _last_persistence_error_cause=None,
+        _compression_adoption_failed=False,
+    )
+    agent._ensure_db_session = lambda: None
+    agent._flush_messages_to_session_db = (
+        AIAgent._flush_messages_to_session_db.__get__(agent, AIAgent)
+    )
+    agent._flush_messages_to_session_db_unlocked = (
+        AIAgent._flush_messages_to_session_db_unlocked.__get__(agent, AIAgent)
+    )
+    return agent
+
+
+def _seed(db, rows=60):
+    if not db._fts_enabled:
+        pytest.skip("FTS5 unavailable in this build")
+    db.create_session("s1", source="cli")
+    for i in range(rows):
+        db.append_message("s1", "user", f"seed row {i} " + "z" * 200)
+
+
+def _stomp_fts_shadow(db_path):
+    """Overwrite the messages_fts shadow b-tree blocks: FTS5 raises SQLITE_CORRUPT_VTAB on the
+    next MATCH / sync-trigger insert while every canonical row stays intact."""
+    raw = sqlite3.connect(str(db_path))
+    raw.execute("UPDATE messages_fts_data SET block = X'DEADBEEFDEADBEEFDEADBEEFDEADBEEF'")
+    raw.commit()
+    raw.close()
+
+
+def _contents(db_path):
+    raw = sqlite3.connect(str(db_path))
+    try:
+        return [r[0] for r in raw.execute("SELECT content FROM messages ORDER BY id").fetchall()]
+    finally:
+        raw.close()
+
+
+def test_turn_flush_survives_fts_only_corruption(tmp_path):
+    """The turn's transcript write succeeds after an FTS-only stomp: the flush reports
+    success (the turn proceeds, no ``session_persistence_failed``) and the exact user
+    message is durable in ``messages``. The derived indexes are detached, the handle is
+    not quarantined."""
+    db_path = tmp_path / "state.db"
+    db = SessionDB(db_path=db_path)
+    try:
+        _seed(db)
+        _stomp_fts_shadow(db_path)
+        agent = _flush_agent(db, "s1")
+
+        ok = agent._flush_messages_to_session_db(
+            [{"role": "user", "content": "lands after stomp"}], []
+        )
+
+        assert ok is True
+        assert agent._last_persistence_error_cause is None
+        assert _contents(db_path)[-1] == "lands after stomp"
+        assert db._db_corrupt is False
+        # Builds whose sync trigger walks the stomped structure record detach the derived
+        # indexes; builds that defer the read pass the insert through untouched. Either way
+        # the canonical write landed, which is the contract.
+        assert db._fts_stale in (True, False)
+        assert db.get_session("s1") is not None
+    finally:
+        db.close()
+
+
+def test_escaped_fts_only_error_is_index_scoped_not_quarantined(tmp_path, monkeypatch):
+    """When the detach itself is refused the FTS-scoped error escapes to the agent. It must
+    classify as ``fts_index`` (guidance names the index, not the file) and must not
+    quarantine the handle or touch the derived indexes."""
+    db_path = tmp_path / "state.db"
+    db = SessionDB(db_path=db_path)
+    try:
+        _seed(db)
+        _stomp_fts_shadow(db_path)
+        monkeypatch.setattr(db, "_enter_fts_fail_open", lambda exc: False)
+        agent = _flush_agent(db, "s1")
+
+        ok = agent._flush_messages_to_session_db(
+            [{"role": "user", "content": "refused detach"}], []
+        )
+        if ok is True:
+            pytest.skip("this SQLite build defers FTS shadow corruption past the insert trigger")
+
+        assert ok is False
+        assert agent._last_persistence_error_cause == "fts_index"
+        assert db._db_corrupt is False
+        assert db._fts_stale is False
+        assert "refused detach" not in _contents(db_path)
+    finally:
+        db.close()