operator-plane: cheapen SSE change detector (shared once/sec signal)

Every SSE client polled COUNT(*) on a fresh EventStore connection each ~1s
(re-applying the 473-line schema unless PG_SKIP_SCHEMA=1). Replaced with one
process-wide _ChangeSignal: a single long-lived connection reads max(timestamp)
(idx_events_timestamp tail seek) at most once per second GLOBALLY behind a lock,
so N clients cost one query/sec. EventStore.max_timestamp() backs it; reader is
injectable for the query-budget test.

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>

api/main.py

@@ -107,6 +107,49 @@ def _catalyst_engine(store, catalyst: dict) -> dict:
             "dilutive": dilutive, "vetoes_today": vetoes}
 
 
+class _ChangeSignal:
+    """One process-wide event-change signal shared by ALL SSE clients.
+
+    Before this, every SSE client ran ``COUNT(*)`` on a FRESH EventStore
+    connection every ~1s (re-applying the 473-line schema unless
+    ``PG_SKIP_SCHEMA=1``). Now a single long-lived connection polls
+    ``max(timestamp)`` (an index-tail seek) at most once per second GLOBALLY,
+    behind a lock — N clients cost one query/sec, not N. ``reader`` is injectable
+    so tests can count underlying queries without a DB.
+    """
+
+    def __init__(self, db_path: str | None = None, *, reader=None, min_interval: float = 1.0):
+        self._db_path = db_path
+        self._reader = reader
+        self._min_interval = float(min_interval)
+        self._lock = threading.Lock()
+        self._store = None
+        self._stamp = None
+        self._checked_at = 0.0
+        self.query_count = 0  # test hook: number of underlying reads performed
+
+    def _read(self):
+        self.query_count += 1
+        if self._reader is not None:
+            return self._reader()
+        if self._store is None:
+            self._store = EventStore(self._db_path)  # long-lived; schema skipped via PG_SKIP_SCHEMA
+        return self._store.max_timestamp()
+
+    def current(self):
+        """Latest change token (max event timestamp), refreshed at most once per
+        ``min_interval`` seconds across all callers."""
+        now = time.monotonic()
+        with self._lock:
+            if self._checked_at == 0.0 or (now - self._checked_at) >= self._min_interval:
+                try:
+                    self._stamp = self._read()
+                except Exception:  # noqa: BLE001 — a transient read must not kill the stream
+                    pass
+                self._checked_at = now
+            return self._stamp
+
+
 class DashboardState:
     """Shared state: event store path + optional live execution service."""
 
@@ -125,6 +168,16 @@ class DashboardState:
         self.research_con = research_con
         self.watch_inject = watch_inject
         self._lock = threading.Lock()
+        # lazily-built, process-wide SSE change signal (shared by all clients)
+        self._change_signal: _ChangeSignal | None = None
+
+    def change_stamp(self):
+        """SSE change token from the shared once-per-second signal (see _ChangeSignal)."""
+        if self._change_signal is None:
+            with self._lock:
+                if self._change_signal is None:
+                    self._change_signal = _ChangeSignal(self.event_db_path)
+        return self._change_signal.current()
 
     def open_store(self) -> EventStore:
         return EventStore(self.event_db_path)
@@ -350,14 +403,15 @@ class DashboardHandler(BaseHTTPRequestHandler):
         self.end_headers()
 
         shutdown = getattr(type(self), "shutdown_event", None)
-        last_count = -1
+        _sentinel = object()
+        last_stamp = _sentinel
         last_heartbeat = 0.0
         try:
             while shutdown is None or not shutdown.is_set():
-                count = self.state.event_count()
+                stamp = self.state.change_stamp()
                 now = time.monotonic()
-                if count != last_count:
-                    last_count = count
+                if stamp != last_stamp:  # last_stamp starts as a unique sentinel
+                    last_stamp = stamp
                     payload = _dump(self.state.snapshots()).decode()
                     self.wfile.write(f"event: snapshot\ndata: {payload}\n\n".encode())
                     self.wfile.flush()

storage/event_store.py

@@ -200,6 +200,16 @@ class EventStore:
             row = cursor.fetchone()
         return int(row[0]) if row else 0
 
+    def max_timestamp(self):
+        """Newest events.timestamp (None on an empty store). The SSE change
+        signal polls this instead of COUNT(*): idx_events_timestamp makes it an
+        index-tail seek, whereas COUNT(*) scans the whole table every tick."""
+        with self._lock:
+            cursor = self.con.cursor()
+            cursor.execute("SELECT max(timestamp) FROM events")
+            row = cursor.fetchone()
+        return row[0] if row else None
+
     def get_symbol_timeline(self, symbol: str) -> list[dict]:
         """Get full timeline for a symbol.
 

tests/unit/api/test_sse_change_signal.py

@@ -0,0 +1,51 @@
+"""The shared SSE change signal must run ONE underlying query per second across
+ALL clients, not one per client per tick (plan 007 step 2)."""
+
+import time
+
+from api.main import _ChangeSignal
+
+
+def test_shared_signal_query_budget():
+    calls = {"n": 0}
+
+    def reader():
+        calls["n"] += 1
+        return calls["n"]  # a monotonic stamp
+
+    sig = _ChangeSignal(reader=reader, min_interval=1.0)
+
+    # Two "clients" hammering current() 10 times total inside the 1s window must
+    # collapse to a single underlying read (the first one).
+    for _ in range(5):
+        sig.current()  # client A
+        sig.current()  # client B
+    assert calls["n"] == 1, f"expected 1 read in the window, got {calls['n']}"
+    assert sig.query_count == 1
+
+
+def test_signal_refreshes_after_interval():
+    calls = {"n": 0}
+
+    def reader():
+        calls["n"] += 1
+        return calls["n"]
+
+    sig = _ChangeSignal(reader=reader, min_interval=0.05)
+    first = sig.current()
+    time.sleep(0.06)
+    second = sig.current()
+    assert first == 1 and second == 2  # re-read once the interval elapsed
+
+
+def test_signal_survives_reader_error():
+    def boom():
+        raise RuntimeError("db blip")
+
+    sig = _ChangeSignal(reader=boom, min_interval=1.0)
+    assert sig.current() is None  # swallowed; stream keeps running
+
+
+if __name__ == "__main__":  # pragma: no cover
+    import pytest
+    pytest.main([__file__, "-q"])