fix(state): refuse SessionDB open and writes on a deleted WAL generation

A live writer can keep a deleted state.db-wal inode while a second opener
mints a fresh WAL at the same path. Fail closed on writable open (before
connect) and on the write-path sidecar identity check so the second
generation is never created.

Co-authored-by: Noa <rainbowgore@users.noreply.github.com>

hermes_state.py

@@ -2267,6 +2267,8 @@ def classify_persistence_error(exc_or_str) -> str:
         return "replaced"
     if isinstance(exc_or_str, StateDbCorruptError):
         return "corrupt"
+    if isinstance(exc_or_str, DeletedWalGenerationError):
+        return "replaced"
     text = str(exc_or_str).lower()
     if "turn lease" in text:
         return "turn_lease"
@@ -2276,6 +2278,8 @@ def classify_persistence_error(exc_or_str) -> str:
         return "compression"
     if "was replaced underneath" in text:
         return "replaced"
+    if "deleted state.db-wal" in text or "deleted state.db-shm" in text:
+        return "replaced"
     # Structural corruption BEFORE the lock and disk buckets: "database disk
     # image is malformed" contains "disk" (and some wrapped corruption
     # strings mention "locked" recovery attempts), so later buckets would
@@ -4351,6 +4355,17 @@ class StateDbReplacedError(RuntimeError):
     """
 
 
+class DeletedWalGenerationError(RuntimeError):
+    """A live process holds a deleted state.db-wal / -shm generation.
+
+    Opening or writing through this handle would mint a second WAL inode
+    (or keep committing on the orphan) — the split-brain that produces
+    intermittent SQLITE_CORRUPT / SQLITE_IOERR. Stop the writers; do not
+    unlink the WAL yourself. ``database.journal_mode: delete`` is operator
+    containment, not a default change.
+    """
+
+
 # SQLite header: 4-byte big-endian application_id at offset 68. Distinct from
 # inode: ``cp`` onto the same path keeps st_ino and truncates+rewrites.
 _STATE_DB_APPLICATION_ID_OFFSET = 68
@@ -4361,6 +4376,14 @@ _STATE_DB_REPLACED_MSG = (
     "gateway pending_messages spool) and restore or reopen after operator "
     "intervention."
 )
+_DELETED_WAL_GENERATION_MSG = (
+    "FATAL: a live process holds a deleted state.db-wal or state.db-shm "
+    "inode while the path names a different (or missing) generation. "
+    "Refusing to open or write so a second WAL cannot be minted. "
+    "Stop the gateway, dashboard, and cron writers that hold the deleted "
+    "sidecar, then reopen. Do not delete the WAL yourself. "
+    "database.journal_mode: delete is operator containment, not a new default."
+)
 
 
 class StateDbCorruptError(sqlite3.DatabaseError):
@@ -4524,6 +4547,91 @@ def _stat_db_file_identity(path: Path) -> "Optional[tuple]":
     return (st.st_dev, st.st_ino)
 
 
+def _stat_sqlite_sidecar_identity(db_path: Path) -> Dict[str, tuple]:
+    """Snapshot ``(st_dev, st_ino)`` for existing WAL/SHM sidecars."""
+    identities: Dict[str, tuple] = {}
+    base = os.fspath(db_path)
+    for suffix in ("-wal", "-shm"):
+        ident = _stat_db_file_identity(Path(base + suffix))
+        if ident is not None:
+            identities[suffix] = ident
+    return identities
+
+
+def _canonical_sqlite_path(path: str) -> str:
+    """Normalize a /proc fd target, stripping the Linux `` (deleted)`` suffix."""
+    return os.path.normcase(os.path.abspath(path.removesuffix(" (deleted)")))
+
+
+def _watched_sqlite_sidecar_paths(db_path) -> Set[str]:
+    base = os.path.abspath(os.fspath(db_path))
+    return {
+        _canonical_sqlite_path(base + "-wal"),
+        _canonical_sqlite_path(base + "-shm"),
+    }
+
+
+def iter_deleted_sqlite_sidecar_holders(
+    db_path, include_self: bool = True
+) -> List[Tuple[int, str]]:
+    """Return processes holding an unlinked ``state.db-wal`` / ``-shm``.
+
+    Linux-only (``/proc/<pid>/fd`` readlink). Windows and other hosts
+    return ``[]`` — Windows cannot unlink a sidecar another process still
+    holds, and macOS does not use the `` (deleted)`` suffix.
+
+    ``include_self=True`` is required on the SessionDB open/write refuse
+    path: the in-process writer that still holds the orphan inode is the
+    one that must not mint a replacement WAL (and must stop committing).
+    ``_foreign_state_db_holders`` keeps skipping this PID for FTS
+    maintenance so a process does not block its own optional repair.
+    """
+    if _IS_WINDOWS or not sys.platform.startswith("linux"):
+        return []
+
+    holders: List[Tuple[int, str]] = []
+    own_pid = os.getpid()
+    watched = _watched_sqlite_sidecar_paths(db_path)
+    try:
+        for pid_str in os.listdir("/proc"):
+            if not pid_str.isdigit():
+                continue
+            pid = int(pid_str)
+            if pid == own_pid and not include_self:
+                continue
+            fd_dir = f"/proc/{pid}/fd"
+            try:
+                fds = os.listdir(fd_dir)
+            except OSError:
+                continue
+            for fd in fds:
+                try:
+                    target = os.readlink(f"{fd_dir}/{fd}")
+                except OSError:
+                    continue
+                if " (deleted)" not in target:
+                    continue
+                if _canonical_sqlite_path(target) in watched:
+                    holders.append((pid, target))
+    except Exception as exc:
+        logger.debug("deleted-WAL holder scan failed for %s: %s", db_path, exc)
+        return holders
+    return holders
+
+
+def refuse_deleted_wal_generation(db_path) -> None:
+    """Raise if any process holds a deleted WAL/SHM generation for *db_path*.
+
+    Called *before* ``sqlite3.connect`` so a second opener cannot mint a
+    replacement WAL inode while a live writer still holds the orphan.
+    """
+    holders = iter_deleted_sqlite_sidecar_holders(db_path, include_self=True)
+    if not holders:
+        return
+    logger.error(_DELETED_WAL_GENERATION_MSG)
+    raise DeletedWalGenerationError(_DELETED_WAL_GENERATION_MSG)
+
+
 # ── Process-wide shared SessionDB registry (#90837) ──
 #
 # The registry itself lives in hermes_state_registry.py — a bounded
@@ -5289,6 +5397,8 @@ class SessionDB(SessionSearchMixin, SessionSchemaMixin, SessionPortabilityMixin)
         # restored file (see StateDbCorruptError).
         self._db_corrupt = False
         self._db_corrupt_reason = ""
+        self._db_sidecar_identity: Dict[str, tuple] = {}
+        self._db_wal_generation_lost = False
         # One-shot guard for the usermerge-floor config write on the
         # incremental FTS merge cadence (see _merge_fts_incrementally).
         self._fts_usermerge_floor_applied = False
@@ -5400,6 +5510,10 @@ class SessionDB(SessionSearchMixin, SessionSchemaMixin, SessionPortabilityMixin)
             # readonly database" from deep inside _init_schema.
             if not read_only:
                 preflight_db_writability(self.db_path, db_label="state.db")
+                # Refuse before the first sqlite3.connect so we cannot mint a
+                # replacement WAL while a live writer still holds a deleted
+                # sidecar inode.
+                refuse_deleted_wal_generation(self.db_path)
 
             # #68474 / #97568: Serialize startup across zero-byte check, quarantine,
             # connect, and schema commit so concurrent openers don't race on an
@@ -5440,6 +5554,7 @@ class SessionDB(SessionSearchMixin, SessionSchemaMixin, SessionPortabilityMixin)
                         raise sqlite3.DatabaseError(msg)
 
             def _connect_and_init():
+                refuse_deleted_wal_generation(self.db_path)
                 self._conn = _connect_tracked_db(
                     str(self.db_path),
                     check_same_thread=False,
@@ -5839,6 +5954,8 @@ class SessionDB(SessionSearchMixin, SessionSchemaMixin, SessionPortabilityMixin)
                 f"structural corruption; refusing to reopen for a {context} "
                 "after close(). "
             )
+        if self._db_wal_generation_lost or self._wal_generation_was_lost():
+            self._halt_deleted_wal_generation()
         logger.warning(
             "state.db connection for %s was closed while a %s was still in "
             "flight — reopening (teardown/worker race, #94736)",
@@ -6346,6 +6463,7 @@ class SessionDB(SessionSearchMixin, SessionSchemaMixin, SessionPortabilityMixin)
     def _record_db_file_identity(self) -> None:
         """Snapshot inode plus the on-disk generation header when present."""
         self._db_file_identity = _stat_db_file_identity(self.db_path)
+        self._db_sidecar_identity = _stat_sqlite_sidecar_identity(self.db_path)
         disk_id = _read_sqlite_application_id(self.db_path)
         if disk_id:
             self._db_file_application_id = disk_id
@@ -6380,11 +6498,57 @@ class SessionDB(SessionSearchMixin, SessionSchemaMixin, SessionPortabilityMixin)
         logger.error(_STATE_DB_REPLACED_MSG)
         raise StateDbReplacedError(_STATE_DB_REPLACED_MSG)
 
+    def _wal_generation_was_lost(self) -> bool:
+        """True when the WAL/SHM generation this instance opened is gone.
+
+        Two independent cheap signals, either sufficient:
+
+        - a sidecar inode recorded at open is missing or has been replaced
+          by a new file at the same path (stat only — no /proc walk);
+        - this process still holds a ``state.db-wal (deleted)`` /
+          ``-shm (deleted)`` fd (``/proc/self/fd`` only).
+
+        A full ``/proc/*/fd`` walk is reserved for
+        :func:`refuse_deleted_wal_generation` on open, where we must see
+        *foreign* deleted holders before ``sqlite3.connect`` mints a new WAL.
+        """
+        recorded = self._db_sidecar_identity or {}
+        base = os.fspath(self.db_path)
+        for suffix, recorded_ident in recorded.items():
+            current = _stat_db_file_identity(Path(base + suffix))
+            if current is None or current != recorded_ident:
+                return True
+        if _IS_WINDOWS or not sys.platform.startswith("linux"):
+            return False
+        watched = _watched_sqlite_sidecar_paths(self.db_path)
+        fd_dir = f"/proc/{os.getpid()}/fd"
+        try:
+            for fd in os.listdir(fd_dir):
+                try:
+                    target = os.readlink(f"{fd_dir}/{fd}")
+                except OSError:
+                    continue
+                if " (deleted)" in target and _canonical_sqlite_path(target) in watched:
+                    return True
+        except OSError:
+            return False
+        return False
+
+    def _halt_deleted_wal_generation(self) -> None:
+        """Stop writes; do not mint or keep committing on a split WAL."""
+        self._db_wal_generation_lost = True
+        logger.error(_DELETED_WAL_GENERATION_MSG)
+        raise DeletedWalGenerationError(_DELETED_WAL_GENERATION_MSG)
+
     def _raise_if_db_replaced(self) -> None:
         if self._db_replaced:
             raise StateDbReplacedError(_STATE_DB_REPLACED_MSG)
+        if self._db_wal_generation_lost:
+            raise DeletedWalGenerationError(_DELETED_WAL_GENERATION_MSG)
         if self._db_file_was_replaced():
             self._halt_db_replaced()
+        if self._wal_generation_was_lost():
+            self._halt_deleted_wal_generation()
 
     @classmethod
     def _is_structural_corruption_error(cls, exc: BaseException) -> bool:
@@ -6533,15 +6697,11 @@ class SessionDB(SessionSearchMixin, SessionSchemaMixin, SessionPortabilityMixin)
         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"),
+            _canonical_sqlite_path(db_path),
+            _canonical_sqlite_path(db_path + "-wal"),
+            _canonical_sqlite_path(db_path + "-shm"),
         }
         holders: List[Tuple[int, str]] = []
 
@@ -6580,7 +6740,7 @@ class SessionDB(SessionSearchMixin, SessionSchemaMixin, SessionPortabilityMixin)
                             target = os.readlink(f"{fd_dir}/{fd}")
                         except OSError:
                             continue
-                        if _canonical(target) in watched:
+                        if _canonical_sqlite_path(target) in watched:
                             holders.append((pid, target))
             except Exception as exc:
                 logger.warning(
@@ -6605,7 +6765,7 @@ class SessionDB(SessionSearchMixin, SessionSchemaMixin, SessionPortabilityMixin)
                 # Linux-specific (systemd units running as root).
                 for opened in info.get("open_files") or ():
                     path = getattr(opened, "path", "")
-                    if path and _canonical(path) in watched:
+                    if path and _canonical_sqlite_path(path) in watched:
                         holders.append((pid, path))
         except Exception as exc:
             logger.warning(
@@ -6690,6 +6850,8 @@ class SessionDB(SessionSearchMixin, SessionSchemaMixin, SessionPortabilityMixin)
         self._raise_if_db_corrupt()
         if self._db_replaced or self._db_file_was_replaced():
             self._halt_db_replaced()
+        if self._db_wal_generation_lost or self._wal_generation_was_lost():
+            self._halt_deleted_wal_generation()
 
         try:
             with self._lock:
@@ -16497,6 +16659,10 @@ class SessionDB(SessionSearchMixin, SessionSchemaMixin, SessionPortabilityMixin)
                 self._conn.execute("PRAGMA wal_checkpoint(TRUNCATE)")
             except Exception as exc:
                 logger.debug("WAL checkpoint (TRUNCATE) after VACUUM failed: %s", exc)
+            # TRUNCATE may replace the WAL inode; adopt the post-VACUUM
+            # sidecars so the write-path generation guard does not halt a
+            # healthy exclusive maintenance connection.
+            self._record_db_file_identity()
         return optimized
 
     def maybe_auto_prune_and_vacuum(

tests/hermes_state/test_deleted_wal_generation_guard.py

@@ -0,0 +1,197 @@
+"""Refuse SessionDB open/write when a deleted WAL generation is still held.
+
+A live writer that keeps the unlinked ``state.db-wal`` inode while a second
+opener would mint a fresh WAL is the split-brain that produces intermittent
+``database disk image is malformed`` / ``disk I/O error``. The store must
+fail closed on both the open and write paths instead of creating the second
+generation.
+"""
+
+import os
+import sqlite3
+import sys
+from pathlib import Path
+
+import pytest
+
+import hermes_state
+from hermes_state import (
+    DeletedWalGenerationError,
+    SessionDB,
+    classify_persistence_error,
+    iter_deleted_sqlite_sidecar_holders,
+    refuse_deleted_wal_generation,
+)
+
+
+@pytest.fixture
+def force_wal(monkeypatch):
+    """Pin WAL so this host's vulnerable SQLite still matches production topology."""
+    monkeypatch.setattr(
+        hermes_state, "is_sqlite_wal_reset_vulnerable", lambda version_info=None: False
+    )
+    monkeypatch.setattr(hermes_state, "resolve_journal_mode", lambda: "wal")
+
+
+def _make_db(path: Path, session_id: str, content: str) -> SessionDB:
+    db = SessionDB(db_path=path)
+    db.create_session(session_id, "cli")
+    db.append_message(session_id, role="user", content=content)
+    return db
+
+
+def _require_wal(db: SessionDB) -> Path:
+    if not db._wal_active:
+        db.close()
+        pytest.skip("WAL not active on this filesystem")
+    wal = Path(os.fspath(db.db_path) + "-wal")
+    if not wal.exists():
+        db.close()
+        pytest.skip("WAL sidecar missing after first write")
+    return wal
+
+
+def _unlink_sidecars(db_path: Path) -> None:
+    for suffix in ("-wal", "-shm"):
+        sidecar = Path(os.fspath(db_path) + suffix)
+        if sidecar.exists():
+            os.unlink(sidecar)
+
+
+def test_classify_deleted_wal_is_replaced_not_disk():
+    err = DeletedWalGenerationError(
+        "FATAL: a live process holds a deleted state.db-wal or state.db-shm "
+        "inode while the path names a different (or missing) generation."
+    )
+    assert classify_persistence_error(err) == "replaced"
+    assert classify_persistence_error(str(err)) == "replaced"
+
+
+def test_iter_holders_empty_on_non_linux(monkeypatch, tmp_path):
+    monkeypatch.setattr(hermes_state, "_IS_WINDOWS", True)
+    assert iter_deleted_sqlite_sidecar_holders(tmp_path / "state.db") == []
+
+
+def test_clean_open_and_second_open_still_work(tmp_path, force_wal):
+    path = tmp_path / "state.db"
+    db = _make_db(path, "s1", "hello")
+    _require_wal(db)
+    db.close()
+    reopened = SessionDB(db_path=path)
+    try:
+        reopened.append_message("s1", role="user", content="second-open")
+        rows = reopened.get_messages("s1")
+        assert any(m["content"] == "second-open" for m in rows)
+    finally:
+        reopened.close()
+
+
+def test_delete_journal_two_writers_still_work(tmp_path, monkeypatch):
+    monkeypatch.setattr(hermes_state, "resolve_journal_mode", lambda: "delete")
+    monkeypatch.setattr(
+        hermes_state, "is_sqlite_wal_reset_vulnerable", lambda version_info=None: False
+    )
+    path = tmp_path / "state.db"
+    a = _make_db(path, "s", "from-a")
+    try:
+        assert not Path(os.fspath(path) + "-wal").exists()
+        b = SessionDB(db_path=path)
+        try:
+            b.append_message("s", role="user", content="from-b")
+            contents = [m["content"] for m in b.get_messages("s")]
+            assert "from-a" in contents
+            assert "from-b" in contents
+        finally:
+            b.close()
+    finally:
+        a.close()
+
+
+@pytest.mark.skipif(
+    not sys.platform.startswith("linux"),
+    reason="deleted-WAL /proc scan is Linux-only",
+)
+def test_iter_finds_self_after_wal_unlink(tmp_path, force_wal):
+    path = tmp_path / "state.db"
+    db = _make_db(path, "s", "held")
+    wal = _require_wal(db)
+    inode_before = wal.stat().st_ino
+    _unlink_sidecars(path)
+    holders = iter_deleted_sqlite_sidecar_holders(path, include_self=True)
+    try:
+        assert holders, "expected this process to still hold the deleted WAL inode"
+        assert any("(deleted)" in target for _pid, target in holders)
+        assert any(target.rstrip(" (deleted)").endswith("-wal") or
+                   target.rstrip(" (deleted)").endswith("-shm")
+                   for _pid, target in holders)
+        assert not wal.exists() or wal.stat().st_ino != inode_before
+    finally:
+        db.close()
+
+
+@pytest.mark.skipif(
+    not sys.platform.startswith("linux"),
+    reason="deleted-WAL /proc scan is Linux-only",
+)
+def test_second_sessiondb_open_refuses_and_does_not_mint_wal(tmp_path, force_wal):
+    path = tmp_path / "state.db"
+    writer = _make_db(path, "s", "before-unlink")
+    wal = _require_wal(writer)
+    inode_before = wal.stat().st_ino
+    _unlink_sidecars(path)
+    assert not wal.exists()
+
+    with pytest.raises(DeletedWalGenerationError, match="deleted state.db-wal"):
+        SessionDB(db_path=path)
+
+    assert not wal.exists(), "open must refuse before sqlite3.connect mints a WAL"
+    # If a WAL somehow reappeared it must not be a new generation.
+    if wal.exists():
+        assert wal.stat().st_ino == inode_before
+    writer.close()
+
+
+@pytest.mark.skipif(
+    not sys.platform.startswith("linux"),
+    reason="deleted-WAL write halt uses Linux unlink semantics",
+)
+def test_writer_halts_after_own_wal_unlinked(tmp_path, force_wal):
+    path = tmp_path / "state.db"
+    db = _make_db(path, "s", "before")
+    _require_wal(db)
+    recorded = db._db_sidecar_identity.get("-wal")
+    assert recorded is not None
+    _unlink_sidecars(path)
+
+    with pytest.raises(DeletedWalGenerationError, match="deleted state.db-wal"):
+        db.append_message("s", role="user", content="after-unlink")
+    assert db._db_wal_generation_lost is True
+
+    with pytest.raises(DeletedWalGenerationError):
+        db.append_message("s", role="user", content="second-after-halt")
+    db.close()
+
+
+@pytest.mark.skipif(
+    not sys.platform.startswith("linux"),
+    reason="deleted-WAL /proc scan is Linux-only",
+)
+def test_refuse_helper_raises_while_deleted_wal_held(tmp_path, force_wal):
+    path = tmp_path / "state.db"
+    raw = sqlite3.connect(str(path))
+    try:
+        raw.execute("PRAGMA journal_mode=WAL")
+        raw.execute("CREATE TABLE t (id INTEGER PRIMARY KEY, v TEXT)")
+        raw.execute("INSERT INTO t VALUES (1, 'held')")
+        raw.commit()
+        wal = Path(str(path) + "-wal")
+        assert wal.exists()
+        os.unlink(wal)
+        shm = Path(str(path) + "-shm")
+        if shm.exists():
+            os.unlink(shm)
+        with pytest.raises(DeletedWalGenerationError):
+            refuse_deleted_wal_generation(path)
+        assert not wal.exists()
+    finally:
+        raw.close()