fix(state): a busy session store reads as busy, not as damaged or empty

A concurrent WAL checkpoint / reset / frame-flush can surface SQLITE_IOERR
to a reader on a perfectly healthy database: a mode=ro connection cannot
perform the WAL recovery the read needs, because recovery writes the -shm
index and read-only mode refuses. The window is millisecond-scale.

Today that one-shot error escapes the SessionDB read-only constructor, and
GET /api/sessions turns it into a 500 the desktop reads as an authoritative
empty list.

Retry it, bounded, in the constructor so every read-only opener is covered —
the sidebar poll, cross-profile aggregation, recall, browse — rather than at
one route. A persistent IOERR still exhausts the budget and propagates.
Remaining transient failures answer 503, so the client keeps the list it has.

On the write path, BEGIN IMMEDIATE can hit the same transient IOERR before
the callback runs. That one is safe to retry on the same connection because
nothing has been mutated; once the callback starts, settlement is unknown and
the error propagates. Never close()+reopen to heal it — close() cancels this
process's POSIX advisory locks on the file for every sibling connection, and
a list poll's reader must stay disposable so a replaced state.db is observed
and the pre-repair forensic backup stays reachable.

Fixes #100436

Co-authored-by: rkfshakti <rkfshakti@users.noreply.github.com>
Co-authored-by: AKAZIK-py <AKAZIK-py@users.noreply.github.com>

hermes_cli/web_routers/sessions.py

@@ -31,7 +31,7 @@ from hermes_cli.web_models import (
     SessionPrune,
     SessionRename,
 )
-from hermes_state import is_malformed_db_error
+from hermes_state import is_malformed_db_error, is_transient_sqlite_error
 
 # Same logger the handlers used before extraction (identical logger object).
 _log = logging.getLogger("hermes_cli.web_server")
@@ -197,6 +197,22 @@ def get_sessions(
             db.close()
     except HTTPException:
         raise
+    except sqlite3.OperationalError as exc:
+        _log.exception("GET /api/sessions failed")
+        # 503, not 500: the store is busy, not gone. The desktop keeps the
+        # sidebar it already has instead of reading a 500 as an authoritative
+        # empty list. Retrying the OPEN here is deliberately not done — the
+        # bounded retry lives in SessionDB's read-only constructor, so every
+        # read-only opener gets it, not just this route.
+        transient = is_transient_sqlite_error(exc)
+        raise HTTPException(
+            status_code=503 if transient else 500,
+            detail=(
+                "Session store is busy (disk I/O or lock). Retry; the list was not cleared."
+                if transient
+                else "Internal server error"
+            ),
+        ) from exc
     except Exception:
         _log.exception("GET /api/sessions failed")
         raise HTTPException(status_code=500, detail="Internal server error")

hermes_state.py

@@ -374,6 +374,20 @@ DEFAULT_DB_PATH = get_hermes_home() / "state.db"
 # query; short enough that transient fd pressure doesn't strand the read pool.
 _READ_OPEN_RETRY_SECONDS = 60.0
 
+# Transient SQLITE_IOERR retry budget for READ-ONLY opens (#100436). A WAL
+# database being actively written (checkpoint, WAL reset/truncate, frame
+# flush) can surface "disk I/O error" to a concurrent ``mode=ro`` reader in
+# a millisecond-wide transition window: the read-only connection cannot
+# perform the WAL recovery a read through a stale or mid-update -shm file
+# needs, because recovery requires writing the -shm index, which mode=ro
+# refuses. The window closes on its own (the writer finishes the transition),
+# so a bounded number of short retries makes the open succeed instead of
+# 500-ing the whole /api/sessions poll (or any other read-only opener).
+# Deliberately NOT attempted on writable opens: a writer owns the
+# transition, so an IOERR there means a real storage/fd problem.
+_READ_ONLY_IOERR_RETRY_ATTEMPTS = 3
+_READ_ONLY_IOERR_RETRY_BACKOFF_S = 0.05
+
 # Hard ceiling on read-only connections ALIVE at once against one database
 # FILE — pooled idle ones and checked-out ones together, summed over every
 # SessionDB in this process that points at that file. See _PathReadBudget.
@@ -2082,6 +2096,52 @@ def is_malformed_db_error(exc: BaseException) -> bool:
     return any(marker in str(exc).lower() for marker in _MALFORMED_DB_MARKERS)
 
 
+# SQLITE_IOERR, matched as a plain substring so wrapped error strings still
+# classify. Shared by the read-only open retry and the write-path BEGIN retry.
+_DISK_IO_ERROR_MARKER = "disk i/o error"
+
+# Broader set for HTTP classification: a read that failed for one of these
+# reasons found the store BUSY, not gone. Callers map it to 503 (retry, the
+# list was not cleared) instead of 500. Corruption is deliberately absent —
+# a malformed store must surface, not be retried into a timeout.
+_TRANSIENT_SQLITE_MARKERS = (
+    _DISK_IO_ERROR_MARKER,
+    "database is locked",
+    "database table is locked",
+    "busy",
+)
+
+
+def is_transient_sqlite_error(exc: BaseException) -> bool:
+    """True when a SQLite failure means "busy right now", not "damaged".
+
+    One predicate so the read paths cannot drift apart on what counts as
+    recoverable: the read-only open retry, and the HTTP 503-vs-500 split on
+    the session-list endpoints, classify the same way.
+    """
+    if not isinstance(exc, sqlite3.OperationalError):
+        return False
+    message = str(exc).lower()
+    return any(marker in message for marker in _TRANSIENT_SQLITE_MARKERS)
+
+
+def _is_transient_read_only_ioerr(exc: sqlite3.OperationalError, *, attempt: int) -> bool:
+    """True when a read-only open should be retried rather than raised.
+
+    A ``mode=ro`` connection cannot perform WAL recovery (recovery needs to
+    write the -shm index, which read-only mode refuses), so a concurrent WAL
+    checkpoint / reset / frame-flush can surface ``SQLITE_IOERR`` ("disk I/O
+    error") to a reader on an otherwise healthy database (#100436). The
+    transition is millisecond-scale, so a bounded number of short retries
+    clears it without changing classification for genuine storage failures —
+    a persistent IOERR still exhausts the budget and propagates.
+    """
+    return (
+        attempt < _READ_ONLY_IOERR_RETRY_ATTEMPTS
+        and _DISK_IO_ERROR_MARKER in str(exc).lower()
+    )
+
+
 def is_malformed_schema_error(exc: BaseException) -> bool:
     """True only when SQLite explicitly reports malformed schema text.
 
@@ -5123,46 +5183,67 @@ class SessionDB(SessionSearchMixin, SessionSchemaMixin, SessionPortabilityMixin)
                 # must already exist + be initialised (callers guard on
                 # db_path.exists()); a SELECT against an empty file raises and
                 # the caller degrades per-profile.
-                self._conn = _connect_tracked_db(
-                    f"file:{self.db_path}?mode=ro",
-                    tracking_path=self.db_path,
-                    uri=True,
-                    check_same_thread=False,
-                    timeout=1.0,
-                    isolation_level=None,
-                )
-                self._conn.row_factory = sqlite3.Row
-                # FTS capability flags normally come from writable schema
-                # initialisation. Probe existing virtual tables with SELECTs
-                # only so read-only search keeps its FTS and trigram paths.
-                # Close the connection on ANY probe failure (e.g. malformed
-                # schema raises DatabaseError, not the OperationalError the
-                # probe handles). The constructor's outer finally also covers
-                # failures before this probe and BaseException paths, so a
-                # leaked tracked connection cannot block _backup_db_file's
-                # raw-copy for the rest of the process — the writable heal
-                # that follows would then repair WITHOUT its forensic backup.
-                try:
-                    apply_database_pragmas(self._conn, db_label="state.db")
-                    cursor = self._conn.cursor()
-                    self._fts_enabled = (
-                        self._fts_table_probe(cursor, "messages_fts") is True
-                    )
-                    if self._fts_enabled:
-                        self._trigram_available = (
-                            self._fts_table_probe(
-                                cursor,
-                                "messages_fts_trigram",
-                            )
-                            is True
-                        )
-                except BaseException:
-                    conn, self._conn = self._conn, None
+                open_attempt = 0
+                while True:
                     try:
-                        conn.close()
-                    except Exception:
-                        pass
-                    raise
+                        self._conn = _connect_tracked_db(
+                            f"file:{self.db_path}?mode=ro",
+                            tracking_path=self.db_path,
+                            uri=True,
+                            check_same_thread=False,
+                            timeout=1.0,
+                            isolation_level=None,
+                        )
+                        self._conn.row_factory = sqlite3.Row
+                        # FTS capability flags normally come from writable schema
+                        # initialisation. Probe existing virtual tables with
+                        # SELECTs only so read-only search keeps its FTS and
+                        # trigram paths. Close the connection on ANY probe
+                        # failure (e.g. malformed schema raises DatabaseError,
+                        # not the OperationalError the probe handles). The
+                        # constructor's outer finally also covers failures
+                        # before this probe and BaseException paths, so a
+                        # leaked tracked connection cannot block
+                        # _backup_db_file's raw-copy for the rest of the
+                        # process — the writable heal that follows would then
+                        # repair WITHOUT its forensic backup.
+                        try:
+                            apply_database_pragmas(self._conn, db_label="state.db")
+                            cursor = self._conn.cursor()
+                            self._fts_enabled = (
+                                self._fts_table_probe(cursor, "messages_fts")
+                                is True
+                            )
+                            if self._fts_enabled:
+                                self._trigram_available = (
+                                    self._fts_table_probe(
+                                        cursor,
+                                        "messages_fts_trigram",
+                                    )
+                                    is True
+                                )
+                        except BaseException:
+                            conn, self._conn = self._conn, None
+                            try:
+                                conn.close()
+                            except Exception:
+                                pass
+                            raise
+                        break
+                    except sqlite3.OperationalError as ioerr:
+                        # A WAL checkpoint / reset / frame-flush in flight on
+                        # the writer side can surface SQLITE_IOERR to a
+                        # concurrent mode=ro reader (it cannot perform the
+                        # recovery the read needs — recovery writes the -shm
+                        # index, which mode=ro refuses). The transition closes
+                        # in milliseconds, so retry a bounded number of times
+                        # before classifying the store as failed (#100436).
+                        if not _is_transient_read_only_ioerr(
+                            ioerr, attempt=open_attempt
+                        ):
+                            raise
+                        open_attempt += 1
+                        time.sleep(_READ_ONLY_IOERR_RETRY_BACKOFF_S)
                 self._record_db_file_identity()
                 initialization_complete = True
                 return
@@ -5930,6 +6011,13 @@ class SessionDB(SessionSearchMixin, SessionSchemaMixin, SessionPortabilityMixin)
         # Set on the first compression-busy collision so the short wait is
         # measured from then, not from the start of the write.
         compression_deadline: Optional[float] = None
+        # One retry for SQLITE_IOERR raised by BEGIN IMMEDIATE itself. The
+        # callback has not run at that point, so there is no durable effect
+        # to replay and the retry is exactly-once safe (#99502's contract).
+        # Once the callback starts, an IOERR leaves the write's settlement
+        # unknown and must propagate — this helper owns non-idempotent
+        # transcript/counter mutations, not just idempotent UPSERTs.
+        ioerr_begin_retried = False
 
         # Transient engine-level error observed on contended WAL appends
         # (dual gateway/agent writers; FTS5 trigram sync holds the write
@@ -5943,6 +6031,7 @@ class SessionDB(SessionSearchMixin, SessionSchemaMixin, SessionPortabilityMixin)
 
         while True:
             self._raise_if_db_replaced()
+            fn_started = False
             try:
                 with self._lock:
                     if self._conn is None:
@@ -5951,6 +6040,7 @@ class SessionDB(SessionSearchMixin, SessionSchemaMixin, SessionPortabilityMixin)
                         self._reopen_after_close_locked(context="write")
                     self._conn.execute("BEGIN IMMEDIATE")
                     try:
+                        fn_started = True
                         result = fn(self._conn)
                         self._conn.commit()
                     except BaseException:
@@ -6003,7 +6093,21 @@ class SessionDB(SessionSearchMixin, SessionSchemaMixin, SessionPortabilityMixin)
                     ) from exc
                 if _is_no_more_rows(exc) and self._sleep_before_write_retry(deadline, patience_s):
                     continue
-                # Non-lock error or patience exhausted — propagate.
+                if (
+                    _DISK_IO_ERROR_MARKER in err_msg
+                    and not fn_started
+                    and not ioerr_begin_retried
+                    and self._sleep_before_write_retry(deadline, patience_s)
+                ):
+                    # BEGIN IMMEDIATE itself hit a transient WAL-transition
+                    # IOERR. Nothing has been mutated, so retrying on the SAME
+                    # connection replays nothing. Never close()+reopen to
+                    # "heal" it: close() cancels this process's POSIX locks on
+                    # the file for every sibling connection (howtocorrupt §2.2).
+                    ioerr_begin_retried = True
+                    continue
+                # Non-lock error, the callback already ran (settlement is
+                # unknown — do not replay), or patience exhausted.
                 raise
             except sqlite3.DatabaseError as exc:
                 if _is_no_more_rows(exc) and self._sleep_before_write_retry(deadline, patience_s):

tests/hermes_cli/test_session_list_reader_disposable.py

@@ -0,0 +1,96 @@
+"""Read-only session-list opens must stay disposable.
+
+Two properties that a "keep the read-only handle for the process lifetime"
+optimisation silently destroys. Both are asserted against the real
+``_open_session_db_at_path`` read path the sidebar poll uses, because both
+failures are invisible in a unit test that mocks the store.
+
+1. **The store on disk is the truth.** Recovering a corrupt ``state.db``
+   is a file swap (``mv state.db state.db.corrupt-…; cp -a recovered.db
+   state.db``) performed while the backend is stopped, but a poll can also
+   race a restore. A reader pinned to the old inode keeps serving
+   pre-recovery rows forever, so the user "recovers" and still sees the
+   broken list.
+
+2. **Forensic backup must stay reachable.** ``offline_file_access`` refuses
+   raw byte access while ANY tracked connection is registered for the path,
+   because a raw ``close()`` would cancel this process's POSIX advisory locks
+   (howtocorrupt §2.2). ``_backup_db_file`` (the copy taken BEFORE a malformed
+   store is repaired) and ``_db_fingerprint`` (the repair-attempt ledger key)
+   both go through it. A never-closed list reader makes both fail for the rest
+   of the process, so a repair runs without its forensic backup and the
+   ledger degrades to a size-only key.
+"""
+
+from __future__ import annotations
+
+import shutil
+
+from hermes_cli.sqlite_safe_read import LiveConnectionError, offline_file_access
+from hermes_cli.web_server import _open_session_db_at_path
+from hermes_state import SessionDB, _db_fingerprint
+
+
+def _ids(db) -> list:
+    return [row["id"] for row in db.list_sessions_rich(limit=10, compact_rows=True)]
+
+
+def test_poll_observes_a_replaced_state_db(tmp_path):
+    db_path = tmp_path / "state.db"
+    old = SessionDB(db_path=db_path)
+    old.create_session("before-recovery", source="cli")
+    old.close()
+
+    first = _open_session_db_at_path(db_path, read_only=True)
+    try:
+        assert _ids(first) == ["before-recovery"]
+    finally:
+        first.close()
+
+    # `hermes sessions recover` writes a clean database, which the operator
+    # then installs over the corrupt one.
+    recovered = tmp_path / "recovered-state.db"
+    rebuilt = SessionDB(db_path=recovered)
+    rebuilt.create_session("after-recovery", source="cli")
+    rebuilt.close()
+
+    for suffix in ("-wal", "-shm"):
+        sidecar = db_path.with_name(db_path.name + suffix)
+        if sidecar.exists():
+            sidecar.unlink()
+    db_path.unlink()
+    shutil.copy2(recovered, db_path)
+
+    second = _open_session_db_at_path(db_path, read_only=True)
+    try:
+        assert _ids(second) == ["after-recovery"]
+    finally:
+        second.close()
+
+
+def test_poll_leaves_forensic_backup_reachable(tmp_path):
+    db_path = tmp_path / "state.db"
+    writer = SessionDB(db_path=db_path)
+    writer.create_session("s1", source="cli")
+    writer.close()
+
+    baseline = _db_fingerprint(db_path)
+    assert baseline is not None
+
+    poll = _open_session_db_at_path(db_path, read_only=True)
+    try:
+        assert _ids(poll) == ["s1"]
+    finally:
+        poll.close()
+
+    # The raw-copy path a malformed-store repair takes before it touches
+    # anything must still be permitted after the poll.
+    try:
+        with offline_file_access(db_path, what="forensic-backup"):
+            pass
+    except LiveConnectionError as exc:  # pragma: no cover - failure detail
+        raise AssertionError(
+            f"a session-list poll left a tracked connection open: {exc}"
+        ) from exc
+
+    assert _db_fingerprint(db_path) == baseline

tests/hermes_cli/test_web_server.py

@@ -319,6 +319,29 @@ class TestWebServerEndpoints:
                 monitor.close()
             writer.close()
 
+    def test_get_sessions_transient_ioerr_is_503(self, monkeypatch):
+        """Busy store, not a gone store: the desktop keeps the list it has."""
+        import sqlite3
+
+        from hermes_cli import web_server
+
+        def boom(*_args, **_kwargs):
+            raise sqlite3.OperationalError("disk I/O error")
+
+        monkeypatch.setattr(web_server, "_open_session_db_for_profile", boom)
+        assert self.client.get("/api/sessions?limit=1&offset=0").status_code == 503
+
+    def test_get_sessions_non_transient_operational_error_is_500(self, monkeypatch):
+        import sqlite3
+
+        from hermes_cli import web_server
+
+        def boom(*_args, **_kwargs):
+            raise sqlite3.OperationalError("no such table: sessions")
+
+        monkeypatch.setattr(web_server, "_open_session_db_for_profile", boom)
+        assert self.client.get("/api/sessions?limit=1&offset=0").status_code == 500
+
     def test_get_status_loads_gateway_config_off_event_loop(self, monkeypatch):
         """Cold gateway config loading must not block the WebSocket loop.
 

tests/test_hermes_state.py

@@ -281,6 +281,83 @@ class TestConnectionLifecycle:
         healed.close()
         assert list(tmp_path.glob("*malformed-backup*"))
 
+    def test_read_only_open_retries_transient_wal_ioerr(self, tmp_path, monkeypatch):
+        """A transient SQLITE_IOERR on a read-only open must retry, not raise.
+
+        A ``mode=ro`` connection cannot perform WAL recovery (recovery would
+        need to write the -shm index, which read-only mode refuses), so a
+        concurrent checkpoint / WAL reset / frame-flush on the writer side can
+        surface "disk I/O error" to a reader on a perfectly healthy database
+        (#100436). The transition window is millisecond-scale; a bounded retry
+        must let the open succeed instead of 500-ing the /api/sessions poll
+        and every other read-only opener.
+        """
+        import sqlite3
+
+        from hermes_cli.sqlite_safe_read import has_live_connection
+
+        db_path = tmp_path / "state.db"
+        writable = SessionDB(db_path=db_path)
+        writable.create_session("wal-race", source="cli")
+        writable.close()
+
+        real_connect = hermes_state._connect_tracked_db
+        attempts = []
+
+        def flaky_connect(*args, **kwargs):
+            attempts.append(kwargs.get("uri"))
+            if len(attempts) == 1:
+                # First open lands inside the writer's WAL transition window.
+                raise sqlite3.OperationalError("disk I/O error")
+            return real_connect(*args, **kwargs)
+
+        monkeypatch.setattr(hermes_state, "_connect_tracked_db", flaky_connect)
+        # Keep the test fast: one backoff tick is enough; the retry budget
+        # itself is exercised by the attempt count below.
+        monkeypatch.setattr(hermes_state, "_READ_ONLY_IOERR_RETRY_BACKOFF_S", 0.0)
+
+        read_only = SessionDB(db_path=db_path, read_only=True)
+        try:
+            assert read_only._fts_enabled is True
+            matches = read_only.search_messages("wal-race")
+        finally:
+            read_only.close()
+
+        assert len(attempts) >= 2, "the transient IOERR must be retried"
+        assert has_live_connection(db_path) is False  # no leaked connections
+
+    def test_read_only_open_exhausts_retry_budget_for_persistent_ioerr(
+        self, tmp_path, monkeypatch
+    ):
+        """A persistent SQLITE_IOERR must exhaust the budget and raise.
+
+        The retry exists to ride out a millisecond WAL transition — a
+        storage layer that keeps failing after the full budget is genuinely
+        broken and must surface the error (and not loop forever).
+        """
+        import sqlite3
+
+        db_path = tmp_path / "state.db"
+        writable = SessionDB(db_path=db_path)
+        writable.create_session("broken-disk", source="cli")
+        writable.close()
+
+        attempts = []
+
+        def bad_connect(*args, **kwargs):
+            attempts.append(1)
+            raise sqlite3.OperationalError("disk I/O error")
+
+        monkeypatch.setattr(hermes_state, "_connect_tracked_db", bad_connect)
+        monkeypatch.setattr(hermes_state, "_READ_ONLY_IOERR_RETRY_BACKOFF_S", 0.0)
+        budget = hermes_state._READ_ONLY_IOERR_RETRY_ATTEMPTS
+
+        with pytest.raises(sqlite3.OperationalError, match="disk I/O error"):
+            SessionDB(db_path=db_path, read_only=True)
+
+        # budget + 1 = the initial attempt plus `budget` retries.
+        assert len(attempts) == budget + 1
+
 
 # =========================================================================
 # Session lifecycle

tests/test_state_db_notadb_fail_closed.py

@@ -1,9 +1,10 @@
 """Tests for fail-closed state.db NOTADB handling and journal-mode EIO retries.
 
-Covers the two independently-valuable pieces salvaged from the state.db
-hardening rollup:
+Covers:
 
 * fail closed when a live write connection reports ``file is not a database``;
+* the write-path SQLITE_IOERR retry boundary: admitted only when the callback
+  has provably not run, never by closing and replaying;
 * transient ``disk i/o error`` retry in ``_on_disk_journal_mode`` so a
   one-shot EIO doesn't push callers onto the fail-closed unknown-mode branch.
 """
@@ -49,6 +50,89 @@ class TestFailClosedAfterNotADb:
             db.close()
 
 
+class TestWriteIoerrRetryBoundary:
+    """IOERR retry is admitted by EFFECT POSITION, not error spelling.
+
+    ``_execute_write`` owns non-idempotent transcript/counter mutations, so
+    replaying its callback is only safe when the first attempt provably did
+    nothing. SQLite does not define ``SQLITE_IOERR`` as pre-effect-only (an
+    IOERR at fsync/commit may or may not have landed), so the admission gate
+    is "did the callback start", not "does the message say disk I/O".
+    """
+
+    def test_ioerr_on_begin_retries_because_the_callback_never_ran(self, tmp_path):
+        db = SessionDB(db_path=tmp_path / "state.db")
+        real_conn = db._conn
+        try:
+
+            class _BeginIoerrOnce:
+                def __init__(self, conn):
+                    self._real = conn
+                    self.begins = 0
+
+                def execute(self, sql, *args, **kwargs):
+                    if str(sql).strip().upper().startswith("BEGIN") and self.begins == 0:
+                        self.begins += 1
+                        raise sqlite3.OperationalError("disk I/O error")
+                    return self._real.execute(sql, *args, **kwargs)
+
+                def __getattr__(self, name):
+                    return getattr(self._real, name)
+
+            proxy = _BeginIoerrOnce(real_conn)
+            db._conn = proxy
+            db.create_session(session_id="s1", source="cli", model="test")
+            assert proxy.begins == 1
+        finally:
+            db._conn = real_conn
+
+        rows = db.list_sessions_rich(limit=10, compact_rows=True)
+        assert [row["id"] for row in rows] == ["s1"]
+        db.close()
+
+    def test_ioerr_after_the_callback_mutates_does_not_replay(self, tmp_path):
+        """Settlement is unknown once the callback has run — surface, don't rerun."""
+        db = SessionDB(db_path=tmp_path / "state.db")
+        try:
+            calls = []
+
+            def mutate_then_fail(conn):
+                calls.append(1)
+                conn.execute(
+                    "INSERT INTO sessions (id, started_at, source) VALUES (?, ?, ?)",
+                    (f"row-{len(calls)}", 1.0, "cli"),
+                )
+                raise sqlite3.OperationalError("disk I/O error")
+
+            with pytest.raises(sqlite3.OperationalError, match="disk I/O error"):
+                db._execute_write(mutate_then_fail)
+
+            assert calls == [1], "a started write must not be replayed"
+            assert db.list_sessions_rich(limit=10, compact_rows=True) == []
+        finally:
+            db.close()
+
+    def test_write_ioerr_never_closes_the_connection(self, tmp_path, monkeypatch):
+        """close() cancels this process's POSIX locks for every sibling fd."""
+        db = SessionDB(db_path=tmp_path / "state.db")
+        try:
+            closed = []
+            monkeypatch.setattr(
+                type(db._conn), "close", lambda self: closed.append(1), raising=False
+            )
+
+            def always_ioerr(conn):
+                raise sqlite3.OperationalError("disk I/O error")
+
+            with pytest.raises(sqlite3.OperationalError):
+                db._execute_write(always_ioerr)
+
+            assert closed == []
+            assert db._conn is not None
+        finally:
+            db.close()
+
+
 class TestOnDiskJournalModeEioRetry:
     def _conn_raising_then(self, failures, result_rows):
         conn = MagicMock()