diff --git a/hermes_state.py b/hermes_state.py
index 4f7b6453461a..b0daccd50854 100644
--- a/hermes_state.py
+++ b/hermes_state.py
@@ -2083,10 +2083,13 @@ def _get_read_conn(self) -> Optional[sqlite3.Connection]:
         if getattr(self._read_local, "failed", False):
             return None
         try:
+            # close() drains worker-created connections from its caller, so
+            # readers must permit cross-thread close after the worker exits.
             conn = _connect_tracked_db(
                 f"file:{self.db_path}?mode=ro",
                 tracking_path=self.db_path,
                 uri=True,
+                check_same_thread=False,
                 timeout=5.0,
                 isolation_level=None,
             )
@@ -2651,6 +2654,11 @@ def close(self):
         for conn in read_conns:
             try:
                 conn.close()
+            except sqlite3.ProgrammingError as exc:
+                logger.warning(
+                    "SessionDB.close() could not close a read connection: %s",
+                    exc,
+                )
             except Exception:
                 pass
         self._read_local.conn = None
diff --git a/tests/test_session_db_read_path_split.py b/tests/test_session_db_read_path_split.py
index 35c54228301a..87065e780944 100644
--- a/tests/test_session_db_read_path_split.py
+++ b/tests/test_session_db_read_path_split.py
@@ -8,6 +8,9 @@
 back to the legacy locked path when WAL or the read connection is missing.
 """
 
+from concurrent.futures import ThreadPoolExecutor
+from contextlib import suppress
+import sqlite3
 import threading
 
 import pytest
@@ -129,3 +132,22 @@ def reader():
         assert done["view"]["window"]
     finally:
         db._lock.release()
+
+
+@pytest.mark.requires_wal
+def test_close_drains_read_conn_created_by_finished_worker(tmp_path):
+    """close() must reclaim a read connection created by another thread."""
+    d = SessionDB(db_path=tmp_path / "state.db")
+    try:
+        with ThreadPoolExecutor(max_workers=1) as pool:
+            conn = pool.submit(d._get_read_conn).result(timeout=5.0)
+        assert conn is not None
+
+        d.close()
+
+        assert not d._read_conns
+        with pytest.raises(sqlite3.ProgrammingError, match="closed database"):
+            conn.execute("SELECT 1")
+    finally:
+        with suppress(Exception):
+            d.close()