fix(state): the deferred FTS rebuild retry is quarantined by the same rule as the checkpoints

retry_deferred_fts_recovery gated only on _db_corrupt ("mirrors _try_wal_checkpoint /
close") — after this PR it no longer mirrored them: on a replaced/lost-generation handle
the periodic housekeeping tick still ran FTS DDL/DML + commit, the same split-brain write
class as the #105670 checkpoint. One SessionDB._quarantine_reason() now decides for the
periodic checkpoint, close(), and the FTS retry, with the halt path's precedence
(replaced before generation loss) and the operator wording in one place.

Test: the periodic-checkpoint case folds into the close test (same setup), which now
also proves the FTS retry returns False without touching the file; the mutation with
main's schema sibling swapped in returns True (a rebuild ran).

hermes_state.py

@@ -1052,7 +1052,8 @@ class SessionDB(
     def _disable_close_time_checkpoint(self) -> None:
         """Best-effort SQLITE_DBCONFIG_NO_CKPT_ON_CLOSE (Python 3.12+): sqlite3's
         close() otherwise runs the internal last-connection checkpoint that wrote
-        the incident's pages under wrong page numbers (see StateDbCorruptError).
+        the incident's pages under wrong page numbers (see StateDbCorruptError and
+        the generation-loss halts).
         <3.12 has no setconfig; the residual checkpoint only carries
         pre-quarantine committed frames, which is tolerable."""
         flag = getattr(sqlite3, "SQLITE_DBCONFIG_NO_CKPT_ON_CLOSE", None)
@@ -1090,6 +1091,19 @@ class SessionDB(
         """Foreign processes holding this DB or its WAL sidecars (see hermes_state_holders)."""
         return _foreign_state_db_holders(self.db_path)
 
+    def _quarantine_reason(self) -> Optional[str]:
+        """Why this handle must not checkpoint or run in-file repair, or None. A corrupted image has
+        torn B-trees; a replaced file or a deleted/replaced WAL generation would checkpoint under
+        wrong page numbers into the main DB -- the shutdown-time cause of #105670. Same precedence
+        as the halt path (replaced is checked before generation loss)."""
+        if self._db_corrupt:
+            return f"structural corruption ({self._db_corrupt_reason})"
+        if self._db_replaced:
+            return "a replaced state.db file"
+        if self._db_wal_generation_lost:
+            return "a deleted WAL generation (split-brain)"
+        return None
+
     def _try_wal_checkpoint(self) -> None:
         """Best-effort PASSIVE WAL checkpoint; never raises. PASSIVE never blocks writers;
         TRUNCATE corrupted B-trees on 65K+ page databases under exclusive-lock I/O pressure.
@@ -1097,8 +1111,7 @@ class SessionDB(
         Previous TRUNCATE strategy caused B-tree corruption on large databases (65K+ pages) due to the
         exclusive-lock I/O pressure from checkpointing thousands of frames at once (issue #45383).
         """
-        # Quarantined: never checkpoint over a damaged image or a stale/replaced generation.
-        if self._db_corrupt or self._db_wal_generation_lost or self._db_replaced:
+        if self._quarantine_reason() is not None:
             return
         try:
             with self._lock:
@@ -1153,16 +1166,8 @@ class SessionDB(
             pass
         with self._lock:
             if self._conn:
-                # Quarantined handles must not checkpoint: a corrupted image has torn B-trees, and a
-                # stale (deleted/replaced) WAL generation would checkpoint under wrong page numbers
-                # into the main DB -- the shutdown-time cause of #105670.
-                quarantine_reason = (
-                    f"structural corruption ({self._db_corrupt_reason})" if self._db_corrupt
-                    else "a deleted WAL generation (split-brain)" if self._db_wal_generation_lost
-                    else "a replaced state.db file" if self._db_replaced
-                    else None
-                )
-                if quarantine_reason:
+                quarantine_reason = self._quarantine_reason()
+                if quarantine_reason is not None:
                     logger.warning(
                         "Skipping the close-time WAL checkpoint for %s: this "
                         "handle observed %s. Take a snapshot of state.db, -wal and -shm "

hermes_state_schema.py

@@ -503,9 +503,10 @@ class SessionSchemaMixin:
         """
         if not self._fts_stale:
             return False
-        if getattr(self, "_db_corrupt", False):
-            # Quarantined: never run FTS DDL/DML against a damaged image (mirrors _try_wal_checkpoint /
-            # close). Reset the backoff so a future un-quarantine starts from the default interval.
+        if self._quarantine_reason() is not None:
+            # Quarantined: never run FTS DDL/DML against a damaged image or a stale/replaced generation
+            # (mirrors _try_wal_checkpoint / close). Reset the backoff so a future un-quarantine starts
+            # from the default interval.
             self._fts_stale_retry_after = 0.0
             self._fts_stale_retry_interval = 0.0
             return False

tests/hermes_state/test_deleted_wal_checkpoint_guard.py

@@ -62,7 +62,7 @@ def _unlink_sidecars(db_path: Path) -> None:
 
 @pytest.mark.linux_only  # deleted-WAL write halt uses Linux unlink semantics
 def test_close_after_halt_runs_no_checkpoint(tmp_path, force_wal):
-    """A writer halted by DeletedWalGenerationError must not checkpoint on close()."""
+    """A writer halted by DeletedWalGenerationError must not checkpoint (periodic or close) nor run FTS repair."""
     path = tmp_path / "state.db"
     db = _make_db(path, "s", "before")
     _require_wal(db)
@@ -72,8 +72,11 @@ def test_close_after_halt_runs_no_checkpoint(tmp_path, force_wal):
         db.append_message("s", role="user", content="after-unlink")
     assert db._db_wal_generation_lost is True
 
-    # close() must not run the explicit PRAGMA wal_checkpoint(PASSIVE).
+    # Neither the periodic checkpoint, the stale-FTS retry, nor close() may touch the file now.
     with patch.object(db._conn, "execute", wraps=db._conn.execute) as mock_execute:
+        db._try_wal_checkpoint()
+        db._fts_stale = True
+        assert db.retry_deferred_fts_recovery() is False
         db.close()
         # No checkpoint call should have been made.
         checkpoint_calls = [
@@ -119,29 +122,3 @@ def test_halt_disables_close_time_checkpoint(tmp_path, force_wal):
             "halt did not call setconfig(SQLITE_DBCONFIG_NO_CKPT_ON_CLOSE, True)"
         )
     db.close()
-
-
-@pytest.mark.linux_only  # deleted-WAL write halt uses Linux unlink semantics
-def test_try_wal_checkpoint_skips_when_generation_lost(tmp_path, force_wal):
-    """The periodic _try_wal_checkpoint() must skip when _db_wal_generation_lost is set."""
-    path = tmp_path / "state.db"
-    db = _make_db(path, "s", "before")
-    _require_wal(db)
-    _unlink_sidecars(path)
-
-    with pytest.raises(DeletedWalGenerationError):
-        db.append_message("s", role="user", content="after-unlink")
-    assert db._db_wal_generation_lost is True
-
-    # Directly call _try_wal_checkpoint() — it must skip silently.
-    with patch.object(db._conn, "execute", wraps=db._conn.execute) as mock_execute:
-        db._try_wal_checkpoint()
-        checkpoint_calls = [
-            call
-            for call in mock_execute.call_args_list
-            if "wal_checkpoint" in str(call).lower()
-        ]
-        assert not checkpoint_calls, (
-            "_try_wal_checkpoint() ran a checkpoint on a quarantined handle"
-        )
-    db.close()