fix(state): stop the housekeeping FTS retry from running on a quarantined SessionDB

retry_deferred_fts_recovery() is called unconditionally on every
housekeeping tick for the life of a long-running gateway process
(#100108). It checks _fts_stale, read_only, and _conn is None, but never
_db_corrupt (bcc2e65818, #101095/#101224): a handle that observed
structural corruption is supposed to stop being touched entirely (see
_try_wal_checkpoint's identical guard, and close()'s skip of the
checkpoint), but this method has no such check.

If a handle both has a deferred stale-FTS breadcrumb AND later trips
quarantine (both are plausible on the same corrupted file — the field
incidents motivating the quarantine feature describe corruption touching
FTS shadow tables and canonical btrees together), every subsequent
housekeeping tick runs a real FTS rebuild (DROP TABLE / CREATE VIRTUAL
TABLE / bulk INSERT) against the file the code has explicitly decided to
stop touching — exactly what quarantine exists to prevent.

Fixed by returning False immediately when _db_corrupt is set, mirroring
_try_wal_checkpoint's "quarantined: never touch a damaged image" guard.
The method's own contract ("never raises") is preserved — no
StateDbCorruptError is raised here, this is a quiet skip like the other
corrupt-aware call sites.

Also resets the backoff bookkeeping (_fts_stale_retry_after,
_fts_stale_retry_interval) in the same early-return, mirroring the
success path's own reset a few lines down (review feedback from
Baophan00 on the PR). Verified empirically before making this change:
_db_corrupt is set to False nowhere in the codebase outside __init__, and
the shared registry's file-replace path always constructs a genuinely new
SessionDB instance rather than clearing the flag on a live one — so no
code path today revives a quarantined handle in place, and leaving the
backoff fields untouched is inert in practice. The reset is still cheap,
harmless, and closes a real footgun for whoever adds an un-quarantine
path later: without it, a handle quarantined mid-backoff would carry a
doubled multi-minute interval into any future retry instead of starting
from the default.

Added a regression test that marks a handle stale, forces the open-time
recovery to defer via a real held rebuild lock (so _fts_stale survives
construction), sets _db_corrupt plus a pre-existing multi-minute backoff,
and asserts the retry is a no-op with both backoff fields reset to 0.0.
Mutation-verified: reverting hermes_state_schema.py makes the retry
actually run the rebuild and return True, and separately makes the
backoff-reset assertions fail with the stale pre-quarantine values still
in place.

(cherry picked from commit 3445da1d98d84bb60cb3799ef59e1fa4c619100f)

hermes_state_schema.py

@@ -681,6 +681,28 @@ class SessionSchemaMixin:
         """
         if not getattr(self, "_fts_stale", False):
             return False
+        if getattr(self, "_db_corrupt", False):
+            # Quarantined: structural corruption was already observed on
+            # this handle, so the only safe policy is to stop touching the
+            # file (mirrors hermes_state.py's _try_wal_checkpoint). A full
+            # FTS rebuild is real DDL/DML against the same damaged image —
+            # exactly what quarantine exists to prevent — and this method
+            # is called unconditionally every housekeeping tick for the
+            # life of a long-running gateway process, so a stale-FTS flag
+            # left set on a now-corrupt handle would otherwise retry the
+            # rebuild forever.
+            #
+            # Reset the backoff bookkeeping too (mirrors the success path's
+            # own reset a few lines down): no code path today clears
+            # _db_corrupt on a live handle, so this is inert in practice,
+            # but leaving a doubled _fts_stale_retry_interval sitting behind
+            # a flag nothing currently clears is a footgun for whoever adds
+            # an un-quarantine/recovery path later — the next real retry
+            # should start from the default backoff, not wherever this
+            # handle's interval happened to be when it was quarantined.
+            self._fts_stale_retry_after = 0.0
+            self._fts_stale_retry_interval = 0.0
+            return False
         if getattr(self, "read_only", False) or getattr(self, "_conn", None) is None:
             return False
         now = time.monotonic()

tests/state/test_fts_rebuild_admission.py

@@ -622,3 +622,48 @@ class TestDeferredFtsRetryInProcess:
             assert ro.retry_deferred_fts_recovery() is False
         finally:
             ro.close()
+
+    def test_retry_skips_quarantined_handle(self, tmp_path, fast_timeout):
+        """A structurally corrupt handle must never run a full FTS rebuild —
+        the housekeeping tick calls this unconditionally for the life of a
+        long-running gateway process, so a stale-FTS flag left set on a
+        now-corrupt handle must not retry the rebuild forever against the
+        damaged image (real DDL/DML the quarantine exists to prevent)."""
+        db_path = tmp_path / "state.db"
+        d = SessionDB(db_path=db_path)
+        if not d._fts_enabled:
+            d.close()
+            pytest.skip("FTS5 unavailable in this build")
+        d.create_session("s1", source="test")
+        d.append_message("s1", "user", "hello quarantine")
+        d.close()
+        self._mark_stale(db_path)
+
+        # Force the open-time recovery to defer (foreign rebuild-lock
+        # holder) so _fts_stale is still True once the handle is open —
+        # mirrors test_retry_is_non_blocking_while_live_holder_and_backs_off.
+        with _rebuild_lock_held_by_other_process(db_path):
+            gw = SessionDB(db_path=db_path)
+        try:
+            assert gw._fts_stale is True
+            gw._db_corrupt = True
+            gw._db_corrupt_reason = "database disk image is malformed"
+            # Simulate a handle that had already been backing off for a
+            # while before it tripped quarantine.
+            gw._fts_stale_retry_after = time.monotonic() + 900.0
+            gw._fts_stale_retry_interval = 900.0
+            assert gw.retry_deferred_fts_recovery() is False
+            # Untouched: still marked stale, triggers still absent — no
+            # rebuild ran against the "damaged" handle.
+            assert gw._fts_stale is True
+            # The backoff bookkeeping is reset too, mirroring the success
+            # path's own reset — a doubled interval left behind a flag
+            # nothing currently clears would otherwise make the next real
+            # retry (if this handle is ever un-quarantined) start from a
+            # stale multi-minute backoff instead of the default.
+            assert gw._fts_stale_retry_after == 0.0
+            assert gw._fts_stale_retry_interval == 0.0
+        finally:
+            gw.close()
+        assert _meta_value(db_path, FTS_STALE_KEY) == "1"
+        assert _base_fts_triggers(db_path) == set()