fix(sse): always advance _ws_inflight_seq on emit, even past cap

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

@@ -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,

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: