Merge branch 'master' into feat/update-channels

CHANGELOG.md

@@ -5,6 +5,12 @@
 
 ### Fixed
 
+- **The background reaper and drain daemon threads can no longer be started twice under a race.** `start_session_channel_reaper` and `start_drain_thread` used a check-then-act (`is_alive()` then create) with no lock, so two concurrent callers could both see "not running" and each spawn a thread — the loser's thread was never referenced by the module and ran forever, un-joinable. Both start paths now perform the liveness check, thread creation, and global assignment atomically under a dedicated lifecycle lock. Stop/join still run outside the lock, so there's no deadlock. Thanks @ai-ag2026. (#5783)
+
+- **A long-running server no longer slowly leaks memory from the background-task completion-dedup map.** Each session that ever had a background task complete (delegation-completion delivery, kanban events) left a permanent entry in an in-memory dedup map — including headless completions where no browser tab was ever open — so the map grew unbounded over a long-lived server's lifetime. The session-channel reaper now sweeps that map by delivery lifecycle (dropping a session's entry once its completion has been delivered and is no longer pending), and session deletion prunes any still-pending entry. Exactly-once delivery is unaffected — it's independently guaranteed by the process registry's per-task consumed marker. Thanks @ai-ag2026. (#5781, #4633)
+
+- **Session-lineage reads and the gateway session-list watcher now open the agent's `state.db` read-only.** Reading session lineage/parent-child metadata and the watcher's cheap change-fingerprint opened a *writable* SQLite handle on the live, potentially multi-GB WAL `state.db` while the agent was streaming into it — adding needless checkpoint/lock surface to a pure read. These three read paths now open a read-only (`mode=ro`) connection, falling back to a writable one (with a warning) only if the read-only open fails on an exotic filesystem, and raising rather than letting the fallback create an empty "ghost" `state.db` at a missing path. Same rationale as the #5455 session-listing read-only path. Thanks @ai-ag2026. (#5780, #5455)
+
 - **The MCP servers/tools list now reflects the active profile instead of the default one.** When viewing MCP servers or tools while a non-default profile was active, the list was read from the ambient (default) profile's config rather than the profile you were actually in, so it could show the wrong profile's MCP servers. Both the server-list and tool-list reads are now scoped to the active profile's config home, matching where the MCP write handlers save. Deployments that point config at an external path via `HERMES_CONFIG_PATH` now also read that override consistently on the active-home path. Thanks @rodboev. (#5772, #5619)
 
 - **Mobile/touch: the chat no longer jumps backward while a reply streams and you're scrolled up reading history (non-virtualized transcripts).** Companion to the desktop and virtualized fixes (#5637/#5638/#5742) for the one code path they didn't cover: with transcript virtualization turned off (#4325 opt-out), a re-render rebuilt every row from scratch, and a fresh off-screen tall user row under `content-visibility: auto` (active on coarse-pointer devices) reserved only its flat placeholder height instead of its real height — so `scrollHeight` shrank, the browser clamped `scrollTop`, and the viewport jumped toward the top. The rebuild now reserves each user row's real height, measured from the still-laid-out rows just before the wipe (with a CJK-aware estimate as the backstop for never-painted rows and a `max(remembered, estimate)` floor so a partial paint can't under-reserve). Desktop is unchanged (the placeholder is inert there). Thanks @allenliang2022. (#5751, #5744)

api/agent_sessions.py

@@ -7,6 +7,38 @@ from pathlib import Path
 logger = logging.getLogger(__name__)
 
 
+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.
+
+    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.
+
+    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))
+
+
 MESSAGING_SOURCES = {
     'discord',
     'email',
@@ -756,7 +788,7 @@ def read_session_lineage_report(db_path: Path, session_id: str | None, max_hops:
         return _empty_lineage_report(sid)
 
     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)")
@@ -895,7 +927,7 @@ def read_session_lineage_metadata(db_path: Path, session_ids: list[str] | set[st
         return {}
 
     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)")

api/background_process.py

@@ -57,6 +57,14 @@ _REAPER_THREAD: Optional[threading.Thread] = None
 _REAPER_STOP = threading.Event()
 _REAPER_INTERVAL_SECS = 60.0
 
+# Serializes the check-then-start of the module's daemon threads
+# (``start_drain_thread`` / ``start_session_channel_reaper``). Without it two
+# concurrent callers can both observe ``is_alive() == False`` and each spawn a
+# thread; the loser's thread is never referenced by the module global and runs
+# forever, un-joinable. A dedicated lock (not the purpose-bound
+# ``SESSION_CHANNELS_LOCK`` / ``_EMIT_COALESCE_LOCK``) keeps this narrow.
+_THREAD_LIFECYCLE_LOCK = threading.Lock()
+
 # T3: per-session coalesce gate for the public bg_task_complete SSE emit.
 # The server-side wakeup path remains immediate; only the browser-observation
 # frame is throttled so a burst of background task completions does not flood an
@@ -377,6 +385,32 @@ def _reaper_loop() -> None:
                     for sid in collected:
                         _LAST_EMIT_TS.pop(sid, None)
                 logger.debug("SessionChannel reaper collected: %s", collected)
+            # Sweep the per-session completion-dedup map by DELIVERY lifecycle,
+            # not channel collection. ``BG_TASK_COMPLETE_EVENTS_SEEN`` gains a
+            # ``session_id -> set[process_id]`` entry the first time a bg task
+            # completes for a session — in ``_process_one``, whether or not any
+            # tab/SSE channel ever existed — and is otherwise never deleted, so it
+            # grows unbounded. Coupling the prune to channel collection (an
+            # earlier version of this fix) missed the dominant case: a headless
+            # completion (task fires, tab closed or never opened) has no channel
+            # to collect. Instead, once a completion has been drained (its
+            # ``session_id`` removed from ``PENDING_BG_TASK_COMPLETIONS``), the
+            # short ``_move_to_finished`` dedup window is closed and the entry is
+            # pure leak — so sweep every delivered (not-pending) session here,
+            # every tick. The registry's own per-``process_id``
+            # ``_completion_consumed`` gate remains the primary idempotency
+            # backstop, so sweeping a delivered session's set can never resurrect
+            # an already-delivered completion (even in the tiny window between
+            # this module's ``SEEN.add`` and ``PENDING.add`` in ``_process_one``).
+            from api import config as _cfg
+
+            with _cfg.BG_TASK_COMPLETE_EVENTS_SEEN_LOCK:
+                for sid in [
+                    s
+                    for s in _cfg.BG_TASK_COMPLETE_EVENTS_SEEN
+                    if s not in _cfg.PENDING_BG_TASK_COMPLETIONS
+                ]:
+                    _cfg.BG_TASK_COMPLETE_EVENTS_SEEN.pop(sid, None)
         except Exception:
             logger.warning("SessionChannel reaper iteration failed", exc_info=True)
         # Wait but wake up promptly on stop.
@@ -387,16 +421,17 @@ def _reaper_loop() -> None:
 def start_session_channel_reaper() -> bool:
     """Start the SessionChannel reaper thread. Idempotent; returns True on first start."""
     global _REAPER_THREAD
-    if _REAPER_THREAD is not None and _REAPER_THREAD.is_alive():
-        return False
-    _REAPER_STOP.clear()
-    _REAPER_THREAD = threading.Thread(
-        target=_reaper_loop,
-        name="hermes-webui-session-channel-reaper",
-        daemon=True,
-    )
-    _REAPER_THREAD.start()
-    return True
+    with _THREAD_LIFECYCLE_LOCK:
+        if _REAPER_THREAD is not None and _REAPER_THREAD.is_alive():
+            return False
+        _REAPER_STOP.clear()
+        _REAPER_THREAD = threading.Thread(
+            target=_reaper_loop,
+            name="hermes-webui-session-channel-reaper",
+            daemon=True,
+        )
+        _REAPER_THREAD.start()
+        return True
 
 
 def stop_session_channel_reaper(timeout: float = 2.0) -> None:
@@ -1328,19 +1363,35 @@ def unregister_process_session(session_key: str) -> None:
         _cfg.PROCESS_SESSION_INDEX.pop(str(session_key), None)
 
 
+def forget_bg_task_completion_dedup(session_id: str) -> None:
+    """Drop a session's ``BG_TASK_COMPLETE_EVENTS_SEEN`` entry.
+
+    Called on session deletion so a session deleted while a completion is still
+    pending (undelivered) — which the reaper's delivery-gated sweep deliberately
+    keeps — can't leak its dedup set forever. Safe for unknown ids (no-op).
+    """
+    if not session_id:
+        return
+    from api import config as _cfg
+
+    with _cfg.BG_TASK_COMPLETE_EVENTS_SEEN_LOCK:
+        _cfg.BG_TASK_COMPLETE_EVENTS_SEEN.pop(str(session_id), None)
+
+
 def start_drain_thread() -> bool:
     """Start the background drain thread idempotently. Returns True on first start."""
     global _DRAIN_THREAD
-    if _DRAIN_THREAD is not None and _DRAIN_THREAD.is_alive():
-        return False
-    _DRAIN_STOP.clear()
-    _DRAIN_THREAD = threading.Thread(
-        target=_drain_loop,
-        name="hermes-webui-bg-task-complete-drain",
-        daemon=True,
-    )
-    _DRAIN_THREAD.start()
-    return True
+    with _THREAD_LIFECYCLE_LOCK:
+        if _DRAIN_THREAD is not None and _DRAIN_THREAD.is_alive():
+            return False
+        _DRAIN_STOP.clear()
+        _DRAIN_THREAD = threading.Thread(
+            target=_drain_loop,
+            name="hermes-webui-bg-task-complete-drain",
+            daemon=True,
+        )
+        _DRAIN_THREAD.start()
+        return True
 
 
 def stop_drain_thread(timeout: float = 2.0) -> None:

api/gateway_watcher.py

@@ -20,7 +20,7 @@ from contextlib import closing
 from pathlib import Path
 
 from api.config import HOME
-from api.agent_sessions import read_importable_agent_session_rows
+from api.agent_sessions import open_state_db_readonly, read_importable_agent_session_rows
 
 logger = logging.getLogger(__name__)
 
@@ -87,7 +87,7 @@ def _cheap_change_fingerprint(db_path: Path) -> str | None:
         'origin_chat_id', 'origin_user_id', 'platform',
     )
     try:
-        with closing(sqlite3.connect(str(db_path))) as conn:
+        with closing(open_state_db_readonly(db_path)) as conn:
             cur = conn.cursor()
             cur.execute("PRAGMA table_info(sessions)")
             cols = {row[1] for row in cur.fetchall()}

api/routes.py

@@ -13868,6 +13868,15 @@ def handle_post(handler, parsed) -> bool:
         # Lock entries in SESSION_AGENT_LOCKS forever.
         with SESSION_AGENT_LOCKS_LOCK:
             SESSION_AGENT_LOCKS.pop(sid, None)
+        # Prune the completion-dedup entry too. The reaper sweeps it once the
+        # completion is delivered (drained from PENDING); a session deleted
+        # while a completion is still pending would otherwise keep its entry.
+        try:
+            from api.background_process import forget_bg_task_completion_dedup
+
+            forget_bg_task_completion_dedup(sid)
+        except Exception:
+            logger.debug("Failed to prune bg-task dedup entry for deleted session %s", sid)
         try:
             from api.terminal import close_terminal
             close_terminal(sid)

tests/test_bg_dedup_reaper_prune.py

@@ -0,0 +1,122 @@
+"""Regression test for the ``BG_TASK_COMPLETE_EVENTS_SEEN`` memory leak (#4633).
+
+The per-session completion-dedup map gained one ``session_id -> set[process_id]``
+entry the first time a background task completed for a session and was never
+removed anywhere in the repo, so it grew unbounded for the server lifetime.
+
+The dedup entry is created in ``_process_one`` for EVERY completion, whether or
+not any SSE channel/tab exists — so coupling the prune to channel collection
+would miss the dominant headless case (task fires, tab closed or never opened).
+Instead the reaper sweeps the map by DELIVERY lifecycle: once a completion has
+been drained (its ``session_id`` removed from ``PENDING_BG_TASK_COMPLETIONS``),
+the short dedup window is closed and the entry is swept. Session deletion prunes
+it too, covering a session deleted while a completion is still pending.
+"""
+
+from __future__ import annotations
+
+import time
+from pathlib import Path
+
+import pytest
+
+
+REPO_ROOT = Path(__file__).resolve().parents[1]
+
+
+def test_reaper_source_sweeps_dedup_map_by_delivery():
+    """The reaper cleanup must sweep BG_TASK_COMPLETE_EVENTS_SEEN under its lock,
+    gated on PENDING_BG_TASK_COMPLETIONS (delivery), not channel collection."""
+    src = (REPO_ROOT / "api" / "background_process.py").read_text(encoding="utf-8")
+    reaper = src[src.index("def _reaper_loop("):]
+    reaper = reaper[: reaper.index("\ndef ", 1)]
+    assert "BG_TASK_COMPLETE_EVENTS_SEEN_LOCK" in reaper, (
+        "reaper must take the dedup-map lock"
+    )
+    assert "BG_TASK_COMPLETE_EVENTS_SEEN.pop(" in reaper, (
+        "reaper must pop dedup entries"
+    )
+    assert "PENDING_BG_TASK_COMPLETIONS" in reaper, (
+        "the sweep must be gated on delivery (PENDING_BG_TASK_COMPLETIONS), "
+        "not on channel collection — otherwise headless completions leak"
+    )
+
+
+def _run_reaper_once(bp, cfg):
+    """Start the real reaper, wait for one sweep pass, stop it."""
+    bp.start_session_channel_reaper()
+    time.sleep(0.4)  # first pass runs immediately (interval sleep is at loop tail)
+    bp.stop_session_channel_reaper()
+
+
+def test_reaper_sweeps_delivered_dedup_entry_without_any_channel():
+    """The dominant headless case: a completed+drained task with NO SSE channel
+    ever — its dedup entry must still be swept (this is what the channel-coupled
+    version missed)."""
+    from api import background_process as bp
+    from api import config as cfg
+
+    sid = "sess-headless-4633"
+    with cfg.BG_TASK_COMPLETE_EVENTS_SEEN_LOCK:
+        cfg.BG_TASK_COMPLETE_EVENTS_SEEN[sid] = {"proc-1"}
+    cfg.PENDING_BG_TASK_COMPLETIONS.discard(sid)  # delivered/drained
+    # No SessionChannel for sid — nothing for channel collection to reap.
+    try:
+        with bp.SESSION_CHANNELS_LOCK:
+            assert sid not in bp.SESSION_CHANNELS
+        _run_reaper_once(bp, cfg)
+        with cfg.BG_TASK_COMPLETE_EVENTS_SEEN_LOCK:
+            assert sid not in cfg.BG_TASK_COMPLETE_EVENTS_SEEN, (
+                "reaper did not sweep the delivered headless dedup entry"
+            )
+    finally:
+        bp.stop_session_channel_reaper()
+        with cfg.BG_TASK_COMPLETE_EVENTS_SEEN_LOCK:
+            cfg.BG_TASK_COMPLETE_EVENTS_SEEN.pop(sid, None)
+
+
+def test_reaper_retains_dedup_entry_while_completion_pending():
+    """An UNDELIVERED completion (still in PENDING) must keep its dedup entry —
+    the _move_to_finished dedup window is still open."""
+    from api import background_process as bp
+    from api import config as cfg
+
+    sid = "sess-pending-4633"
+    with cfg.BG_TASK_COMPLETE_EVENTS_SEEN_LOCK:
+        cfg.BG_TASK_COMPLETE_EVENTS_SEEN[sid] = {"proc-2"}
+    cfg.PENDING_BG_TASK_COMPLETIONS.add(sid)  # undelivered
+    try:
+        _run_reaper_once(bp, cfg)
+        with cfg.BG_TASK_COMPLETE_EVENTS_SEEN_LOCK:
+            assert sid in cfg.BG_TASK_COMPLETE_EVENTS_SEEN, (
+                "reaper swept a still-pending (undelivered) dedup entry"
+            )
+    finally:
+        bp.stop_session_channel_reaper()
+        cfg.PENDING_BG_TASK_COMPLETIONS.discard(sid)
+        with cfg.BG_TASK_COMPLETE_EVENTS_SEEN_LOCK:
+            cfg.BG_TASK_COMPLETE_EVENTS_SEEN.pop(sid, None)
+
+
+def test_session_delete_prunes_dedup_entry():
+    """Deleting a session prunes its dedup entry even if the completion never
+    delivered (so a delete-while-pending session can't leak forever)."""
+    from api import background_process as bp
+    from api import config as cfg
+
+    sid = "sess-delete-4633"
+    with cfg.BG_TASK_COMPLETE_EVENTS_SEEN_LOCK:
+        cfg.BG_TASK_COMPLETE_EVENTS_SEEN[sid] = {"proc-3"}
+    cfg.PENDING_BG_TASK_COMPLETIONS.add(sid)  # undelivered, but being deleted
+    try:
+        bp.forget_bg_task_completion_dedup(sid)
+        with cfg.BG_TASK_COMPLETE_EVENTS_SEEN_LOCK:
+            assert sid not in cfg.BG_TASK_COMPLETE_EVENTS_SEEN
+    finally:
+        cfg.PENDING_BG_TASK_COMPLETIONS.discard(sid)
+        with cfg.BG_TASK_COMPLETE_EVENTS_SEEN_LOCK:
+            cfg.BG_TASK_COMPLETE_EVENTS_SEEN.pop(sid, None)
+
+
+if __name__ == "__main__":  # pragma: no cover - manual invocation
+    raise SystemExit(pytest.main([__file__, "-v"]))

tests/test_bg_thread_start_lock.py

@@ -0,0 +1,98 @@
+"""Regression test for the thread-start check-then-act race in background_process.
+
+``start_drain_thread`` and ``start_session_channel_reaper`` checked
+``is_alive()`` and then created + started the daemon thread WITHOUT a lock, so
+two concurrent callers could both observe "not alive" and each spawn a thread.
+The loser's thread was never stored in the module global and ran forever,
+un-joinable. Both check-then-start sequences are now serialized under
+``_THREAD_LIFECYCLE_LOCK``, so exactly one thread is ever created.
+
+The test drives N callers through a barrier so they hit the check-and-start
+simultaneously, and asserts exactly one caller reports "started" (returns True)
+and exactly one live thread exists.
+"""
+
+from __future__ import annotations
+
+import threading
+
+import pytest
+
+
+# (start_fn, stop_fn, loop_attr, thread_attr, thread_name_fragment)
+_CASES = [
+    (
+        "start_drain_thread",
+        "stop_drain_thread",
+        "_drain_loop",
+        "_DRAIN_THREAD",
+        "bg-task-complete-drain",
+    ),
+    (
+        "start_session_channel_reaper",
+        "stop_session_channel_reaper",
+        "_reaper_loop",
+        "_REAPER_THREAD",
+        "session-channel-reaper",
+    ),
+]
+
+
+@pytest.mark.parametrize(
+    "start_name,stop_name,loop_attr,thread_attr,name_frag", _CASES
+)
+def test_concurrent_start_creates_exactly_one_thread(
+    monkeypatch, start_name, stop_name, loop_attr, thread_attr, name_frag
+):
+    from api import background_process as bp
+
+    release = threading.Event()
+
+    def _blocking_loop() -> None:
+        # Keep the started thread alive during the assertions so a genuinely
+        # started thread reads as is_alive()==True deterministically.
+        release.wait(5.0)
+
+    monkeypatch.setattr(bp, loop_attr, _blocking_loop, raising=True)
+    # Fresh slate: no pre-existing thread reference.
+    monkeypatch.setattr(bp, thread_attr, None, raising=True)
+
+    start_fn = getattr(bp, start_name)
+    n = 12
+    barrier = threading.Barrier(n)
+    results: list[bool] = []
+    results_lock = threading.Lock()
+
+    def _caller() -> None:
+        barrier.wait()
+        started = start_fn()
+        with results_lock:
+            results.append(started)
+
+    callers = [threading.Thread(target=_caller) for _ in range(n)]
+    try:
+        for c in callers:
+            c.start()
+        for c in callers:
+            c.join(timeout=5.0)
+
+        # Exactly one caller won the create; the rest saw a live thread.
+        assert sum(1 for r in results if r) == 1, (
+            f"expected exactly one start to win, got {results}"
+        )
+        # Exactly one live daemon thread carrying this loop's name.
+        live = [
+            t
+            for t in threading.enumerate()
+            if name_frag in t.name and t.is_alive()
+        ]
+        assert len(live) == 1, f"expected 1 live {name_frag} thread, got {len(live)}"
+        # The module global points at that one live thread.
+        assert getattr(bp, thread_attr) is live[0]
+    finally:
+        release.set()
+        getattr(bp, stop_name)()
+
+
+if __name__ == "__main__":  # pragma: no cover - manual invocation
+    raise SystemExit(pytest.main([__file__, "-v"]))

tests/test_issue5455_lineage_readonly_reads.py

@@ -0,0 +1,173 @@
+"""Regression tests for #5455 — extend read-only state.db opens to the
+remaining pure-read projections.
+
+The session-listing path already opens the live agent ``state.db`` read-only
+(``file:...?mode=ro``) so a write-capable handle doesn't add checkpoint/lock
+surface while the agent streams into the same WAL DB
+(see ``test_issue5455_listing_readonly_connection``). The lineage-report and
+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
+
+import pytest
+
+import api.agent_sessions as agent_sessions
+from api.agent_sessions import (
+    open_state_db_readonly,
+    read_session_lineage_metadata,
+    read_session_lineage_report,
+)
+
+
+def _make_lineage_db(path):
+    conn = sqlite3.connect(str(path))
+    conn.execute(
+        """
+        CREATE TABLE sessions (
+            id TEXT PRIMARY KEY, parent_session_id TEXT, end_reason TEXT,
+            source TEXT, session_source TEXT, title TEXT,
+            started_at REAL, ended_at REAL
+        )
+        """
+    )
+    conn.executemany(
+        "INSERT INTO sessions (id, parent_session_id, end_reason, source, session_source, "
+        "title, started_at, ended_at) VALUES (?,?,?,?,?,?,?,?)",
+        [
+            ("parent-1", None, "compressed", "cli", "cli", "Root", 1000.0, 1100.0),
+            ("child-1", "parent-1", None, "cli", "cli", "Cont", 1100.0, None),
+        ],
+    )
+    conn.commit()
+    conn.close()
+
+
+def _record_connects(monkeypatch):
+    """Spy on agent_sessions.sqlite3.connect, recording target + uri per call."""
+    real_connect = sqlite3.connect
+    calls = []
+
+    def spy(target, *args, **kwargs):
+        calls.append({"target": str(target), "uri": bool(kwargs.get("uri"))})
+        return real_connect(target, *args, **kwargs)
+
+    monkeypatch.setattr(agent_sessions.sqlite3, "connect", spy)
+    return calls
+
+
+# ── open_state_db_readonly (the shared helper) ───────────────────────────────
+
+def test_helper_opens_read_only_uri(tmp_path, monkeypatch):
+    db = tmp_path / "state.db"
+    _make_lineage_db(db)
+    calls = _record_connects(monkeypatch)
+
+    with closing(open_state_db_readonly(db)) as conn:
+        assert conn.execute("SELECT COUNT(*) FROM sessions").fetchone()[0] == 2
+
+    assert calls and calls[0]["uri"] is True
+    assert "mode=ro" in calls[0]["target"]
+
+
+def test_helper_connection_rejects_writes(tmp_path):
+    db = tmp_path / "state.db"
+    _make_lineage_db(db)
+    with closing(open_state_db_readonly(db)) as conn:
+        with pytest.raises(sqlite3.OperationalError):
+            conn.execute("INSERT INTO sessions (id) VALUES ('x')")
+
+
+def test_helper_encodes_special_path_chars(tmp_path, monkeypatch):
+    db_dir = tmp_path / "state dir #1"
+    db_dir.mkdir()
+    db = db_dir / "state?.db"
+    _make_lineage_db(db)
+    calls = _record_connects(monkeypatch)
+
+    with closing(open_state_db_readonly(db)):
+        pass
+
+    assert calls[0]["target"].startswith("file://")
+    assert "%20" in calls[0]["target"]  # space
+    assert "%23" in calls[0]["target"]  # '#'
+    assert "%3F" in calls[0]["target"]  # '?'
+    assert calls[0]["target"].endswith("?mode=ro")
+
+
+def test_helper_falls_back_to_writable_and_logs(tmp_path, monkeypatch, caplog):
+    db = tmp_path / "state.db"
+    _make_lineage_db(db)
+    real_connect = sqlite3.connect
+
+    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)
+
+    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
+
+
+def test_helper_raises_on_missing_db_instead_of_creating_a_ghost(tmp_path):
+    """A missing path must raise FileNotFoundError, not have the writable
+    fallback silently create an empty state.db there."""
+    missing = tmp_path / "does_not_exist" / "state.db"
+    with pytest.raises(FileNotFoundError):
+        open_state_db_readonly(missing)
+    assert not missing.exists(), "helper created a ghost state.db for a missing path"
+
+
+# ── lineage reads route through the helper ───────────────────────────────────
+
+def test_lineage_report_opens_read_only(tmp_path, monkeypatch):
+    db = tmp_path / "state.db"
+    _make_lineage_db(db)
+    calls = _record_connects(monkeypatch)
+
+    report = read_session_lineage_report(db, "child-1")
+
+    assert report  # a non-empty report came back
+    assert calls and calls[0]["uri"] is True
+    assert "mode=ro" in calls[0]["target"]
+
+
+def test_lineage_metadata_opens_read_only(tmp_path, monkeypatch):
+    db = tmp_path / "state.db"
+    _make_lineage_db(db)
+    calls = _record_connects(monkeypatch)
+
+    meta = read_session_lineage_metadata(db, ["child-1", "parent-1"])
+
+    assert isinstance(meta, dict)
+    assert calls and calls[0]["uri"] is True
+    assert "mode=ro" in calls[0]["target"]
+
+
+# ── gateway-watcher fingerprint projection routes through the helper ─────────
+
+def test_gateway_watcher_fingerprint_opens_read_only(tmp_path, monkeypatch):
+    # The 5s watcher poll (_cheap_change_fingerprint) calls open_state_db_readonly,
+    # which resolves sqlite3.connect in agent_sessions' namespace — so the same
+    # spy catches it. The DB needs a `source` column or the fingerprint bails to None.
+    import api.gateway_watcher as gateway_watcher
+
+    db = tmp_path / "state.db"
+    _make_lineage_db(db)
+    calls = _record_connects(monkeypatch)
+
+    fp = gateway_watcher._cheap_change_fingerprint(db)
+
+    assert fp is not None  # a fingerprint (not the schema-bail None) was produced
+    assert calls and calls[0]["uri"] is True
+    assert "mode=ro" in calls[0]["target"]