From ef9341b2ab1e59ae84cd3437a24cca00c7bdefbd Mon Sep 17 00:00:00 2001 From: Anthony Ruiz Date: Sun, 6 Sep 2026 14:12:12 +0000 Subject: [PATCH] fix(sqlite): keep state.db readers strictly read-only and stop idle polling MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Readers never upgrade to writers. open_state_db_readonly() opens file:...?mode=ro only; a read-only open failure propagates instead of silently retrying with a write-capable handle on the multi-GiB WAL database the agent is streaming into. The session listing no longer self-heals idx_messages_session with CREATE INDEX (a writer lock held for minutes on a large messages table) — a missing index degrades to the pre-aggregated path. Cron sidebar info, insights and deep health checks in api/routes.py use the same read-only opener. Watcher (api/gateway_watcher.py): _poll_once returns without touching the database when nobody is subscribed, and forces a fresh full projection on the next subscription. Maintenance (scripts/ensure_state_db_read_indexes.py): the covering read indexes (idx_messages_session, idx_messages_session_role with NOCASE, idx_sessions_webui_fingerprint) are created only under --confirm-drained, optionally holding an exclusive flock on --lock-file for deployments that serialise agent turns on a lock file; existing indexes are verified for shape/collation and coverage (EXPLAIN QUERY PLAN), never silently replaced. Tests: tests/test_state_db_readonly_readers.py (read-only failure never retries writable; no poll without subscribers; listing creates no index; index tool requires drain, is idempotent, works without a lock file and refuses a held lock; cron sidebar read is mode=ro). Neighbouring tests that asserted the retired writable fallback / self-heal writer (test_issue3887_index_prime, test_issue5455_listing_readonly_connection, test_issue5455_lineage_readonly_reads, test_issue2628_cli_sessions_perf, test_issue1494_state_db_fd_leak) now assert the read-only contract; test_issue3506 subscribes before polling. --- api/agent_sessions.py | 57 +------- api/gateway_watcher.py | 6 + api/routes.py | 7 +- scripts/ensure_state_db_read_indexes.py | 98 ++++++++++++++ tests/test_issue1494_state_db_fd_leak.py | 6 +- tests/test_issue2628_cli_sessions_perf.py | 3 +- tests/test_issue3506_memory_and_watcher.py | 1 + tests/test_issue3887_index_prime.py | 34 +++-- .../test_issue5455_lineage_readonly_reads.py | 27 ++-- ...t_issue5455_listing_readonly_connection.py | 45 ++++--- tests/test_state_db_readonly_readers.py | 124 ++++++++++++++++++ 11 files changed, 303 insertions(+), 105 deletions(-) create mode 100644 scripts/ensure_state_db_read_indexes.py create mode 100644 tests/test_state_db_readonly_readers.py diff --git a/api/agent_sessions.py b/api/agent_sessions.py index 97086795..310df64d 100644 --- a/api/agent_sessions.py +++ b/api/agent_sessions.py @@ -24,30 +24,17 @@ def open_state_db_readonly(db_path: Path, log: logging.Logger | None = None) -> 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(read_only_uri, uri=True) MESSAGING_SOURCES = { @@ -549,21 +536,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() @@ -613,11 +586,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: @@ -625,21 +594,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})" diff --git a/api/gateway_watcher.py b/api/gateway_watcher.py index b52f5b0c..0354c337 100644 --- a/api/gateway_watcher.py +++ b/api/gateway_watcher.py @@ -317,6 +317,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 7d30ab90..0804cb2e 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)") @@ -11897,7 +11898,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; @@ -12532,7 +12533,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/scripts/ensure_state_db_read_indexes.py b/scripts/ensure_state_db_read_indexes.py new file mode 100644 index 00000000..87aad2cd --- /dev/null +++ b/scripts/ensure_state_db_read_indexes.py @@ -0,0 +1,98 @@ +#!/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 ``flock`` on that path so that a deployment +which serialises agent turns on a lock file cannot start a turn mid-rebuild. + +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 fcntl +import json +from pathlib import Path +import sqlite3 + +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() + handle = open(lock_file, "a+") + try: + fcntl.flock(handle, fcntl.LOCK_EX | fcntl.LOCK_NB) + 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): + with closing(sqlite3.connect(db_path.as_uri() + "?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..90dbaf2c 100644 --- a/tests/test_issue2628_cli_sessions_perf.py +++ b/tests/test_issue2628_cli_sessions_perf.py @@ -212,7 +212,8 @@ def test_importable_agent_rows_push_sidebar_limit_into_sql(tmp_path): 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 + # Listing is a pure read: index creation lives in the drained maintenance tool. + assert "CREATE INDEX IF NOT EXISTS idx_messages_session" not 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 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 fc2cb263..06bdaa9e 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 @@ -92,24 +91,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): @@ -124,7 +126,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) @@ -133,12 +135,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_state_db_readonly_readers.py b/tests/test_state_db_readonly_readers.py new file mode 100644 index 00000000..25dd8d20 --- /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 fcntl +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): + 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