fix(state): harden state.db against corruption — write lock, WAL protections, reconnect self-heal

cron/lifecycle_guard.py

@@ -169,6 +169,13 @@ def _iter_command_segments(command: str) -> Iterator[list[str]]:
 
         segment: list[str] = []
         for token in tokens:
+            if "\x00" in token:
+                # NUL byte (tokenized out of binary content) is never a
+                # real command/script path. Drop it so downstream
+                # Path(...) ops can't raise embedded-null-byte
+                # (#76762 follow-up 2: _resolve_terminal_script_path /
+                # _iter_referenced_shell_scripts still crashed).
+                continue
             if token and set(token) <= _CONTROL_CHARS:
                 if segment:
                     yield segment
@@ -429,6 +436,11 @@ def _read_referenced_script(path: Path) -> tuple[Optional[str], bool]:
     """Return ``(text, unsafe)`` using bounded, regular-file-only reads."""
     flags = os.O_RDONLY | getattr(os, "O_NONBLOCK", 0)
     try:
+        # A NUL byte in the path (tokenized out of binary content) makes
+        # os.open raise ValueError, not OSError — treat it like an unreadable
+        # path so the guard can never crash the caller (#76762 follow-up).
+        if "\x00" in str(path):
+            return None, False
         descriptor = os.open(path, flags)
     except (OSError, ValueError):
         # OSError: unreadable / missing / over-long paths. ValueError: an

gateway/delivery_ledger.py

@@ -50,6 +50,7 @@
 from typing import Any, Dict, Iterator, List, Optional
 
 from hermes_constants import get_hermes_home
+from hermes_cli.state_db_write_lock import state_db_write_lock
 
 logger = logging.getLogger(__name__)
 
@@ -92,7 +93,15 @@ def _connect() -> sqlite3.Connection:
 def _initialize_schema(conn: sqlite3.Connection) -> None:
     from hermes_state import apply_wal_with_fallback
 
-    apply_wal_with_fallback(conn, db_label="state.db (delivery_ledger)")
+    # SessionDB owns initial WAL activation.  A short-lived ledger connection
+    # must only verify that ownership state, never re-run journal_mode=WAL
+    # against the gateway's live WAL/SHM files.
+    row = conn.execute("PRAGMA journal_mode").fetchone()
+    mode = str(row[0]).strip().lower() if row and row[0] is not None else ""
+    if mode != "wal":
+        apply_wal_with_fallback(
+            conn, db_label="state.db (delivery_ledger)", require_wal=True
+        )
     conn.execute(
         """CREATE TABLE IF NOT EXISTS delivery_obligations (
             obligation_id TEXT PRIMARY KEY,
@@ -124,12 +133,14 @@ def _transaction() -> Iterator[sqlite3.Connection]:
     bug was #69567 / PR #69594). ``record_obligation`` runs on every outbound
     final response, so this ledger is the highest-frequency leaker.
     """
-    conn = _connect()
-    try:
-        with conn:
-            yield conn
-    finally:
-        conn.close()
+    path = _db_path()
+    with state_db_write_lock(path):
+        conn = _connect()
+        try:
+            with conn:
+                yield conn
+        finally:
+            conn.close()
 
 
 def _owner_stamp() -> tuple[int, Optional[int]]:

hermes_cli/state_db_write_lock.py

@@ -0,0 +1,30 @@
+"""Cross-process write ownership for Hermes' canonical ``state.db``.
+
+SQLite WAL supports concurrent readers but only one writer.  Hermes has a
+long-lived SessionDB plus a few durable ledgers which historically opened
+their own write connections.  SQLite serialises the transactions eventually,
+but it cannot make independently managed WAL/checkpoint lifecycles safe.
+
+This lock is deliberately only for write transactions and schema changes.
+Read-only SessionDB connections remain lock-free and retain WAL concurrency.
+"""
+
+from __future__ import annotations
+
+import fcntl
+from contextlib import contextmanager
+from pathlib import Path
+from typing import Iterator
+
+
+@contextmanager
+def state_db_write_lock(db_path: Path) -> Iterator[None]:
+    """Serialize a state.db write transaction across Hermes processes."""
+    lock_path = db_path.with_name(f"{db_path.name}.write.lock")
+    lock_path.parent.mkdir(parents=True, exist_ok=True)
+    with lock_path.open("a+") as lock_file:
+        fcntl.flock(lock_file.fileno(), fcntl.LOCK_EX)
+        try:
+            yield
+        finally:
+            fcntl.flock(lock_file.fileno(), fcntl.LOCK_UN)