fix(state): close the SessionDB lock gate's blind spot on its own mixin files

test_no_locked_readers_gate.py (#97676) parses hermes_state.py's SessionDB
class body with ast and flags any method that holds the writer lock
around a pure-read query — Pattern C, where every concurrent turn's
persistence convoys behind an unrelated read. #97676 converted 39 such
methods and closed detection blind spots for alias/variable-SQL readers.

But SessionDB is declared as
`class SessionDB(SessionSearchMixin, SessionSchemaMixin, SessionPortabilityMixin)`,
and the gate only ever opened hermes_state.py — it never parsed the three
mixin files those base classes are defined in, so a locked reader
declared there was structurally invisible to the scanner regardless of
how good the alias/variable-SQL detection got.

Applying the gate's exact scanning logic to the three mixin files
directly turns up 9 genuine pure-read methods still holding the writer
lock, none in #97676's converted list:

- hermes_state_search.py: _fts_teardown_trash_step, fts_optimize_available,
  optimize_fts_storage, list_recent_user_messages
- hermes_state_portability.py: distinct_session_cwds, list_cron_job_runs,
  _get_session_rich_rows_batch, list_skill_scaffolded_sessions,
  get_first_assistant_text

_get_session_rich_rows_batch is a hot path: it backs list_sessions_rich's
compression-tip resolution and the web server's session-search hydration
across every gateway install — its own docstring already claims "same
read-your-writes guarantee as list_sessions_rich", but list_sessions_rich
was already using _read_ctx() (its guarantee comes from flush_token_counts()
before the read, not from holding the writer lock) while this method's
implementation never caught up to match.

Converted all 9 to `with self._read_ctx() as conn:`, the exact pattern
#97676 used, verified each is a genuine pure read with no hidden writes
by tracing every helper call it makes.

Extended the gate itself (_ALL_STATE_SOURCES) to scan all three mixin
files under their own class names, plus hermes_state.py, so this blind
spot can't silently reopen. Added a regression test
(test_scan_all_state_sources_visits_every_mixin_file) that plants a
synthetic violation in a mixin-shaped file and asserts the scanner still
catches it — a change that reverts the file list back to one file passes
the existing sabotage test but fails this one.

Mutation-verified: with the gate's new scope but the old (unconverted)
mixin sources, test_no_locked_pure_readers fails and names all 9 real
violations with correct file/line. Restored the fix; it passes clean.

hermes_state_portability.py

@@ -54,8 +54,8 @@ class SessionPortabilityMixin:
         where = "cwd IS NOT NULL AND TRIM(cwd) != ''"
         if not include_archived:
             where += " AND archived = 0"
-        with self._lock:
-            rows = self._conn.execute(
+        with self._read_ctx() as conn:
+            rows = conn.execute(
                 "SELECT cwd AS cwd, COUNT(*) AS sessions, "
                 "MAX(COALESCE(ended_at, started_at, 0)) AS last_active "
                 f"FROM sessions WHERE {where} GROUP BY cwd"
@@ -119,8 +119,8 @@ class SessionPortabilityMixin:
             ORDER BY s.started_at DESC, s.id DESC
             LIMIT ? OFFSET ?
         """
-        with self._lock:
-            cursor = self._conn.execute(query, (prefix, prefix_hi, limit, offset))
+        with self._read_ctx() as conn:
+            cursor = conn.execute(query, (prefix, prefix_hi, limit, offset))
             rows = cursor.fetchall()
 
         runs: List[Dict[str, Any]] = []
@@ -202,8 +202,8 @@ class SessionPortabilityMixin:
             {prompt_join}
             WHERE s.id IN ({placeholders})
         """
-        with self._lock:
-            cursor = self._conn.execute(query, ids)
+        with self._read_ctx() as conn:
+            cursor = conn.execute(query, ids)
             rows = cursor.fetchall()
         result: Dict[str, Dict[str, Any]] = {}
         for row in rows:
@@ -229,8 +229,8 @@ class SessionPortabilityMixin:
         Returns ``id``, ``title``, and the full first-turn ``content`` so a
         caller can re-derive what the user typed. Newest first.
         """
-        with self._lock:
-            rows = self._conn.execute(
+        with self._read_ctx() as conn:
+            rows = conn.execute(
                 """
                 SELECT s.id, s.title, m.content
                 FROM sessions s
@@ -254,8 +254,8 @@ class SessionPortabilityMixin:
         Pairs with :meth:`list_skill_scaffolded_sessions` so a re-title can feed
         the titler the same (request, reply) shape the live path uses.
         """
-        with self._lock:
-            row = self._conn.execute(
+        with self._read_ctx() as conn:
+            row = conn.execute(
                 "SELECT content FROM messages "
                 "WHERE session_id = ? AND role = 'assistant' AND content IS NOT NULL "
                 "ORDER BY timestamp, id LIMIT 1",

hermes_state_search.py

@@ -191,9 +191,9 @@ class SessionSearchMixin:
         comparison, so they keep the legacy chunked ``LIMIT`` delete —
         those shadow tables are small by construction.
         """
-        with self._lock:
+        with self._read_ctx() as conn:
             trash = [
-                r[0] for r in self._conn.execute(
+                r[0] for r in conn.execute(
                     "SELECT name FROM sqlite_master WHERE type = 'table' "
                     "AND name LIKE ? ESCAPE '\\'",
                     (self._FTS_TRASH_PREFIX.replace("_", "\\_") + "%",),
@@ -651,15 +651,15 @@ class SessionSearchMixin:
         unavailable)."""
         if not self._fts_enabled or self.read_only:
             return False
-        with self._lock:
-            if self._db_has_legacy_inline_fts(self._conn):
+        with self._read_ctx() as conn:
+            if self._db_has_legacy_inline_fts(conn):
                 return True
             # Interrupted optimize: demotion already removed the legacy
             # vtables (so the check above is False), but the transition is
             # unfinished until the backfill markers are cleared and the
             # demoted trash tables are torn down. Search stays complete
             # through the gap supplement meanwhile; re-running resumes.
-            if self._conn.execute(
+            if conn.execute(
                 "SELECT 1 FROM state_meta "
                 "WHERE key = 'fts_rebuild_high_water' LIMIT 1"
             ).fetchone():
@@ -667,17 +667,17 @@ class SessionSearchMixin:
             # CJK-bigram index work — only offerable when THIS process can
             # tokenize: a pending backfill (markers set at creation on a
             # populated DB) or a stale index awaiting a from-scratch rebuild.
-            if self._fts_cjk_loaded and self._conn.execute(
+            if self._fts_cjk_loaded and conn.execute(
                 "SELECT 1 FROM state_meta WHERE key IN "
                 f"('fts_cjk_rebuild_high_water', '{FTS_CJK_STALE_KEY}') LIMIT 1"
             ).fetchone():
                 return True
-            if self._has_fts_trash(self._conn):
+            if self._has_fts_trash(conn):
                 return True
             # Pre-fix crash window: empty external-content index with
             # messages still present, no markers, no trash (teardown already
             # finished or never needed). Re-run seeds markers and backfills.
-            return self._fts_external_index_empty_with_messages(self._conn)
+            return self._fts_external_index_empty_with_messages(conn)
 
     def _demote_legacy_fts_to_trash(self) -> int:
         """Demote the legacy inline FTS vtables and stage their shadow tables
@@ -876,13 +876,13 @@ class SessionSearchMixin:
         # still empty against a non-empty messages table. Pre-fix code could
         # tear down trash and settle after a no-op backfill when markers were
         # missing — permanent search-index loss for historical rows.
-        with self._lock:
-            still_pending = self._conn.execute(
+        with self._read_ctx() as conn:
+            still_pending = conn.execute(
                 "SELECT 1 FROM state_meta "
                 "WHERE key = 'fts_rebuild_high_water' LIMIT 1"
             ).fetchone() is not None
-            still_trash = self._has_fts_trash(self._conn)
-            empty_index = self._fts_external_index_empty_with_messages(self._conn)
+            still_trash = self._has_fts_trash(conn)
+            empty_index = self._fts_external_index_empty_with_messages(conn)
         if still_pending or still_trash or empty_index:
             reason = (
                 "backfill_incomplete" if still_pending or empty_index
@@ -1131,8 +1131,8 @@ class SessionSearchMixin:
         # excludes handoffs with a DB pick that includes them, soft-deleting
         # the wrong turn.
         fetch_limit = int(limit) * 2 + 5
-        with self._lock:
-            cursor = self._conn.execute(
+        with self._read_ctx() as conn:
+            cursor = conn.execute(
                 "SELECT id, timestamp, content FROM messages "
                 "WHERE session_id = ? AND role = 'user'"
                 f"{active_clause}{display_clause} "

tests/state/test_no_locked_readers_gate.py

@@ -17,6 +17,13 @@ convoying on the writer lock. Methods that write under the lock are the
 lock's legitimate users and pass. New violations fail with the method
 name and the fix (route through ``_read_ctx()``).
 
+``SessionDB`` itself is declared in ``hermes_state.py`` as
+``class SessionDB(SessionSearchMixin, SessionSchemaMixin,
+SessionPortabilityMixin)`` — its actual methods live across four files.
+A gate that only opens ``hermes_state.py`` never sees a locked reader
+declared in one of the three mixin files, so ``_ALL_STATE_SOURCES`` scans
+each of them under their own class name.
+
 Deliberately NOT flagged:
 - methods that INSERT/UPDATE/DELETE/REPLACE under the lock (writers);
 - read-modify-write methods (the read is ordered against its own write);
@@ -32,7 +39,18 @@ from pathlib import Path
 
 import pytest
 
-_STATE_PY = Path(__file__).resolve().parents[2] / "hermes_state.py"
+_REPO_ROOT = Path(__file__).resolve().parents[2]
+_STATE_PY = _REPO_ROOT / "hermes_state.py"
+
+# SessionDB's own class body lives in hermes_state.py; the rest of its
+# methods come from these mixins (see module docstring). Each entry is
+# (source file, class name to scan in that file).
+_ALL_STATE_SOURCES: list[tuple[Path, str]] = [
+    (_STATE_PY, "SessionDB"),
+    (_REPO_ROOT / "hermes_state_search.py", "SessionSearchMixin"),
+    (_REPO_ROOT / "hermes_state_schema.py", "SessionSchemaMixin"),
+    (_REPO_ROOT / "hermes_state_portability.py", "SessionPortabilityMixin"),
+]
 
 _WRITE_RE = re.compile(
     r"^\s*(INSERT|UPDATE|DELETE|REPLACE|CREATE|DROP|ALTER|VACUUM|BEGIN|COMMIT|ANALYZE)\b",
@@ -124,17 +142,19 @@ def _is_self_lock_with(item: ast.withitem) -> bool:
     )
 
 
-def _scan_locked_readers(state_py: "Path | None" = None) -> list[str]:
+def _scan_locked_readers(
+    state_py: "Path | None" = None, class_name: str = "SessionDB"
+) -> list[str]:
     target = state_py if state_py is not None else _STATE_PY
     tree = ast.parse(target.read_text(encoding="utf-8"))
     violations: list[str] = []
 
     session_db = None
     for node in tree.body:
-        if isinstance(node, ast.ClassDef) and node.name == "SessionDB":
+        if isinstance(node, ast.ClassDef) and node.name == class_name:
             session_db = node
             break
-    assert session_db is not None, "SessionDB class not found"
+    assert session_db is not None, f"{class_name} class not found in {target}"
 
     for method in session_db.body:
         if not isinstance(method, (ast.FunctionDef, ast.AsyncFunctionDef)):
@@ -197,9 +217,22 @@ def _scan_locked_readers(state_py: "Path | None" = None) -> list[str]:
     return violations
 
 
+def _scan_all_state_sources() -> list[str]:
+    """Run ``_scan_locked_readers`` over every file that contributes methods
+    to ``SessionDB`` — the class body in ``hermes_state.py`` plus each mixin
+    it inherits from (see module docstring). Violations are prefixed with
+    their source filename since methods can share names across mixins.
+    """
+    violations: list[str] = []
+    for path, class_name in _ALL_STATE_SOURCES:
+        for v in _scan_locked_readers(path, class_name):
+            violations.append(f"{path.name}: {v}")
+    return violations
+
+
 class TestNoPureReadersUnderWriterLock:
     def test_no_locked_pure_readers(self):
-        violations = _scan_locked_readers()
+        violations = _scan_all_state_sources()
         assert violations == [], (
             "Pure-read SessionDB methods holding the writer lock "
             "(Pattern C — every concurrent turn's persistence convoys "
@@ -235,3 +268,24 @@ class TestNoPureReadersUnderWriterLock:
         assert flagged == {
             "guilty_reader", "guilty_alias_reader", "guilty_variable_sql"
         }, violations
+
+    def test_scan_all_state_sources_visits_every_mixin_file(self, tmp_path):
+        """Sabotage self-check for the multi-file scope itself: a locked
+        reader planted in a MIXIN file (not hermes_state.py) must still be
+        caught. Guards against the gate's scope silently narrowing back to
+        one file — exactly how the real 2026-08 gap (9 locked readers across
+        three mixin files, invisible to the single-file scanner) happened.
+        """
+        mixin_sabotage = (
+            "class FakeMixin:\n"
+            "    def guilty_mixin_reader(self):\n"
+            "        with self._lock:\n"
+            "            return self._conn.execute(\"SELECT 1\").fetchone()\n"
+        )
+        p = tmp_path / "fake_mixin.py"
+        p.write_text(mixin_sabotage, encoding="utf-8")
+
+        violations = [
+            f"{p.name}: {v}" for v in _scan_locked_readers(p, "FakeMixin")
+        ]
+        assert any("guilty_mixin_reader" in v for v in violations), violations