diff --git a/tests/test_app_js.py b/tests/test_app_js.py index f64ceb4b..7a3a42e0 100644 --- a/tests/test_app_js.py +++ b/tests/test_app_js.py @@ -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) diff --git a/tests/test_sse_reconnect_replay.py b/tests/test_sse_reconnect_replay.py index 65f725a9..3a54b3d6 100644 --- a/tests/test_sse_reconnect_replay.py +++ b/tests/test_sse_reconnect_replay.py @@ -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" + ) # --------------------------------------------------------------------------- diff --git a/turnstone/console/static/coordinator/coordinator.js b/turnstone/console/static/coordinator/coordinator.js index d5adf22a..fb463107 100644 --- a/turnstone/console/static/coordinator/coordinator.js +++ b/turnstone/console/static/coordinator/coordinator.js @@ -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 diff --git a/turnstone/core/session_routes.py b/turnstone/core/session_routes.py index 743a681c..7ac9991e 100644 --- a/turnstone/core/session_routes.py +++ b/turnstone/core/session_routes.py @@ -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`` diff --git a/turnstone/core/session_ui_base.py b/turnstone/core/session_ui_base.py index ca97bc37..c8682889 100644 --- a/turnstone/core/session_ui_base.py +++ b/turnstone/core/session_ui_base.py @@ -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