fix(sse): address PR #561 review round 2

5 follow-up comments from Copilot, all valid:

1. **CRITICAL — snap_seq race with split writer** (concurrency, 001).
   Round-1's fix lifted snapshot capture into
   register_listener_with_replay under nested locks, but the
   WRITER side (on_content_token / on_reasoning_token) still
   released _ws_lock before calling _enqueue (which bumps
   _event_id under _listeners_lock).  A reader could
   interleave between writer's release and writer's _enqueue:
   capture inflight WITH the new text, read STALE _event_id,
   return snap_seq < new_event_id.  The new event's live emit
   then has _seq > snap_seq, slips past the dedup filter, and
   double-renders text the snapshot already contained.

   Fix: move self._enqueue(...) INSIDE the with self._ws_lock:
   block in both token writers.  The inflight mutation and the
   _event_id advancement are now atomic against any snapshot
   reader.  Lock order _ws_lock (outer) → _listeners_lock
   (inner via _enqueue) matches the snapshot helpers, so no
   deadlock.  Fan-out's put_nowait calls happen under
   _ws_lock for token writers — microsecond cost per listener,
   acceptable for the correctness guarantee.

2. **NIT — stale comment ref to buffered[-1]._event_id** (docs, 002).
   The comment referenced a local var (buffered) that lives in
   register_listener_with_replay, not in the events handler.
   Reworded to describe the cutoff in terms of the last replayed
   event id and the atomic-against-writers registration.

3. **MODERATE — 401 branch leaves reconnect loop** (bug, 003).
   The coord's onerror schedules a 5 s CLOSED-state recovery timer
   unconditionally.  In the 401-expired-session branch we close
   evtSource and showLogin — but the timer still fires 5 s later,
   observes !evtSource, and calls scheduleReconnect(),
   which opens a new EventSource that 401s again → infinite
   reconnect loop while the login overlay is up.  Fix: cancel
   reconnectTimer in the 401 branch.

4. **MODERATE — race test was vacuous** (test_coverage, 004).
   The previous regression test drained the listener queue after
   register_listener_with_replay returned, but the helper
   doesn't backfill buffered events into the queue, so the loop
   was almost always a no-op and the assertion never executed.
   Rewrote with a monkey-patched _enqueue that sleeps 50 ms
   before bumping _event_id — widens the race window
   deterministically.  Verified: the test FAILS on pre-fix code
   (snap.content has marker but snap.seq=0 < final_event_id=1)
   and PASSES on post-fix code (writer holds _ws_lock through
   _enqueue, so the reader blocks until writer fully done).
   Also pinned the no-backfill contract so a future change adding
   listener-queue backfill remembers to keep snap_seq the
   high-water mark.

5. **NIT — except Exception too broad in test** (best_practices, 005).
   Tightened except Exception: to except queue.Empty: so
   unexpected exceptions aren't silently swallowed in the drain
   loop.

Tests:
- 86 tests in test_sse_reconnect_replay.py + test_session_ui_base.py
  pass (existing 84 + 2 new race regressions).
- Full non-live suite: 6347 passed, 15 skipped, no regressions.
- Ruff + mypy clean on changed .py files; JS parses.

tests/test_sse_reconnect_replay.py

@@ -341,36 +341,129 @@ def test_truncated_path_snapshot_captures_real_snap_seq() -> None:
     assert "t9" in snap["content"]
 
 
-def test_truncated_path_filters_already_in_snapshot_tokens() -> None:
-    """End-to-end: a token landing in the listener queue between
-    ``register_listener_with_replay`` returning and the live drain's
-    first read must be filtered by the handler's
-    ``_seq <= snap_seq`` dedup, because its text is ALSO in the
-    snapshot we just emitted.  Without the snap_seq fix this test
-    would observe a duplicate live token after the snapshot."""
-    import collections
+def test_snap_seq_high_water_mark_holds_under_writer_race() -> None:
+    """Regression for PR #561 review comment 1.
+
+    The invariant: every token whose text appears in
+    ``snapshot["content"]`` (or ``"reasoning"``) must have its
+    ``_event_id`` <= ``snapshot["seq"]``.  Equivalently, any token
+    that fires AFTER the snapshot was captured must have
+    ``_event_id > snap_seq``.  Otherwise the events handler's
+    ``_seq <= snap_seq`` live-drain filter would let the new token
+    through AND its text would already be in the snapshot text →
+    double-render.
+
+    The pre-fix race: ``on_content_token`` took ``_ws_lock``,
+    appended to inflight, released ``_ws_lock``, then called
+    ``_enqueue`` (which bumps ``_event_id``).  A snapshot reader
+    interleaving between the release and the ``_enqueue`` would
+    capture inflight (with the new text) and read a STALE
+    ``_event_id``.  Snap_seq below new event's id → filter slips →
+    double-render.
+
+    The race window in plain Python is narrow (a few bytecodes
+    between lock release and the ``_enqueue`` call), so a pure
+    barrier-based race rarely hits it.  This test injects a
+    deterministic sleep into ``_enqueue`` via monkey-patch to
+    widen the window enough to be reliably observed under the
+    pre-fix code path — AND to be reliably AVOIDED under the
+    post-fix code path (because the post-fix
+    ``on_content_token`` calls ``_enqueue`` while still holding
+    ``_ws_lock``, so the snapshot reader can't acquire
+    ``_ws_lock`` until the writer is fully done).
+    """
+    import queue
+    import threading
+    import time
 
     ui = _make_ui()
-    ui._event_buffer = collections.deque(maxlen=3)
-    for i in range(10):
-        ui.on_content_token(f"t{i}")
-    lq, _, status, _, _, snap = ui.register_listener_with_replay(1)
-    assert status == "truncated"
-    captured_seq = snap["seq"]
-    # Simulate the race: at this point the listener queue already
-    # holds the buffered token events.  Drain them and confirm
-    # every one has ``_seq <= snap_seq`` — i.e. the handler's filter
-    # would correctly drop them.
+    marker = "RACE-MARKER"
+    original_enqueue = ui._enqueue
+
+    # Widen the race window: sleep just BEFORE the original
+    # ``_enqueue`` runs (which is where ``_event_id`` would advance).
+    # Post-fix this sleep happens while the writer still holds
+    # ``_ws_lock`` — readers block.  Pre-fix the writer has
+    # released ``_ws_lock`` before reaching this monkey-patch, so
+    # the reader gets a clean window to capture an inconsistent
+    # ``(inflight, _event_id)`` pair.
+    def slow_enqueue(data: dict[str, Any]) -> None:
+        time.sleep(0.05)  # 50 ms — orders of magnitude wider than the GIL switch interval
+        return original_enqueue(data)
+
+    ui._enqueue = slow_enqueue  # type: ignore[method-assign]
+
+    snap_box: dict[str, Any] = {}
+    writer_done = threading.Event()
+
+    def _writer() -> None:
+        ui.on_content_token(marker)
+        writer_done.set()
+
+    def _reader() -> None:
+        # Give the writer time to enter ``on_content_token`` and
+        # (pre-fix) release ``_ws_lock`` before the snapshot.  50 ms
+        # is conservative; 5 ms would also work in practice.
+        time.sleep(0.025)
+        _, _, _, _, _, snap = ui.register_listener_with_replay(0)
+        snap_box["snap"] = snap
+        snap_box["event_id_at_snapshot_return"] = ui._event_id
+
+    wt = threading.Thread(target=_writer)
+    rt = threading.Thread(target=_reader)
+    wt.start()
+    rt.start()
+    wt.join(timeout=5)
+    rt.join(timeout=5)
+    assert writer_done.is_set(), "writer thread did not complete"
+
+    snap = snap_box["snap"]
+    final_event_id = ui._event_id
+
+    # Core invariant: if the snapshot's content includes the marker
+    # text, snap.seq must be >= the writer's final _event_id.
+    # Pre-fix this fails (snap.seq=0 while final_event_id=1 and
+    # snap.content="RACE-MARKER"); post-fix the reader can't acquire
+    # ``_ws_lock`` until the writer completes, so snap is either
+    # (content="", seq=0) — reader won first — or
+    # (content="RACE-MARKER", seq=1) — writer won first.
+    assert marker in snap["content"] or snap["content"] == "", (
+        f"unexpected snap content: {snap['content']!r}"
+    )
+    if marker in snap["content"]:
+        assert snap["seq"] >= final_event_id, (
+            f"snap captured '{marker}' but snap.seq={snap['seq']} < "
+            f"final _event_id={final_event_id}; the live emission of "
+            f"this token would slip past the events handler's "
+            f"_seq <= snap_seq filter and double-render text the "
+            f"snapshot already contained.  Pre-fix race window "
+            f"opened by ``_enqueue`` running outside ``_ws_lock``."
+        )
+
+    # Sanity: also exercise the post-truncated drain shape so the
+    # test file pins both the contract AND the no-backfill behaviour
+    # (a future change that adds backfill into the listener queue
+    # must keep the dedup invariant above true).
+    ui2 = _make_ui()
+    for j in range(5):
+        ui2.on_content_token(f"x{j}")
+    lq, _, status, _, _, snap2 = ui2.register_listener_with_replay(0)
+    captured_seq = snap2["seq"]
+    drained = 0
     while True:
         try:
             ev = lq.get_nowait()
-        except Exception:
+        except queue.Empty:
             break
+        drained += 1
         if ev.get("type") == "content":
-            assert ev["_seq"] <= captured_seq, (
-                f"token _seq={ev['_seq']} > snap_seq={captured_seq}; "
-                "would slip past the live-drain dedup and double-render"
-            )
+            assert ev["_seq"] <= captured_seq, f"token _seq={ev['_seq']} > snap_seq={captured_seq}"
+    assert drained == 0, (
+        f"register_listener_with_replay backfilled {drained} events "
+        f"into the listener queue; if intentional, the dedup "
+        f"invariant above must still hold and this assertion should "
+        f"be updated."
+    )
 
 
 # ---------------------------------------------------------------------------

turnstone/console/static/coordinator/coordinator.js

@@ -2011,6 +2011,17 @@
               /* noop */
             }
             evtSource = null;
+            // Cancel the pending CLOSED-state recovery timer (set
+            // below).  Without this, 5 s later the timer would
+            // observe ``!evtSource`` and call ``scheduleReconnect``,
+            // which would open a new EventSource that gets 401 again
+            // → infinite reconnect loop while the login overlay is
+            // up.  The login flow re-arms ``connectSSE`` after a
+            // successful sign-in via its own callback path.
+            if (reconnectTimer) {
+              clearTimeout(reconnectTimer);
+              reconnectTimer = null;
+            }
             showLogin("Session expired. Please sign in to reconnect.");
           }
         },
@@ -2027,7 +2038,8 @@
       // lastEventId via the URL query param, so replay still
       // works across the manual reconnect).  Cancel/replace the
       // existing timer so successive onerror fires don't pile up
-      // multiple checks for the same source.
+      // multiple checks for the same source.  The 401 branch above
+      // ALSO cancels this timer when it fires — see comment there.
       if (reconnectTimer) clearTimeout(reconnectTimer);
       reconnectTimer = setTimeout(function () {
         reconnectTimer = null;

turnstone/core/session_routes.py

@@ -1514,10 +1514,11 @@ def make_events_handler(cfg: SessionEndpointConfig) -> Handler:
                 # synthetic snapshot/state_change/history emission is
                 # skipped by the events handler.  No live-dedup
                 # filtering required because the buffered events
-                # themselves are the cutoff — anything past
-                # ``buffered[-1]._event_id`` is genuinely new live
-                # traffic that lands in the listener queue after the
-                # buffer snapshot.
+                # themselves are the cutoff — anything past the last
+                # replayed event id is genuinely new live traffic
+                # that lands in the listener queue after the buffer
+                # snapshot was taken (atomic-against-writers under
+                # the registration's nested locks).
                 in_progress_snap = {"content": "", "reasoning": "", "seq": 0}
 
         # Per-kind executor for the blocking ``client_queue.get``

turnstone/core/session_ui_base.py

@@ -1737,12 +1737,25 @@ class SessionUIBase:
         snapshot text up to the cap and then live tokens past it,
         with a visual gap equal to the past-cap chunk. No silent
         drop of subsequent tokens.
+
+        **Lock coupling**: ``_enqueue`` is called WHILE still
+        holding ``_ws_lock`` so the inflight append AND the
+        ``_event_id`` advancement happen atomically against a
+        snapshot reader.  Without this coupling a reader could
+        capture the inflight (with the new text) and read
+        ``_event_id`` BEFORE the writer's ``_enqueue`` bumped it,
+        producing a ``snap_seq`` lower than the new event's
+        ``_event_id``.  The new event would then slip past the
+        ``_seq <= snap_seq`` live-drain dedup and double-render
+        the text the snapshot already contained.  Acquisition
+        order ``_ws_lock`` (outer) → ``_listeners_lock`` (inner via
+        ``_enqueue``) matches the snapshot helpers, so no deadlock.
         """
         with self._ws_lock:
             if self._ws_inflight_reasoning_size < _MAX_TURN_CONTENT_CHARS:
                 self._ws_inflight_reasoning.append(text)
                 self._ws_inflight_reasoning_size += len(text)
-        self._enqueue({"type": "reasoning", "text": text})
+            self._enqueue({"type": "reasoning", "text": text})
 
     def on_content_token(self, text: str) -> None:
         """Append to both turn-content buffers (capped) + enqueue.
@@ -1758,18 +1771,23 @@ class SessionUIBase:
         is stamped by :meth:`_enqueue` against the per-ws
         ``_event_id`` counter, which advances on EVERY emit
         regardless of cap state — see :meth:`on_reasoning_token` for
-        the full rationale.
+        the full rationale, including why ``_enqueue`` runs while
+        still holding ``_ws_lock`` (the lock coupling that makes
+        ``snap_seq`` a true high-water mark for the snapshot text).
 
-        The cap-check + append + size-update run under ``_ws_lock``
-        so a concurrent
+        The cap-check + append + size-update + enqueue all run under
+        ``_ws_lock`` so a concurrent
         :meth:`snapshot_and_consume_state_payload` IDLE/ERROR drain or
         a concurrent :meth:`register_listener_with_in_progress_snapshot`
-        can't see a torn list mid-append. In production this is
-        single-writer-per-ws (the worker thread) but the snapshot
+        / :meth:`register_listener_with_replay` sees a consistent
+        ``(inflight_content, _event_id)`` pair.  In production this
+        is single-writer-per-ws (the worker thread) but the snapshot
         reader runs from coord's adapter via ``mgr.set_state``;
         without the lock the writer's append could land in an
-        orphaned list reference the snapshot just swapped out. Lock
-        hold is microseconds.
+        orphaned list reference the snapshot just swapped out, AND
+        the inflight/counter pair could de-sync.  Lock hold is
+        microseconds (the fan-out's ``put_nowait`` calls are O(N
+        listeners) but each is a single non-blocking enqueue).
         """
         with self._ws_lock:
             if self._ws_turn_content_size < _MAX_TURN_CONTENT_CHARS:
@@ -1778,7 +1796,7 @@ class SessionUIBase:
             if self._ws_inflight_content_size < _MAX_TURN_CONTENT_CHARS:
                 self._ws_inflight_content.append(text)
                 self._ws_inflight_content_size += len(text)
-        self._enqueue({"type": "content", "text": text})
+            self._enqueue({"type": "content", "text": text})
 
     def on_stream_end(self) -> None:
         with self._ws_lock: