fix(state): use PASSIVE checkpoint for periodic WAL flush to prevent B-tree corruption

TRUNCATE checkpoint every 50 writes causes B-tree corruption on large
databases (65K+ pages) due to the exclusive-lock I/O pressure from
checkpointing thousands of frames at once.

Switch periodic _try_wal_checkpoint() to PASSIVE mode which does not
require an exclusive lock and cannot corrupt pages under I/O pressure.
Keep TRUNCATE in close() and pre-VACUUM paths where it is safe
(infrequent, controlled conditions).

Also replace silent `except Exception: pass` with logged warnings for
checkpoint failures so operators can detect early corruption signals.

Fixes #45383

hermes_state.py

@@ -986,7 +986,7 @@ class SessionDB:
     _WRITE_MAX_RETRIES = 15
     _WRITE_RETRY_MIN_S = 0.020   # 20ms
     _WRITE_RETRY_MAX_S = 0.150   # 150ms
-    # Attempt a PASSIVE WAL checkpoint every N successful writes.
+    # Attempt a WAL checkpoint every N successful writes (PASSIVE mode).
     _CHECKPOINT_EVERY_N_WRITES = 50
     # Merge fragmented FTS5 segments every N successful writes. The message
     # triggers append one segment per insert; left unmaintained these grow
@@ -1301,35 +1301,34 @@ class SessionDB:
         )
 
     def _try_wal_checkpoint(self) -> None:
-        """Best-effort TRUNCATE WAL checkpoint.  Never raises.
+        """Best-effort PASSIVE WAL checkpoint.  Never raises.
 
-        Flushes committed WAL frames back into the main DB file and
-        truncates the WAL file to zero bytes.  Keeps the WAL from
-        growing unbounded when many processes hold persistent
-        connections.
+        Flushes committed WAL frames back into the main DB file without
+        requiring an exclusive lock.  PASSIVE is safe for frequent
+        periodic use because it does not block concurrent writers and
+        cannot corrupt B-tree pages under I/O pressure.
 
-        PASSIVE checkpoint was previously used here, but it never
-        truncates the WAL file — the file stays at its high-water
-        mark until an explicit TRUNCATE is called (which only
-        happened inside the infrequent vacuum()).
+        PASSIVE does not truncate the WAL file — it stays at its
+        high-water mark.  WAL truncation happens in :meth:`close`
+        (TRUNCATE) and pre-VACUUM checkpoints, which run infrequently
+        under controlled conditions.
 
-        TRUNCATE may block writers briefly while checkpointing, but
-        _try_wal_checkpoint is called off the hot path (every 50
-        writes) and already runs under ``self._lock``, so the
-        additional hold time is negligible.
+        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).
         """
         try:
             with self._lock:
                 result = self._conn.execute(
-                    "PRAGMA wal_checkpoint(TRUNCATE)"
+                    "PRAGMA wal_checkpoint(PASSIVE)"
                 ).fetchone()
                 if result and result[1] > 0:
                     logger.debug(
                         "WAL checkpoint: %d/%d pages checkpointed",
                         result[2], result[1],
                     )
-        except Exception:
-            pass  # Best effort — never fatal.
+        except Exception as exc:
+            logger.warning("WAL checkpoint (PASSIVE) failed: %s", exc)
 
     def _try_optimize_fts(self) -> None:
         """Best-effort FTS5 segment merge. Never raises.
@@ -1357,8 +1356,8 @@ class SessionDB:
             if self._conn:
                 try:
                     self._conn.execute("PRAGMA wal_checkpoint(TRUNCATE)")
-                except Exception:
-                    pass
+                except Exception as exc:
+                    logger.debug("WAL checkpoint (TRUNCATE) at close failed: %s", exc)
                 self._conn.close()
                 self._conn = None
 
@@ -7015,8 +7014,8 @@ class SessionDB:
             # Best-effort WAL checkpoint first, then VACUUM.
             try:
                 self._conn.execute("PRAGMA wal_checkpoint(TRUNCATE)")
-            except Exception:
-                pass
+            except Exception as exc:
+                logger.debug("WAL checkpoint (TRUNCATE) before VACUUM failed: %s", exc)
             self._conn.execute("VACUUM")
         return optimized
 

tests/test_wal_checkpoint_strategy.py

@@ -0,0 +1,138 @@
+"""Tests for SessionDB WAL checkpoint strategy (issue #45383).
+
+Verifies that periodic checkpoints use PASSIVE mode (safe for large DBs)
+while close() and pre-VACUUM paths still use TRUNCATE.
+"""
+
+import sqlite3
+import logging
+from unittest.mock import MagicMock, patch
+
+import pytest
+
+from hermes_state import SessionDB
+
+
+@pytest.fixture()
+def db(tmp_path):
+    """Create a SessionDB with a temp database file."""
+    db_path = tmp_path / "test_state.db"
+    session_db = SessionDB(db_path=db_path)
+    yield session_db
+    try:
+        session_db.close()
+    except Exception:
+        pass
+
+
+class TestTryWalCheckpointPassive:
+    """_try_wal_checkpoint() should use PASSIVE mode for periodic use."""
+
+    def test_checkpoint_uses_passive_mode(self, db):
+        """PASSIVE checkpoint does not require exclusive lock — safe for large DBs."""
+        # Capture the real connection's execute before mocking
+        real_conn = db._conn
+        execute_calls = []
+
+        def tracking_execute(sql, *args, **kwargs):
+            execute_calls.append(sql)
+            return real_conn.execute(sql, *args, **kwargs)
+
+        # sqlite3.Connection.execute is read-only (C extension) — replace _conn
+        mock_conn = MagicMock()
+        mock_conn.execute.side_effect = tracking_execute
+        mock_conn.fetchone.return_value = None
+        db._conn = mock_conn
+
+        db._try_wal_checkpoint()
+
+        passive_calls = [c for c in execute_calls if "wal_checkpoint(PASSIVE)" in c]
+        truncate_calls = [c for c in execute_calls if "wal_checkpoint(TRUNCATE)" in c]
+        assert len(passive_calls) == 1, (
+            f"Expected 1 PASSIVE checkpoint call, got {len(passive_calls)}"
+        )
+        assert len(truncate_calls) == 0, (
+            "Periodic checkpoint should NOT use TRUNCATE"
+        )
+
+    def test_checkpoint_logs_warning_on_failure(self, db, caplog):
+        """Failed PASSIVE checkpoint logs a warning instead of silent pass."""
+        mock_conn = MagicMock()
+        mock_conn.execute.side_effect = sqlite3.OperationalError("disk I/O error")
+        db._conn = mock_conn
+
+        with caplog.at_level(logging.WARNING):
+            db._try_wal_checkpoint()
+
+        assert any("WAL checkpoint (PASSIVE) failed" in r.message for r in caplog.records), (
+            f"Expected warning log about PASSIVE checkpoint failure, got: {caplog.text}"
+        )
+
+    def test_checkpoint_returns_result_on_success(self, db):
+        """Successful PASSIVE checkpoint does not raise."""
+        db._try_wal_checkpoint()
+
+
+class TestCloseUsesTruncate:
+    """close() should still use TRUNCATE to shrink WAL on shutdown."""
+
+    def test_close_uses_truncate_mode(self, db):
+        """TRUNCATE at close is safe — no concurrent writers during shutdown."""
+        real_conn = db._conn
+        execute_calls = []
+
+        def tracking_execute(sql, *args, **kwargs):
+            execute_calls.append(sql)
+            return real_conn.execute(sql, *args, **kwargs)
+
+        mock_conn = MagicMock()
+        mock_conn.execute.side_effect = tracking_execute
+        db._conn = mock_conn
+
+        db.close()
+
+        truncate_calls = [c for c in execute_calls if "wal_checkpoint(TRUNCATE)" in c]
+        assert len(truncate_calls) == 1, (
+            f"Expected 1 TRUNCATE checkpoint at close, got {len(truncate_calls)}"
+        )
+
+    def test_close_logs_debug_on_failure(self, db, caplog):
+        """Failed TRUNCATE at close logs debug (not warning — close is best-effort)."""
+        mock_conn = MagicMock()
+        mock_conn.execute.side_effect = sqlite3.OperationalError("database is locked")
+        db._conn = mock_conn
+
+        with caplog.at_level(logging.DEBUG):
+            db.close()
+
+        assert any("WAL checkpoint (TRUNCATE) at close failed" in r.message for r in caplog.records), (
+            f"Expected debug log about TRUNCATE failure at close, got: {caplog.text}"
+        )
+
+
+class TestCheckpointFrequency:
+    """Checkpoint triggers every N writes."""
+
+    def test_checkpoint_triggers_at_interval(self, db):
+        """_try_wal_checkpoint is called every _CHECKPOINT_EVERY_N_WRITES writes."""
+        call_count = [0]
+        original = db._try_wal_checkpoint
+
+        def counting_checkpoint():
+            call_count[0] += 1
+            original()
+
+        db._try_wal_checkpoint = counting_checkpoint
+
+        # Write exactly _CHECKPOINT_EVERY_N_WRITES sessions to trigger one checkpoint
+        n = db._CHECKPOINT_EVERY_N_WRITES
+        import time as _time
+        for i in range(n):
+            db._execute_write(lambda conn, _i=i: conn.execute(
+                "INSERT INTO sessions (id, source, started_at) VALUES (?, ?, ?)",
+                (f"sess_{_i}", "test", _time.time()),
+            ))
+
+        assert call_count[0] == 1, (
+            f"Expected 1 checkpoint after {n} writes, got {call_count[0]}"
+        )