fix(sse): address PR #542 review findings

Four issues raised on the merged PR #542, evaluated and fixed: 1. **Truncated-path snap_seq bug (Copilot low-confidence, VALID).** make_events_handler's truncated branch set snap_seq = 0, disabling the live-drain _seq <= snap_seq dedup filter. Any token writer racing between register_listener_with_replay returning and the live drain's first read would land in BOTH the listener queue AND the captured snapshot text (the snapshot is emitted via in_progress_snapshot as the recovery floor), so the client double-renders. Fix lifts the snapshot capture INTO register_listener_with_replay under the same nested-lock acquire as the listener registration + buffer slice + counter read, so the returned snapshot["seq"] is the exact high-water mark the snapshot text corresponds to. Handler now uses snapshot["seq"] as snap_seq on truncated, dropping any token event with _seq <= snap_seq from the live emit. 2. **Lock-held string join in truncated path (Copilot, VALID).** "".join(ui_base._ws_inflight_content) ran inside the with ui_base._ws_lock: block, holding the lock for the duration of the join and blocking on-token writers. Fix (folded into #1's refactor): the new register_listener_with_replay copies the inflight lists under lock and joins outside, matching the existing pattern in register_listener_with_in_progress_snapshot. 3. **_strip_js_comments docstring misclaim (Copilot, VALID).** Docstring claimed the helper preserves "string/regex literals" but the implementation only tracks string delimiters. Fix: docstring updated to call out the regex-literal limitation explicitly + note that current callers don't scan regions containing regex literals. Extending the tracker is left for a future caller that needs it. 4. **Coord scheduleReconnect dead-code regression (Copilot, VALID).** After the PR-D refactor, scheduleReconnect() had no remaining call sites — which meant reconnectAttempts never incremented, wasReconnecting was always false, AND there was no fallback when the browser transitioned the source to CLOSED (hard 4xx after retries, intermediary tearing the connection down with prejudice, etc.). The first failure mode silently broke the post-gap replace-mode refresh of children / tasks / wait indicator / live-badge cache; the second left the coord permanently disconnected on non-transient failures. Fix: - Introduce disconnectedSinceLastOpen flag set in onerror, cleared in onopen. wasReconnecting reads it (with the legacy reconnectAttempts > 0 fallback for the scheduleReconnect-driven case), so the post-gap refresh fires after every reconnect including the common native-reconnect path. - Re-introduce CLOSED-state recovery: onerror schedules a 5 s delayed check via reconnectTimer; if the source is still CLOSED at that point, call scheduleReconnect(), which opens a new EventSource (threading the saved lastEventId via the URL query param so replay still works across the manual reconnect). Cancel/replace successive timers so onerror floods don't pile up multiple checks for the same source. Tests: - New test_truncated_path_snapshot_captures_real_snap_seq pins the snap_seq fix at the helper boundary. - New test_truncated_path_filters_already_in_snapshot_tokens pins the end-to-end dedup invariant — would have caught the double-render under the old code. - Existing test_sse_reconnect_replay.py call sites updated for the new 6-tuple return of register_listener_with_replay. - All 86 tests in those two files pass; full non-live suite (6347 tests) passes; ruff + mypy clean on changed files; both JS files parse-check.
2026-08-12 23:12:23 -06:00 · 2026-05-22 19:08:23 -07:00
parent 83a602fef7
commit 5b449e502d
5 changed files with 228 additions and 53 deletions
@@ -1225,14 +1225,24 @@ def test_dead_sse_defensive_reconnect_registered() -> None:


 def _strip_js_comments(src: str) -> str:
-    """Strip ``//`` and ``/* */`` comments while preserving string/regex
-    literals and keeping byte length identical (comments replaced with
-    spaces).  ``_slice_balanced_body`` doesn't skip comments, so an
+    """Strip ``//`` and ``/* */`` comments while preserving string
+    literal contents (``"..."``, ``'...'``, `` `...` ``) and keeping
+    byte length identical (comments replaced with spaces).
+
+    Limitation — does NOT detect regex literals (``/pattern/flags``).
+    A ``//`` inside a regex like ``/abc//`` would be misread as the
+    start of a line comment.  Safe today because the regions we scan
+    (SSE-handler ``onerror`` bodies, ``connectSSE`` /
+    ``connectGlobalSSE`` function bodies) don't contain regex
+    literals; if a future caller wants to scan a region with regex
+    literals, extend the tracker first.
+
+    Motivation: ``_slice_balanced_body`` doesn't skip comments, so an
    apostrophe inside a comment (``can't``, ``don't``) opens a fake
    string state that swallows braces until the next ``'``.  The new
-    onerror handlers carry these comments routinely; stripping comments
-    before brace-walking removes the hazard without re-architecting
-    the existing slice helper.
+    onerror handlers carry these comments routinely; stripping
+    comments before brace-walking removes the hazard without
+    re-architecting the existing slice helper.
    """
    out: list[str] = []
    n = len(src)
@@ -98,7 +98,7 @@ def test_replay_holds_events_through_empty_listeners_period() -> None:
    for i in range(10):
        ui._enqueue({"type": "tool_started", "name": f"t{i}"})
    # Reconnect-style register with Last-Event-ID=0 (client saw nothing).
-    lq, replay, status, lost, earliest = ui.register_listener_with_replay(0)
+    lq, replay, status, lost, earliest, _ = ui.register_listener_with_replay(0)
    assert status == "replay_ok"
    assert lost == 0
    assert earliest == 1
@@ -115,7 +115,7 @@ def test_replay_with_last_event_id_skips_already_seen_events() -> None:
    ui = _make_ui()
    for i in range(8):
        ui._enqueue({"type": "tool_started", "name": f"t{i}"})
-    lq, replay, status, lost, earliest = ui.register_listener_with_replay(5)
+    lq, replay, status, lost, earliest, _ = ui.register_listener_with_replay(5)
    assert status == "replay_ok"
    assert lost == 0
    assert [ev["_event_id"] for ev in replay] == [6, 7, 8]
@@ -134,7 +134,7 @@ def test_replay_truncated_when_last_event_id_predates_buffer() -> None:
    for i in range(20):
        ui._enqueue({"type": "tool_started", "name": f"t{i}"})
    # Buffer now holds ids 16..20 (5 most recent of 20 emitted).
-    lq, replay, status, lost, earliest = ui.register_listener_with_replay(3)
+    lq, replay, status, lost, earliest, _ = ui.register_listener_with_replay(3)
    assert status == "truncated"
    assert earliest == 16
    assert lost == 12  # earliest-1 - last_event_id = 15 - 3
@@ -146,7 +146,7 @@ def test_replay_empty_buffer_returns_replay_ok_empty() -> None:
    A spurious ``replay_truncated`` envelope on a freshly-opened
    workstream would be confusing and incorrect."""
    ui = _make_ui()
-    lq, replay, status, lost, earliest = ui.register_listener_with_replay(0)
+    lq, replay, status, lost, earliest, _ = ui.register_listener_with_replay(0)
    assert status == "replay_ok"
    assert replay == []
    assert lost == 0
@@ -161,7 +161,7 @@ def test_replay_registers_listener_atomically_with_buffer_snapshot() -> None:
    and the live queue, and never in NEITHER."""
    ui = _make_ui()
    ui._enqueue({"type": "tool_started", "name": "before"})
-    lq, replay, _, _, _ = ui.register_listener_with_replay(0)
+    lq, replay, _, _, _, _ = ui.register_listener_with_replay(0)
    # Now fire after registration — must arrive live, NOT in replay.
    ui._enqueue({"type": "tool_started", "name": "after"})
    assert [ev["name"] for ev in replay] == ["before"]
@@ -213,7 +213,7 @@ def test_event_id_does_not_skip_when_listener_queue_full() -> None:
    for i in range(10):
        ui._enqueue({"type": "tool_started", "name": f"t{i}"})
    # Replay from id=0 — fresh listener gets all 10, ids 1..10 dense.
-    _, replay, status, _, _ = ui.register_listener_with_replay(0)
+    _, replay, status, _, _, _ = ui.register_listener_with_replay(0)
    assert status == "replay_ok"
    assert [ev["_event_id"] for ev in replay] == list(range(1, 11))

@@ -239,7 +239,7 @@ def test_cross_thread_writer_and_replay_observer_consistent() -> None:
    def _reader() -> None:
        # Wait briefly so the writer is mid-flight.
        threading.Event().wait(0.001)
-        _, replay, status, _, earliest = ui.register_listener_with_replay(0)
+        _, replay, status, _, earliest, _ = ui.register_listener_with_replay(0)
        snap_box["replay"] = replay
        snap_box["status"] = status
        snap_box["earliest"] = earliest
@@ -277,7 +277,7 @@ def test_event_id_persists_across_turn_boundaries() -> None:
    seq_after = ui._event_id
    assert seq_after > seq_before, "counter regressed across turn boundary"
    # Replay from mid-turn-N must still serve turn-N+1's content.
-    _, replay, status, _, _ = ui.register_listener_with_replay(seq_before)
+    _, replay, status, _, _, _ = ui.register_listener_with_replay(seq_before)
    assert status == "replay_ok"
    assert len(replay) == 1
    assert replay[0]["text"] == "turn-N+1 tok1"
@@ -294,10 +294,83 @@ def test_replay_ok_skips_in_progress_snapshot_path() -> None:
    ui.on_content_token("partial ")
    # Replay path: returns replay_ok and a synthetic snap is NOT taken
    # (we test the handler-side behavior in the handler tests below).
-    lq, replay, status, _, _ = ui.register_listener_with_replay(0)
+    lq, replay, status, _, _, snap = ui.register_listener_with_replay(0)
    assert status == "replay_ok"
    # The buffered event carries the partial content as a content event.
    assert any(ev.get("type") == "content" for ev in replay)
+    # Snapshot is captured atomically too (used on truncated path to
+    # drive live-drain ``_seq <= snap_seq`` dedup); for replay_ok the
+    # caller ignores it but the contract returns one regardless.
+    assert isinstance(snap, dict)
+    assert snap["seq"] >= 1
+
+
+def test_truncated_path_snapshot_captures_real_snap_seq() -> None:
+    """Regression for PR #542 review comment 1 (Copilot, low-confidence).
+
+    On the truncated path the caller used to set ``snap_seq=0``, which
+    disabled the events handler's live-drain ``_seq <= snap_seq``
+    dedup.  A token writer racing between
+    ``register_listener_with_replay`` returning and the live drain's
+    first read would land in the listener queue AND in the captured
+    snapshot text, causing the client to render the token twice
+    (once via the ``in_progress_snapshot`` content text, once via the
+    live event delivery).
+
+    The fix lifts the snapshot capture INTO
+    ``register_listener_with_replay`` under the same nested-lock
+    acquire as the listener registration + buffer slice + counter
+    read, so ``snap_seq`` returned in the snapshot is the exact
+    high-water mark the snapshot text corresponds to."""
+    import collections
+
+    ui = _make_ui()
+    ui._event_buffer = collections.deque(maxlen=3)
+    # Fire enough events to trigger truncation on reconnect with a
+    # stale ``Last-Event-ID``.
+    for i in range(10):
+        ui.on_content_token(f"t{i}")
+    _, _, status, _, _, snap = ui.register_listener_with_replay(1)
+    assert status == "truncated"
+    # The snapshot's seq must be the LATEST event_id, not 0 — that's
+    # what gates the live-drain dedup filter in the events handler.
+    assert snap["seq"] == ui._event_id
+    assert snap["seq"] >= 10
+    # And the content is captured (not empty).
+    assert "t0" in snap["content"]
+    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
+
+    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.
+    while True:
+        try:
+            ev = lq.get_nowait()
+        except Exception:
+            break
+        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"
+            )


 # ---------------------------------------------------------------------------
@@ -139,6 +139,15 @@

  let evtSource = null;
  let reconnectAttempts = 0;
+  // Flag set in onerror, cleared in onopen.  Drives the "did we
+  // just recover from a gap?" decision in onopen so the replace-
+  // mode refresh of children/tasks/wait/badge caches fires on
+  // every reconnect — including the common case where native
+  // EventSource auto-reconnect handles the underlying SSE transition
+  // without scheduleReconnect running (which used to be the only
+  // place reconnectAttempts incremented; that path is rarely hit
+  // now that native reconnect handles transient errors).
+  let disconnectedSinceLastOpen = false;
  // Saved high-water mark for the manual-reconnect path.  The
  // EventSource constructor can't set custom headers, so when we
  // construct a fresh source we thread ``?last_event_id=N`` instead
@@ -1910,7 +1919,15 @@
    // reconnectAttempts in onopen — child_ws_* events dispatched while
    // we were disconnected aren't replayed by the events SSE handler,
    // so the client has to pull authoritative state after any gap.
-    const wasReconnecting = reconnectAttempts > 0;
+    // Snapshot whether this connect attempt follows a prior
+    // disconnect.  Native EventSource auto-reconnect no longer
+    // routes through scheduleReconnect on the transient-error path,
+    // so the legacy ``reconnectAttempts > 0`` check is always false
+    // after PR-D — use ``disconnectedSinceLastOpen`` (set by onerror,
+    // cleared by onopen below) as the authoritative "was-gap" flag.
+    // Falls back to the legacy semantic for the genuinely manual
+    // case (scheduleReconnect-driven reconnect after CLOSED state).
+    const wasReconnecting = disconnectedSinceLastOpen || reconnectAttempts > 0;
    let url = "/v1/api/workstreams/" + encodeURIComponent(wsId) + "/events";
    if (lastEventId) {
      url += "?last_event_id=" + encodeURIComponent(lastEventId);
@@ -1918,6 +1935,9 @@
    evtSource = new EventSource(url, { withCredentials: true });
    evtSource.onopen = function () {
      reconnectAttempts = 0;
+      // Clear the "was disconnected" flag now that the gap is
+      // closed.  Future onerror fires will set it again.
+      disconnectedSinceLastOpen = false;
      setSseStatus("live", "ok");
      // Lift the disconnected dim treatment + restore the last known
      // counters; the replay phase will overwrite with authoritative
@@ -1971,6 +1991,7 @@
      // reconnect, which is exactly the reconnect-with-replay defect
      // PR-D ships to fix.  See
      // tests/test_app_js.py::test_coord_connectsse_onerror_preserves_native_reconnect.
+      disconnectedSinceLastOpen = true;
      setSseStatus("disconnected", "err");
      // Dim the status bar so a stale reading doesn't read as live.
      statusBarEl.classList.add("ws-sb-disconnected");
@@ -1979,9 +2000,7 @@
      // must log in), so we DO close + showLogin in that branch.
      // Transient errors (network blips, intermediary timeouts) just
      // let native reconnect run — no scheduleReconnect needed
-      // because the source isn't dead.  scheduleReconnect remains
-      // available as the final fallback for the truly-CLOSED case
-      // (covered by the auth.js callback path).
+      // because the source isn't dead.
      var probe = typeof authFetch === "function" ? authFetch : fetch;
      probe("/v1/api/workstreams/" + encodeURIComponent(wsId)).then(
        function (r) {
@@ -1996,6 +2015,26 @@
          }
        },
      );
+      // CLOSED-state recovery: native auto-reconnect covers the
+      // transient case (source stays in CONNECTING and eventually
+      // re-opens).  But if the browser gives up — hard 4xx after
+      // retries, intermediary tearing the connection down with
+      // prejudice, etc. — the source transitions to CLOSED and
+      // there is no further native recovery.  Schedule a delayed
+      // check that calls scheduleReconnect if the source is still
+      // CLOSED at that point; scheduleReconnect's exp-backoff +
+      // jitter then opens a new EventSource (threading the saved
+      // 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.
+      if (reconnectTimer) clearTimeout(reconnectTimer);
+      reconnectTimer = setTimeout(function () {
+        reconnectTimer = null;
+        if (!evtSource || evtSource.readyState === EventSource.CLOSED) {
+          scheduleReconnect();
+        }
+      }, 5000);
    };
    evtSource.onmessage = function (event) {
      // Capture lastEventId BEFORE JSON.parse so a malformed event
@@ -1489,21 +1489,35 @@ def make_events_handler(cfg: SessionEndpointConfig) -> Handler:
                replay_status,
                lost_count,
                earliest_available_id,
+                snapshot,
            ) = ui_base.register_listener_with_replay(last_event_id)
            if replay_status == "truncated":
-                # Capture the snapshot too — it's the recovery floor
-                # when the buffer can't fill the gap.  Listener is
-                # already registered by ``register_listener_with_replay``;
-                # take the snapshot bits only.
-                with ui_base._ws_lock:  # noqa: SLF001 — same-module-level access pattern
-                    captured_content = "".join(ui_base._ws_inflight_content)  # noqa: SLF001
-                    captured_reasoning = "".join(ui_base._ws_inflight_reasoning)  # noqa: SLF001
-                in_progress_snap = {
-                    "content": captured_content,
-                    "reasoning": captured_reasoning,
-                    "seq": 0,  # not used on truncated path; fresh-style yields don't filter
-                }
+                # Truncated → emit ``replay_truncated`` envelope, then
+                # the snapshot is the recovery floor.  ``snap_seq``
+                # MUST come from the snapshot capture (not 0), because
+                # writers can race between
+                # ``register_listener_with_replay`` returning and our
+                # first live-drain read: any token event landing in
+                # the listener queue between registration and the
+                # captured ``_event_id`` is ALSO covered by the
+                # snapshot's content/reasoning text, and would
+                # double-render without the ``_seq <= snap_seq`` dedup
+                # filter on the live path.  The helper captured both
+                # under the same nested-lock acquire, so this
+                # ``snap_seq`` is exactly the high-water mark
+                # corresponding to the snapshot text.
+                in_progress_snap = snapshot
+                snap_seq = snapshot["seq"]
            else:
+                # ``replay_ok``: the buffered events ARE the partial
+                # token stream (no separate snapshot needed); the
+                # 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.
                in_progress_snap = {"content": "", "reasoning": "", "seq": 0}

        # Per-kind executor for the blocking ``client_queue.get``
@@ -482,15 +482,24 @@ class SessionUIBase:
        self,
        last_event_id: int,
        maxsize: int = _DEFAULT_LISTENER_QUEUE_MAX,
-    ) -> tuple[queue.Queue[dict[str, Any]], list[dict[str, Any]], str, int, int]:
-        """Register a listener AND capture buffered events for replay.
+    ) -> tuple[
+        queue.Queue[dict[str, Any]],
+        list[dict[str, Any]],
+        str,
+        int,
+        int,
+        dict[str, Any],
+    ]:
+        """Register a listener AND capture buffered events for replay
+        AND snapshot the per-turn inflight content/reasoning + snap_seq
+        in one atomic-against-writers step.

        Used by :func:`make_events_handler` when the client sends
        ``Last-Event-ID`` (header or ``?last_event_id=`` query-param
        fallback for the manual-reconnect path).  Returns

            ``(client_queue, replay_events, status, lost_count,
-            earliest_available_id)``
+            earliest_available_id, snapshot)``

        where ``status`` is one of ``"replay_ok"`` (caller emits the
        replay events then drops into live drain, skipping
@@ -499,19 +508,32 @@ class SessionUIBase:
        envelope then falls through to the fresh-connect replay path
        as the recovery floor — the snapshot picks up the partial
        content/reasoning that the evicted events would have carried).
+        ``snapshot`` has the same shape as
+        :meth:`register_listener_with_in_progress_snapshot`'s second
+        return value: ``{"content": str, "reasoning": str, "seq": int}``.

-        Atomicity contract: under ``_listeners_lock`` we both snapshot
-        the buffer AND register the listener.  A writer's
-        :meth:`_enqueue` takes the same lock, so events either
+        Atomicity contract: under ``_ws_lock`` (outer) + ``_listeners_lock``
+        (inner) — matches writer order in :meth:`on_content_token` —
+        we snapshot the buffer, the listener registration, the
+        inflight content/reasoning, AND the ``_event_id`` counter as
+        a consistent tuple.  Writers' :meth:`_enqueue` blocks on
+        ``_listeners_lock`` for the duration, so events either
          - land in the buffer snapshot but NOT the listener queue
            (writer ran before our lock acquire — caught by the
-            replay slice), or
+            replay slice on the ``replay_ok`` path, or by the
+            content snapshot on the ``truncated`` path), or
          - land in the listener queue but NOT the buffer snapshot
            (writer ran after our lock release — live drain handles
            them, ``_event_id`` is strictly above
-            ``earliest_available_id``).
+            ``earliest_available_id`` AND strictly above
+            ``snapshot["seq"]``).
        No event is double-delivered, none is lost across the
-        registration boundary.
+        registration boundary.  Crucially, the truncated path can
+        now use ``snapshot["seq"]`` as the live-drain ``snap_seq``
+        filter — the events handler's existing ``_seq <= snap_seq``
+        dedup catches any token event that landed in the listener
+        queue AND was covered by the snapshot's content/reasoning
+        text (prevents double-rendering after a truncated emit).

        ``last_event_id`` semantics:
          - ``< earliest_available_id - 1`` → ``"truncated"``.
@@ -528,26 +550,43 @@ class SessionUIBase:
        the cold-start case (the ws just bootstrapped with no events
        ever) and the all-quiet case (a long-idle ws past which all
        events fall out of the buffer cap, but in practice the buffer
-        starts evicting only after 2000 events have been enqueued —
-        which means the counter is >= 2000 and the client's
-        last_event_id is below earliest, so they get ``truncated``
-        instead).  We can't distinguish the two without a separate
-        ``highest_evicted_id`` tracker; treating empty as ``replay_ok``
-        is the safe choice for the genuine cold-start case (no false
-        ``replay_truncated`` envelopes on freshly-opened workstreams).
+        starts evicting only after the cap is hit — which means the
+        counter is at the cap and the client's last_event_id is below
+        earliest, so they get ``truncated`` instead).  We can't
+        distinguish the two without a separate ``highest_evicted_id``
+        tracker; treating empty as ``replay_ok`` is the safe choice
+        for the genuine cold-start case (no false ``replay_truncated``
+        envelopes on freshly-opened workstreams).
        """
        client_queue: queue.Queue[dict[str, Any]] = queue.Queue(maxsize=maxsize)
-        with self._listeners_lock:
-            buffered = list(self._event_buffer)
-            self._listeners.append(client_queue)
+        # Lock order matches writer: ``_ws_lock`` outer, ``_listeners_lock``
+        # inner.  Both inflight buffers AND the buffer slice AND the
+        # ``_event_id`` counter AND the listener registration captured
+        # as one atomic against any concurrent ``_enqueue``.  The string
+        # joins for content/reasoning happen OUTSIDE the locks (bounded
+        # at ``_MAX_TURN_CONTENT_CHARS`` but O(n) over fragments — not
+        # worth blocking on-token writers for the duration).  See
+        # the per-fresh-path helper for the same rationale.
+        with self._ws_lock:
+            captured_content = list(self._ws_inflight_content)
+            captured_reasoning = list(self._ws_inflight_reasoning)
+            with self._listeners_lock:
+                buffered = list(self._event_buffer)
+                self._listeners.append(client_queue)
+                snap_seq = self._event_id
+        snapshot: dict[str, Any] = {
+            "content": "".join(captured_content),
+            "reasoning": "".join(captured_reasoning),
+            "seq": snap_seq,
+        }
        if not buffered:
-            return client_queue, [], "replay_ok", 0, 0
+            return client_queue, [], "replay_ok", 0, 0, snapshot
        earliest_id = buffered[0][0]
        if last_event_id < earliest_id - 1:
            lost_count = (earliest_id - 1) - last_event_id
-            return client_queue, [], "truncated", lost_count, earliest_id
+            return client_queue, [], "truncated", lost_count, earliest_id, snapshot
        replay_events = [ev for eid, ev in buffered if eid > last_event_id]
-        return client_queue, replay_events, "replay_ok", 0, earliest_id
+        return client_queue, replay_events, "replay_ok", 0, earliest_id, snapshot

    # ------------------------------------------------------------------
    # Approval / plan blocking gates