fix(gateway-watcher): fence stale projections after final subscriber removal

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,31 +288,32 @@ class GatewayWatcher:
         self._idle_wake.set()
         return q
 
-    def unsubscribe(self, q: queue.Queue):
-        """Remove a subscriber queue.
-
-        When the last subscriber leaves, drop the cheap fingerprint and parity
-        timestamp so the next subscription runs a fresh projection. The poll
-        loop parks *before* calling ``_poll_once`` once nobody is subscribed,
-        so the equivalent reset inside ``_poll_once`` is not reached on the
-        normal disconnect path.
-        """
-        with self._sub_lock:
+    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
+            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:
@@ -320,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:
@@ -340,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
@@ -371,14 +371,20 @@ class GatewayWatcher:
         if sessions is None:
             return False
         current_hash = _snapshot_hash(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
-            self._notify_subscribers(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

@@ -5,16 +5,19 @@
 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. The reset now happens in
-``unsubscribe()`` when the last subscriber leaves.
+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"
@@ -101,6 +104,89 @@ def test_unsubscribe_of_unknown_queue_does_not_reset(tmp_path):
     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")