diff --git a/api/agent_sessions.py b/api/agent_sessions.py
index f682c8f3..2a4f3143 100644
--- a/api/agent_sessions.py
+++ b/api/agent_sessions.py
@@ -2,8 +2,10 @@
 import json
 import logging
 import sqlite3
+import sys
 from contextlib import closing
-from pathlib import Path
+from pathlib import Path, PurePosixPath, PureWindowsPath
+from urllib.parse import quote
 
 logger = logging.getLogger(__name__)
 
@@ -19,36 +21,65 @@ logger = logging.getLogger(__name__)
 _SOURCE_COLUMN_WARNED_DB_PATHS: set[str] = set()
 
 
+def state_db_file_uri(db_path, platform: str | None = None) -> str:
+    """Build the ``file:`` URI (no query string) for an absolute ``state.db`` path.
+
+    ``Path.as_uri()`` is not usable here: for a UNC share it emits
+    ``file://server/share/state.db`` and SQLite rejects any non-local URI
+    authority unless compiled with ``SQLITE_ALLOW_URI_AUTHORITY`` (the
+    CPython 3.11-3.13 Windows builds are not). SQLite instead accepts the
+    *empty*-authority spelling ``file:////server/share/state.db``, whose path
+    ``//server/share/...`` the Windows VFS opens as the UNC name. Drive-letter
+    paths keep the documented ``file:///C:/...`` form and POSIX paths are
+    unchanged. Only ``/`` survives unescaped, so ``?`` and ``#`` in path
+    components cannot leak into the query string.
+
+    ``platform`` defaults to the running interpreter; tests pass it explicitly
+    so the Windows shapes are checked from any host.
+    """
+    platform = platform or sys.platform
+    if platform == "win32":
+        win = PureWindowsPath(str(db_path))
+        drive = win.drive
+        if drive.startswith("\\\\?\\"):
+            # Extended-length prefix: "\\?\C:" or "\\?\UNC\server\share".
+            drive = drive[4:]
+            if drive.upper().startswith("UNC\\"):
+                drive = "\\\\" + drive[4:]
+        parts = win.parts[1:]  # drop the anchor, keep the path components
+        if drive.startswith("\\\\"):
+            host_share = drive[2:].replace("\\", "/")
+            posix_path = "//" + "/".join((host_share, *parts))
+        else:
+            posix_path = "/" + "/".join((drive, *parts))
+        # ``:`` stays literal so the drive letter keeps SQLite's documented
+        # ``file:///C:/...`` shape (``Path.as_uri()`` leaves it unescaped too).
+        return "file://" + quote(posix_path, safe="/:")
+    posix_path = PurePosixPath(str(db_path)).as_posix()
+    return "file://" + quote(posix_path, safe="/")
+
+
+def state_db_readonly_uri(db_path, platform: str | None = None) -> str:
+    """Strict read-only (``mode=ro``) form of :func:`state_db_file_uri`."""
+    return state_db_file_uri(db_path, platform=platform) + "?mode=ro"
+
+
 def open_state_db_readonly(db_path: Path, log: logging.Logger | None = None) -> sqlite3.Connection:
     """Open the live agent ``state.db`` read-only for a pure-read projection.
 
     Same rationale as the session-listing path (#5455): a write-capable handle
     on the multi-GB, WAL ``state.db`` while the agent streams into it adds
     needless checkpoint/lock surface. The read-only ``file:...?mode=ro`` URI
-    avoids that. Falls back to a writable connection (and warns) if the
-    read-only open fails, so callers never lose data on exotic filesystems.
+    avoids that. Read failures propagate; a reader never upgrades to a writer.
 
     The caller must ensure ``db_path`` exists — this raises ``FileNotFoundError``
-    for a missing path rather than letting the writable fallback below create an
-    empty, writable ``state.db`` there (a ghost DB in the agent's HOME). The
-    fallback is only for an *existing* DB whose read-only open fails on an exotic
-    filesystem, so a real read never loses data.
+    for a missing path rather than creating a ghost database.
 
     Callers own the returned connection (wrap it in ``contextlib.closing``).
     """
-    log = log or logger
     if not db_path.exists():
         raise FileNotFoundError(f"agent state.db not found: {db_path}")
-    read_only_uri = f"{db_path.resolve().as_uri()}?mode=ro"
-    try:
-        return sqlite3.connect(read_only_uri, uri=True)
-    except sqlite3.Error as exc:
-        log.warning(
-            "agent state.db read-only open failed for %s; falling back to writable connection: %s",
-            db_path,
-            exc,
-        )
-        return sqlite3.connect(str(db_path))
+    return sqlite3.connect(state_db_readonly_uri(db_path.resolve()), uri=True)
 
 
 MESSAGING_SOURCES = {
@@ -609,21 +640,7 @@ def read_importable_agent_session_rows(
         return []
 
     log = log or logger
-    # Open read-only for this projection/listing path: it is a pure read, and
-    # holding a write-capable handle on the live (multi-GB, WAL) state.db while
-    # the agent streams into it adds needless checkpoint/lock surface (#5455).
-    # The defensive index self-heal below still runs, but through a separate
-    # short-lived writable connection on the rare missing-index path only.
-    read_only_uri = f"{db_path.resolve().as_uri()}?mode=ro"
-    try:
-        conn = sqlite3.connect(read_only_uri, uri=True)
-    except sqlite3.Error as exc:
-        log.warning(
-            "agent session listing read-only open failed for %s; falling back to writable connection: %s",
-            db_path,
-            exc,
-        )
-        conn = sqlite3.connect(str(db_path))
+    conn = open_state_db_readonly(db_path, log=log)
     with closing(conn):
         conn.row_factory = sqlite3.Row
         cur = conn.cursor()
@@ -674,11 +691,7 @@ def read_importable_agent_session_rows(
         use_messages_join = messages_has_session_id
         count_col = 'id' if 'id' in message_cols else 'session_id'
 
-        # Defensive index prime (#3887). The normal candidate-ordering shape uses
-        # the agent's standard ``idx_messages_session ON messages(session_id,
-        # timestamp)`` index; without it, large cron-only scans degrade badly.
-        # Writable dbs self-heal by recreating the index. Read-only or locked dbs
-        # fall back to the pre-aggregated cron-only path below instead of failing.
+        # Index creation belongs to explicit drained maintenance, never a read.
         messages_index_present = False
         if messages_has_session_id and messages_has_timestamp:
             try:
@@ -686,21 +699,7 @@ def read_importable_agent_session_rows(
                 messages_index_present = any(str(row[1]) == "idx_messages_session" for row in cur.fetchall())
             except sqlite3.Error:
                 messages_index_present = False
-            if not messages_index_present:
-                # Self-heal via a separate writable connection so the common
-                # (index-present) path keeps its read-only handle. On a truly
-                # read-only/locked db this fails and we degrade to the
-                # pre-aggregated cron-only path below, exactly as before.
-                try:
-                    with closing(sqlite3.connect(str(db_path))) as _heal:
-                        _heal.execute(
-                            "CREATE INDEX IF NOT EXISTS idx_messages_session "
-                            "ON messages(session_id, timestamp)"
-                        )
-                        _heal.commit()
-                    messages_index_present = True
-                except sqlite3.Error:
-                    pass  # read-only db / locked / older schema — degrade gracefully
+
 
         if use_messages_join:
             actual_count_expr = f"COUNT(m.{count_col})"
@@ -726,7 +725,6 @@ def read_importable_agent_session_rows(
 
         where_clauses = ["s.source IS NOT NULL"]
         params: list[object] = []
-        included = ()
         if include_sources:
             included = tuple(str(source) for source in include_sources if source)
             if included:
@@ -740,10 +738,16 @@ def read_importable_agent_session_rows(
                 where_clauses.append(f"s.source NOT IN ({placeholders})")
                 params.extend(excluded)
 
+        # Without ``idx_messages_session`` the correlated ``MAX(mx.timestamp)``
+        # candidate ordering rescans ``messages`` once per session (seconds on
+        # a few thousand sessions). Every missing-index projection — default
+        # sidebar, gateway watcher, cron/webhook/kanban views — orders through
+        # one read-only pre-aggregation pass instead; the listing never creates
+        # the index itself (that is drained maintenance, see
+        # ``scripts/ensure_state_db_read_indexes.py``).
         use_preaggregated_candidate_order = (
             use_messages_join
             and messages_has_timestamp
-            and included == ("cron",)
             and not messages_index_present
         )
         if use_preaggregated_candidate_order:
diff --git a/api/gateway_watcher.py b/api/gateway_watcher.py
index 635969ec..dc93ecb9 100644
--- a/api/gateway_watcher.py
+++ b/api/gateway_watcher.py
@@ -328,6 +328,12 @@ class GatewayWatcher:
         protects projection fields (notably role-derived CLI visibility) that
         the agent's existing index cannot see.
         """
+        with self._sub_lock:
+            has_subscribers = bool(self._subscribers)
+        if not has_subscribers:
+            # Force fresh projection on the next subscription, without touching DB.
+            self._last_full_projection_at = None
+            return False
         db_path = self._state_db_path
         # A watcher may start before the agent has created state.db. Publishing an
         # empty first snapshot would make an already-rendered sidebar disappear;
diff --git a/api/routes.py b/api/routes.py
index b1823387..fcd31a0f 100644
--- a/api/routes.py
+++ b/api/routes.py
@@ -46,6 +46,7 @@ from api.agent_sessions import (
     _looks_like_default_cli_title,
     is_cli_session_row,
     is_cli_session_row_visible,
+    open_state_db_readonly,
     read_session_lineage_report,
 )
 from api.compression_anchor import visible_messages_for_anchor
@@ -320,7 +321,7 @@ def _latest_cron_session_info_for_jobs(
     if not db_path or not Path(db_path).exists():
         return {jid: {"session_id": "", "message_count": None} for jid in requested}
     try:
-        with closing(sqlite3.connect(str(db_path))) as conn:
+        with closing(open_state_db_readonly(db_path)) as conn:
             conn.row_factory = sqlite3.Row
             cur = conn.cursor()
             cur.execute("PRAGMA table_info(sessions)")
@@ -12031,7 +12032,7 @@ def _handle_insights(handler, parsed) -> bool:
         from api.models import _active_state_db_path
         db_path = _active_state_db_path()
         if db_path and db_path.exists():
-            with closing(sqlite3.connect(str(db_path))) as conn:
+            with closing(open_state_db_readonly(db_path)) as conn:
                 conn.row_factory = sqlite3.Row
                 cur = conn.cursor()
                 # cache_read_tokens may not exist on older agent state DBs;
@@ -12666,7 +12667,7 @@ def _deep_health_checks(stream_check: dict | None = None) -> tuple[dict, bool]:
                 "ms": round((time.time() - t0) * 1000, 1),
             }
         else:
-            with closing(sqlite3.connect(str(db_path))) as conn:
+            with closing(open_state_db_readonly(db_path)) as conn:
                 conn.execute("PRAGMA schema_version").fetchone()
             checks["state_db"] = {
                 "status": "ok",
diff --git a/docs/troubleshooting.md b/docs/troubleshooting.md
index 92481a53..833fbd6a 100644
--- a/docs/troubleshooting.md
+++ b/docs/troubleshooting.md
@@ -231,6 +231,36 @@ For a foreground `python3 bootstrap.py`, stop it with Ctrl-C and start it again.
 
 ---
 
+## Agent sessions list slowly (or the `state.db` read index is missing)
+
+**Symptom.** The sidebar's imported/CLI session list takes seconds per refresh on a large Hermes profile, or a log line says a `state.db` read failed. Sessions still appear; nothing is lost.
+
+**Why.** Every WebUI reader of the agent's `state.db` (session listing, transcript reads, lineage, gateway watcher, cron sidebar, insights, health) opens it strictly read-only (`file:...?mode=ro`). A reader never upgrades to a write-capable handle and never creates an index: on a multi-GiB `messages` table `CREATE INDEX` holds the SQLite writer lock for minutes and stalls the agent streaming into the same WAL database. When the agent's standard `idx_messages_session` index is missing (older agent, hand-rebuilt or re-imported DB), listings degrade to a bounded one-pass pre-aggregation — slower than the indexed seek, but read-only. A read-only open failure propagates to the caller's existing error boundary (the listing returns empty for that profile) instead of silently reopening the file writable.
+
+**Diagnostic.**
+
+```bash
+sqlite3 "file:$HOME/.hermes/state.db?mode=ro" "PRAGMA index_list(messages)"
+```
+
+`idx_messages_session` should be listed. If it is not, the agent has not created it and WebUI will not create it for you.
+
+**Fix.** Create the covering read indexes in an explicit drained maintenance window: stop the agent (and any gateway/cron runner) writing to that `state.db`, then run:
+
+```bash
+python3 scripts/ensure_state_db_read_indexes.py --db ~/.hermes/state.db --confirm-drained
+```
+
+- `--confirm-drained` is mandatory: it is your assertion that no agent turn is running against the database. The tool does not verify it.
+- `--lock-file PATH` (optional) additionally holds an exclusive non-blocking lock on `PATH` (`flock` on POSIX, `msvcrt.locking` on Windows) for deployments that serialise agent turns on a lock file; a held lock makes the tool exit without touching the database. Without `--lock-file` no lock primitive is required, so the script runs on native Windows as well. On Windows the lock covers byte 0 of `PATH`; a new or empty lock file is initialised with one byte first (an existing lock file is never rewritten).
+- The tool opens the database `mode=rw` (never `rwc`): a mistyped path raises instead of creating an empty database. Indexes are created inside one `BEGIN IMMEDIATE` transaction and rolled back on any error.
+- It is idempotent and prints a JSON status per index (`created` / `existing`). An existing index with a different table, key shape or collation is reported as `Incompatible index` and never replaced; an index that is not covering (`EXPLAIN QUERY PLAN`) is reported as `Index is not covering`.
+- Windows UNC profiles (`HERMES_HOME=\\server\share\...`) are supported: readers and this tool build the empty-authority URI `file:////server/share/state.db` that the bundled SQLite accepts.
+
+**When to file a bug.** File a WebUI bug if the listing stays slow after the tool reports `existing` for `idx_messages_session`, if the tool reports `Incompatible index` on an untouched agent-created database, or if a read-only open fails on a local path. Include the tool's JSON output, the `PRAGMA index_list(messages)` result, and the sanitized error text.
+
+---
+
 ## 404 after login when password auth is enabled
 
 **Symptom.** After enabling password authentication (`HERMES_WEBUI_PASSWORD`), logging in redirects to `/sessions` and the browser shows a `404 not found` error instead of the chat interface.
diff --git a/scripts/ensure_state_db_read_indexes.py b/scripts/ensure_state_db_read_indexes.py
new file mode 100644
index 00000000..31dce342
--- /dev/null
+++ b/scripts/ensure_state_db_read_indexes.py
@@ -0,0 +1,132 @@
+#!/usr/bin/env python3
+"""Create the covering read indexes on the agent ``state.db`` in an explicit,
+drained maintenance window.
+
+The WebUI read paths (session listing, lineage reads, gateway watcher, cron
+sidebar, insights, health) never create indexes themselves: ``CREATE INDEX``
+on a multi-GiB ``messages`` table holds the SQLite writer lock for minutes,
+which stalls the agent streaming into the same WAL database. Index
+maintenance is therefore an operator action run while the agent is idle.
+
+Usage::
+
+    python scripts/ensure_state_db_read_indexes.py --db ~/.hermes/state.db \\
+        --confirm-drained [--lock-file /path/to/agent-activity.lock]
+
+``--confirm-drained`` is mandatory. When ``--lock-file`` is given, the tool
+takes an exclusive non-blocking lock on that path (``flock`` on POSIX,
+``msvcrt.locking`` on Windows) so that a deployment which serialises agent
+turns on a lock file cannot start a turn mid-rebuild. Without ``--lock-file``
+no lock primitive is needed, so the script runs on every supported platform.
+
+Existing indexes are verified for table, key shape and collation and must be
+covering (``EXPLAIN QUERY PLAN``); an incompatible index is reported, never
+silently replaced.
+"""
+import argparse
+from contextlib import closing, nullcontext
+import json
+import os
+from pathlib import Path
+import sqlite3
+import sys
+
+try:  # POSIX
+    import fcntl
+except ImportError:  # native Windows
+    fcntl = None
+try:  # Windows
+    import msvcrt
+except ImportError:  # POSIX
+    msvcrt = None
+
+_REPO_ROOT = Path(__file__).resolve().parents[1]
+if str(_REPO_ROOT) not in sys.path:
+    sys.path.insert(0, str(_REPO_ROOT))
+from api.agent_sessions import state_db_file_uri  # same UNC-aware URI as the readers
+
+INDEXES = {
+    "idx_messages_session": ("messages", (("session_id", "BINARY"), ("timestamp", "BINARY"))),
+    "idx_messages_session_role": ("messages", (("session_id", "BINARY"), ("role", "NOCASE"))),
+    "idx_sessions_webui_fingerprint": ("sessions", tuple((c, "BINARY") for c in
+        ("source", "id", "message_count", "last_activity_at"))),
+}
+
+
+def _exclusive_lock(lock_file):
+    if lock_file is None:
+        return nullcontext()
+    if fcntl is None and msvcrt is None:
+        raise RuntimeError("--lock-file needs fcntl (POSIX) or msvcrt (Windows); "
+                           "neither is available on this interpreter")
+    handle = open(lock_file, "a+")
+    try:
+        if fcntl is not None:
+            fcntl.flock(handle, fcntl.LOCK_EX | fcntl.LOCK_NB)
+        else:
+            # Lock one byte at offset 0 without blocking; raises OSError when held.
+            # ``a+`` may have just created an empty file, and Windows cannot lock
+            # a range beyond EOF, so make sure byte 0 exists first. An existing
+            # (deployment-owned) lock file is left untouched. Write the byte on
+            # the descriptor: a text-mode ``handle.write("\n")`` becomes two
+            # bytes (``\r\n``) on native Windows.
+            if os.fstat(handle.fileno()).st_size == 0:
+                os.write(handle.fileno(), b"\0")
+                handle.flush()
+            handle.seek(0)
+            msvcrt.locking(handle.fileno(), msvcrt.LK_NBLCK, 1)
+    except BaseException:
+        handle.close()
+        raise
+    return closing(handle)
+
+
+def ensure_read_indexes(db_path, *, confirmed_drained=False, lock_file=None):
+    if not confirmed_drained:
+        raise RuntimeError("requires --confirm-drained")
+    db_path = Path(db_path).resolve(strict=True)
+    statuses = {}
+    with _exclusive_lock(lock_file):
+        # mode=rw (not rwc): never create a ghost database at a mistyped path.
+        with closing(sqlite3.connect(state_db_file_uri(db_path) + "?mode=rw", uri=True)) as db:
+            db.execute("BEGIN IMMEDIATE")
+            try:
+                for name, (table, keys) in INDEXES.items():
+                    existing = db.execute("SELECT tbl_name FROM sqlite_master WHERE name=?", (name,)).fetchone()
+                    if existing:
+                        actual = tuple((r[2], r[4]) for r in db.execute(f"PRAGMA index_xinfo({name})") if r[5])
+                        flags = next((r for r in db.execute(f"PRAGMA index_list({table})") if r[1] == name), None)
+                        if existing[0] != table or actual != keys or not flags or flags[2] or flags[4]:
+                            raise RuntimeError(f"Incompatible index: {name}")
+                        statuses[name] = "existing"
+                    else:
+                        columns = ", ".join(f"{col} COLLATE {collation}" for col, collation in keys)
+                        db.execute(f"CREATE INDEX {name} ON {table}({columns})")
+                        statuses[name] = "created"
+                    columns = ", ".join(col for col, _ in keys)
+                    plan = db.execute(f"EXPLAIN QUERY PLAN SELECT {columns} FROM {table} INDEXED BY {name}").fetchall()
+                    if not any(f"COVERING INDEX {name}" in row[3] for row in plan):
+                        raise RuntimeError(f"Index is not covering: {name}")
+                db.commit()
+            except BaseException:
+                db.rollback()
+                raise
+    return statuses
+
+
+def main():
+    parser = argparse.ArgumentParser(description=__doc__,
+                                     formatter_class=argparse.RawDescriptionHelpFormatter)
+    parser.add_argument("--db", type=Path, required=True, help="path to the agent state.db")
+    parser.add_argument("--lock-file", type=Path, default=None,
+                        help="optional lock file to hold exclusively (flock) while indexes are built")
+    parser.add_argument("--confirm-drained", action="store_true",
+                        help="assert that no agent turn is running against this database")
+    args = parser.parse_args()
+    result = ensure_read_indexes(args.db, confirmed_drained=args.confirm_drained,
+                                 lock_file=args.lock_file)
+    print(json.dumps(result, sort_keys=True))
+
+
+if __name__ == "__main__":
+    main()
diff --git a/tests/test_issue1494_state_db_fd_leak.py b/tests/test_issue1494_state_db_fd_leak.py
index 1691e43d..6ba92759 100644
--- a/tests/test_issue1494_state_db_fd_leak.py
+++ b/tests/test_issue1494_state_db_fd_leak.py
@@ -163,9 +163,9 @@ def test_read_importable_agent_session_rows_closes_connection(tmp_path, tracking
         read_importable_agent_session_rows(db)
 
     _assert_all_closed(tracking_sqlite, "read_importable_agent_session_rows")
-    # #5455: missing-index self-heal uses one separate short-lived RW connection
-    # in addition to the five read-only listing connections.
-    assert len(tracking_sqlite.instances) == 6
+    # #5455: the listing is strictly read-only, including when indexes are
+    # missing — five calls, five read-only connections, no self-heal writer.
+    assert len(tracking_sqlite.instances) == 5
 
 
 def test_read_session_lineage_metadata_closes_connection(tmp_path, tracking_sqlite):
diff --git a/tests/test_issue2628_cli_sessions_perf.py b/tests/test_issue2628_cli_sessions_perf.py
index 70f4513f..6f31f031 100644
--- a/tests/test_issue2628_cli_sessions_perf.py
+++ b/tests/test_issue2628_cli_sessions_perf.py
@@ -1,15 +1,14 @@
 """Regression coverage for capped CLI/agent session sidebar scans (#2628)."""
 
-import pathlib
 import sqlite3
 import time
+from contextlib import contextmanager
 
 import pytest
 
 import api.agent_sessions as agent_sessions
 
 _REAL_SQLITE_CONNECT = sqlite3.connect
-REPO_ROOT = pathlib.Path(__file__).resolve().parents[1]
 
 
 def _make_state_db(path, *, sessions=80, messages_per_session=3, create_messages_index=True, source="cli", session_source="cli"):
@@ -169,6 +168,24 @@ def _make_connect_with_progress_budget(*, budget_ops, interval=1):
     return _connect
 
 
+@contextmanager
+def _capture_listing_sql():
+    """Record the SQL statements the listing runs (trace callback, read-only)."""
+    statements = []
+
+    def _tracing_connect(*args, **kwargs):
+        conn = _REAL_SQLITE_CONNECT(*args, **kwargs)
+        conn.set_trace_callback(statements.append)
+        return conn
+
+    original_connect = agent_sessions.sqlite3.connect
+    agent_sessions.sqlite3.connect = _tracing_connect
+    try:
+        yield statements
+    finally:
+        agent_sessions.sqlite3.connect = original_connect
+
+
 def _make_connect_with_progress_counter(*, interval=1):
     steps = {"count": 0}
 
@@ -198,24 +215,59 @@ def test_importable_agent_rows_push_sidebar_limit_into_sql(tmp_path):
     db = tmp_path / "state.db"
     _make_state_db(db, sessions=120, messages_per_session=5)
 
-    rows = agent_sessions.read_importable_agent_session_rows(db, limit=20, exclude_sources=("webui",))
+    with _capture_listing_sql() as statements:
+        rows = agent_sessions.read_importable_agent_session_rows(db, limit=20, exclude_sources=("webui",))
 
     assert len(rows) == 20
     assert [row["id"] for row in rows][:3] == ["cli_perf_0119", "cli_perf_0118", "cli_perf_0117"]
     assert {row["actual_message_count"] for row in rows} == {5}
 
-    src = (REPO_ROOT / "api" / "agent_sessions.py").read_text()
-    assert "WITH candidates AS" in src
-    assert "JOIN candidates c ON c.id = s.id" in src
-    assert "latest_messages AS" in src
-    assert "LEFT JOIN latest_messages lm ON lm.session_id = s.id" in src
-    assert 'included == ("cron",)' in src
-    assert "not messages_index_present" in src
-    assert "PRAGMA index_list(messages)" in src
-    assert "CREATE INDEX IF NOT EXISTS idx_messages_session" in src
-    assert "_CRON_PREAGGREGATE_CANDIDATE_ORDER_MIN_MESSAGES" not in src
-    assert "MAX(mx.timestamp) FROM messages mx WHERE mx.session_id = s.id" in src
-    assert "candidate_limit = max(result_limit * 8, result_limit)" in src
+    candidates_sql = next(
+        statement
+        for statement in statements
+        if "LIMIT" in statement and "FROM sessions s" in statement
+    )
+    # The cap is applied inside the candidate subquery, not in Python after an
+    # unbounded aggregate: the oversampled window (limit * 8 = 160 here) is
+    # part of the statement SQLite actually runs, and it bounds the candidate
+    # subquery that the outer projection joins.
+    assert candidates_sql.count("LIMIT") == 1, candidates_sql
+    window_end = candidates_sql.index("LIMIT")
+    window = candidates_sql[:window_end]
+    assert "JOIN candidates" not in window, candidates_sql
+    cap = int(candidates_sql[window_end:].split(None, 2)[1])
+    assert cap == 160, candidates_sql
+
+
+def test_importable_agent_rows_candidate_window_overfilters_then_fills(tmp_path):
+    """Rows dropped after projection must not shrink the returned page.
+
+    The SQL candidate window is oversampled (limit * 8) precisely so sessions
+    filtered client-side of the SQL (invisible CLI rows here) only consume
+    candidates, not result slots: a page that would be short if the cap were
+    the plain ``limit`` still fills to ``limit`` visible rows.
+    """
+    db = tmp_path / "state.db"
+    _make_state_db(db, sessions=120, messages_per_session=5)
+
+    # 30 newest sessions are CLI rows that the projection hides (ended, default
+    # title, no user turn visible): they must fall out of the window without
+    # starving the visible page below them.
+    conn = sqlite3.connect(str(db))
+    for i in range(90, 120):
+        conn.execute(
+            "UPDATE sessions SET title = 'cli session', ended_at = 1.0, end_reason = 'timeout' WHERE id = ?",
+            (f"cli_perf_{i:04d}",),
+        )
+    conn.commit()
+    conn.close()
+
+    rows = agent_sessions.read_importable_agent_session_rows(db, limit=20, exclude_sources=("webui",))
+
+    assert len(rows) == 20
+    assert rows[0]["id"] == "cli_perf_0089"
+    assert rows[-1]["id"] == "cli_perf_0070"
+    assert {row["actual_message_count"] for row in rows} == {5}
 
 
 def test_importable_agent_rows_candidate_ordering_stays_under_progress_budget(tmp_path, monkeypatch):
diff --git a/tests/test_issue3506_memory_and_watcher.py b/tests/test_issue3506_memory_and_watcher.py
index 5fa60884..46eb0040 100644
--- a/tests/test_issue3506_memory_and_watcher.py
+++ b/tests/test_issue3506_memory_and_watcher.py
@@ -534,6 +534,7 @@ def test_poll_loop_skips_projection_when_unchanged(tmp_path, monkeypatch):
 
     monkeypatch.setattr(gw, "_get_agent_sessions_from_db", fake_projection)
     w = gw.GatewayWatcher(state_db_path=db)
+    w.subscribe()  # Idle watchers deliberately do not poll without observers.
 
     assert w._poll_once(now=1.0) is True
     assert projected == [True]
diff --git a/tests/test_issue3887_index_prime.py b/tests/test_issue3887_index_prime.py
index 399f3406..d3641989 100644
--- a/tests/test_issue3887_index_prime.py
+++ b/tests/test_issue3887_index_prime.py
@@ -74,28 +74,38 @@ def _messages_indexes(path):
     return {r[0] for r in rows}
 
 
-def test_prime_creates_missing_index(tmp_path):
-    """A db missing idx_messages_session gets it primed on the first scan."""
+def test_listing_does_not_create_missing_index(tmp_path, monkeypatch):
+    """A missing index must never turn a sidebar read into a schema write.
+
+    CREATE INDEX on a multi-GiB ``messages`` table holds the SQLite writer lock
+    for minutes; index maintenance belongs to the explicit drained tool
+    (``scripts/ensure_state_db_read_indexes.py``), not to a listing.
+    """
     db = tmp_path / "state.db"
     _full_schema_db(db)
     assert "idx_messages_session" not in _messages_indexes(db)
+    connect_calls = []
+    real_connect = agent_sessions.sqlite3.connect
+
+    def recording_connect(*args, **kwargs):
+        connect_calls.append((args, kwargs))
+        return real_connect(*args, **kwargs)
+
+    monkeypatch.setattr(agent_sessions.sqlite3, "connect", recording_connect)
 
     rows = agent_sessions.read_importable_agent_session_rows(
         db, limit=20, exclude_sources=None
     )
+    # Snapshot before the verifier below opens its own (non-listing) handle.
+    listing_connect_calls = list(connect_calls)
 
     # Listing still returns the sessions ...
     assert {r["id"] for r in rows} == {"sess0", "sess1", "sess2"}
-    # ... and the index now exists on (session_id, timestamp).
-    assert "idx_messages_session" in _messages_indexes(db)
-    conn = sqlite3.connect(str(db))
-    try:
-        sql = conn.execute(
-            "SELECT sql FROM sqlite_master WHERE name='idx_messages_session'"
-        ).fetchone()[0]
-    finally:
-        conn.close()
-    assert "session_id" in sql and "timestamp" in sql
+    # ... without mutating the schema or opening a write-capable connection.
+    assert "idx_messages_session" not in _messages_indexes(db)
+    assert len(listing_connect_calls) == 1
+    assert "mode=ro" in str(listing_connect_calls[0][0][0])
+    assert listing_connect_calls[0][1].get("uri") is True
 
 
 def test_prime_is_noop_when_index_exists(tmp_path):
diff --git a/tests/test_issue5455_lineage_readonly_reads.py b/tests/test_issue5455_lineage_readonly_reads.py
index 35d3af71..be2ec777 100644
--- a/tests/test_issue5455_lineage_readonly_reads.py
+++ b/tests/test_issue5455_lineage_readonly_reads.py
@@ -9,7 +9,6 @@ lineage-metadata reads, and the gateway-watcher fingerprint projection (a 5s
 poll), were still opening a read-WRITE connection. This shared them onto the
 same ``open_state_db_readonly`` helper.
 """
-import logging
 import sqlite3
 from contextlib import closing
 
@@ -98,25 +97,25 @@ def test_helper_encodes_special_path_chars(tmp_path, monkeypatch):
     assert calls[0]["target"].endswith("?mode=ro")
 
 
-def test_helper_falls_back_to_writable_and_logs(tmp_path, monkeypatch, caplog):
+def test_helper_never_falls_back_to_writable(tmp_path, monkeypatch):
+    """A read-only open failure propagates; the helper never retries with a
+    write-capable handle on the live WAL database."""
     db = tmp_path / "state.db"
     _make_lineage_db(db)
-    real_connect = sqlite3.connect
+    calls = []
 
     def fail_read_only(target, *args, **kwargs):
-        if kwargs.get("uri"):
-            raise sqlite3.OperationalError("synthetic read-only URI failure")
-        return real_connect(target, *args, **kwargs)
+        calls.append({"target": str(target), "uri": bool(kwargs.get("uri"))})
+        raise sqlite3.OperationalError("synthetic read-only URI failure")
 
     monkeypatch.setattr(agent_sessions.sqlite3, "connect", fail_read_only)
-    with caplog.at_level(logging.WARNING, logger="api.agent_sessions"):
-        conn = open_state_db_readonly(db)
-    try:
-        assert conn.execute("SELECT COUNT(*) FROM sessions").fetchone()[0] == 2
-    finally:
-        conn.close()
-    assert "read-only open failed" in caplog.text
-    assert "synthetic read-only URI failure" in caplog.text
+    with pytest.raises(sqlite3.OperationalError, match="synthetic read-only URI failure"):
+        open_state_db_readonly(db)
+
+    # Exactly one attempt, and it was the read-only URI form.
+    assert len(calls) == 1
+    assert calls[0]["uri"] is True
+    assert calls[0]["target"].endswith("?mode=ro")
 
 
 def test_helper_raises_on_missing_db_instead_of_creating_a_ghost(tmp_path):
diff --git a/tests/test_issue5455_listing_readonly_connection.py b/tests/test_issue5455_listing_readonly_connection.py
index 9502fae1..799638a0 100644
--- a/tests/test_issue5455_listing_readonly_connection.py
+++ b/tests/test_issue5455_listing_readonly_connection.py
@@ -6,13 +6,12 @@ re-run a defensive ``CREATE INDEX`` self-heal on every sidebar build. Holding a
 write-capable handle while the agent streams into the same DB adds needless
 checkpoint/lock surface.
 
-The listing path now opens the DB read-only (``file:...?mode=ro``) and only
-self-heals a missing ``idx_messages_session`` through a SEPARATE short-lived
-writable connection. With the index present (the normal case) the read path
-performs zero writes; when the index is missing the self-heal still runs and
-rows still come back.
+The listing path opens the DB read-only (``file:...?mode=ro``) and never
+upgrades to a writable handle: a read-only open failure propagates, and a
+missing ``idx_messages_session`` degrades to the pre-aggregated path instead
+of being self-healed. Index maintenance is an explicit drained operation
+(``scripts/ensure_state_db_read_indexes.py``).
 """
-import logging
 import sqlite3
 
 import api.agent_sessions as agent_sessions
@@ -163,24 +162,27 @@ def test_listing_read_only_uri_encodes_special_path_chars(tmp_path, monkeypatch)
     assert calls[0]["target"].endswith("?mode=ro")
 
 
-def test_read_only_open_fallback_is_logged(tmp_path, monkeypatch, caplog):
+def test_read_only_open_failure_never_retries_with_writable_connection(tmp_path, monkeypatch):
     db = tmp_path / "state.db"
     _make_db(db, with_index=True)
-    real_connect = sqlite3.connect
+    calls = []
 
     def fail_read_only(target, *args, **kwargs):
-        if kwargs.get("uri"):
-            raise sqlite3.OperationalError("synthetic read-only URI failure")
-        return real_connect(target, *args, **kwargs)
+        calls.append({"target": str(target), "uri": bool(kwargs.get("uri"))})
+        raise sqlite3.OperationalError("synthetic read-only URI failure")
 
     monkeypatch.setattr(agent_sessions.sqlite3, "connect", fail_read_only)
 
-    with caplog.at_level(logging.WARNING, logger="api.agent_sessions"):
-        out = read_importable_agent_session_rows(db, exclude_sources=None)
+    try:
+        read_importable_agent_session_rows(db, exclude_sources=None)
+        raise AssertionError("read-only open failure unexpectedly recovered")
+    except sqlite3.OperationalError as exc:
+        assert "synthetic read-only URI failure" in str(exc)
 
-    assert "cli-1" in {r["id"] for r in out}
-    assert "read-only open failed" in caplog.text
-    assert "synthetic read-only URI failure" in caplog.text
+    # Exactly one attempt, and it was the read-only URI form.
+    assert len(calls) == 1
+    assert calls[0]["uri"] is True
+    assert "mode=ro" in calls[0]["target"]
 
 
 def test_index_present_performs_no_writable_connection(tmp_path, monkeypatch):
@@ -195,7 +197,7 @@ def test_index_present_performs_no_writable_connection(tmp_path, monkeypatch):
     assert all(c["uri"] and "mode=ro" in c["target"] for c in calls), calls
 
 
-def test_missing_index_self_heals_via_separate_connection(tmp_path, monkeypatch):
+def test_missing_index_degrades_without_writable_connection(tmp_path, monkeypatch):
     db = tmp_path / "state.db"
     _make_db(db, with_index=False)
     calls = _record_connects(monkeypatch)
@@ -204,12 +206,13 @@ def test_missing_index_self_heals_via_separate_connection(tmp_path, monkeypatch)
 
     # Rows still come back...
     assert "cli-1" in {r["id"] for r in out}
-    # ...and the self-heal ran through a separate writable (non-ro) connection.
-    assert any((not c["uri"]) and "mode=ro" not in c["target"] for c in calls), calls
-    # The index now exists on disk.
+    # ...without any implicit schema-maintenance writer.
+    assert calls
+    assert all(c["uri"] and "mode=ro" in c["target"] for c in calls), calls
+    # A missing index stays a maintenance concern, not a listing side effect.
     verify = sqlite3.connect(str(db))
     try:
         names = {row[1] for row in verify.execute("PRAGMA index_list(messages)")}
     finally:
         verify.close()
-    assert "idx_messages_session" in names
+    assert "idx_messages_session" not in names
diff --git a/tests/test_pr7445_readonly_review_followups.py b/tests/test_pr7445_readonly_review_followups.py
new file mode 100644
index 00000000..4239afaa
--- /dev/null
+++ b/tests/test_pr7445_readonly_review_followups.py
@@ -0,0 +1,367 @@
+"""Review follow-ups for the strictly read-only ``state.db`` readers (PR #7445).
+
+Contract under test:
+
+* every missing-index projection (default sidebar, webhook, kanban, …), not
+  only the cron view, orders its candidate window through the read-only
+  pre-aggregated ``latest_messages`` CTE — never through the correlated
+  ``MAX(mx.timestamp)`` scalar subquery that rescans ``messages`` per session;
+  the selected rows and their order stay identical to the indexed listing;
+* ``state_db_readonly_uri`` builds a ``file:`` URI SQLite accepts on every
+  supported platform: POSIX unchanged, Windows drive letters as
+  ``file:///C:/...``, UNC shares with an *empty* authority
+  (``file:////server/share/...``) because CPython's bundled SQLite rejects any
+  non-local authority, extended-length ``\\\\?\\`` prefixes stripped; the
+  query string is always the strict ``mode=ro``;
+* ``scripts/ensure_state_db_read_indexes.py`` imports and runs its no-lock path
+  on a platform without ``fcntl`` (native Windows), and refuses ``--lock-file``
+  there with a clear error instead of an ``ImportError`` at import time;
+* on the ``msvcrt`` branch the tool locks byte 0 of the lock file only after
+  making sure that byte exists, so a fresh (empty) ``--lock-file`` is acquired
+  without relying on the CRT's beyond-EOF locking behaviour; that byte is
+  written as exactly one byte even under the ``\\n`` → ``\\r\\n`` translation
+  native Windows applies to text-mode handles.
+"""
+import importlib
+import os
+import sqlite3
+import sys
+from contextlib import closing
+
+import pytest
+
+import api.agent_sessions as agent_sessions
+
+
+# --------------------------------------------------------------------------
+# 1. Missing-index projections use the pre-aggregated candidate ordering.
+# --------------------------------------------------------------------------
+
+def _make_state_db(path, *, sessions, source, with_index):
+    with closing(sqlite3.connect(str(path))) as conn:
+        conn.executescript(
+            """
+            CREATE TABLE sessions(
+                id TEXT PRIMARY KEY, title TEXT, model TEXT, message_count INTEGER,
+                started_at REAL, source TEXT, session_source TEXT,
+                parent_session_id TEXT, ended_at REAL, end_reason TEXT);
+            CREATE TABLE messages(
+                id INTEGER PRIMARY KEY, session_id TEXT, role TEXT,
+                content TEXT, timestamp REAL);
+            """
+        )
+        for i in range(sessions):
+            sid = f"{source}-{i:04d}"
+            conn.execute(
+                "INSERT INTO sessions VALUES (?,?,?,?,?,?,?,NULL,NULL,NULL)",
+                (sid, f"Session {i}", "m", 3, 1000.0 + i, source, source),
+            )
+            for j in range(3):
+                conn.execute(
+                    "INSERT INTO messages(session_id, role, content, timestamp) VALUES (?,?,?,?)",
+                    (sid, "user" if j == 0 else "assistant", "x", 1000.0 + i + j / 10),
+                )
+        # A resumed old session: started first, but its last message is newest.
+        conn.execute(
+            "INSERT INTO sessions VALUES (?,?,?,?,?,?,?,NULL,NULL,NULL)",
+            (f"{source}-resumed", "Resumed", "m", 2, 1.0, source, source),
+        )
+        conn.execute(
+            "INSERT INTO messages(session_id, role, content, timestamp) VALUES (?,?,?,?)",
+            (f"{source}-resumed", "user", "old", 1.0),
+        )
+        conn.execute(
+            "INSERT INTO messages(session_id, role, content, timestamp) VALUES (?,?,?,?)",
+            (f"{source}-resumed", "assistant", "recent", 5000.0),
+        )
+        if with_index:
+            conn.execute("CREATE INDEX idx_messages_session ON messages(session_id, timestamp)")
+        conn.commit()
+
+
+def _capture_listing_sql(monkeypatch):
+    """Record the SQL statements the listing runs (trace callback, read-only)."""
+    statements = []
+    real_connect = sqlite3.connect
+
+    def tracing_connect(*args, **kwargs):
+        conn = real_connect(*args, **kwargs)
+        conn.set_trace_callback(statements.append)
+        return conn
+
+    monkeypatch.setattr(agent_sessions.sqlite3, "connect", tracing_connect)
+    return statements
+
+
+def _candidate_query_plan(db_path, statements):
+    listing_sql = [s for s in statements if "FROM sessions s" in s]
+    assert len(listing_sql) == 1, listing_sql
+    with closing(sqlite3.connect(f"file:{db_path}?mode=ro", uri=True)) as conn:
+        return [row[3] for row in conn.execute("EXPLAIN QUERY PLAN " + listing_sql[0])]
+
+
+@pytest.mark.parametrize(
+    ("source", "kwargs"),
+    [
+        ("cli", {}),  # default /api/sessions projection
+        ("webhook", {"exclude_sources": None, "include_sources": ("webhook",)}),
+        ("kanban", {"exclude_sources": None, "include_sources": ("kanban",)}),
+        ("cli", {"exclude_sources": None}),  # gateway watcher / diagnostic view
+    ],
+)
+def test_missing_index_projection_uses_preaggregated_ordering(tmp_path, monkeypatch, source, kwargs):
+    indexed = tmp_path / "indexed.db"
+    unindexed = tmp_path / "unindexed.db"
+    _make_state_db(indexed, sessions=60, source=source, with_index=True)
+    _make_state_db(unindexed, sessions=60, source=source, with_index=False)
+
+    expected = agent_sessions.read_importable_agent_session_rows(indexed, limit=20, **kwargs)
+    assert expected[0]["id"] == f"{source}-resumed"
+    assert len(expected) == 20
+
+    statements = _capture_listing_sql(monkeypatch)
+    actual = agent_sessions.read_importable_agent_session_rows(unindexed, limit=20, **kwargs)
+
+    # Same rows, same order as the indexed listing.
+    assert [r["id"] for r in actual] == [r["id"] for r in expected]
+    assert [r["actual_message_count"] for r in actual] == [r["actual_message_count"] for r in expected]
+    assert [r["last_activity"] for r in actual] == [r["last_activity"] for r in expected]
+
+    plan = _candidate_query_plan(unindexed, statements)
+    # One pre-aggregation pass over messages, never a per-candidate rescan.
+    assert not any("CORRELATED SCALAR SUBQUERY" in step for step in plan), plan
+    assert any("latest_messages" in step for step in plan), plan
+    # The listing stayed a pure read: no index appeared.
+    with closing(sqlite3.connect(str(unindexed))) as conn:
+        assert conn.execute("PRAGMA index_list(messages)").fetchall() == []
+
+
+def test_indexed_projection_keeps_the_correlated_index_seek(tmp_path, monkeypatch):
+    """With ``idx_messages_session`` present the indexed seek stays in place."""
+    db = tmp_path / "state.db"
+    _make_state_db(db, sessions=10, source="cli", with_index=True)
+    statements = _capture_listing_sql(monkeypatch)
+    rows = agent_sessions.read_importable_agent_session_rows(db, limit=20)
+    assert rows[0]["id"] == "cli-resumed"
+    plan = _candidate_query_plan(db, statements)
+    assert not any("latest_messages" in step for step in plan), plan
+    assert any("idx_messages_session" in step for step in plan), plan
+
+
+# --------------------------------------------------------------------------
+# 2. Read-only URI construction is platform-aware and strictly mode=ro.
+# --------------------------------------------------------------------------
+
+@pytest.mark.parametrize(
+    ("platform", "path", "expected"),
+    [
+        ("linux", "/home/me/.hermes/state.db", "file:///home/me/.hermes/state.db?mode=ro"),
+        ("darwin", "/Users/me/state dir #1/state?.db",
+         "file:///Users/me/state%20dir%20%231/state%3F.db?mode=ro"),
+        ("win32", r"C:\Users\me\.hermes\state.db", "file:///C:/Users/me/.hermes/state.db?mode=ro"),
+        ("win32", r"C:\a b\state#?.db", "file:///C:/a%20b/state%23%3F.db?mode=ro"),
+        ("win32", r"\\server\share\hermes\state.db", "file:////server/share/hermes/state.db?mode=ro"),
+        ("win32", r"\\server\share\a b\state#.db", "file:////server/share/a%20b/state%23.db?mode=ro"),
+        ("win32", r"\\?\C:\x\state.db", "file:///C:/x/state.db?mode=ro"),
+        ("win32", r"\\?\UNC\server\share\state.db", "file:////server/share/state.db?mode=ro"),
+    ],
+)
+def test_state_db_readonly_uri_by_platform(platform, path, expected):
+    assert agent_sessions.state_db_readonly_uri(path, platform=platform) == expected
+
+
+def test_state_db_readonly_uri_never_carries_a_host_authority():
+    uri = agent_sessions.state_db_readonly_uri(r"\\server\share\state.db", platform="win32")
+    assert uri.startswith("file:////")
+    assert "file://server" not in uri
+    assert uri.endswith("?mode=ro")
+
+
+def test_open_state_db_readonly_uses_platform_uri_builder(tmp_path, monkeypatch):
+    path = tmp_path / "state.db"
+    with closing(sqlite3.connect(str(path))) as conn:
+        conn.execute("CREATE TABLE t(x)")
+    calls = []
+    real_connect = sqlite3.connect
+
+    def recording(target, *args, **kwargs):
+        calls.append((str(target), kwargs))
+        return real_connect(target, *args, **kwargs)
+
+    monkeypatch.setattr(agent_sessions.sqlite3, "connect", recording)
+    with closing(agent_sessions.open_state_db_readonly(path)) as conn:
+        assert conn.execute("SELECT COUNT(*) FROM t").fetchone() == (0,)
+    assert calls == [(agent_sessions.state_db_readonly_uri(path.resolve()), {"uri": True})]
+    assert calls[0][0].endswith("?mode=ro")
+
+
+def test_empty_authority_uri_opens_on_this_platform(tmp_path):
+    """SQLite accepts the empty-authority form the UNC path relies on."""
+    path = tmp_path / "state.db"
+    with closing(sqlite3.connect(str(path))) as conn:
+        conn.execute("CREATE TABLE t(x)")
+    uri = agent_sessions.state_db_readonly_uri(path.resolve())
+    with closing(sqlite3.connect(uri, uri=True)) as conn:
+        assert conn.execute("SELECT COUNT(*) FROM t").fetchone() == (0,)
+        with pytest.raises(sqlite3.OperationalError):
+            conn.execute("INSERT INTO t VALUES (1)")
+
+
+# --------------------------------------------------------------------------
+# 3. The index maintenance script imports and runs without fcntl.
+# --------------------------------------------------------------------------
+
+def _maintenance_schema(path):
+    with closing(sqlite3.connect(str(path))) as conn:
+        conn.executescript(
+            """
+            CREATE TABLE sessions(source TEXT, id TEXT, message_count INTEGER, last_activity_at REAL);
+            CREATE TABLE messages(session_id TEXT, timestamp REAL, role TEXT);
+            """
+        )
+
+
+@pytest.fixture
+def maintenance_without_fcntl(monkeypatch):
+    """Import the maintenance module as a platform without ``fcntl`` sees it."""
+    name = "scripts.ensure_state_db_read_indexes"
+    monkeypatch.setitem(sys.modules, "fcntl", None)
+    monkeypatch.delitem(sys.modules, name, raising=False)  # restores a prior real import
+    module = importlib.import_module(name)
+    try:
+        yield module
+    finally:
+        # Drop the fcntl-less copy before monkeypatch restores the original entry.
+        if sys.modules.get(name) is module:
+            del sys.modules[name]
+
+
+def test_maintenance_imports_and_runs_without_fcntl(tmp_path, maintenance_without_fcntl):
+    module = maintenance_without_fcntl
+    path = tmp_path / "state.db"
+    _maintenance_schema(path)
+    result = module.ensure_read_indexes(path, confirmed_drained=True)
+    assert set(result.values()) == {"created"}
+    with closing(sqlite3.connect(str(path))) as conn:
+        names = {row[1] for row in conn.execute("PRAGMA index_list(messages)")}
+    assert "idx_messages_session" in names
+
+
+def test_maintenance_refuses_lock_file_without_a_lock_primitive(tmp_path, maintenance_without_fcntl, monkeypatch):
+    module = maintenance_without_fcntl
+    monkeypatch.setitem(sys.modules, "msvcrt", None)
+    path = tmp_path / "state.db"
+    _maintenance_schema(path)
+    with pytest.raises(RuntimeError, match="--lock-file"):
+        module.ensure_read_indexes(path, confirmed_drained=True, lock_file=tmp_path / "turns.lock")
+    with closing(sqlite3.connect(str(path))) as conn:
+        assert conn.execute("PRAGMA index_list(messages)").fetchall() == []
+
+
+def test_maintenance_cli_without_lock_file_works_without_fcntl(tmp_path, maintenance_without_fcntl, monkeypatch, capsys):
+    module = maintenance_without_fcntl
+    path = tmp_path / "state.db"
+    _maintenance_schema(path)
+    monkeypatch.setattr(sys, "argv", ["ensure_state_db_read_indexes.py", "--db", str(path), "--confirm-drained"])
+    module.main()
+    out = capsys.readouterr().out
+    assert '"idx_messages_session": "created"' in out
+
+
+# --------------------------------------------------------------------------
+# 4. The Windows lock branch acquires a fresh (empty) lock file.
+# --------------------------------------------------------------------------
+
+class _StrictMsvcrt:
+    """Stand-in for ``msvcrt`` that, like Windows, refuses to lock past EOF.
+
+    Records every ``locking`` call with the file size at the current offset so
+    the test can prove byte 0 existed when the lock was taken.
+    """
+
+    LK_NBLCK = 2
+    LK_UNLCK = 0
+
+    def __init__(self):
+        self.calls = []
+
+    def locking(self, fd, mode, nbytes):
+        offset = os.lseek(fd, 0, os.SEEK_CUR)
+        size = os.fstat(fd).st_size
+        self.calls.append((mode, offset, nbytes, size))
+        if mode == self.LK_NBLCK and offset + nbytes > size:
+            raise OSError(36, "Resource deadlock avoided: lock region beyond end of file")
+
+
+@pytest.fixture
+def windows_text_mode(maintenance_without_fcntl, monkeypatch):
+    """Make the tool's text-mode ``open`` translate ``\\n`` to ``\\r\\n`` as on native Windows."""
+    real_open = open
+
+    def crlf_text_open(file, mode="r", *args, **kwargs):
+        if "b" not in mode:
+            kwargs.setdefault("newline", "\r\n")
+        return real_open(file, mode, *args, **kwargs)
+
+    monkeypatch.setattr(maintenance_without_fcntl, "open", crlf_text_open, raising=False)
+
+
+def test_windows_lock_branch_acquires_a_fresh_empty_lock_file(tmp_path, maintenance_without_fcntl, windows_text_mode, monkeypatch):
+    module = maintenance_without_fcntl
+    assert module.fcntl is None  # the fixture already routes the tool to the msvcrt branch
+    fake = _StrictMsvcrt()
+    monkeypatch.setattr(module, "msvcrt", fake)
+
+    path = tmp_path / "state.db"
+    _maintenance_schema(path)
+    lock = tmp_path / "turns.lock"
+    assert not lock.exists()
+
+    result = module.ensure_read_indexes(path, confirmed_drained=True, lock_file=lock)
+
+    assert set(result.values()) == {"created"}
+    # Exactly one non-blocking lock on byte 0, taken while that byte existed
+    # and was the only one: no newline translation may inflate the file.
+    assert fake.calls == [(fake.LK_NBLCK, 0, 1, 1)]
+    assert lock.stat().st_size == 1
+    with closing(sqlite3.connect(str(path))) as conn:
+        names = {row[1] for row in conn.execute("PRAGMA index_list(messages)")}
+    assert "idx_messages_session" in names
+
+
+def test_windows_lock_branch_keeps_an_existing_lock_file_intact(tmp_path, maintenance_without_fcntl, monkeypatch):
+    module = maintenance_without_fcntl
+    fake = _StrictMsvcrt()
+    monkeypatch.setattr(module, "msvcrt", fake)
+
+    path = tmp_path / "state.db"
+    _maintenance_schema(path)
+    lock = tmp_path / "turns.lock"
+    lock.write_bytes(b"deployment-owned\n")
+
+    module.ensure_read_indexes(path, confirmed_drained=True, lock_file=lock)
+
+    assert fake.calls == [(fake.LK_NBLCK, 0, 1, len(b"deployment-owned\n"))]
+    assert lock.read_bytes() == b"deployment-owned\n"
+
+
+def test_windows_lock_branch_reports_a_held_lock_without_touching_the_db(tmp_path, maintenance_without_fcntl, windows_text_mode, monkeypatch):
+    module = maintenance_without_fcntl
+
+    class _HeldMsvcrt(_StrictMsvcrt):
+        def locking(self, fd, mode, nbytes):
+            super().locking(fd, mode, nbytes)
+            raise PermissionError(13, "Permission denied: lock held")
+
+    fake = _HeldMsvcrt()
+    monkeypatch.setattr(module, "msvcrt", fake)
+    path = tmp_path / "state.db"
+    _maintenance_schema(path)
+    lock = tmp_path / "turns.lock"
+
+    with pytest.raises(PermissionError):
+        module.ensure_read_indexes(path, confirmed_drained=True, lock_file=lock)
+
+    assert fake.calls == [(fake.LK_NBLCK, 0, 1, 1)]
+    with closing(sqlite3.connect(str(path))) as conn:
+        assert conn.execute("PRAGMA index_list(messages)").fetchall() == []
diff --git a/tests/test_state_db_readonly_readers.py b/tests/test_state_db_readonly_readers.py
new file mode 100644
index 00000000..c7d1c7a0
--- /dev/null
+++ b/tests/test_state_db_readonly_readers.py
@@ -0,0 +1,124 @@
+"""state.db readers stay strictly read-only and idle watchers do not poll.
+
+Contract under test:
+
+* ``open_state_db_readonly`` opens ``file:...?mode=ro`` once and propagates a
+  failure — it never retries with a write-capable handle;
+* the gateway watcher does not touch the database while nobody is subscribed;
+* the session listing never runs ``CREATE INDEX`` (a writer lock held for
+  minutes on a large ``messages`` table) — a missing index degrades instead;
+* index maintenance is an explicit drained operation
+  (``scripts/ensure_state_db_read_indexes.py``);
+* the cron sidebar read in ``api/routes.py`` uses the read-only opener.
+"""
+import sqlite3
+from contextlib import closing
+
+import pytest
+
+
+def test_readonly_open_failure_never_retries_writable(tmp_path, monkeypatch):
+    from api import agent_sessions
+    path = tmp_path / "state.db"
+    path.touch()
+    calls = []
+
+    def connect(*args, **kwargs):
+        calls.append((args, kwargs))
+        raise sqlite3.OperationalError("readonly unavailable")
+
+    monkeypatch.setattr(agent_sessions.sqlite3, "connect", connect)
+    with pytest.raises(sqlite3.OperationalError):
+        agent_sessions.open_state_db_readonly(path)
+    assert len(calls) == 1
+    assert calls[0][1] == {"uri": True}
+
+
+def test_no_subscribers_means_no_database_poll(tmp_path, monkeypatch):
+    from api import gateway_watcher as watcher
+    path = tmp_path / "state.db"
+    path.touch()
+    instance = watcher.GatewayWatcher(hermes_home=tmp_path)
+    monkeypatch.setattr(watcher, "_cheap_change_fingerprint", lambda *a: pytest.fail("polled DB"))
+    monkeypatch.setattr(watcher, "_get_agent_sessions_from_db", lambda *a: pytest.fail("projected payload"))
+    assert instance._poll_once(now=500) is False
+
+
+def test_listing_does_not_create_indexes(tmp_path):
+    from api.agent_sessions import read_importable_agent_session_rows
+    path = tmp_path / "state.db"
+    with closing(sqlite3.connect(path)) as db:
+        db.executescript("""
+        CREATE TABLE sessions(id TEXT PRIMARY KEY, title TEXT, model TEXT,
+          message_count INTEGER, started_at REAL, source TEXT);
+        CREATE TABLE messages(id INTEGER PRIMARY KEY, session_id TEXT, timestamp REAL, role TEXT);
+        INSERT INTO sessions VALUES ('one', 'Test', 'test', 2, 1, 'cli');
+        INSERT INTO messages VALUES (1, 'one', 2, 'user');
+        INSERT INTO messages VALUES (2, 'one', 3, 'assistant');
+        """)
+    rows = read_importable_agent_session_rows(path)
+    assert rows[0]["id"] == "one"
+    with closing(sqlite3.connect(path)) as db:
+        assert db.execute("PRAGMA index_list(messages)").fetchall() == []
+
+
+def _maintenance_schema(path):
+    with closing(sqlite3.connect(path)) as db:
+        db.executescript("""
+        CREATE TABLE sessions(source TEXT, id TEXT, message_count INTEGER, last_activity_at REAL);
+        CREATE TABLE messages(session_id TEXT, timestamp REAL, role TEXT);
+        """)
+
+
+def test_index_maintenance_requires_drain_and_is_idempotent(tmp_path):
+    from scripts.ensure_state_db_read_indexes import ensure_read_indexes
+    path = tmp_path / "state.db"
+    _maintenance_schema(path)
+    with pytest.raises(RuntimeError, match="confirm-drained"):
+        ensure_read_indexes(path)
+    lock = tmp_path / "turns.lock"
+    first = ensure_read_indexes(path, confirmed_drained=True, lock_file=lock)
+    second = ensure_read_indexes(path, confirmed_drained=True, lock_file=lock)
+    assert set(first.values()) == {"created"}
+    assert set(second.values()) == {"existing"}
+    assert "idx_messages_session_role" in first
+
+
+def test_index_maintenance_without_lock_file_still_requires_drain(tmp_path):
+    from scripts.ensure_state_db_read_indexes import ensure_read_indexes
+    path = tmp_path / "state.db"
+    _maintenance_schema(path)
+    result = ensure_read_indexes(path, confirmed_drained=True)
+    assert set(result.values()) == {"created"}
+
+
+def test_index_maintenance_refuses_when_lock_is_held(tmp_path):
+    fcntl = pytest.importorskip("fcntl", reason="flock-held scenario needs a POSIX interpreter")
+    from scripts.ensure_state_db_read_indexes import ensure_read_indexes
+    path = tmp_path / "state.db"
+    _maintenance_schema(path)
+    lock = tmp_path / "turns.lock"
+    with open(lock, "a+") as holder:
+        fcntl.flock(holder, fcntl.LOCK_EX | fcntl.LOCK_NB)
+        with pytest.raises(BlockingIOError):
+            ensure_read_indexes(path, confirmed_drained=True, lock_file=lock)
+    with closing(sqlite3.connect(path)) as db:
+        assert db.execute("PRAGMA index_list(messages)").fetchall() == []
+
+
+def test_cron_sidebar_read_uses_readonly_connection(tmp_path, monkeypatch):
+    from api import routes
+    path = tmp_path / 'state.db'
+    with closing(sqlite3.connect(path)) as db:
+        db.execute('CREATE TABLE sessions(id TEXT, source TEXT)')
+    monkeypatch.setattr(routes, '_active_state_db_path', lambda: path)
+    connect = sqlite3.connect
+    calls = []
+    def recording(*args, **kwargs):
+        calls.append((args, kwargs))
+        return connect(*args, **kwargs)
+    monkeypatch.setattr(sqlite3, 'connect', recording)
+    routes._latest_cron_session_info_for_jobs(['test'])
+    assert len(calls) == 1
+    assert 'mode=ro' in calls[0][0][0]
+    assert calls[0][1].get('uri') is True