fix(state): close gate blind spots — alias + variable-SQL readers (simplify findings)

Reviewer findings from the formal /simplify-code pass, all verified:

1. Scanner blind spots (HIGH): five more locked pure readers were
   invisible to the v1 gate — get_compression_fallback_streak and
   get_compression_ineffective_count hid behind `conn = self._conn`
   aliasing; list_gateway_sessions, find_session_by_origin and
   search_sessions hid behind SQL held in variables/f-strings (the
   scanner required unknown == 0 to flag). All five converted to
   _read_ctx(); the scanner now (a) tracks self._conn aliases and
   (b) flags lock blocks with NO proven write instead of silently
   skipping unprovable SQL. Sabotage self-check extended to pin all
   three detection classes (literal, alias, variable-SQL) plus a
   mixed variable-SQL writer that must NOT fire.

2. get_meta reverted to the writer lock: its inline comment (present
   on main) documents a real read-your-writes dependency —
   fts_rebuild_step reads rebuild progress that a pooled WAL reader
   cannot see mid-transaction. The blanket conversion had overridden
   a documented design decision; it is now the single justified
   _ALLOWED_LOCKED_READERS entry, replacing the dead
   _enter_fts_fail_open entry (whose lock block counts 3 writes and
   never needed allowlisting).

Strengthened scanner on pre-conversion main: 43 violations
(39 pure-read + 4 no-proven-write). This branch: zero.

Suites: gate 2/2; tests/test_hermes_state.py + tests/state/ 331
passed (same 3 pre-existing FTS-rebuild reds as clean main); 433
passed across the 12 consumer suites of the five newly-converted
methods (compression anti-thrash, session search, status, scheduler).

hermes_state.py

@@ -6490,8 +6490,8 @@ class SessionDB(SessionSearchMixin, SessionSchemaMixin, SessionPortabilityMixin)
         if active_only:
             query += " AND ended_at IS NULL"
         query += " ORDER BY last_active DESC"
-        with self._lock:
-            rows = self._conn.execute(query, params).fetchall()
+        with self._read_ctx() as conn:
+            rows = conn.execute(query, params).fetchall()
         return [self._session_row_dict(r) for r in rows]
 
     def find_session_by_origin(
@@ -6524,8 +6524,8 @@ class SessionDB(SessionSearchMixin, SessionSchemaMixin, SessionPortabilityMixin)
             query += " AND COALESCE(thread_id, '') = ?"
             params.append(str(thread_id))
         query += " ORDER BY started_at DESC"
-        with self._lock:
-            rows = [dict(r) for r in self._conn.execute(query, params).fetchall()]
+        with self._read_ctx() as conn:
+            rows = [dict(r) for r in conn.execute(query, params).fetchall()]
         if not rows:
             return None
         if user_id:
@@ -7599,8 +7599,7 @@ class SessionDB(SessionSearchMixin, SessionSchemaMixin, SessionPortabilityMixin)
         """Return the persisted deterministic-fallback streak."""
         if not session_id:
             return 0
-        with self._lock:
-            conn = self._conn
+        with self._read_ctx() as conn:
             if conn is None:
                 return 0
             row = conn.execute(
@@ -7680,8 +7679,7 @@ class SessionDB(SessionSearchMixin, SessionSchemaMixin, SessionPortabilityMixin)
         """
         if not session_id:
             return 0
-        with self._lock:
-            conn = self._conn
+        with self._read_ctx() as conn:
             if conn is None:
                 return 0
             row = conn.execute(
@@ -13177,8 +13175,8 @@ class SessionDB(SessionSearchMixin, SessionSchemaMixin, SessionPortabilityMixin)
             params.extend(ws_params)
         where_sql = f" WHERE {' AND '.join(where_clauses)}" if where_clauses else ""
         params.extend([limit, offset])
-        with self._lock:
-            cursor = self._conn.execute(
+        with self._read_ctx() as conn:
+            cursor = conn.execute(
                 f"{select_with_last_active}"
                 f"{where_sql} "
                 "ORDER BY last_active DESC, s.started_at DESC, s.id DESC LIMIT ? OFFSET ?",
@@ -14283,8 +14281,8 @@ class SessionDB(SessionSearchMixin, SessionSchemaMixin, SessionPortabilityMixin)
         # committed data — a pending write transaction's uncommitted
         # meta writes would be invisible.  This is a cheap point lookup,
         # not the convoy bottleneck the read-path split targets.
-        with self._read_ctx() as conn:
-            row = conn.execute(
+        with self._lock:
+            row = self._conn.execute(
                 "SELECT value FROM state_meta WHERE key = ?", (key,)
             ).fetchone()
         if row is None:

tests/state/test_no_locked_readers_gate.py

@@ -50,10 +50,11 @@ _READ_RE = re.compile(r"^\s*(SELECT|PRAGMA)\b", re.IGNORECASE)
 # the reason. Keep this list SHRINKING — never add to it without the same
 # scrutiny a new blocking call would get.
 _ALLOWED_LOCKED_READERS: dict[str, str] = {
-    # _enter_fts_fail_open reads schema state under the lock as part of the
-    # fail-open WRITE transition (the writes live in sibling statements the
-    # simple statement scanner attributes to other calls).
-    "_enter_fts_fail_open": "fail-open transition; lock orders vs FTS writes",
+    # get_meta stays on the writer lock BY DESIGN (see its inline comment):
+    # fts_rebuild_step reads rebuild progress before entering a write
+    # transaction, and a pooled WAL reader sees only committed data — the
+    # writer's own just-staged meta updates would be invisible to it.
+    "get_meta": "read-your-writes: rebuild progress read before write txn",
 }
 
 
@@ -76,16 +77,41 @@ def _first_sql_text(call: ast.Call) -> str | None:
     return text.strip()
 
 
-def _is_self_conn_execute(call: ast.Call) -> bool:
+def _is_self_conn_execute(call: ast.Call, aliases: set[str]) -> bool:
+    """Match ``self._conn.execute*`` and ``<alias>.execute*`` where the
+    alias was bound from ``self._conn`` (``conn = self._conn``)."""
     f = call.func
-    return (
+    if not (
         isinstance(f, ast.Attribute)
         and f.attr in ("execute", "executemany", "executescript")
-        and isinstance(f.value, ast.Attribute)
-        and f.value.attr == "_conn"
-        and isinstance(f.value.value, ast.Name)
-        and f.value.value.id == "self"
-    )
+    ):
+        return False
+    target = f.value
+    if (
+        isinstance(target, ast.Attribute)
+        and target.attr == "_conn"
+        and isinstance(target.value, ast.Name)
+        and target.value.id == "self"
+    ):
+        return True
+    return isinstance(target, ast.Name) and target.id in aliases
+
+
+def _collect_conn_aliases(method: ast.AST) -> set[str]:
+    """Names bound from ``self._conn`` anywhere in the method body."""
+    aliases: set[str] = set()
+    for node in ast.walk(method):
+        if isinstance(node, ast.Assign) and isinstance(node.value, ast.Attribute):
+            v = node.value
+            if (
+                v.attr == "_conn"
+                and isinstance(v.value, ast.Name)
+                and v.value.id == "self"
+            ):
+                for t in node.targets:
+                    if isinstance(t, ast.Name):
+                        aliases.add(t.id)
+    return aliases
 
 
 def _is_self_lock_with(item: ast.withitem) -> bool:
@@ -113,6 +139,7 @@ def _scan_locked_readers(state_py: "Path | None" = None) -> list[str]:
     for method in session_db.body:
         if not isinstance(method, (ast.FunctionDef, ast.AsyncFunctionDef)):
             continue
+        aliases = _collect_conn_aliases(method)
         for node in ast.walk(method):
             if not isinstance(node, ast.With):
                 continue
@@ -120,10 +147,18 @@ def _scan_locked_readers(state_py: "Path | None" = None) -> list[str]:
                 continue
             reads, writes, unknown = 0, 0, 0
             for inner in ast.walk(node):
-                if isinstance(inner, ast.Call) and _is_self_conn_execute(inner):
+                if isinstance(inner, ast.Call) and _is_self_conn_execute(inner, aliases):
                     word_full = _first_sql_text(inner)
                     word = word_full.split(None, 1)[0].upper() if word_full else None
                     if word is None:
+                        # SQL held in a variable or built f-string: the
+                        # scanner cannot prove it reads. A lock block whose
+                        # ONLY statements are unprovable is still flagged
+                        # below — writers name their verbs in literals
+                        # throughout this file, so opacity correlates with
+                        # composed SELECTs, and silently skipping these is
+                        # how 5 readers hid from the first version of this
+                        # gate.
                         unknown += 1
                     elif _PRAGMA_WRITE_RE.match(word_full or ""):
                         writes += 1
@@ -150,12 +185,14 @@ def _scan_locked_readers(state_py: "Path | None" = None) -> list[str]:
                         )
                     ):
                         writes += 1
-            if reads > 0 and writes == 0 and unknown == 0:
+            if writes == 0 and (reads > 0 or unknown > 0):
                 if method.name not in _ALLOWED_LOCKED_READERS:
+                    kind = "pure-read" if unknown == 0 else "no-proven-write"
                     violations.append(
-                        f"{method.name} (line {node.lineno}): pure-read "
+                        f"{method.name} (line {node.lineno}): {kind} "
                         f"body under `with self._lock:` — route through "
-                        f"_read_ctx() instead"
+                        f"_read_ctx() instead (or add a justified "
+                        f"allowlist entry)"
                     )
     return violations
 
@@ -179,10 +216,22 @@ class TestNoPureReadersUnderWriterLock:
             "    def guilty_reader(self):\n"
             "        with self._lock:\n"
             "            return self._conn.execute(\"SELECT 1\").fetchone()\n"
+            "    def guilty_alias_reader(self):\n"
+            "        with self._lock:\n"
+            "            conn = self._conn\n"
+            "            return conn.execute(\"SELECT 2\").fetchone()\n"
+            "    def guilty_variable_sql(self, query):\n"
+            "        with self._lock:\n"
+            "            return self._conn.execute(query).fetchall()\n"
+            "    def innocent_variable_writer(self, query):\n"
+            "        with self._lock:\n"
+            "            self._conn.execute(query)\n"
+            "            self._conn.execute(\"UPDATE t SET x = 2\")\n"
         )
         p = tmp_path / "fake_state.py"
         p.write_text(sabotage, encoding="utf-8")
         violations = _scan_locked_readers(p)
-        assert len(violations) == 1
-        assert "guilty_reader" in violations[0]
-        assert "innocent_writer" not in violations[0]
+        flagged = {v.split(" ")[0] for v in violations}
+        assert flagged == {
+            "guilty_reader", "guilty_alias_reader", "guilty_variable_sql"
+        }, violations