fix(state): defer FTS rebuild under foreign WAL holders

hermes_state.py

@@ -4211,6 +4211,59 @@ class SessionDB(SessionSearchMixin, SessionSchemaMixin, SessionPortabilityMixin)
         msg = str(exc).lower()
         return "fts5" in msg and "corrupt" in msg
 
+    def _foreign_state_db_holders(self) -> List[Tuple[int, str]]:
+        """Return foreign processes holding this DB or its WAL sidecars.
+
+        Automatic FTS repair is structural maintenance, not an ordinary WAL
+        write.  It must not run while another process remains attached: a
+        sidecar reset under that holder can leave the two processes writing
+        through different WAL inodes.  ``psutil`` reads the kernel's open-file
+        table, including Linux ``(deleted)`` descriptors, so this also catches
+        a split brain already in progress.
+
+        A scan failure is represented as an unknown holder.  Skipping optional
+        automatic maintenance is safer than assuming quiescence; canonical
+        writes continue through the stale-FTS fail-open path.
+        """
+        # The split-brain mechanism requires POSIX unlink semantics: Windows
+        # refuses to replace SQLite sidecars while another process has them
+        # open.  Avoid psutil.open_files() there; querying arbitrary Windows
+        # processes can block for minutes on device-backed handles.
+        if _IS_WINDOWS:
+            return []
+        if psutil is None:
+            return [(-1, "open-file scan unavailable")]
+
+        def _canonical(path: str) -> str:
+            clean = path.removesuffix(" (deleted)")
+            return os.path.normcase(os.path.abspath(clean))
+
+        db_path = os.path.abspath(os.fspath(self.db_path))
+        watched = {
+            _canonical(db_path),
+            _canonical(db_path + "-wal"),
+            _canonical(db_path + "-shm"),
+        }
+        holders: List[Tuple[int, str]] = []
+        try:
+            for process in psutil.process_iter(["pid", "open_files"]):
+                info = process.info
+                pid = int(info["pid"])
+                if pid == os.getpid():
+                    continue
+                for opened in info.get("open_files") or ():
+                    path = getattr(opened, "path", "")
+                    if path and _canonical(path) in watched:
+                        holders.append((pid, path))
+        except Exception as exc:
+            logger.warning(
+                "Could not prove state.db has no foreign holders; deferring "
+                "automatic FTS maintenance: %s",
+                exc,
+            )
+            return holders or [(-1, f"open-file scan failed: {exc}")]
+        return holders
+
     def _try_runtime_fts_rebuild(self, exc: sqlite3.DatabaseError) -> bool:
         """One-shot in-place FTS rebuild after a corrupt-index write failure.
 
@@ -4234,6 +4287,15 @@ class SessionDB(SessionSearchMixin, SessionSchemaMixin, SessionPortabilityMixin)
         if not self._is_fts_write_corruption_error(exc):
             return False
         self._fts_runtime_rebuild_attempted = True
+        foreign_holders = self._foreign_state_db_holders()
+        if foreign_holders:
+            logger.warning(
+                "Skipping automatic state.db FTS rebuild while foreign "
+                "processes hold the database or WAL sidecars (%s); detaching "
+                "FTS sync so canonical writes can continue.",
+                foreign_holders,
+            )
+            return False
         logger.warning(
             "state.db write failed with an FTS-corruption error (%s) — "
             "attempting one-shot in-place FTS rebuild; canonical message "

hermes_state_schema.py

@@ -368,6 +368,15 @@ class SessionSchemaMixin:
 
     def _recover_stale_fts(self, cursor: sqlite3.Cursor, *, legacy: bool) -> bool:
         """Atomically rebuild stale base/trigram indexes and resume syncing."""
+        foreign_holders = self._foreign_state_db_holders()
+        if foreign_holders:
+            logger.warning(
+                "Deferred stale state.db FTS rebuild while foreign processes "
+                "hold the database or WAL sidecars (%s); canonical writes and "
+                "LIKE search remain available.",
+                foreign_holders,
+            )
+            return False
         try:
             trigram_status = self._fts_table_probe(cursor, "messages_fts_trigram")
         except sqlite3.DatabaseError:

tests/state/test_fts_runtime_rebuild.py

@@ -14,9 +14,11 @@ until a later open atomically rebuilds the index and restores the triggers.
 """
 
 import sqlite3
+from types import SimpleNamespace
 
 import pytest
 
+import hermes_state
 from hermes_state import (
     FTS_STALE_KEY,
     LEGACY_FTS_SQL,
@@ -84,6 +86,47 @@ def _base_fts_triggers(db_path):
 
 
 class TestRuntimeFtsRebuild:
+    def test_foreign_holder_detection_includes_deleted_wal(
+        self, db, tmp_path, monkeypatch
+    ):
+        db_path = tmp_path / "state.db"
+
+        class FakePsutil:
+            @staticmethod
+            def process_iter(_attrs):
+                return iter(
+                    (
+                        SimpleNamespace(
+                            info={
+                                "pid": 111,
+                                "open_files": [SimpleNamespace(path=str(db_path))],
+                            }
+                        ),
+                        SimpleNamespace(
+                            info={
+                                "pid": 222,
+                                "open_files": [
+                                    SimpleNamespace(path=f"{db_path}-wal (deleted)")
+                                ],
+                            }
+                        ),
+                        SimpleNamespace(
+                            info={
+                                "pid": 333,
+                                "open_files": [SimpleNamespace(path=str(tmp_path / "other.db"))],
+                            }
+                        ),
+                    )
+                )
+
+        monkeypatch.setattr(hermes_state, "psutil", FakePsutil)
+        monkeypatch.setattr(hermes_state, "_IS_WINDOWS", False)
+        monkeypatch.setattr(hermes_state.os, "getpid", lambda: 111)
+
+        assert db._foreign_state_db_holders() == [
+            (222, f"{db_path}-wal (deleted)")
+        ]
+
     def test_corruption_error_classification_covers_both_sqlite_messages(self):
         """SQLite's message for a corrupt FTS index varies by version: older
         builds raise the generic malformed-image error, newer builds raise an
@@ -242,6 +285,30 @@ class TestRuntimeFtsRebuild:
         assert _meta_value(db_path, FTS_STALE_KEY) == "1"
         assert _base_fts_triggers(db_path) == set()
 
+    def test_foreign_holder_skips_runtime_rebuild_and_fails_open(
+        self, db, tmp_path, monkeypatch
+    ):
+        if not db._fts_enabled:
+            pytest.skip("FTS5 unavailable in this build")
+        db_path = tmp_path / "state.db"
+        db.create_session("s1", source="test")
+        db.append_message("s1", "user", "seed")
+        _corrupt_fts(db_path)
+
+        monkeypatch.setattr(
+            db,
+            "_foreign_state_db_holders",
+            lambda: [(4242, str(db_path) + "-wal")],
+            raising=False,
+        )
+
+        db.append_message("s1", "user", "canonical survives foreign holder")
+
+        assert _message_contents(db_path)[-1] == "canonical survives foreign holder"
+        assert db._fts_stale is True
+        assert _meta_value(db_path, FTS_STALE_KEY) == "1"
+        assert _base_fts_triggers(db_path) == set()
+
     def test_stale_search_preserves_not_semantics(self, db, tmp_path, monkeypatch):
         if not db._fts_enabled:
             pytest.skip("FTS5 unavailable in this build")
@@ -324,6 +391,39 @@ class TestRuntimeFtsRebuild:
         finally:
             reopened.close()
 
+    def test_foreign_holder_defers_startup_stale_rebuild(
+        self, db, tmp_path, monkeypatch
+    ):
+        if not db._fts_enabled:
+            pytest.skip("FTS5 unavailable in this build")
+        db_path = tmp_path / "state.db"
+        db.create_session("s1", source="test")
+        db.append_message("s1", "user", "seed")
+        _corrupt_fts(db_path)
+        monkeypatch.setattr(
+            db,
+            "rebuild_fts",
+            lambda: (_ for _ in ()).throw(sqlite3.DatabaseError("still corrupt")),
+        )
+        db.append_message("s1", "user", "before restart")
+        db.close()
+
+        monkeypatch.setattr(
+            SessionDB,
+            "_foreign_state_db_holders",
+            lambda self: [(4242, str(db_path) + "-wal")],
+            raising=False,
+        )
+        reopened = SessionDB(db_path=db_path)
+        try:
+            assert reopened._fts_stale is True
+            assert _meta_value(db_path, FTS_STALE_KEY) == "1"
+            assert _base_fts_triggers(db_path) == set()
+            reopened.append_message("s1", "user", "after deferred recovery")
+            assert _message_contents(db_path)[-1] == "after deferred recovery"
+        finally:
+            reopened.close()
+
     def test_legacy_inline_fts_fails_open_and_recovers(self, tmp_path, monkeypatch):
         db_path = tmp_path / "legacy-state.db"
         raw = sqlite3.connect(str(db_path))