Kanban dispatcher handling of corrupt board databases

Fixes on record

Merged to main. A stamp counts the issues a fix is linked to.

• #28439 fix(gateway): quiet corrupt kanban dispatcher boards (#26490)

  merged to main

  files: gateway/run.py, tests/hermes_cli/test_kanban_core_functionality.py

  Show the diff

  • gateway/run.py

    @@ -37,6 +37,7 @@
     import tempfile
     import threading
     import time
    +import sqlite3
     from collections import OrderedDict
     from contextvars import copy_context
     from pathlib import Path
    @@ -4830,6 +4831,28 @@ async def _kanban_dispatcher_watcher(self) -> None:
             HEALTH_WINDOW = 6
             bad_ticks = 0
             last_warn_at = 0
    +        disabled_corrupt_boards: dict[str, tuple[str, int | None, int | None]] = {}
    +
    +        def _board_db_fingerprint(slug: str) -> tuple[str, int | None, int | None]:
    +            path = _kb.kanban_db_path(slug)
    +            try:
    +                resolved = str(path.expanduser().resolve())
    +            except Exception:
    +                resolved = str(path)
    +            try:
    +                stat = path.stat()
    +            except OSError:
    +                return (resolved, None, None)
    +            return (resolved, stat.st_mtime_ns, stat.st_size)
    +
    +        def _is_corrupt_board_db_error(exc: Exception) -> bool:
    +            if not isinstance(exc, sqlite3.DatabaseError):
    +                return False
    +            msg = str(exc).lower()
    +            return (
    +                "file is not a database" in msg
    +                or "database disk image is malformed" in msg
    +            )
     
             def _tick_once_for_board(slug: str) -> "Optional[object]":
                 """Run one dispatch_once for a specific board.
    @@ -4841,6 +4864,16 @@ def _tick_once_for_board(slug: str) -> "Optional[object]":
                 connection handle or accidentally claim across each other.
                 """
                 conn = None
    +            fingerprint = _board_db_fingerprint(slug)
    +            disabled_fingerprint = disabled_corrupt_boards.get(slug)
    +            if disabled_fingerprint == fingerprint:
    +                return None
    +            if disabled_fingerprint is not None:
    +                logger.info(
    +                    "kanban dispatcher: board %s database changed; retrying dispatch",
    +                    slug,
    +                )
    +                disabled_corrupt_boards.pop(slug, None)
                 try:
                     conn = _kb.connect(board=slug)
                     # `connect()` runs the schema + idempotent migration on
    @@ -4856,6 +4889,21 @@ def _tick_once_for_board(slug: str) -> "Optional[object]":
                         max_in_progress=max_in_progress,
                         failure_limit=failure_limit,
                     )
    +            except sqlite3.DatabaseError as exc:
    +                if _is_corrupt_board_db_error(exc):
    +                    disabled_corrupt_boards[slug] = fingerprint
    +                    logger.error(
    +                        "kanban dispatcher: board %s database %s is not a valid "
    +                        "SQLite database; disabling dispatch for this board "
    +                        "until the file changes or the gateway restarts. Move "
    +                        "or restore the file, then run `hermes kanban init` if "
    +                        "you need a fresh board.",
    +                        slug,
    +                        fingerprint[0],
    +                    )
    +                    return None
    +                logger.exception("kanban dispatcher: tick failed on board %s", slug)
    +                return None
                 except Exception:
                     logger.exception("kanban dispatcher: tick failed on board %s", slug)
                     return None

  • tests/hermes_cli/test_kanban_core_functionality.py

    @@ -3594,6 +3594,82 @@ def test_gateway_dispatcher_watcher_env_truthy_uses_config(monkeypatch):
         )
     
     
    +def test_gateway_dispatcher_disables_corrupt_board_without_traceback(
    +    monkeypatch, tmp_path, caplog
    +):
    +    """Corrupt board DBs log one actionable error and stop retrying per tick."""
    +    import asyncio
    +    import logging
    +    import sqlite3
    +
    +    from gateway.run import GatewayRunner
    +    import hermes_cli.config as _cfg_mod
    +    import hermes_cli.kanban_db as _kb
    +
    +    runner = object.__new__(GatewayRunner)
    +    runner._running = True
    +    corrupt_db = tmp_path / "kanban.db"
    +    corrupt_db.write_text("not sqlite", encoding="utf-8")
    +
    +    monkeypatch.setattr(
    +        _cfg_mod,
    +        "load_config",
    +        lambda: {
    +            "kanban": {
    +                "dispatch_in_gateway": True,
    +                "dispatch_interval_seconds": 1,
    +            }
    +        },
    +    )
    +    monkeypatch.setattr(
    +        _kb,
    +        "list_boards",
    +        lambda include_archived=False: [{"slug": _kb.DEFAULT_BOARD}],
    +    )
    +    monkeypatch.setattr(
    +        _kb,
    +        "read_board_metadata",
    +        lambda slug: {"slug": slug},
    +    )
    +    monkeypatch.setattr(_kb, "kanban_db_path", lambda board=None: corrupt_db)
    +
    +    calls = {"connect": 0, "to_thread": 0}
    +
    +    def _connect(*args, **kwargs):
    +        calls["connect"] += 1
    +        raise sqlite3.DatabaseError("file is not a database")
    +
    +    async def _to_thread(fn, *args, **kwargs):
    +        calls["to_thread"] += 1
    +        result = fn(*args, **kwargs)
    +        if calls["to_thread"] >= 4:
    +            runner._running = False
    +        return result
    +
    +    async def _sleep(_delay):
    +        return None
    +
    +    monkeypatch.setattr(_kb, "connect", _connect)
    +    monkeypatch.setattr("gateway.run.asyncio.to_thread", _to_thread)
    +    monkeypatch.setattr("gateway.run.asyncio.sleep", _sleep)
    +
    +    with caplog.at_level(logging.ERROR, logger="gateway.run"):
    +        asyncio.run(
    +            asyncio.wait_for(
    +                runner._kanban_dispatcher_watcher(),
    +                timeout=3.0,
    +            )
    +        )
    +
    +    messages = [record.getMessage() for record in caplog.records]
    +    assert sum("not a valid SQLite database" in msg for msg in messages) == 1
    +    assert not any("tick failed on board" in msg for msg in messages)
    +    assert not any(record.exc_info for record in caplog.records)
    +    # First tick connect + two ready-queue probes. The second dispatch tick
    +    # skips connect because the corrupt board fingerprint is disabled.
    +    assert calls["connect"] == 3
    +
    +
     # ---------------------------------------------------------------------------
     # Hallucination gate (created_cards verify + prose scan)
     # ---------------------------------------------------------------------------

• #33482 fix(kanban): batch-salvage 7 SQLite corruption hardening fixes from #32857

  merged to main

  files: gateway/run.py, hermes_cli/kanban_db.py, hermes_state.py, scripts/release.py, tests/hermes_cli/test_kanban_core_functionality.py, tests/hermes_cli/test_kanban_db.py, tests/test_hermes_state.py, tests/test_hermes_state_wal_fallback.py

  Show the diff

  • gateway/run.py

    @@ -5691,6 +5691,19 @@ def _auto_decompose_tick() -> int:
                 "kanban dispatcher: embedded in gateway (interval=%.1fs)", interval
             )
             while self._running:
    +            try:
    +                # Reap zombie children before per-board work so a board DB
    +                # failure cannot block cleanup of unrelated workers.
    +                pids = await asyncio.to_thread(_kb.reap_worker_zombies)
    +                if pids:
    +                    logger.info(
    +                        "kanban dispatcher: reaped %d zombie worker(s), pids=%s",
    +                        len(pids),
    +                        pids,
    +                    )
    +            except Exception:
    +                logger.exception("kanban dispatcher: zombie reaper failed")
    +
                 try:
                     if auto_decompose_enabled:
                         await asyncio.to_thread(_auto_decompose_tick)

  • hermes_cli/kanban_db.py

    @@ -134,6 +134,34 @@ def _resolve_claim_ttl_seconds(ttl_seconds: Optional[int] = None) -> int:
         return DEFAULT_CLAIM_TTL_SECONDS
     
     
    +# Grace period after a task transitions to ``running`` during which
    +# ``detect_crashed_workers`` skips the ``_pid_alive`` check. Covers the
    +# fork() → /proc-visibility window where liveness can transiently report
    +# False for a freshly-spawned worker. The 15-minute claim TTL still
    +# catches genuinely-crashed workers; this only suppresses false positives
    +# during the launch window.
    +DEFAULT_CRASH_GRACE_SECONDS = 30
    +
    +
    +def _resolve_crash_grace_seconds() -> int:
    +    """Return the crash-detection grace period in seconds.
    +
    +    Reads ``HERMES_KANBAN_CRASH_GRACE_SECONDS`` from the environment;
    +    falls back to ``DEFAULT_CRASH_GRACE_SECONDS`` when absent, empty,
    +    non-integer, or negative. A value of 0 restores immediate-reclaim
    +    behaviour (useful for tests).
    +    """
    +    raw = os.environ.get("HERMES_KANBAN_CRASH_GRACE_SECONDS", "").strip()
    +    if raw:
    +        try:
    +            parsed = int(raw)
    +        except ValueError:
    +            parsed = -1
    +        if parsed >= 0:
    +            return parsed
    +    return DEFAULT_CRASH_GRACE_SECONDS
    +
    +
     # Worker-context caps so build_worker_context() stays bounded on
     # pathological boards (retry-heavy tasks, comment storms, giant
     # summaries). Values chosen to fit a typical 100k-char LLM prompt with
    @@ -1181,8 +1209,17 @@ def connect(
                 # See hermes_state._WAL_INCOMPAT_MARKERS for detection logic.
                 from hermes_state import apply_wal_with_fallback
                 apply_wal_with_fallback(conn, db_label=f"kanban.db ({path.name})")
    -            conn.execute("PRAGMA synchronous=NORMAL")
    +            # FULL (was NORMAL): fsync before each checkpoint to narrow the
    +            # crash window that can leave a b-tree page header torn.
    +            conn.execute("PRAGMA synchronous=FULL")
    +            conn.execute("PRAGMA wal_autocheckpoint=100")
                 conn.execute("PRAGMA foreign_keys=ON")
    +            # Zero freed pages so a later torn write cannot expose stale
    +            # cell content; persisted in the DB header for new DBs.
    +            conn.execute("PRAGMA secure_delete=ON")
    +            # Surface corrupt cells as read errors instead of silent
    +            # wrong-data returns.
    +            conn.execute("PRAGMA cell_size_check=ON")
                 needs_init = resolved not in _INITIALIZED_PATHS
                 if needs_init:
                     # Idempotent: runs CREATE TABLE IF NOT EXISTS + the additive
    @@ -1466,6 +1503,45 @@ def _migrate_add_optional_columns(conn: sqlite3.Connection) -> None:
             )
     
     
    +def _check_file_length_invariant(conn: sqlite3.Connection) -> None:
    +    """Read the SQLite header page_count and compare against actual file size.
    +
    +    Raises sqlite3.DatabaseError if the file is shorter than the header claims
    +    (torn-extend corruption).
    +    """
    +    try:
    +        row = conn.execute("PRAGMA database_list").fetchone()
    +        if row is None:
    +            return
    +        path_str = row[2]  # column 2 is the file path; empty for in-memory DBs
    +        if not path_str:
    +            return  # in-memory or unnamed DB; skip
    +        path = path_str
    +        page_size = conn.execute("PRAGMA page_size").fetchone()[0]
    +        file_size = os.path.getsize(path)
    +        with open(path, "rb") as f:
    +            f.seek(28)
    +            header_bytes = f.read(4)
    +        if len(header_bytes) < 4:
    +            return  # can't read header; skip
    +        header_page_count = int.from_bytes(header_bytes, "big")
    +        if header_page_count == 0:
    +            return  # new/empty DB; skip
    +        actual_pages = file_size // page_size
    +        if actual_pages < header_page_count:
    +            raise sqlite3.DatabaseError(
    +                f"torn-extend detected: page count mismatch on {path}: "
    +                f"header claims {header_page_count} pages, "
    +                f"file has {actual_pages} pages "
    +                f"(missing {header_page_count - actual_pages} pages, "
    +                f"file_size={file_size}, page_size={page_size})"
    +            )
    +    except sqlite3.DatabaseError:
    +        raise
    +    except Exception:
    +        pass  # I/O errors during check are non-fatal; let normal ops continue
    +
    +
     @contextlib.contextmanager
     def write_txn(conn: sqlite3.Connection):
         """Context manager for an IMMEDIATE write transaction.
    @@ -1473,15 +1549,28 @@ def write_txn(conn: sqlite3.Connection):
         Use for any multi-statement write (creating a task + link, claiming a
         task + recording an event, etc.).  A claim CAS inside this context is
         atomic -- at most one concurrent writer can succeed.
    +
    +    The explicit ROLLBACK on exception is wrapped in try/except so that
    +    a SQLite auto-rollback (which leaves no active transaction) does not
    +    shadow the original exception with a spurious rollback error.
         """
         conn.execute("BEGIN IMMEDIATE")
         try:
             yield conn
         except Exception:
    -        conn.execute("ROLLBACK")
    +        try:
    +            conn.execute("ROLLBACK")
    +        except sqlite3.OperationalError:
    +            # SQLite has already auto-rolled-back the transaction (typical
    +            # under EIO, lock contention, or corruption). Nothing to undo;
    +            # do not let this secondary failure shadow the real one.
    +            pass
             raise
         else:
             conn.execute("COMMIT")
    +        # Post-commit file-length check: header page_count must match actual file pages.
    +        # A discrepancy means a torn-extend — raise now rather than silently corrupt.
    +        _check_file_length_invariant(conn)
     
     
     # ---------------------------------------------------------------------------
    @@ -4169,6 +4258,30 @@ def _classify_worker_exit(pid: int) -> "tuple[str, Optional[int]]":
         return ("unknown", None)
     
     
    +def reap_worker_zombies() -> "list[int]":
    +    """Reap all zombie children of this process without blocking.
    +
    +    Returns the list of reaped PIDs. Safe to call when there are no
    +    children (returns []). No-op on Windows.
    +    """
    +    if os.name == "nt":
    +        return []
    +    reaped: "list[int]" = []
    +    try:
    +        while True:
    +            try:
    +                pid, status = os.waitpid(-1, os.WNOHANG)
    +            except ChildProcessError:
    +                break
    +            if pid == 0:
    +                break
    +            _record_worker_exit(pid, status)
    +            reaped.append(pid)
    +    except Exception:
    +        pass
    +    return reaped
    +
    +
     def _pid_alive(pid: Optional[int]) -> bool:
         """Return True if ``pid`` is still running on this host.
     
    @@ -4635,7 +4748,7 @@ def detect_crashed_workers(conn: sqlite3.Connection) -> list[str]:
         # (task_id, pid, claimer, protocol_violation, error_text)
         with write_txn(conn):
             rows = conn.execute(
    -            "SELECT id, worker_pid, claim_lock FROM tasks "
    +            "SELECT id, worker_pid, claim_lock, started_at FROM tasks "
                 "WHERE status = 'running' AND worker_pid IS NOT NULL"
             ).fetchall()
             host_prefix = f"{_claimer_id().split(':', 1)[0]}:"
    @@ -4644,6 +4757,14 @@ def detect_crashed_workers(conn: sqlite3.Connection) -> list[str]:
                 lock = row["claim_lock"] or ""
                 if not lock.startswith(host_prefix):
                     continue
    +            # Skip liveness check inside the launch-window grace period
    +            # so a freshly-spawned worker isn't reclaimed before its PID
    +            # is visible on /proc.
    +            started_at = row["started_at"] if "started_at" in row.keys() else None
    +            if started_at is not None:
    +                grace = _resolve_crash_grace_seconds()
    +                if time.time() - started_at < grace:
    +                    continue
                 if _pid_alive(row["worker_pid"]):
                     continue
     
    @@ -5125,38 +5246,9 @@ def dispatch_once(
         ``board`` pins workspace/log/db resolution for this tick to a specific
         board. When omitted, the current-board resolution chain is used.
         """
    -    # Reap zombie children from previously spawned workers.
    -    # The gateway-embedded dispatcher is the parent of every worker spawned
    -    # via _default_spawn (start_new_session=True only detaches the
    -    # controlling tty, not the parent). Without an explicit waitpid, each
    -    # completed worker becomes a <defunct> entry that lingers until gateway
    -    # exit. WNOHANG keeps this non-blocking; ChildProcessError means no
    -    # children to reap. Bounded: at most one tick's worth of completions
    -    # can be in <defunct> at once.
    -    #
    -    # We also record the exit status keyed by pid, so
    -    # ``detect_crashed_workers`` can distinguish a worker that exited
    -    # cleanly without calling ``kanban_complete`` / ``kanban_block``
    -    # (protocol violation — auto-block) from a real crash (OOM killer,
    -    # SIGKILL, non-zero exit — existing counter behavior).
    -    #
    -    # Windows has no zombies / no os.WNOHANG — subprocess.Popen handles
    -    # are freed when the Python object is garbage-collected or .wait() is
    -    # called explicitly.  The kanban dispatcher discards the Popen handle
    -    # after spawn (``_default_spawn`` → abandon), so on Windows there's
    -    # nothing to reap here — skip the whole block.
    -    if os.name != "nt":
    -        try:
    -            while True:
    -                try:
    -                    _pid, _status = os.waitpid(-1, os.WNOHANG)
    -                except ChildProcessError:
    -                    break
    -                if _pid == 0:
    -                    break
    -                _record_worker_exit(_pid, _status)
    -        except Exception:
    -            pass
    +    # Reap zombie children from previously spawned workers. See
    +    # reap_worker_zombies() for the full rationale.
    +    reap_worker_zombies()
     
         result = DispatchResult()
         result.reclaimed = release_stale_claims(conn)

  • hermes_state.py

    @@ -54,7 +54,6 @@
     _WAL_INCOMPAT_MARKERS = (
         "locking protocol",       # SQLITE_PROTOCOL on NFS/SMB
         "not authorized",         # Some FUSE mounts block WAL pragma outright
    -    "disk i/o error",         # Flaky network FS during WAL setup
     )
     
     # Last SessionDB() init error, per-process.  Surfaced in /resume and
    @@ -125,6 +124,27 @@ def format_session_db_unavailable(prefix: str = "Session database not available"
         return f"{prefix}: {cause}{hint}."
     
     
    +def _on_disk_journal_mode(conn: sqlite3.Connection) -> Optional[str]:
    +    """Read the journal mode from the SQLite DB header on disk.
    +
    +    Returns the mode string (e.g. ``"wal"``, ``"delete"``), or ``None``
    +    if the value cannot be determined (new DB, or PRAGMA read failed).
    +    """
    +    try:
    +        row = conn.execute("PRAGMA journal_mode").fetchone()
    +    except sqlite3.OperationalError:
    +        return None
    +    if row is None:
    +        return None
    +    mode = row[0]
    +    if isinstance(mode, bytes):  # defensive: sqlite3 occasionally returns bytes
    +        try:
    +            mode = mode.decode("ascii")
    +        except UnicodeDecodeError:
    +            return None
    +    return str(mode).strip().lower() if mode is not None else None
    +
    +
     def apply_wal_with_fallback(
         conn: sqlite3.Connection,
         *,
    @@ -147,7 +167,18 @@ def apply_wal_with_fallback(
     
         Shared by :class:`SessionDB` and ``hermes_cli.kanban_db.connect`` so
         both databases get identical fallback behavior.
    +
    +    Never downgrades to DELETE if the on-disk DB header reports WAL — see _on_disk_journal_mode.
         """
    +    # Read-only probe — no flock, no checkpoint, no WAL/SHM unlink.
    +    # Skipping the set-pragma prevents WAL-init from unlinking files other connections hold open.
    +    try:
    +        current_mode = conn.execute("PRAGMA journal_mode").fetchone()
    +        if current_mode and current_mode[0] == "wal":
    +            return "wal"
    +    except sqlite3.OperationalError:
    +        pass
    +
         try:
             conn.execute("PRAGMA journal_mode=WAL")
             return "wal"
    @@ -156,6 +187,10 @@ def apply_wal_with_fallback(
             if not any(marker in msg for marker in _WAL_INCOMPAT_MARKERS):
                 # Unrelated OperationalError — don't silently swallow.
                 raise
    +        # Don't downgrade if another process already set WAL on disk.
    +        existing = _on_disk_journal_mode(conn)
    +        if existing == "wal":
    +            raise
             _log_wal_fallback_once(db_label, exc)
             conn.execute("PRAGMA journal_mode=DELETE")
             return "delete"

  • scripts/release.py

    @@ -71,6 +71,8 @@
         "[redacted:email]": "Strontvod",
         "[redacted:email]": "someaka",
         "[redacted:email]": "Jaggia",
    +    "[redacted:email]": "steveonjava",
    +    "[redacted:email]": "steveonjava",
         "[redacted:email]": "simpolism",
         "[redacted:email]": "simpolism",
         "[redacted:email]": "simpolism",

  • tests/hermes_cli/test_kanban_core_functionality.py

    @@ -35,7 +35,19 @@ def kanban_home(tmp_path, monkeypatch):
         home = tmp_path / ".hermes"
         home.mkdir()
         monkeypatch.setenv("HERMES_HOME", str(home))
    +    # Existing crash-detection tests pre-date the grace window; pin to 0
    +    # so they keep their immediate-reclaim semantics.
    +    monkeypatch.setenv("HERMES_KANBAN_CRASH_GRACE_SECONDS", "0")
         monkeypatch.setattr(Path, "home", lambda: tmp_path)
    +    # Disable the detect_crashed_workers grace period for legacy tests in
    +    # this file that claim a task and immediately expect
    +    # ``detect_crashed_workers`` to act on it. The grace period (30s by
    +    # default, see ``DEFAULT_CRASH_GRACE_SECONDS``) prevents the
    +    # multi-dispatcher reap race in production; setting it to 0 here
    +    # restores the pre-fix instant-reclaim semantics these tests were
    +    # written against. The grace-period itself is covered by dedicated
    +    # tests in tests/hermes_cli/test_kanban_db.py.
    +    monkeypatch.setenv("HERMES_KANBAN_CRASH_GRACE_SECONDS", "0")
         kb.init_db()
         return home
     
    @@ -3655,9 +3667,16 @@ def _connect(*args, **kwargs):
             raise sqlite3.DatabaseError("file is not a database")
     
         async def _to_thread(fn, *args, **kwargs):
    +        # PR salvage (#32857 commit 7): the dispatcher now reaps zombies at
    +        # the top of each tick via ``asyncio.to_thread(_kb.reap_worker_zombies)``
    +        # BEFORE the per-board tick work. Each tick now issues 3 ``to_thread``
    +        # calls (reaper + ``_tick_once`` + ``_ready_nonempty``) instead of 2,
    +        # so this counter must reach 6 to allow the same 2 dispatch ticks the
    +        # pre-reaper test expected at 4. Connect counts in the assertion below
    +        # are unchanged.
             calls["to_thread"] += 1
             result = fn(*args, **kwargs)
    -        if calls["to_thread"] >= 4:
    +        if calls["to_thread"] >= 6:
                 runner._running = False
             return result
     

  • tests/hermes_cli/test_kanban_db.py

    @@ -6,6 +6,7 @@
     import os
     import sqlite3
     import time
    +import unittest.mock
     from pathlib import Path
     
     import pytest
    @@ -564,6 +565,80 @@ def test_detect_crashed_workers_isolated_failure_normal_retry(
                 )
     
     
    +def test_detect_crashed_workers_skips_freshly_claimed_tasks(
    +    kanban_home, monkeypatch,
    +):
    +    """Grace period prevents reclaim of freshly-started tasks."""
    +    import hermes_cli.kanban_db as _kb
    +
    +    monkeypatch.setattr(_kb, "_pid_alive", lambda _pid: False)
    +    monkeypatch.delenv("HERMES_KANBAN_CRASH_GRACE_SECONDS", raising=False)
    +
    +    now = 1_000_000.0
    +    monkeypatch.setattr(_kb.time, "time", lambda: now)
    +
    +    with kb.connect() as conn:
    +        host = _kb._claimer_id().split(":", 1)[0]
    +        tid = kb.create_task(conn, title="grace test", assignee="a")
    +        conn.execute(
    +            "UPDATE tasks SET status='running', worker_pid=?, "
    +            "claim_lock=?, started_at=? WHERE id=?",
    +            (99999, f"{host}:w", int(now), tid),
    +        )
    +        conn.commit()
    +
    +        # With time = now (just claimed), grace period should suppress reclaim.
    +        crashed = kb.detect_crashed_workers(conn)
    +        assert tid not in crashed, "should not reclaim freshly-started task"
    +
    +        # With time = now + 60 (past default 30s grace), should reclaim.
    +        monkeypatch.setattr(_kb.time, "time", lambda: now + 60)
    +        crashed = kb.detect_crashed_workers(conn)
    +        assert tid in crashed, "should reclaim task past grace period"
    +
    +
    +def test_detect_crashed_workers_grace_period_env_override(
    +    kanban_home, monkeypatch,
    +):
    +    """HERMES_KANBAN_CRASH_GRACE_SECONDS env var adjusts the window."""
    +    import hermes_cli.kanban_db as _kb
    +
    +    monkeypatch.setattr(_kb, "_pid_alive", lambda _pid: False)
    +    monkeypatch.setenv("HERMES_KANBAN_CRASH_GRACE_SECONDS", "5")
    +
    +    now = 2_000_000.0
    +
    +    with kb.connect() as conn:
    +        host = _kb._claimer_id().split(":", 1)[0]
    +        tid = kb.create_task(conn, title="env override test", assignee="a")
    +        conn.execute(
    +            "UPDATE tasks SET status='running', worker_pid=?, "
    +            "claim_lock=?, started_at=? WHERE id=?",
    +            (99999, f"{host}:w", int(now), tid),
    +        )
    +        conn.commit()
    +
    +        # 3s after claim: within 5s grace → no reclaim.
    +        monkeypatch.setattr(_kb.time, "time", lambda: now + 3)
    +        assert tid not in kb.detect_crashed_workers(conn)
    +
    +        # 6s after claim: past 5s grace → reclaim.
    +        monkeypatch.setattr(_kb.time, "time", lambda: now + 6)
    +        assert tid in kb.detect_crashed_workers(conn)
    +
    +
    +def test_resolve_crash_grace_seconds_handles_bad_env(monkeypatch):
    +    """Bad env values fall back to DEFAULT_CRASH_GRACE_SECONDS."""
    +    import hermes_cli.kanban_db as _kb
    +
    +    for bad_val in ("notanumber", "-5", ""):
    +        monkeypatch.setenv("HERMES_KANBAN_CRASH_GRACE_SECONDS", bad_val)
    +        result = _kb._resolve_crash_grace_seconds()
    +        assert result == _kb.DEFAULT_CRASH_GRACE_SECONDS, (
    +            f"expected default for {bad_val!r}, got {result}"
    +        )
    +
    +
     def test_max_runtime_uses_current_run_start_after_retry(kanban_home, monkeypatch):
         """A retry should get a fresh max-runtime window.
     
    @@ -2097,17 +2172,31 @@ def test_latest_summaries_batch_omits_tasks_without_summary(kanban_home):
     # NFS / network-filesystem fallback (see hermes_state.apply_wal_with_fallback)
     # ---------------------------------------------------------------------------
     
    -def test_connect_falls_back_to_delete_on_locking_protocol(kanban_home, caplog):
    +def test_connect_falls_back_to_delete_on_locking_protocol(tmp_path, monkeypatch, caplog):
         """kanban_db.connect() must handle ``locking protocol`` on NFS/SMB.
     
         Without this fallback, the gateway's kanban dispatcher crashes every
         60s and the kanban migration (``consecutive_failures`` ADD COLUMN) is
         retried forever — which is what the real-world user report shows
         (see hermes-agent issue #22032).
    +
    +    NOTE: We do NOT use the ``kanban_home`` fixture here because that
    +    fixture pre-initializes the DB via ``kb.init_db()`` — putting the
    +    file in WAL on disk. The Bug D safety guard now refuses to downgrade
    +    to DELETE when the on-disk header is already WAL, so testing the
    +    NFS-fallback path requires a truly-fresh DB file (NFS scenario in
    +    production: first connection of the first process ever to touch the
    +    file, where downgrading is safe because nobody else has WAL state
    +    yet).
         """
         import sqlite3 as _sqlite3
         from unittest.mock import patch as _patch
     
    +    home = tmp_path / ".hermes"
    +    home.mkdir()
    +    monkeypatch.setenv("HERMES_HOME", str(home))
    +    monkeypatch.setattr(Path, "home", lambda: tmp_path)
    +
         # Clear module cache so a fresh connect() is attempted
         kb._INITIALIZED_PATHS.clear()
     
    @@ -3339,3 +3428,380 @@ def test_maybe_emit_scratch_tip_skips_non_scratch_workspaces(kanban_home, caplog
                 ).fetchall()
                 assert "tip_scratch_workspace" not in [e["kind"] for e in events]
     
    +
    +# ---------------------------------------------------------------------------
    +# Connection pragmas (secure_delete, cell_size_check, synchronous=FULL)
    +# ---------------------------------------------------------------------------
    +
    +
    +def test_connect_sets_secure_delete_on(tmp_path):
    +    """secure_delete=ON must be active on every new connection."""
    +    db_path = tmp_path / "kanban.db"
    +    kb._INITIALIZED_PATHS.discard(str(db_path.resolve()))
    +    with kb.connect(db_path=db_path) as conn:
    +        row = conn.execute("PRAGMA secure_delete").fetchone()
    +    assert row[0] == 1, f"expected secure_delete=1, got {row[0]}"
    +
    +
    +def test_connect_sets_cell_size_check_on(tmp_path):
    +    """cell_size_check=ON must be active on every new connection."""
    +    db_path = tmp_path / "kanban.db"
    +    kb._INITIALIZED_PATHS.discard(str(db_path.resolve()))
    +    with kb.connect(db_path=db_path) as conn:
    +        row = conn.execute("PRAGMA cell_size_check").fetchone()
    +    assert row[0] == 1, f"expected cell_size_check=1, got {row[0]}"
    +
    +
    +def test_connect_sets_synchronous_full(tmp_path):
    +    """synchronous must be FULL (=2), not NORMAL (=1)."""
    +    db_path = tmp_path / "kanban.db"
    +    kb._INITIALIZED_PATHS.discard(str(db_path.resolve()))
    +    with kb.connect(db_path=db_path) as conn:
    +        row = conn.execute("PRAGMA synchronous").fetchone()
    +    assert row[0] == 2, f"expected synchronous=2 (FULL), got {row[0]}"
    +
    +
    +def test_connect_pragmas_applied_on_reconnect(tmp_path):
    +    """All three pragmas must be re-applied on every connect(), not just the first."""
    +    db_path = tmp_path / "kanban.db"
    +    kb._INITIALIZED_PATHS.discard(str(db_path.resolve()))
    +    # First connection: write a task and close.
    +    with kb.connect(db_path=db_path) as conn:
    +        kb.create_task(conn, title="reconnect-check")
    +    # Force re-init path by discarding path cache.
    +    kb._INITIALIZED_PATHS.discard(str(db_path.resolve()))
    +    # Second connection: pragmas must still be applied.
    +    with kb.connect(db_path=db_path) as conn:
    +        assert conn.execute("PRAGMA secure_delete").fetchone()[0] == 1
    +        assert conn.execute("PRAGMA cell_size_check").fetchone()[0] == 1
    +        assert conn.execute("PRAGMA synchronous").fetchone()[0] == 2
    +
    +
    +
    +def test_pragmas_not_accidentally_disabled_by_migrate_path(tmp_path):
    +    """Migration path must not reset connection pragmas."""
    +    db_path = tmp_path / "legacy.db"
    +    kb._INITIALIZED_PATHS.discard(str(db_path.resolve()))
    +    # Initialise with a fresh connect so schema + init run.
    +    with kb.connect(db_path=db_path) as conn:
    +        kb.create_task(conn, title="pre-migration-task")
    +    # Simulate a re-entry through the init/migration path by discarding path cache.
    +    kb._INITIALIZED_PATHS.discard(str(db_path.resolve()))
    +    with kb.connect(db_path=db_path) as conn:
    +        assert conn.execute("PRAGMA secure_delete").fetchone()[0] == 1
    +        assert conn.execute("PRAGMA cell_size_check").fetchone()[0] == 1
    +        assert conn.execute("PRAGMA synchronous").fetchone()[0] == 2
    +
    +# write_txn — rollback handler must not mask the original exception
    +# ---------------------------------------------------------------------------
    +
    +
    +def test_write_txn_preserves_original_exception_when_rollback_fails(kanban_home):
    +    """When a write inside write_txn raises an OperationalError that SQLite
    +    has already auto-rolled-back (e.g. ``disk I/O error``,
    +    ``database is locked``, ``database disk image is malformed``), the
    +    explicit ROLLBACK in ``write_txn.__exit__`` itself raises
    +    ``cannot rollback - no transaction is active``. The original cause
    +    must NOT be masked by the secondary rollback failure — operators rely
    +    on the original cause to diagnose the underlying issue.
    +    """
    +
    +    class FailingConnWrapper:
    +        """Delegate to a real connection, simulating an EIO during an INSERT
    +        that SQLite has already auto-rolled-back."""
    +
    +        def __init__(self, real):
    +            self._real = real
    +            self._fail_armed = True
    +
    +        def execute(self, sql, *args, **kwargs):
    +            if (
    +                self._fail_armed
    +                and sql.lstrip().upper().startswith("INSERT")
    +                and "task_events" in sql.lower()
    +            ):
    +                self._fail_armed = False  # one-shot
    +                # Simulate SQLite auto-rolling back the transaction by
    +                # issuing a real ROLLBACK now. After this, BEGIN IMMEDIATE
    +                # is no longer active and an explicit ROLLBACK would error.
    +                try:
    +                    self._real.execute("ROLLBACK")
    +                except sqlite3.OperationalError:
    +                    pass
    +                raise sqlite3.OperationalError("disk I/O error")
    +            return self._real.execute(sql, *args, **kwargs)
    +
    +        def __getattr__(self, name):
    +            return getattr(self._real, name)
    +
    +    with kb.connect() as conn:
    +        wrapper = FailingConnWrapper(conn)
    +        with pytest.raises(sqlite3.OperationalError) as excinfo:
    +            with kb.write_txn(wrapper):
    +                kb._append_event(wrapper, "t_bogus", "promoted", None)
    +
    +    msg = str(excinfo.value)
    +    assert "disk I/O error" in msg, (
    +        f"write_txn masked the original exception with rollback failure; "
    +        f"got {msg!r} (expected to contain 'disk I/O error')"
    +    )
    +    assert "cannot rollback" not in msg, (
    +        f"write_txn surfaced the rollback failure instead of the original "
    +        f"OperationalError; got {msg!r}"
    +    )
    +def test_write_txn_healthy_commit_no_exception(tmp_path):
    +    """Normal commit does not trigger the torn-extend check."""
    +    from hermes_cli.kanban_db import connect, write_txn, create_task
    +    db = tmp_path / "test.db"
    +    conn = connect(db_path=db)
    +    # Should not raise
    +    with write_txn(conn) as c:
    +        c.execute(
    +            "INSERT INTO tasks (id, title, assignee, status, priority, created_at) "
    +            "VALUES ('t_test01', 'test task', 'tester', 'todo', 0, 1234567890)"
    +        )
    +    row = conn.execute("SELECT title FROM tasks WHERE id='t_test01'").fetchone()
    +    assert row["title"] == "test task"
    +    conn.close()
    +
    +
    +def test_write_txn_raises_on_truncated_file(tmp_path):
    +    """A mocked smaller file size triggers the torn-extend check."""
    +    from hermes_cli.kanban_db import connect, write_txn
    +    import hermes_cli.kanban_db as kanban_db_module
    +    db = tmp_path / "test.db"
    +    conn = connect(db_path=db)
    +    # Get actual page size so we can fake a smaller file
    +    page_size = conn.execute("PRAGMA page_size").fetchone()[0]
    +    original_getsize = os.path.getsize
    +
    +    def fake_getsize(path):
    +        # Return a size that implies at least 1 fewer page than header claims
    +        real_size = original_getsize(path)
    +        return max(0, real_size - page_size)
    +
    +    with pytest.raises(sqlite3.DatabaseError, match="torn-extend|page count mismatch"):
    +        with unittest.mock.patch("hermes_cli.kanban_db.os.path.getsize", side_effect=fake_getsize):
    +            with write_txn(conn) as c:
    +                c.execute(
    +                    "INSERT INTO tasks (id, title, assignee, status, priority, created_at) "
    +                    "VALUES ('t_test02', 'test task 2', 'tester', 'todo', 0, 1234567890)"
    +                )
    +    conn.close()
    +
    +
    +def test_write_txn_post_commit_check_fires_every_call(tmp_path):
    +    """The invariant check runs on every write_txn call."""
    +    from hermes_cli.kanban_db import connect, write_txn
    +    import hermes_cli.kanban_db as kanban_db_module
    +    db = tmp_path / "test.db"
    +    conn = connect(db_path=db)
    +    call_count = 0
    +    real_check = kanban_db_module._check_file_length_invariant
    +
    +    def counting_check(c):
    +        nonlocal call_count
    +        call_count += 1
    +        real_check(c)
    +
    +    with unittest.mock.patch.object(kanban_db_module, "_check_file_length_invariant", counting_check):
    +        for i in range(3):
    +            with write_txn(conn) as c:
    +                c.execute(
    +                    f"INSERT INTO tasks (id, title, assignee, status, priority, created_at) "
    +                    f"VALUES ('t_fire{i:02d}', 'task {i}', 'tester', 'todo', 0, 1234567890)"
    +                )
    +    assert call_count == 3
    +    conn.close()
    +
    +
    +def test_connect_sets_wal_autocheckpoint_100(tmp_path):
    +    """connect() sets wal_autocheckpoint to 100."""
    +    from hermes_cli.kanban_db import connect
    +    db = tmp_path / "test.db"
    +    conn = connect(db_path=db)
    +    val = conn.execute("PRAGMA wal_autocheckpoint").fetchone()[0]
    +    assert val == 100
    +    conn.close()
    +
    +
    +def test_write_txn_check_reads_correct_header_fields(tmp_path):
    +    """Synthetic DB file with mismatched header page_count triggers the check."""
    +    import struct
    +    from hermes_cli.kanban_db import connect, write_txn, _check_file_length_invariant
    +    db = tmp_path / "synthetic.db"
    +    conn = connect(db_path=db)
    +    page_size = conn.execute("PRAGMA page_size").fetchone()[0]
    +    conn.close()
    +    # Now corrupt the file: claim N pages but truncate to N-1 pages
    +    with open(db, "rb") as f:
    +        data = bytearray(f.read())
    +    # Read current page_count from header bytes 28-31
    +    real_page_count = struct.unpack(">I", data[28:32])[0]
    +    if real_page_count < 2:
    +        # Need at least 2 pages to fake a truncation
    +        pytest.skip("DB too small for synthetic truncation test")
    +    # Truncate to N-1 pages
    +    truncated = bytes(data[: (real_page_count - 1) * page_size])
    +    with open(db, "wb") as f:
    +        f.write(truncated)
    +    # Now open and check — should raise
    +    # We can't use connect() because _validate_sqlite_header may block; use a raw connection
    +    raw_conn = sqlite3.connect(str(db), isolation_level=None)
    +    with pytest.raises(sqlite3.DatabaseError, match="torn-extend|page count mismatch"):
    +        _check_file_length_invariant(raw_conn)
    +    raw_conn.close()
    +
    +
    +# ---------------------------------------------------------------------------
    +# reap_worker_zombies() tests
    +# ---------------------------------------------------------------------------
    +
    +
    +def test_reap_worker_zombies_returns_count():
    +    """reap_worker_zombies() returns the list of reaped PIDs."""
    +    from unittest.mock import patch
    +
    +    fake_pids = [12345, 67890, 11111]
    +    call_count = [0]
    +
    +    def fake_waitpid(pid, flags):
    +        if call_count[0] < len(fake_pids):
    +            p = fake_pids[call_count[0]]
    +            call_count[0] += 1
    +            return p, 0
    +        return 0, 0
    +
    +    with patch("hermes_cli.kanban_db.os.waitpid", side_effect=fake_waitpid):
    +        with patch("hermes_cli.kanban_db._record_worker_exit"):
    +            pids = kb.reap_worker_zombies()
    +    assert pids == [12345, 67890, 11111]
    +
    +
    +def test_reap_worker_zombies_noop_on_windows(monkeypatch):
    +    """reap_worker_zombies() returns 0 and never calls os.waitpid on Windows."""
    +    from unittest.mock import patch
    +
    +    monkeypatch.setattr("hermes_cli.kanban_db.os.name", "nt")
    +    with patch("hermes_cli.kanban_db.os.waitpid") as mock_waitpid:
    +        result = kb.reap_worker_zombies()
    +    mock_waitpid.assert_not_called()
    +    assert result == []
    +
    +
    +def test_reap_worker_zombies_noop_no_children():
    +    """reap_worker_zombies() returns 0 without error when there are no children."""
    +    from unittest.mock import patch
    +
    +    with patch("hermes_cli.kanban_db.os.waitpid", side_effect=ChildProcessError):
    +        result = kb.reap_worker_zombies()
    +    assert result == []
    +
    +
    +def test_reap_worker_zombies_records_exit_status():
    +    """reap_worker_zombies() calls _record_worker_exit for each reaped pid."""
    +    from unittest.mock import patch
    +
    +    calls = []
    +    call_count = [0]
    +
    +    def fake_waitpid(pid, flags):
    +        call_count[0] += 1
    +        if call_count[0] == 1:
    +            return 12345, 0
    +        return 0, 0
    +
    +    with patch("hermes_cli.kanban_db.os.waitpid", side_effect=fake_waitpid):
    +        with patch(
    +            "hermes_cli.kanban_db._record_worker_exit",
    +            side_effect=lambda p, s: calls.append((p, s)),
    +        ):
    +            kb.reap_worker_zombies()
    +
    +    assert calls == [(12345, 0)]
    +
    +
    +def test_reap_worker_zombies_handles_waitpid_os_error():
    +    """reap_worker_zombies() does not propagate generic OSError from os.waitpid."""
    +    from unittest.mock import patch
    +
    +    with patch("hermes_cli.kanban_db.os.waitpid", side_effect=OSError("test error")):
    +        result = kb.reap_worker_zombies()
    +    assert result == []
    +
    +
    +def test_zombie_reaper_runs_despite_board_connect_failure():
    +    """reap_worker_zombies runs even when a board tick raises an error."""
    +    from unittest.mock import patch
    +
    +    call_count = [0]
    +
    +    def fake_waitpid(pid, flags):
    +        call_count[0] += 1
    +        if call_count[0] <= 2:
    +            return [12345, 67890][call_count[0] - 1], 0
    +        return 0, 0
    +
    +    with patch("hermes_cli.kanban_db.os.waitpid", side_effect=fake_waitpid):
    +        with patch("hermes_cli.kanban_db._record_worker_exit"):
    +            # Simulate a board tick failure before reaping
    +            try:
    +                raise sqlite3.OperationalError("disk I/O error")
    +            except sqlite3.OperationalError:
    +                pass
    +
    +            # Reaper still runs independently
    +            pids = kb.reap_worker_zombies()
    +
    +    assert pids == [12345, 67890]
    +
    +
    +def test_zombie_reaper_survives_all_boards_failing():
    +    """reap_worker_zombies runs each tick regardless of board tick failures."""
    +    from unittest.mock import patch
    +
    +    total_reaped = 0
    +
    +    def make_fake_waitpid(zombie_pids):
    +        call_count = [0]
    +
    +        def fake_waitpid(pid, flags):
    +            if call_count[0] < len(zombie_pids):
    +                p = zombie_pids[call_count[0]]
    +                call_count[0] += 1
    +                return p, 0
    +            return 0, 0
    +
    +        return fake_waitpid
    +
    +    # 5 ticks, 2 zombies per tick = 10 total
    +    for tick in range(5):
    +        pids = [tick * 100 + 1, tick * 100 + 2]
    +        with patch(
    +            "hermes_cli.kanban_db.os.waitpid", side_effect=make_fake_waitpid(pids)
    +        ):
    +            with patch("hermes_cli.kanban_db._record_worker_exit"):
    +                pids = kb.reap_worker_zombies()
    +        total_reaped += len(pids)
    +
    +    assert total_reaped == 10
    +
    +
    +def test_dispatch_once_still_reaps_via_extracted_fn(kanban_home):
    +    """The reaper inside dispatch_once still works after refactor to reap_worker_zombies()."""
    +    from unittest.mock import patch
    +
    +    call_count = [0]
    +
    +    def fake_waitpid(pid, flags):
    +        call_count[0] += 1
    +        if call_count[0] == 1:
    +            return 99999, 0
    +        return 0, 0
    +
    +    with patch("hermes_cli.kanban_db.os.waitpid", side_effect=fake_waitpid):
    +        with patch("hermes_cli.kanban_db._record_worker_exit"):
    +            with patch("hermes_cli.kanban_db.os.name", "posix"):
    +                pids = kb.reap_worker_zombies()
    +
    +    assert pids == [99999]

  • tests/test_hermes_state.py

    @@ -3021,3 +3021,223 @@ def test_v10_to_v11_upgrade_backfills_tool_fields(self, tmp_path):
             finally:
                 session_db.close()
     
    +
    +# ---------------------------------------------------------------------------
    +# apply_wal_with_fallback — read-only probe tests
    +# ---------------------------------------------------------------------------
    +
    +
    +class TestApplyWalProbe:
    +    """Unit tests for the journal_mode probe in apply_wal_with_fallback."""
    +
    +    def test_skips_set_pragma_when_already_wal(self, tmp_path):
    +        """Already-WAL connection must not trigger the set-pragma."""
    +        import sqlite3
    +        from hermes_state import apply_wal_with_fallback
    +
    +        class _TracingConn(sqlite3.Connection):
    +            def __init__(self, *a, **kw):
    +                super().__init__(*a, **kw)
    +                self.executed = []
    +
    +            def execute(self, sql, params=()):
    +                self.executed.append(sql)
    +                return super().execute(sql, params)
    +
    +        db_path = tmp_path / "wal.db"
    +        # Prime the file into WAL mode first.
    +        with sqlite3.connect(str(db_path)) as seed:
    +            seed.execute("PRAGMA journal_mode=WAL")
    +
    +        conn = _TracingConn(str(db_path))
    +        try:
    +            result = apply_wal_with_fallback(conn)
    +        finally:
    +            conn.close()
    +
    +        assert result == "wal"
    +        # Only the probe should have fired; the set-pragma must NOT appear.
    +        assert any("PRAGMA journal_mode" == sql.strip() for sql in conn.executed), (
    +            "probe PRAGMA should have run"
    +        )
    +        assert not any("journal_mode=WAL" in sql for sql in conn.executed), (
    +            "set-pragma must not run when already in WAL mode"
    +        )
    +
    +    def test_sets_wal_on_fresh_connection(self, tmp_path):
    +        """Probe sees 'delete', then set-pragma runs and returns 'wal'."""
    +        import sqlite3
    +        from hermes_state import apply_wal_with_fallback
    +
    +        class _TracingConn(sqlite3.Connection):
    +            def __init__(self, *a, **kw):
    +                super().__init__(*a, **kw)
    +                self.executed = []
    +
    +            def execute(self, sql, params=()):
    +                self.executed.append(sql)
    +                return super().execute(sql, params)
    +
    +        db_path = tmp_path / "fresh.db"
    +        conn = _TracingConn(str(db_path))
    +        try:
    +            result = apply_wal_with_fallback(conn)
    +        finally:
    +            conn.close()
    +
    +        assert result == "wal"
    +        assert any("journal_mode=WAL" in sql for sql in conn.executed), (
    +            "set-pragma must fire on a fresh (non-WAL) connection"
    +        )
    +
    +    def test_apply_wal_concurrent_connects_no_eio(self, tmp_path):
    +        """20 threads calling connect() on the same DB must not see disk I/O error."""
    +        import sys
    +        import threading
    +        import sqlite3
    +        from hermes_state import apply_wal_with_fallback
    +
    +        db_path = tmp_path / "concurrent.db"
    +        errors = []
    +
    +        def _connect_cycle():
    +            for _ in range(5):
    +                try:
    +                    conn = sqlite3.connect(str(db_path))
    +                    apply_wal_with_fallback(conn)
    +                    conn.close()
    +                except sqlite3.OperationalError as exc:
    +                    if "disk i/o error" in str(exc).lower():
    +                        errors.append(exc)
    +
    +        threads = [threading.Thread(target=_connect_cycle) for _ in range(20)]
    +        for t in threads:
    +            t.start()
    +        for t in threads:
    +            t.join()
    +
    +        assert not errors, f"disk I/O errors from concurrent connects: {errors}"
    +
    +        # Linux-only: no (deleted) WAL/SHM FDs should accumulate.
    +        if sys.platform == "linux":
    +            import os
    +
    +            fd_dir = f"/proc/{os.getpid()}/fd"
    +            deleted_fds = []
    +            for fd_name in os.listdir(fd_dir):
    +                try:
    +                    target = os.readlink(os.path.join(fd_dir, fd_name))
    +                    if "(deleted)" in target and (
    +                        "wal" in target.lower() or "shm" in target.lower()
    +                    ):
    +                        deleted_fds.append(target)
    +                except OSError:
    +                    pass
    +            assert not deleted_fds, f"stale deleted WAL/SHM FDs: {deleted_fds}"
    +
    +    def test_fallback_to_delete_still_works(self, tmp_path):
    +        """When set-pragma raises a WAL-incompat error, falls back to DELETE."""
    +        import sqlite3
    +        from hermes_state import apply_wal_with_fallback
    +
    +        class _IncompatConn(sqlite3.Connection):
    +            def __init__(self, *a, **kw):
    +                super().__init__(*a, **kw)
    +                self._call_count = 0
    +
    +            def execute(self, sql, params=()):
    +                self._call_count += 1
    +                # First call is the read probe; let it return "delete".
    +                # Second call is the set-pragma; raise a WAL-incompat error.
    +                if "journal_mode=WAL" in sql:
    +                    raise sqlite3.OperationalError("locking protocol")
    +                return super().execute(sql, params)
    +
    +        db_path = tmp_path / "incompat.db"
    +        conn = _IncompatConn(str(db_path))
    +        try:
    +            result = apply_wal_with_fallback(conn, db_label="test.db")
    +        finally:
    +            conn.close()
    +
    +        assert result == "delete"
    +
    +    def test_probe_failure_falls_through_to_set_pragma(self, tmp_path):
    +        """When the read probe raises OperationalError, fall through to set-pragma."""
    +        import sqlite3
    +        from hermes_state import apply_wal_with_fallback
    +
    +        class _ProbeFails(sqlite3.Connection):
    +            def __init__(self, *a, **kw):
    +                super().__init__(*a, **kw)
    +                self._first = True
    +
    +            def execute(self, sql, params=()):
    +                if self._first and "journal_mode" in sql and "WAL" not in sql:
    +                    self._first = False
    +                    raise sqlite3.OperationalError("simulated probe failure")
    +                return super().execute(sql, params)
    +
    +        db_path = tmp_path / "probe_fail.db"
    +        conn = _ProbeFails(str(db_path))
    +        try:
    +            result = apply_wal_with_fallback(conn)
    +        finally:
    +            conn.close()
    +
    +        # Despite probe failure, set-pragma must still run and succeed.
    +        assert result == "wal"
    +
    +    def test_no_downgrade_from_wal_to_delete_on_eio(self, tmp_path):
    +        """OperationalError NOT in _WAL_INCOMPAT_MARKERS must propagate, not downgrade."""
    +        import sqlite3
    +        import pytest
    +        from hermes_state import apply_wal_with_fallback
    +
    +        class _EIOConn(sqlite3.Connection):
    +            def __init__(self, *a, **kw):
    +                super().__init__(*a, **kw)
    +                self._first = True
    +
    +            def execute(self, sql, params=()):
    +                # Let the probe succeed (returns "delete" for fresh DB).
    +                if "journal_mode=WAL" in sql:
    +                    raise sqlite3.OperationalError("some unexpected hardware failure")
    +                return super().execute(sql, params)
    +
    +        db_path = tmp_path / "eio.db"
    +        conn = _EIOConn(str(db_path))
    +        try:
    +            with pytest.raises(
    +                sqlite3.OperationalError, match="some unexpected hardware failure"
    +            ):
    +                apply_wal_with_fallback(conn)
    +        finally:
    +            conn.close()
    +
    +    def test_returns_wal_not_delete_from_probe(self, tmp_path):
    +        """Early-return only on 'wal'; 'delete' or 'memory' must fall through to set-pragma."""
    +        import sqlite3
    +        from hermes_state import apply_wal_with_fallback
    +
    +        class _TracingConn(sqlite3.Connection):
    +            def __init__(self, *a, **kw):
    +                super().__init__(*a, **kw)
    +                self.executed = []
    +
    +            def execute(self, sql, params=()):
    +                self.executed.append(sql)
    +                return super().execute(sql, params)
    +
    +        # Fresh DB is in "delete" mode — probe returns "delete", must NOT early-return.
    +        db_path = tmp_path / "delete_mode.db"
    +        conn = _TracingConn(str(db_path))
    +        try:
    +            result = apply_wal_with_fallback(conn)
    +        finally:
    +            conn.close()
    +
    +        assert result == "wal"
    +        assert any("journal_mode=WAL" in sql for sql in conn.executed), (
    +            "set-pragma must fire when probe returns 'delete'"
    +        )

  • tests/test_hermes_state_wal_fallback.py

    @@ -110,15 +110,79 @@ def test_falls_back_on_not_authorized(self, tmp_path):
             assert mode == "delete"
             conn.close()
     
    -    def test_falls_back_on_disk_io_error(self, tmp_path):
    -        """Flaky network FS → disk I/O error → still fall back."""
    +    def test_reraises_on_disk_io_error(self, tmp_path):
    +        """Transient EIO from ``PRAGMA journal_mode=WAL`` must NOT silently
    +        downgrade to DELETE.
    +
    +        Regression for "Bug D": treating transient EIO as a permanent
    +        WAL-incompat marker produced the mixed-journal-mode-across-processes
    +        corruption pattern (process A downgrades to DELETE, sibling
    +        processes successfully set WAL, SQLite corrupts the file because
    +        the two locking protocols are documented as incompatible). EIO is
    +        usually transient (page-cache pressure, lock contention, brief
    +        storage hiccups); the right behavior is to re-raise so the caller
    +        can retry, not to walk the DB into a permanently downgraded state.
    +        """
             conn, _ = _open_blocking(
                 tmp_path / "flaky.db", reason="disk I/O error", isolation_level=None
             )
    -        mode = apply_wal_with_fallback(conn)
    -        assert mode == "delete"
    +        with pytest.raises(sqlite3.OperationalError, match="disk I/O error"):
    +            apply_wal_with_fallback(conn)
             conn.close()
     
    +    def test_does_not_downgrade_when_disk_says_wal(self, tmp_path):
    +        """Refuse to downgrade an already-WAL DB even if the set-pragma path
    +        would have raised a downgrade-eligible marker.
    +
    +        With the WAL-skip patch, the read-only probe short-circuits before
    +        ``PRAGMA journal_mode=WAL`` ever runs on an already-WAL connection,
    +        so the set-pragma path is unreachable here and ``attempts`` stays 0.
    +        Either outcome (skip-via-probe OR re-raise-on-disk-check) preserves
    +        the property this test guards: we never silently DELETE-downgrade
    +        a WAL-mode file. The on-disk guard remains in place as
    +        belt-and-suspenders for any future code path that bypasses the
    +        probe.
    +        """
    +        # Prime the file in WAL mode using a normal connection
    +        primer = sqlite3.connect(
    +            str(tmp_path / "already-wal.db"), isolation_level=None
    +        )
    +        try:
    +            primer.execute("PRAGMA journal_mode=WAL")
    +            primer.execute("CREATE TABLE t (x INTEGER)")
    +            primer.execute("INSERT INTO t VALUES (1)")
    +            assert (
    +                primer.execute("PRAGMA journal_mode").fetchone()[0].lower() == "wal"
    +            )
    +        finally:
    +            primer.close()
    +
    +        # New connection whose set-WAL pragma would raise "locking protocol"
    +        # if it were ever called. With the WAL-skip patch the probe sees
    +        # journal_mode=wal and returns early, so set-WAL is never attempted.
    +        conn, attempts = _open_blocking(
    +            tmp_path / "already-wal.db",
    +            reason="locking protocol",
    +            isolation_level=None,
    +        )
    +        result = apply_wal_with_fallback(conn)
    +        assert result == "wal", (
    +            "must report wal mode (either skipped via probe or refused downgrade)"
    +        )
    +        assert attempts[0] == 0, (
    +            "set-WAL pragma must not run when the on-disk header already says wal"
    +        )
    +        conn.close()
    +
    +        # And the file is STILL WAL on disk — nothing got rewritten
    +        check = sqlite3.connect(str(tmp_path / "already-wal.db"))
    +        try:
    +            assert (
    +                check.execute("PRAGMA journal_mode").fetchone()[0].lower() == "wal"
    +            )
    +        finally:
    +            check.close()
    +
         def test_reraises_unrelated_operational_error(self, tmp_path):
             """Non-WAL-compat errors must NOT be silently swallowed by the fallback."""
             conn, _ = _open_blocking(