fix(ui): align system-turn row event_id with its SSE event (no double-render)

A first-class operator-context system turn (metacognition nudge, output-guard
finding, interjection, watch result) was persisted stamped with the event-id
counter's PRE-emit value, then its live `on_system_turn` SSE event was emitted
with the post-increment id — so the row sat one below its own event.  On an
in-flight-orphan `/history` resume, `_resume_cursor_and_trim` derives the SSE
replay cursor from the row's id; being one low, the replay redelivered the
turn's own `system_turn` event and the frontend (no dedup) painted the
operator bubble twice.  Reliable for coordinator-spawned children (opened
mid-task) and self-healing on rehydrate — a non-persisted, live-only double.

- `SessionUIBase._enqueue` returns the monotonic `_event_id` it assigns;
  `on_system_turn` returns it; `_append_system_turn` emits the hook first and
  persists the row with that id (fallback to the current cursor for non-SSE
  UIs / a throwing hook).  Now row.event_id == its own SSE event id.
- `project_history_messages` surfaces each row's `event_id` so the frontend
  can dedup.
- app.js: tag each SSE event with its id, reset a per-pane rendered-id set on
  `replayHistory`, and skip a `system_turn` already painted from `/history`
  (belt-and-braces against any future cursor skew).
- Regression tests pin the row/event id alignment, the `/history` emit, and
  the FE dedup.

tests/test_app_js.py

@@ -279,6 +279,20 @@ def test_replay_renders_system_turn_via_add_system_context() -> None:
     )
 
 
+def test_system_turn_dedups_against_history_by_event_id() -> None:
+    """The live ``system_turn`` handler skips an event already painted from
+    ``/history`` (matched by ``_event_id``), so an SSE replay that redelivers
+    it past the resume cursor doesn't double-render the operator bubble —
+    belt-and-braces for the row-vs-event id-alignment fix."""
+    body = _APP_JS.read_text(encoding="utf-8")
+    assert re.search(r"_renderedSystemEventIds\s*\.\s*has\(", body), (
+        "the system_turn handler must skip an event whose id was already rendered from /history."
+    )
+    assert re.search(r"_renderedSystemEventIds\s*\.\s*add\(", body), (
+        "replayHistory (and the live handler) must record system-turn ids for the dedup set."
+    )
+
+
 def test_retry_walk_skips_operator_context_cards() -> None:
     """Interactive twin of the coord retry-skip guard.
     ``_attachRetryToLastAssistant`` walks back past ``.operator-context`` rows

tests/test_history_projection.py

@@ -78,6 +78,19 @@ class TestSystemTurnProjection:
         )
         assert "meta" not in history[0]
 
+    def test_event_id_surfaces_when_set(self) -> None:
+        """``_event_id`` → top-level ``event_id`` so the frontend can dedup a
+        ``/history``-painted system turn against an SSE replay that redelivers
+        it (the resume-cursor seam)."""
+        history = project_history_messages(
+            [{"role": "system", "_source": "start", "content": "x", "_event_id": 7}]
+        )
+        assert history[0]["event_id"] == 7
+
+    def test_event_id_absent_when_unset(self) -> None:
+        history = project_history_messages([{"role": "user", "content": "hello"}])
+        assert "event_id" not in history[0]
+
     def test_legacy_reminders_column_not_projected(self) -> None:
         """A pre-migration row that still carries ``_reminders`` must NOT
         surface a ``reminders`` field — the projection dropped that lane."""

tests/test_sse_cursor_resume.py

@@ -21,14 +21,18 @@ from __future__ import annotations
 import collections
 import os
 import tempfile
-from typing import Any
+from typing import TYPE_CHECKING, Any
 
 os.environ.setdefault("TURNSTONE_JWT_SECRET", "x" * 32)
 
+from tests._session_helpers import make_session
 from turnstone.core.session_routes import _resume_cursor_and_trim
 from turnstone.core.session_ui_base import SessionUIBase
 from turnstone.core.storage._sqlite import SQLiteBackend
 
+if TYPE_CHECKING:
+    import pytest
+
 
 class _ConcreteUI(SessionUIBase):
     pass
@@ -194,6 +198,44 @@ def test_can_replay_from_truncated_false() -> None:
     assert ui.can_replay_from(2) is False  # cursor evicted → would be truncated
 
 
+# ---------------------------------------------------------------------------
+# Operator-context system turn — row event_id == its own SSE event id
+# (the metacognition-nudge double-render regression)
+# ---------------------------------------------------------------------------
+
+
+def test_on_system_turn_returns_buffered_event_id() -> None:
+    """``on_system_turn`` returns the SSE ``_event_id`` it assigned — the same
+    id stamped on the buffered event — so ``_append_system_turn`` persists the
+    row with the id matching its own live event."""
+    ui = _ConcreteUI(ws_id="ws", user_id="u")
+    ui._enqueue({"type": "content"})  # advance the counter
+    eid = ui.on_system_turn("ground yourself", "start", None)
+    assert eid == ui._event_buffer[-1][0]
+    assert ui._event_buffer[-1][1]["type"] == "system_turn"
+
+
+def test_append_system_turn_stamps_row_with_its_sse_event_id(
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    """Regression: a system turn's persisted row carries the SAME ``event_id``
+    as its live ``on_system_turn`` event.  Stamping the row with the pre-emit
+    counter left it one below its own event, so an in-flight-orphan
+    ``/history`` resume cursor derived from the row re-replayed the live event
+    and the operator bubble rendered twice (the metacognition-nudge double)."""
+    session = make_session()
+    ui = session.ui  # NullUI is a real SessionUIBase → increments _event_id
+    captured: dict[str, Any] = {}
+    monkeypatch.setattr(
+        "turnstone.core.session.save_message",
+        lambda *a, **k: captured.update(event_id=k.get("event_id")),
+    )
+    ui._enqueue({"type": "content"})  # advance past the prior turn
+    session._append_system_turn("start", "ground yourself")
+    assert captured["event_id"] == ui._event_buffer[-1][0]
+    assert ui._event_buffer[-1][1]["type"] == "system_turn"
+
+
 # ---------------------------------------------------------------------------
 # Storage: event_id round-trip, get_max_event_id, _event_id reseed
 # ---------------------------------------------------------------------------

turnstone/core/history_decoration.py

@@ -552,6 +552,19 @@ def project_history_messages(
         if attachments_meta:
             entry["attachments"] = attachments_meta
 
+        # Expose the row's SSE event id — the monotonic per-ws Last-Event-ID
+        # cursor (migration 059), NOT a ring-buffer index: ``_enqueue`` only
+        # ever increments it and it's re-seeded from the persisted max on UI
+        # rebuild, so it never wraps or repeats (the deque ring buffer evicts
+        # old *entries*, bounding replay reach, but ids keep climbing).  Lets
+        # the frontend dedup a turn it already painted from ``/history``
+        # against the same turn redelivered by an SSE replay — belt-and-braces
+        # alongside the resume-cursor fix (a system turn's row id now matches
+        # its own live ``system_turn`` event id, so the cursor no longer
+        # re-replays it).
+        if isinstance(msg.get("_event_id"), int):
+            entry["event_id"] = msg["_event_id"]
+
         # (3) ``_source`` side-channel → top-level ``source``.  On a user
         #     row it drives the ``.msg.user.system-nudge`` marker
         #     (wake-driven empty turns); on a first-class operator-context

turnstone/core/session.py

@@ -803,7 +803,7 @@ class SessionUI(Protocol):
     def on_error(self, message: str) -> None: ...
     def on_system_turn(
         self, content: str, source: str, meta: dict[str, Any] | None = None
-    ) -> None: ...
+    ) -> int | None: ...
     def on_state_change(self, state: str) -> None: ...
     def on_rename(self, name: str) -> None: ...
     def on_intent_verdict(self, verdict: dict[str, Any]) -> None:
@@ -3728,13 +3728,15 @@ class ChatSession:
         it after the user / tool / assistant turn it advises.
 
         Mirrors :meth:`_append_user_turn`'s bookkeeping: pushes a
-        ``_msg_tokens`` estimate (parallel to ``self.messages``), persists
-        the row via ``save_message(ws, "system", content, source=source)``
-        so reconnecting tabs replay the same bubble, and fires the live
-        ``on_system_turn`` SSE hook so multi-tab mirrors render it in
-        lockstep.  Hook failures are logged and swallowed — the in-memory
-        append + persist are the load-bearing ops; a UI implementation
-        throwing here must not abort the turn.
+        ``_msg_tokens`` estimate (parallel to ``self.messages``), fires the
+        live ``on_system_turn`` SSE hook so multi-tab mirrors render it in
+        lockstep, then persists the row via
+        ``save_message(ws, "system", content, source=source)`` so
+        reconnecting tabs replay the same bubble.  The hook runs *before* the
+        persist so the row can be stamped with its own SSE event id (see the
+        inline note for why the ordering matters).  Hook failures are logged
+        and swallowed — the in-memory append + persist are the load-bearing
+        ops; a UI implementation throwing here must not abort the turn.
 
         *source* must be one of :data:`tool_advisory.SYSTEM_TURN_SOURCES`;
         extra *meta* is the turn's structured per-kind data (e.g.
@@ -3750,18 +3752,25 @@ class ChatSession:
         self.messages.append(turn_from_dict(turn))
         self._msg_tokens.append(max(1, int(self._msg_char_count(turn) / self._chars_per_token)))
         meta_json = json.dumps(meta) if meta else None
+        # Fire the live SSE hook BEFORE persisting so the row carries the SAME
+        # event_id its ``system_turn`` event carries.  ``on_system_turn``
+        # returns the id ``_enqueue`` assigned (``None`` for non-SSE UIs / test
+        # doubles).  The hook stays best-effort: on failure the persist below
+        # still runs and the row falls back to the current cursor (no live
+        # event was delivered to double anyway).
+        emitted_event_id: int | None = None
+        try:
+            emitted_event_id = self.ui.on_system_turn(content, source, meta or None)
+        except Exception:
+            log.warning("ui.on_system_turn failed; system turn still appended", exc_info=True)
         save_message(
             self._ws_id,
             "system",
             content,
             source=source,
-            event_id=self._ui_event_id(),
+            event_id=emitted_event_id if isinstance(emitted_event_id, int) else self._ui_event_id(),
             meta=meta_json,
         )
-        try:
-            self.ui.on_system_turn(content, source, meta or None)
-        except Exception:
-            log.warning("ui.on_system_turn failed; system turn still appended", exc_info=True)
 
     # -- Main generation loop ------------------------------------------------
 

turnstone/core/session_routes.py

@@ -2690,8 +2690,9 @@ def _resume_cursor_and_trim(
 
     Pure + defensive — reads only ``role`` / ``tool_calls`` /
     ``tool_call_id`` / ``_event_id``.  The ``_event_id`` side-channel
-    survives reconstruct → decorate → extract_reasoning and is dropped by
-    ``project_history_messages`` (which runs on the returned list).
+    survives reconstruct → decorate → extract_reasoning; this runs on the
+    pre-projection list, and ``project_history_messages`` then surfaces it
+    as the top-level ``event_id`` for the frontend.
     """
     if awaiting_approval or not messages:
         return messages, None

turnstone/core/session_ui_base.py

@@ -406,7 +406,7 @@ class SessionUIBase:
     # Listener plumbing (SSE)
     # ------------------------------------------------------------------
 
-    def _enqueue(self, data: dict[str, Any]) -> None:
+    def _enqueue(self, data: dict[str, Any]) -> int:
         """Fan ``data`` out to every registered listener queue.
 
         Stamps ``ws_id`` on the payload if not already present so the
@@ -426,6 +426,12 @@ class SessionUIBase:
         ``(event_id, listeners, buffer)`` tuple — no event is
         fanned out to a not-yet-registered listener AND missing from
         the replay buffer.
+
+        Returns the monotonic ``_event_id`` assigned to this event so a
+        caller that also persists the same turn (e.g.
+        ``ChatSession._append_system_turn``) can stamp the row with the
+        matching id, keeping the ``/history`` resume cursor and the live
+        event stream aligned.
         """
         if "ws_id" not in data:
             data = {**data, "ws_id": self.ws_id}
@@ -444,6 +450,7 @@ class SessionUIBase:
         for lq in snapshot:
             with contextlib.suppress(queue.Full):
                 lq.put_nowait(data)
+        return event_id
 
     def _register_listener(
         self, maxsize: int = _DEFAULT_LISTENER_QUEUE_MAX
@@ -2363,7 +2370,9 @@ class SessionUIBase:
     def on_error(self, message: str) -> None:
         self._enqueue({"type": "error", "message": message})
 
-    def on_system_turn(self, content: str, source: str, meta: dict[str, Any] | None = None) -> None:
+    def on_system_turn(
+        self, content: str, source: str, meta: dict[str, Any] | None = None
+    ) -> int | None:
         """Surface a first-class operator-context system turn as its own
         UI element.
 
@@ -2382,8 +2391,12 @@ class SessionUIBase:
         ``watch_name`` / command / poll counters) so the frontend can rebuild
         per-kind rendering (the watch-result card).  ``None`` for kinds with
         no structured data.
+
+        Returns the SSE ``_event_id`` assigned to the emitted event so the
+        caller persists the row with the matching id (``None`` for UIs
+        without an event stream).
         """
-        self._enqueue(
+        return self._enqueue(
             {"type": "system_turn", "content": content, "source": source, "meta": meta or None}
         )
 

turnstone/ui/static/app.js

@@ -802,6 +802,11 @@ class Pane {
         this._lastEventId = this.evtSource.lastEventId;
       }
       const data = JSON.parse(e.data);
+      // Tag the event with its own SSE id so the system_turn handler can
+      // dedup a turn already painted from /history against the same turn
+      // redelivered by an SSE replay.  e.lastEventId is this event's id;
+      // buffered events (system_turn included) always carry one.
+      if (e.lastEventId) data._event_id = e.lastEventId;
       this.handleEvent(data);
     };
 
@@ -1200,7 +1205,7 @@ class Pane {
         this.addErrorMessage(evt.message);
         break;
 
-      case "system_turn":
+      case "system_turn": {
         // First-class operator-context system turn (output-guard finding,
         // user interjection, metacognitive nudge — see
         // tool_advisory.make_system_turn).  Consolidates the legacy
@@ -1209,12 +1214,30 @@ class Pane {
         // so by the time this SSE event arrives the related turn already
         // rendered).  ``evt.source`` carries the kind for the bubble label;
         // ``evt.meta`` the structured per-kind fields (watch-result card).
+        // Dedup: if this turn was already painted from /history (its row id
+        // matches this event's id) an SSE replay redelivered it — skip.  With
+        // the resume-cursor fix this shouldn't recur, but the guard keeps the
+        // /history+replay seam idempotent for system turns regardless.
+        const sysEid = evt._event_id != null ? String(evt._event_id) : null;
+        if (
+          sysEid &&
+          this._renderedSystemEventIds &&
+          this._renderedSystemEventIds.has(sysEid)
+        ) {
+          break;
+        }
         this.addSystemContext(
           evt.content || "",
           evt.source || "",
           evt.meta || null,
         );
+        if (sysEid) {
+          if (!this._renderedSystemEventIds)
+            this._renderedSystemEventIds = new Set();
+          this._renderedSystemEventIds.add(sysEid);
+        }
         break;
+      }
 
       case "message_queued":
         // Confirmation from server that a queued message was accepted.
@@ -1923,6 +1946,10 @@ class Pane {
 
   replayHistory(messages) {
     this.messagesEl.replaceChildren();
+    // Reset the per-pane dedup set: ids of operator-context system turns
+    // already painted from /history.  A later SSE replay that redelivers one
+    // (resume-cursor overlap) is skipped by the system_turn handler.
+    this._renderedSystemEventIds = new Set();
     if (!messages.length) {
       this.showEmptyState();
       return;
@@ -2162,6 +2189,9 @@ class Pane {
           msg.source || "",
           msg.meta || null,
         );
+        if (msg.event_id != null) {
+          this._renderedSystemEventIds.add(String(msg.event_id));
+        }
         lastToolBlock = null;
       }
     }