From e8eca2ec9b4cc4ed9a1da3cdee99fadec3f1fe65 Mon Sep 17 00:00:00 2001 From: Patrick Buckley Date: Fri, 8 May 2026 18:10:14 -0700 Subject: [PATCH] fix(sse): always advance _ws_inflight_seq on emit, even past cap MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Copilot caught a real bug in the cap+seq interaction: the previous shape only advanced ``_ws_inflight_seq`` when the buffer actually appended, on the theory that "every _seq corresponds to a buffered fragment" was a useful invariant. It wasn't — once the buffer hit its cap, seq stalled at the high-water-pre-cap, so a subscriber that registered AFTER the cap was hit would capture ``snap_seq == stalled_seq``, and every subsequent live token (also tagged with the stalled seq) would be filter-dropped by the events handler's ``seq <= snap_seq`` dedup. Silent loss of the entire post-cap stream for refresh-past-cap tabs. Fix: advance seq on every emit, regardless of buffer cap. The cap is a buffer-size limit, not a stop-streaming signal. Past-cap tokens are absent from the snapshot's text payload (the buffer was truncated at cap) but the live stream past them is now correctly delivered — refresh-after-cap renders snapshot-up-to-cap then live tokens past it, with a visual gap equal to the past-cap chunk and no silent drop of subsequent tokens. Test ``test_inflight_seq_increments_only_on_actual_append`` enforced the buggy invariant and is renamed/flipped to ``test_inflight_seq_advances_on_every_emit_even_at_cap``. Added ``test_subscriber_after_cap_hit_receives_subsequent_tokens`` (and the reasoning equivalent) as direct regressions for the silent-token-loss scenario. --- tests/test_session_ui_base.py | 78 +++++++++++++++++++++++++++---- turnstone/core/session_ui_base.py | 46 ++++++++++-------- 2 files changed, 97 insertions(+), 27 deletions(-) diff --git a/tests/test_session_ui_base.py b/tests/test_session_ui_base.py index 6b2e44a1..09be5ae5 100644 --- a/tests/test_session_ui_base.py +++ b/tests/test_session_ui_base.py @@ -900,12 +900,15 @@ def test_on_reasoning_token_writes_to_inflight_buffer_only() -> None: assert ui._ws_turn_content == [] -def test_inflight_seq_increments_only_on_actual_append() -> None: - """Cap-hit content tokens do NOT bump the seq — preserves the - invariant that every ``_seq`` corresponds to a buffered fragment. - Cap-hit live events get the prior seq and are dropped by the - snapshot dedup filter on refresh (matches the multi-turn cap - behaviour today: capped tokens not visible to refresh).""" +def test_inflight_seq_advances_on_every_emit_even_at_cap() -> None: + """Cap-hit content tokens MUST advance ``_ws_inflight_seq``, + even though the buffer rejected the append. If seq stalled at + high-water-pre-cap, a subscriber registering AFTER the cap is + hit would capture ``snap_seq == stalled_seq`` and every + subsequent live token (also tagged with the stalled seq) would + be filter-dropped by the events handler — silently losing the + rest of the stream. The cap is a buffer-size limit, not a + "stop streaming" signal.""" from turnstone.core.session_ui_base import _MAX_TURN_CONTENT_CHARS ui = _make_ui() @@ -913,12 +916,71 @@ def test_inflight_seq_increments_only_on_actual_append() -> None: while ui._ws_inflight_content_size < _MAX_TURN_CONTENT_CHARS: ui.on_content_token(chunk) seq_at_cap = ui._ws_inflight_seq - # Cap-hit token: seq must NOT increment. + + # Cap-hit token: seq MUST advance (no buffer append, but the + # event still gets a fresh seq for the dedup filter). ui.on_content_token(chunk) - assert ui._ws_inflight_seq == seq_at_cap + assert ui._ws_inflight_seq == seq_at_cap + 1 + # Buffer remains bounded — the cap-hit token is NOT in inflight. assert ui._ws_inflight_content_size <= _MAX_TURN_CONTENT_CHARS + len(chunk) +def test_subscriber_after_cap_hit_receives_subsequent_tokens() -> None: + """Regression for Copilot's cap+seq finding: a subscriber that + connects AFTER the inflight buffer is at cap must still receive + live tokens past the cap. Past-cap tokens are absent from + ``snap.content`` (the snapshot text was truncated at cap) but + the live stream past them must NOT be filter-dropped.""" + from turnstone.core.session_ui_base import _MAX_TURN_CONTENT_CHARS + + ui = _make_ui() + chunk = "x" * 1024 + while ui._ws_inflight_content_size < _MAX_TURN_CONTENT_CHARS: + ui.on_content_token(chunk) + # Stream a few tokens PAST the cap before subscribing. + for _ in range(3): + ui.on_content_token(chunk) + + lq, snap = ui.register_listener_with_in_progress_snapshot() + snap_seq = snap["seq"] + + # Live token past cap. + ui.on_content_token(chunk) + ev = lq.get_nowait() + assert ev["type"] == "content" + # The critical invariant: seq advances per-emit, so the new + # event's _seq is strictly greater than the snap_seq the + # subscriber captured. Without this, the events handler's + # ``seq <= snap_seq`` filter would drop every token past the + # cap (silent token loss for refresh-past-cap). + assert ev["_seq"] > snap_seq, ( + f"Token past cap has _seq={ev['_seq']} which is <= " + f"snap_seq={snap_seq} — would be silently dropped after a " + f"refresh past the cap." + ) + + +def test_subscriber_after_reasoning_cap_hit_receives_subsequent_tokens() -> None: + """Same invariant as content cap: reasoning subscribers past + cap must keep receiving live reasoning tokens.""" + from turnstone.core.session_ui_base import _MAX_TURN_CONTENT_CHARS + + ui = _make_ui() + chunk = "x" * 1024 + while ui._ws_inflight_reasoning_size < _MAX_TURN_CONTENT_CHARS: + ui.on_reasoning_token(chunk) + for _ in range(3): + ui.on_reasoning_token(chunk) + + lq, snap = ui.register_listener_with_in_progress_snapshot() + snap_seq = snap["seq"] + + ui.on_reasoning_token(chunk) + ev = lq.get_nowait() + assert ev["type"] == "reasoning" + assert ev["_seq"] > snap_seq + + def test_on_turn_committed_resets_inflight_after_commit() -> None: """``on_turn_committed`` fires immediately after each ``messages.append(assistant_msg)`` in the send loop. Without it, diff --git a/turnstone/core/session_ui_base.py b/turnstone/core/session_ui_base.py index 9dc5e020..66d71485 100644 --- a/turnstone/core/session_ui_base.py +++ b/turnstone/core/session_ui_base.py @@ -1297,23 +1297,31 @@ class SessionUIBase: def on_reasoning_token(self, text: str) -> None: """Append to the inflight reasoning buffer (capped) + enqueue. - Mirrors :meth:`on_content_token`'s shape — append under - ``_ws_lock``, increment ``_ws_inflight_seq`` only on actual - append, then enqueue with ``_seq`` so the events handler can - dedup live events that are already in a fresh subscriber's - snapshot. Cap-hit tokens reuse the prior seq; the live filter - drops them on refresh, matching the cap-hit semantics for - content (no behaviour regression vs today's no-buffer path). + Mirrors :meth:`on_content_token`'s shape. ``_ws_inflight_seq`` + advances on EVERY emit — even when the buffer cap rejected + the append — so the dedup filter in :func:`make_events_handler` + stays correct for subscribers that register after the cap is + hit. If seq stalled at the high-water-pre-cap, those late + subscribers would capture ``snap_seq == high-water`` and + every subsequent live token (with the same stalled seq) + would be filter-dropped as "already in your snapshot", + silently losing the rest of the stream. The cap is a + buffer-size limit, NOT a "stop streaming" signal. + + Tokens past the cap are absent from ``snap.reasoning`` (the + snapshot text was truncated at cap) but the live stream + continues normally past them — refresh-after-cap renders the + 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. """ seq: int = 0 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._ws_inflight_seq += 1 - seq = self._ws_inflight_seq - else: - seq = self._ws_inflight_seq + self._ws_inflight_seq += 1 + seq = self._ws_inflight_seq self._enqueue({"type": "reasoning", "text": text, "_seq": seq}) def on_content_token(self, text: str) -> None: @@ -1326,10 +1334,12 @@ class SessionUIBase: :meth:`on_turn_start`) — fuels the SSE ``in_progress_snapshot`` event a reconnecting client sees on mid-stream refresh. - Both caps are checked independently; ``_ws_inflight_seq`` is - incremented only when the inflight buffer actually appended, - so cap-hit tokens reuse the prior seq and get dropped by the - live filter on refresh (matching the multi-turn cap behaviour). + Both caps are checked independently. ``_ws_inflight_seq`` + advances on EVERY emit — even when the inflight cap rejected + the append — so a subscriber that registers after the cap is + hit doesn't have every subsequent live token filter-dropped + against a stalled ``snap_seq``. See + :meth:`on_reasoning_token` for the full rationale. The cap-check + append + size-update + seq-bump run under ``_ws_lock`` so a concurrent @@ -1350,10 +1360,8 @@ 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._ws_inflight_seq += 1 - seq = self._ws_inflight_seq - else: - seq = self._ws_inflight_seq + self._ws_inflight_seq += 1 + seq = self._ws_inflight_seq self._enqueue({"type": "content", "text": text, "_seq": seq}) def on_stream_end(self) -> None: