Merge pull request #7773 from nesquena/stage/7761-watcher-fence

Release: #7761 gateway watcher stale-projection fence (@franksong2702)

CHANGELOG.md

@@ -28,6 +28,10 @@
 
 ### Fixed
 
+- **A stale in-flight projection can no longer reach a gateway watcher after its last subscriber
+  leaves.** Final unsubscribe and queue eviction now invalidate the cache and fence projections that
+  were already in flight, without holding the lock across database reads or SSE writes, so a client
+  that re-subscribes gets a fresh snapshot instead of a stale one. (#7761 by @ruizanthony)
 - **A burst of "session busy" refusals no longer drops a background-task completion.** When
   `start_session_turn()` refused a completion wake-up with a transient 409 (Agent runtime stale,
   process wake-ups paused, or the session busy with another turn), the bridge released the durable

api/gateway_watcher.py

@@ -203,6 +203,8 @@ class GatewayWatcher:
     ):
         self._subscribers: list[queue.Queue] = []
         self._sub_lock = threading.Lock()
+        # Final removal invalidates any projection admitted by an earlier cohort.
+        self._subscriber_epoch = 0
         self._stop_event = threading.Event()
         # Wakes a poll loop parked because nobody is subscribed (subscribe/stop).
         self._idle_wake = threading.Event()
@@ -286,21 +288,32 @@ class GatewayWatcher:
         self._idle_wake.set()
         return q
 
-    def unsubscribe(self, q: queue.Queue):
-        """Remove a subscriber queue."""
-        with self._sub_lock:
-            try:
-                self._subscribers.remove(q)
-            except ValueError:
-                pass
+    def _remove_subscriber_locked(self, q: queue.Queue) -> bool:
+        """Remove a known queue under _sub_lock; fence polls on the last removal."""
+        try:
+            self._subscribers.remove(q)
+        except ValueError:
+            return False
+        if not self._subscribers:
+            self._subscriber_epoch += 1
+            self._last_cheap_fp = ''
+            self._last_full_projection_at = None
+        return True
 
-    def _notify_subscribers(self, sessions: list):
+    def unsubscribe(self, q: queue.Queue):
+        """Remove a subscriber queue, invalidating projection on the last removal."""
+        with self._sub_lock:
+            self._remove_subscriber_locked(q)
+
+    def _notify_subscribers(self, sessions: list, *, epoch: int | None = None):
         """Push change event to all subscribers."""
         event = {
             'type': 'sessions_changed',
             'sessions': sessions,
         }
         with self._sub_lock:
+            if epoch is not None and epoch != self._subscriber_epoch:
+                return  # An old poll must not notify a new subscriber cohort.
             dead = []
             for q in self._subscribers:
                 try:
@@ -310,10 +323,7 @@ class GatewayWatcher:
                 except Exception:
                     dead.append(q)
             for q in dead:
-                try:
-                    self._subscribers.remove(q)
-                except ValueError:
-                    pass
+                self._remove_subscriber_locked(q)
                 # Send a None sentinel so the SSE handler unblocks, closes,
                 # and lets the browser's EventSource auto-reconnect.
                 try:
@@ -330,9 +340,9 @@ class GatewayWatcher:
         """
         with self._sub_lock:
             has_subscribers = bool(self._subscribers)
+            admission_epoch = self._subscriber_epoch
         if not has_subscribers:
-            # Force fresh projection on the next subscription, without touching DB.
-            self._last_full_projection_at = None
+            # Final removal already invalidated the cache under the same lock.
             return False
         db_path = self._state_db_path
         # A watcher may start before the agent has created state.db. Publishing an
@@ -361,14 +371,20 @@ class GatewayWatcher:
         if sessions is None:
             return False
         current_hash = _snapshot_hash(sessions)
-        if cheap_fp is not None:
-            self._last_cheap_fp = cheap_fp
-        self._last_full_projection_at = current_time
-
-        if current_hash != self._last_hash:
-            self._last_hash = current_hash
-            self._last_sessions = sessions
-            self._notify_subscribers(sessions)
+        with self._sub_lock:
+            if admission_epoch != self._subscriber_epoch or not self._subscribers:
+                return False  # Never restore a cache invalidated during the DB read.
+            if cheap_fp is not None:
+                self._last_cheap_fp = cheap_fp
+            self._last_full_projection_at = current_time
+            if current_hash != self._last_hash:
+                changed = True
+                self._last_hash = current_hash
+                self._last_sessions = sessions
+            else:
+                changed = False
+        if changed:
+            self._notify_subscribers(sessions, epoch=admission_epoch)
         return True
 
     def _poll_loop(self):

docs/sse-streams.md

@@ -56,6 +56,21 @@ infer it from the status code. Clients that only need session updates should
 use `/api/session/stream` directly; the gateway probe is only relevant for
 clients that display CLI/TUI/messaging sessions.
 
+## Gateway watcher projection lifecycle
+
+The optional gateway stream's watcher owns a cheap database fingerprint and a
+timestamp for its last full session projection. While clients are subscribed,
+an unchanged fingerprint can skip the expensive read until the bounded parity
+interval. With no subscribers the poll loop parks instead of reading the DB.
+When the final subscriber leaves (including eviction of a full or failed
+consumer queue), the watcher invalidates both cache fields. A reconnect must
+therefore trigger a fresh projection even when the database fingerprint has not
+changed and the parity interval has not elapsed. Final removal also advances a
+subscriber epoch: an older in-flight DB read cannot restore the invalidated
+cache or publish its result to a later subscriber cohort. The subscriber lock
+protects removal, epoch checks and cache commits, but is not held during DB
+reads, so disconnect is not blocked by a slow projection.
+
 ## Heartbeats and proxy behavior
 
 - All long-lived streams emit SSE keepalive comment lines on the

tests/test_gateway_watcher_resubscribe_fresh.py

@@ -0,0 +1,222 @@
+"""Regression: a re-subscribing SSE client gets a fresh watcher projection.
+
+``_poll_once`` resets the parity timestamp when it runs with no subscribers
+("force fresh projection on the next subscription"), but since the idle-park
+change (#7694) the poll loop parks *before* calling ``_poll_once`` once the
+last subscriber leaves, so that reset is not reached on a normal disconnect.
+A tab that reconnects inside the parity interval then resumes with the stale
+cheap fingerprint and skips the projection. Final removal invalidates the
+cache, and an epoch fences projections already in flight.
+"""
+from __future__ import annotations
+
+import importlib
+import sqlite3
+import threading
+import time
+from pathlib import Path
+
+import pytest
+
+
+def _make_db(tmp_path: Path) -> Path:
+    db = tmp_path / "state.db"
+    conn = sqlite3.connect(str(db))
+    conn.executescript(
+        """
+        CREATE TABLE sessions (
+            id TEXT PRIMARY KEY,
+            source TEXT NOT NULL,
+            session_source TEXT,
+            model TEXT,
+            started_at REAL NOT NULL,
+            ended_at REAL,
+            end_reason TEXT,
+            parent_session_id TEXT,
+            message_count INTEGER DEFAULT 0,
+            title TEXT,
+            archived INTEGER DEFAULT 0
+        );
+        CREATE TABLE messages (
+            id INTEGER PRIMARY KEY AUTOINCREMENT,
+            session_id TEXT NOT NULL,
+            role TEXT NOT NULL,
+            content TEXT,
+            timestamp REAL NOT NULL
+        );
+        """
+    )
+    conn.execute(
+        "INSERT INTO sessions (id, source, model, started_at, message_count, title) "
+        "VALUES ('tg1', 'telegram', 'm', ?, 1, 'Chat')",
+        (time.time(),),
+    )
+    conn.execute(
+        "INSERT INTO messages (session_id, role, content, timestamp) VALUES ('tg1', 'user', 'x', ?)",
+        (time.time(),),
+    )
+    conn.commit()
+    conn.close()
+    return db
+
+
+def test_resubscribe_after_last_unsubscribe_forces_fresh_projection(tmp_path):
+    gw = importlib.import_module("api.gateway_watcher")
+    watcher = gw.GatewayWatcher(state_db_path=_make_db(tmp_path))
+    q = watcher.subscribe()
+    assert watcher._poll_once(now=1.0) is True
+
+    watcher.unsubscribe(q)
+    assert watcher._last_cheap_fp == ""
+    assert watcher._last_full_projection_at is None
+
+    watcher.subscribe()
+    # Well inside PROJECTION_PARITY_INTERVAL and with an unchanged state.db.
+    assert watcher._poll_once(now=2.0) is True
+
+
+def test_unsubscribe_keeps_fingerprint_while_other_subscribers_remain(tmp_path):
+    gw = importlib.import_module("api.gateway_watcher")
+    watcher = gw.GatewayWatcher(state_db_path=_make_db(tmp_path))
+    first = watcher.subscribe()
+    watcher.subscribe()
+    assert watcher._poll_once(now=1.0) is True
+    fingerprint = watcher._last_cheap_fp
+    assert fingerprint
+
+    watcher.unsubscribe(first)
+    assert watcher._last_cheap_fp == fingerprint
+    assert watcher._last_full_projection_at == 1.0
+    assert watcher._poll_once(now=2.0) is False
+
+
+def test_unsubscribe_of_unknown_queue_does_not_reset(tmp_path):
+    import queue as _queue
+
+    gw = importlib.import_module("api.gateway_watcher")
+    watcher = gw.GatewayWatcher(state_db_path=_make_db(tmp_path))
+    watcher.subscribe()
+    assert watcher._poll_once(now=1.0) is True
+    fingerprint = watcher._last_cheap_fp
+
+    watcher.unsubscribe(_queue.Queue())
+    assert watcher._last_cheap_fp == fingerprint
+    assert watcher._last_full_projection_at == 1.0
+
+
+def test_slow_consumer_eviction_invalidates_populated_cache(tmp_path):
+    gw = importlib.import_module("api.gateway_watcher")
+    watcher = gw.GatewayWatcher(state_db_path=_make_db(tmp_path))
+    q = watcher.subscribe()
+    assert watcher._poll_once(now=1.0) is True
+    assert watcher._last_cheap_fp
+    assert watcher._last_full_projection_at == 1.0
+
+    for _ in range(q.maxsize - q.qsize()):
+        q.put_nowait({"type": "existing"})
+    watcher._notify_subscribers([])
+    assert not watcher._has_subscribers()
+    assert watcher._last_cheap_fp == ""
+    assert watcher._last_full_projection_at is None
+    watcher.unsubscribe(q)  # The late SSE cleanup cannot undo the reset.
+    watcher.subscribe()
+    assert watcher._poll_once(now=2.0) is True
+
+
+@pytest.mark.parametrize("removal", ["unsubscribe", "slow_consumer"])
+@pytest.mark.parametrize("reconnect_while_blocked", [False, True])
+def test_inflight_projection_cannot_restore_cache_after_last_removal(
+    tmp_path, monkeypatch, removal, reconnect_while_blocked,
+):
+    """The DB read must not serialize final removal or commit across its epoch."""
+    gw = importlib.import_module("api.gateway_watcher")
+    watcher = gw.GatewayWatcher(state_db_path=_make_db(tmp_path))
+    q = watcher.subscribe()
+    entered = threading.Event()
+    release = threading.Event()
+    projections = []
+    errors = []
+    real_projection = gw._get_agent_sessions_from_db
+
+    def blocked_projection(path):
+        projections.append(path)
+        if len(projections) == 1:
+            entered.set()
+            assert release.wait(5), "test never released the DB read"
+        return real_projection(path)
+
+    def poll():
+        try:
+            watcher._poll_once(now=1.0)
+        except BaseException as exc:
+            errors.append(exc)
+
+    monkeypatch.setattr(gw, "_get_agent_sessions_from_db", blocked_projection)
+    worker = threading.Thread(target=poll, daemon=True)
+    worker.start()
+    new_q = None
+    try:
+        assert entered.wait(5), "poll never reached the DB read"
+        if removal == "unsubscribe":
+            watcher.unsubscribe(q)
+        else:
+            for _ in range(q.maxsize):
+                q.put_nowait({"type": "existing"})
+            watcher._notify_subscribers([])
+            # Handler teardown of an evicted queue must be harmless.
+            watcher.unsubscribe(q)
+        assert not watcher._has_subscribers()
+        assert watcher._last_cheap_fp == ""
+        assert watcher._last_full_projection_at is None
+        if reconnect_while_blocked:
+            new_q = watcher.subscribe()
+    finally:
+        release.set()
+        worker.join(timeout=5)
+    assert not worker.is_alive(), "poll did not finish"
+    assert not errors, errors
+    assert watcher._last_cheap_fp == ""
+    assert watcher._last_full_projection_at is None
+    if new_q is not None:
+        assert new_q.empty(), "old projection must not notify a new subscriber"
+    else:
+        watcher.subscribe()
+    assert watcher._poll_once(now=2.0) is True
+    assert len(projections) == 2
+    assert watcher._last_cheap_fp
+    assert watcher._last_full_projection_at == 2.0
+
+
+def test_poll_loop_reprojects_for_reconnecting_subscriber(tmp_path, monkeypatch):
+    """Through the real loop: disconnect, park, reconnect -> projection runs again."""
+    gw = importlib.import_module("api.gateway_watcher")
+    db = _make_db(tmp_path)
+    projections: list[float] = []
+    real_projection = gw._get_agent_sessions_from_db
+
+    def tracing_projection(path):
+        projections.append(time.monotonic())
+        return real_projection(path)
+
+    monkeypatch.setattr(gw, "_get_agent_sessions_from_db", tracing_projection)
+    watcher = gw.GatewayWatcher(state_db_path=db)
+    watcher.POLL_INTERVAL = 0.05
+    watcher.start()
+    try:
+        q = watcher.subscribe()
+        deadline = time.monotonic() + 2.0
+        while not projections and time.monotonic() < deadline:
+            time.sleep(0.01)
+        assert len(projections) == 1
+
+        watcher.unsubscribe(q)
+        time.sleep(watcher.POLL_INTERVAL * 4)  # loop parks; no further projection
+        assert len(projections) == 1
+
+        watcher.subscribe()
+        deadline = time.monotonic() + 2.0
+        while len(projections) < 2 and time.monotonic() < deadline:
+            time.sleep(0.01)
+        assert len(projections) == 2, "reconnecting subscriber must get a fresh projection"
+    finally:
+        watcher.stop()