fix(state): WAL probe fallback, bare-SELECT locking, writable_schema finally, read-pool self-heal

hermes_state.py

@@ -1318,7 +1318,12 @@ def _apply_delete_for_wal_reset_bug(
                 "this process does not exclusively own"
             )
         _log_wal_reset_bug_once(db_label, kept_wal=True, indeterminate=True)
-        return "wal"
+        # Report "delete", not "wal", for the indeterminate probe: claiming
+        # "wal" would set _wal_active=True and enable the lock-free read pool
+        # on a DB that may actually be in rollback-journal mode (raw
+        # SQLITE_BUSY on reads); claiming "delete" only costs queuing on the
+        # write lock — slow but correct.
+        return "delete"
 
     actual = ""
     try:
@@ -3667,28 +3672,42 @@ def _read_ctx(self):
         """
         conn = self._checkout_read_conn()
         if conn is not None:
+            broken = False
             try:
                 yield conn
+            except sqlite3.DatabaseError:
+                # A DatabaseError marks the pooled read connection as broken
+                # or stale (backing file replaced / truncated — the same
+                # scenario _reconnect_after_notadb self-heals on the write
+                # side). Destroy it instead of returning it to the pool so
+                # the next checkout reopens; otherwise the LIFO order hands
+                # the same broken connection to every subsequent query until
+                # process restart.
+                broken = True
+                raise
             finally:
-                returned = False
-                with self._read_conns_lock:
-                    if not self._read_conns_closed:
-                        try:
-                            self._read_pool.put_nowait(conn)
-                            returned = True
-                        except queue.Full:
-                            pass
-                if not returned:
-                    # close() has already drained the pool, so this connection
-                    # is surplus. Close it here — dropping it on the floor is
-                    # what leaked the fd.
-                    #
-                    # queue.Full is now unreachable in practice (permits and
-                    # maxsize are both _READ_POOL_MAX, so there can never be a
-                    # ninth connection to return), but the branch stays: it is
-                    # load-bearing if those two ever drift apart, and a leak is
-                    # the failure mode it prevents.
+                if broken:
                     self._close_read_conn(conn)
+                else:
+                    returned = False
+                    with self._read_conns_lock:
+                        if not self._read_conns_closed:
+                            try:
+                                self._read_pool.put_nowait(conn)
+                                returned = True
+                            except queue.Full:
+                                pass
+                    if not returned:
+                        # close() has already drained the pool, so this connection
+                        # is surplus. Close it here — dropping it on the floor is
+                        # what leaked the fd.
+                        #
+                        # queue.Full is now unreachable in practice (permits and
+                        # maxsize are both _READ_POOL_MAX, so there can never be a
+                        # ninth connection to return), but the branch stays: it is
+                        # load-bearing if those two ever drift apart, and a leak is
+                        # the failure mode it prevents.
+                        self._close_read_conn(conn)
             return
         with self._lock:
             yield self._conn
@@ -6663,11 +6682,15 @@ def get_compression_lock_holder(self, session_id: str) -> Optional[str]:
         if not session_id:
             return None
         now = time.time()
-        row = self._conn.execute(
-            "SELECT holder FROM compression_locks "
-            "WHERE session_id = ? AND expires_at >= ?",
-            (session_id, now),
-        ).fetchone()
+        # Read via _read_ctx (degrades to self._lock when WAL is inactive) —
+        # a bare self._conn read joins any in-flight BEGIN IMMEDIATE
+        # transaction and can see rows that later roll back.
+        with self._read_ctx() as conn:
+            row = conn.execute(
+                "SELECT holder FROM compression_locks "
+                "WHERE session_id = ? AND expires_at >= ?",
+                (session_id, now),
+            ).fetchone()
         if row is None:
             return None
         return row["holder"] if isinstance(row, sqlite3.Row) else row[0]
@@ -6741,11 +6764,16 @@ def clear_session_activity_labels(self, session_id: str) -> None:
         # No-op fast path: skip the transaction when there is nothing to
         # clear. Read-only, no write lock.
         try:
-            row = self._conn.execute(
-                "SELECT last_activity_description, last_activity_provenance "
-                "FROM sessions WHERE id = ?",
-                (session_id,),
-            ).fetchone()
+            # Read via _read_ctx (degrades to self._lock when WAL is
+            # inactive) — a bare self._conn read joins any in-flight
+            # BEGIN IMMEDIATE transaction and can see rows that later
+            # roll back.
+            with self._read_ctx() as conn:
+                row = conn.execute(
+                    "SELECT last_activity_description, last_activity_provenance "
+                    "FROM sessions WHERE id = ?",
+                    (session_id,),
+                ).fetchone()
         except sqlite3.Error:
             row = None
         if row is not None:
@@ -12919,12 +12947,17 @@ def get_handoff_state(self, session_id: str) -> Optional[Dict[str, Any]]:
         no handoff record.
         """
         try:
-            cur = self._conn.execute(
-                "SELECT handoff_state, handoff_platform, handoff_error "
-                "FROM sessions WHERE id = ?",
-                (session_id,),
-            )
-            row = cur.fetchone()
+            # Read via _read_ctx (degrades to self._lock when WAL is
+            # inactive) — a bare self._conn read joins any in-flight
+            # BEGIN IMMEDIATE transaction and can see rows that later
+            # roll back.
+            with self._read_ctx() as conn:
+                cur = conn.execute(
+                    "SELECT handoff_state, handoff_platform, handoff_error "
+                    "FROM sessions WHERE id = ?",
+                    (session_id,),
+                )
+                row = cur.fetchone()
             if not row:
                 return None
             return {
@@ -12941,15 +12974,20 @@ def list_pending_handoffs(self) -> List[Dict[str, Any]]:
         Used by the gateway's handoff watcher.
         """
         try:
-            cur = self._conn.execute(
-                "SELECT s.*, "
-                "COALESCE(sp.prompt, s.system_prompt) AS _system_prompt_resolved "
-                "FROM sessions s "
-                "LEFT JOIN system_prompts sp ON sp.hash = s.system_prompt_hash "
-                "WHERE s.handoff_state = 'pending' "
-                "ORDER BY s.started_at ASC"
-            )
-            return [self._session_row_dict(r) for r in cur.fetchall()]
+            # Read via _read_ctx (degrades to self._lock when WAL is
+            # inactive) — a bare self._conn read joins any in-flight
+            # BEGIN IMMEDIATE transaction and can see rows that later
+            # roll back.
+            with self._read_ctx() as conn:
+                cur = conn.execute(
+                    "SELECT s.*, "
+                    "COALESCE(sp.prompt, s.system_prompt) AS _system_prompt_resolved "
+                    "FROM sessions s "
+                    "LEFT JOIN system_prompts sp ON sp.hash = s.system_prompt_hash "
+                    "WHERE s.handoff_state = 'pending' "
+                    "ORDER BY s.started_at ASC"
+                )
+                return [self._session_row_dict(r) for r in cur.fetchall()]
         except Exception:
             return []
 

hermes_state_search.py

@@ -699,12 +699,18 @@ def _stage(conn):
             ).fetchone())
             if had:
                 conn.execute("PRAGMA writable_schema=ON")
-                conn.execute(
-                    "DELETE FROM sqlite_master WHERE type = 'table' "
-                    "AND name IN ('messages_fts', 'messages_fts_trigram') "
-                    "AND sql LIKE 'CREATE VIRTUAL TABLE%'"
-                )
-                conn.execute("PRAGMA writable_schema=RESET")
+                try:
+                    conn.execute(
+                        "DELETE FROM sqlite_master WHERE type = 'table' "
+                        "AND name IN ('messages_fts', 'messages_fts_trigram') "
+                        "AND sql LIKE 'CREATE VIRTUAL TABLE%'"
+                    )
+                finally:
+                    # writable_schema is a connection-level switch that does
+                    # not roll back with the transaction — RESET must run even
+                    # when the DELETE raises, or the long-lived write
+                    # connection stays degraded until process exit.
+                    conn.execute("PRAGMA writable_schema=RESET")
                 shadows = [
                     r[0] for r in conn.execute(
                         "SELECT name FROM sqlite_master WHERE type = 'table' "