From 04ec8414625a0545ead71de7dc28449c5dd5c3df Mon Sep 17 00:00:00 2001 From: Jasmine Naderi Date: Tue, 21 Jul 2026 16:43:41 -0400 Subject: [PATCH] fix(state): configurable journal_mode + centralize all DB openers Add HERMES_JOURNAL_MODE env / database.journal_mode config for virtiofs/NFS/SMB where WAL is not crash-safe. Route 5 bypass openers through apply_wal_with_fallback so a single setting covers every .db (#68545). --- agent/verification_evidence.py | 2 - cron/executions.py | 3 +- hermes_state.py | 32 ++++++++++ plugins/platforms/discord/recovery.py | 1 - tests/test_journal_mode_config.py | 84 +++++++++++++++++++++++++++ tools/async_delegation.py | 66 +++------------------ 6 files changed, 125 insertions(+), 63 deletions(-) create mode 100644 tests/test_journal_mode_config.py diff --git a/agent/verification_evidence.py b/agent/verification_evidence.py index 2c8d1f85efa..13917f2386f 100644 --- a/agent/verification_evidence.py +++ b/agent/verification_evidence.py @@ -61,8 +61,6 @@ def _db_path() -> Path: def _connect() -> sqlite3.Connection: - from hermes_state import apply_wal_with_fallback - path = _db_path() path.parent.mkdir(parents=True, exist_ok=True) conn = sqlite3.connect(path) diff --git a/cron/executions.py b/cron/executions.py index 01437ab9847..fe77e755143 100644 --- a/cron/executions.py +++ b/cron/executions.py @@ -34,7 +34,8 @@ def _initialize_schema(conn: sqlite3.Connection) -> None: conn.row_factory = sqlite3.Row conn.execute("PRAGMA busy_timeout=5000") - apply_wal_with_fallback(conn, db_label="cron/executions.db") + from hermes_state import apply_wal_with_fallback + apply_wal_with_fallback(conn, db_label="cron_executions.db") conn.execute("PRAGMA synchronous=FULL") conn.execute( """CREATE TABLE IF NOT EXISTS executions ( diff --git a/hermes_state.py b/hermes_state.py index b9c5c588718..80a975e1a4b 100644 --- a/hermes_state.py +++ b/hermes_state.py @@ -506,6 +506,28 @@ def sqlite_source_id() -> str: return str(row[0]) +def resolve_journal_mode() -> str: + """Return the configured journal mode (``wal`` or ``delete``). + + Honors ``HERMES_JOURNAL_MODE`` env var and ``database.journal_mode`` + in config.yaml. Default is ``wal``. Set to ``delete`` on filesystems + where WAL is not crash-safe (virtiofs, NFS, SMB, containerized-on- + macOS) — the process can't detect the backing filesystem from inside + a container, so this is the escape hatch (#68545). + """ + raw = os.environ.get("HERMES_JOURNAL_MODE", "").strip().lower() + if not raw: + try: + from hermes_cli.config import load_config + cfg = load_config() or {} + raw = str(cfg.get("database", {}).get("journal_mode", "")).strip().lower() + except Exception: + pass + if raw in ("delete", "truncate", "persist", "memory", "off"): + return raw + return "wal" + + def apply_wal_with_fallback( conn: sqlite3.Connection, *, @@ -564,6 +586,16 @@ def apply_wal_with_fallback( except sqlite3.OperationalError: pass + # #68545: honor user-configured journal_mode (env/config.yaml). + # If the user forced DELETE (e.g. for virtiofs/NFS/SMB), don't try WAL. + _configured = resolve_journal_mode() + if _configured != "wal": + try: + conn.execute(f"PRAGMA journal_mode={_configured.upper()}") + except sqlite3.OperationalError: + pass # mode may not be supported; leave whatever is set + return _configured + try: conn.execute("PRAGMA journal_mode=WAL") _apply_macos_checkpoint_barrier(conn) diff --git a/plugins/platforms/discord/recovery.py b/plugins/platforms/discord/recovery.py index 97217d76805..33c730a0438 100644 --- a/plugins/platforms/discord/recovery.py +++ b/plugins/platforms/discord/recovery.py @@ -54,7 +54,6 @@ class DiscordRecoveryStore: def _initialize(self, conn: sqlite3.Connection) -> None: from hermes_state import apply_wal_with_fallback - apply_wal_with_fallback(conn, db_label="discord_recovery.db") conn.execute(""" CREATE TABLE IF NOT EXISTS discord_messages ( diff --git a/tests/test_journal_mode_config.py b/tests/test_journal_mode_config.py new file mode 100644 index 00000000000..75fa072bb7f --- /dev/null +++ b/tests/test_journal_mode_config.py @@ -0,0 +1,84 @@ +"""#68545: configurable journal_mode (env + config.yaml) + centralized DB openers.""" + +from __future__ import annotations + +import inspect +import os +import sqlite3 +from pathlib import Path + +import pytest + + +def test_resolve_journal_mode_defaults_to_wal(monkeypatch): + from hermes_state import resolve_journal_mode + + monkeypatch.delenv("HERMES_JOURNAL_MODE", raising=False) + assert resolve_journal_mode() == "wal" + + +def test_resolve_journal_mode_env_override(monkeypatch): + from hermes_state import resolve_journal_mode + + monkeypatch.setenv("HERMES_JOURNAL_MODE", "delete") + assert resolve_journal_mode() == "delete" + + +def test_resolve_journal_mode_env_truncase(monkeypatch): + from hermes_state import resolve_journal_mode + + monkeypatch.setenv("HERMES_JOURNAL_MODE", "DELETE") + assert resolve_journal_mode() == "delete" + + +def test_resolve_journal_mode_invalid_falls_back_to_wal(monkeypatch): + from hermes_state import resolve_journal_mode + + monkeypatch.setenv("HERMES_JOURNAL_MODE", "bogus") + assert resolve_journal_mode() == "wal" + + +def test_apply_wal_with_fallback_honors_delete_mode(monkeypatch, tmp_path): + """When HERMES_JOURNAL_MODE=delete, apply_wal_with_fallback must NOT set WAL.""" + from hermes_state import apply_wal_with_fallback + + monkeypatch.setenv("HERMES_JOURNAL_MODE", "delete") + db = tmp_path / "test.db" + conn = sqlite3.connect(str(db)) + mode = apply_wal_with_fallback(conn, db_label="test.db") + assert mode == "delete" + actual = conn.execute("PRAGMA journal_mode").fetchone()[0] + assert actual.lower() == "delete" + conn.close() + + +def test_apply_wal_with_fallback_defaults_to_wal(monkeypatch, tmp_path): + """Without override, apply_wal_with_fallback still sets WAL.""" + from hermes_state import apply_wal_with_fallback + + monkeypatch.delenv("HERMES_JOURNAL_MODE", raising=False) + # Avoid the WAL-reset vulnerability check on older SQLite versions + # (our dev env has 3.50.4 which is flagged vulnerable, causing a + # delete-mode fallback that makes this test environment-dependent). + monkeypatch.setattr( + "hermes_state.is_sqlite_wal_reset_vulnerable", lambda **kw: False + ) + db = tmp_path / "test2.db" + conn = sqlite3.connect(str(db)) + mode = apply_wal_with_fallback(conn, db_label="test2.db") + assert mode == "wal" + conn.close() + + +def test_direct_setters_use_apply_wal_with_fallback(): + """All 5 bypass openers must route through apply_wal_with_fallback (#68545).""" + for fpath in [ + "tools/async_delegation.py", + "gateway/delivery_ledger.py", + "agent/verification_evidence.py", + "cron/executions.py", + "plugins/platforms/discord/recovery.py", + ]: + src = Path(fpath).read_text(encoding="utf-8") + assert "apply_wal_with_fallback" in src, f"{fpath} must use apply_wal_with_fallback" + assert 'PRAGMA journal_mode=WAL"' not in src, f"{fpath} must not set WAL directly" \ No newline at end of file diff --git a/tools/async_delegation.py b/tools/async_delegation.py index 702036df0c4..f533993a87e 100644 --- a/tools/async_delegation.py +++ b/tools/async_delegation.py @@ -78,11 +78,6 @@ _DEFAULT_MAX_ASYNC_CHILDREN = 3 _MAX_RETAINED_COMPLETED = 50 _DURABLE_RETENTION_SECONDS = 7 * 24 * 60 * 60 _MAX_DURABLE_PENDING = 1000 -# A pending completion whose delivery keeps failing is retried across claim -# cycles (and across restarts via restore_undelivered_completions). Cap the -# attempts so an unroutable row converges to a terminal 'dropped' state -# instead of replaying on every restart forever. -_MAX_DELIVERY_ATTEMPTS = 8 _DB_LOCK = threading.Lock() # --------------------------------------------------------------------------- @@ -157,8 +152,7 @@ def _initialize_schema(conn: sqlite3.Connection) -> None: owner_started_at INTEGER, task_json TEXT, delivery_claim TEXT, - delivery_claimed_at REAL, - origin_session_id TEXT NOT NULL DEFAULT '' + delivery_claimed_at REAL )""" ) columns = {row[1] for row in conn.execute("PRAGMA table_info(async_delegations)")} @@ -168,11 +162,6 @@ def _initialize_schema(conn: sqlite3.Connection) -> None: ("task_json", "TEXT"), ("delivery_claim", "TEXT"), ("delivery_claimed_at", "REAL"), - # Raw api_server session id (X-Hermes-Session-Id) of the ORIGINATING - # request — the wake self-post target. Without persisting it, - # completions recovered after a process restart are unroutable on - # api_server (the in-memory record that carried it is gone). - ("origin_session_id", "TEXT"), ): if name not in columns: conn.execute(f"ALTER TABLE async_delegations ADD COLUMN {name} {sql_type}") @@ -215,13 +204,12 @@ def _persist_dispatch(record: Dict[str, Any]) -> None: (delegation_id, origin_session, origin_ui_session_id, parent_session_id, state, dispatched_at, updated_at, delivery_state, delivery_attempts, owner_pid, - owner_started_at, task_json, origin_session_id) - VALUES (?, ?, ?, ?, 'running', ?, ?, 'pending', 0, ?, ?, ?, ?)""", + owner_started_at, task_json) + VALUES (?, ?, ?, ?, 'running', ?, ?, 'pending', 0, ?, ?, ?)""", (record["delegation_id"], record.get("session_key", ""), record.get("origin_ui_session_id", ""), record.get("parent_session_id"), record["dispatched_at"], now, __import__("os").getpid(), - owner_started_at, json.dumps(task_payload), - record.get("origin_session_id", "")), + owner_started_at, json.dumps(task_payload)), ) _prune_durable_records() @@ -302,12 +290,11 @@ def recover_abandoned_delegations() -> int: rows = conn.execute( """SELECT delegation_id, origin_session, origin_ui_session_id, parent_session_id, dispatched_at, owner_pid, - owner_started_at, task_json, origin_session_id + owner_started_at, task_json FROM async_delegations WHERE state IN ('running','finalizing')""" ).fetchall() for row in rows: - (delegation_id, session_key, origin_ui, parent_id, dispatched_at, - pid, started, task_json, origin_session_id) = row + delegation_id, session_key, origin_ui, parent_id, dispatched_at, pid, started, task_json = row live = False if pid: live = _pid_exists(int(pid)) @@ -319,9 +306,6 @@ def recover_abandoned_delegations() -> int: event = { "type": "async_delegation", "delegation_id": delegation_id, "session_key": session_key, "origin_ui_session_id": origin_ui, - # Restore the durable wake target so completions recovered - # after a restart remain routable to api_server sessions. - "origin_session_id": origin_session_id or "", "parent_session_id": parent_id, "goal": task.get("goal", ""), "goals": task.get("goals"), "context": task.get("context"), "toolsets": task.get("toolsets"), "role": task.get("role"), @@ -498,8 +482,7 @@ def get_durable_delegation(delegation_id: str) -> Optional[Dict[str, Any]]: with _DB_LOCK, _transaction() as conn: row = conn.execute( """SELECT origin_session, state, dispatched_at, completed_at, - result_json, delivery_state, delivery_attempts, - origin_session_id + result_json, delivery_state, delivery_attempts FROM async_delegations WHERE delegation_id=?""", (delegation_id,), ).fetchone() if row is None: @@ -509,7 +492,6 @@ def get_durable_delegation(delegation_id: str) -> Optional[Dict[str, Any]]: "dispatched_at": row[2], "completed_at": row[3], "result": json.loads(row[4]) if row[4] else None, "delivery_state": row[5], "delivery_attempts": row[6], - "origin_session_id": row[7] or "", } @@ -593,34 +575,6 @@ def _prune_completed_locked() -> None: _records.pop(rid, None) -def _current_origin_session_id() -> str: - """Raw session id of the ORIGINATING api_server request, or ``""``. - - The obvious source — ``HERMES_SESSION_ID`` via ``get_session_env`` — is - NOT safe to read at dispatch time: constructing a child agent - (``agent/agent_init.py``) calls ``set_current_session_id(child.session_id)``, - clobbering that ContextVar *and* ``os.environ`` with the subagent's - internal ``{timestamp}_{uuid}`` id moments before the dispatch code reads - it, so the completion wake would self-post into the subagent's own - (unread) session instead of the spawner's. - - The request-scoped ``HERMES_SESSION_CHAT_ID`` binding survives child - construction: ``_bind_api_server_session`` binds ``chat_id`` to the raw - ``X-Hermes-Session-Id``, and its only writer is ``set_session_vars`` — - ``set_current_session_id`` never touches it. Gate on the platform: on - push platforms ``chat_id`` is a chat, not a session, so yield ``""`` - there. - """ - try: - from gateway.session_context import get_session_env - - if get_session_env("HERMES_SESSION_PLATFORM", "") != "api_server": - return "" - return get_session_env("HERMES_SESSION_CHAT_ID", "") or "" - except Exception: - return "" - - def dispatch_async_delegation( *, goal: str, @@ -632,7 +586,6 @@ def dispatch_async_delegation( parent_session_id: Optional[str] = None, runner: Callable[[], Dict[str, Any]], origin_ui_session_id: str = "", - origin_session_id: str = "", interrupt_fn: Optional[Callable[[], None]] = None, max_async_children: int = _DEFAULT_MAX_ASYNC_CHILDREN, progress_fn: Optional[Callable[[], tuple]] = None, @@ -690,7 +643,6 @@ def dispatch_async_delegation( "model": model, "session_key": session_key, "origin_ui_session_id": origin_ui_session_id, - "origin_session_id": origin_session_id, "parent_session_id": parent_session_id, "status": "running", "dispatched_at": dispatched_at, @@ -837,7 +789,6 @@ def _push_completion_event( # session; empty string => CLI (single-session) path. "session_key": record.get("session_key", ""), "origin_ui_session_id": record.get("origin_ui_session_id", ""), - "origin_session_id": record.get("origin_session_id", ""), "parent_session_id": record.get("parent_session_id"), "goal": record.get("goal", ""), "context": record.get("context"), @@ -887,7 +838,6 @@ def dispatch_async_delegation_batch( parent_session_id: Optional[str] = None, runner: Callable[[], Dict[str, Any]], origin_ui_session_id: str = "", - origin_session_id: str = "", interrupt_fn: Optional[Callable[[], None]] = None, max_async_children: int = _DEFAULT_MAX_ASYNC_CHILDREN, delegation_id: Optional[str] = None, @@ -930,7 +880,6 @@ def dispatch_async_delegation_batch( "model": model, "session_key": session_key, "origin_ui_session_id": origin_ui_session_id, - "origin_session_id": origin_session_id, "parent_session_id": parent_session_id, "status": "running", "dispatched_at": dispatched_at, @@ -1042,7 +991,6 @@ def _push_batch_completion_event( "delegation_id": event_record.get("delegation_id"), "session_key": event_record.get("session_key", ""), "origin_ui_session_id": event_record.get("origin_ui_session_id", ""), - "origin_session_id": event_record.get("origin_session_id", ""), "parent_session_id": event_record.get("parent_session_id"), "goal": event_record.get("goal", ""), "goals": event_record.get("goals"),