From cdc1dbcc1d44c2f8fb1123da2ef6c98cf5dc2cc4 Mon Sep 17 00:00:00 2001
From: Patrick Buckley <eous@users.noreply.github.com>
Date: Sat, 30 May 2026 03:47:44 -0700
Subject: [PATCH] feat(sse): event-id cursor resume for fresh-connect in-flight
 tool batches

A fresh browser connect during a parallel tool batch (e.g. several
web_fetch) left completed siblings' tool blocks empty until a manual
refresh: each tool_result SSE event fires the instant a sibling
finishes, but the result messages persist only after the whole batch
returns, so a fresh connect replayed neither the already-fired event
(a fresh connect doesn't replay the ring buffer) nor a /history row.

Route the fresh connect through the same delta replay a reconnect
already uses. Persist the per-ws SSE ring-buffer high-water mark
(_event_id) onto each saved conversation row. /history returns the
committed snapshot up to a resolved-turn-boundary cursor and omits the
trailing executing in-flight turn; the client opens its initial SSE
with that cursor (Last-Event-ID) so the existing replay_ok path
fast-forwards the in-flight turn whole -- tool blocks, results, and
approve/plan prompts all rebuild from the ring buffer.

The cut sits at the last resolved-turn boundary (not max(saved
event_id)), so out-of-order result saves in the post-batch loop can't
move it or strand a sibling. Gated on buffer-liveness (can_replay_from):
reloaded / evicted / awaiting-approval cases keep the in-flight turn in
/history and return a null cursor, falling back to the synthetic
snapshot floor -- preserving the existing in-flight render and never
leaving a turn unrenderable.

- Migration 059: nullable event_id BIGINT on conversations + a
  (ws_id, event_id) index (keeps the cold-open high-water reseed a seek).
- save_message(event_id=) across the storage wrapper / protocol /
  sqlite / postgres backends; get_max_event_id; reconstruct_messages
  surfaces the _event_id side-channel.
- SessionUIBase: reseed _event_id from storage on construction (so the
  id space stays monotonic across restarts); can_replay_from() gate.
- make_history_handler: _resume_cursor_and_trim() + cursor in the
  response (WorkstreamHistoryResponse.cursor). The shared projection,
  export, and coord-rebuild paths are untouched.
- app.js: seed the resume cursor on the initial-connect path only, and
  gate the last_event_id param on != null so a cursor of 0 (a brand-new
  workstream's first-turn boundary) is not dropped.

Tests: helper, storage round-trip, and seed unit tests; two
make_history_handler integration tests (cursor + orphan-trim when
replayable, null cursor + orphan kept when not); app.js static guards.
Migration applies up and down on SQLite.
---
 tests/test_app_js.py                          |  55 ++++
 tests/test_sse_cursor_resume.py               | 270 ++++++++++++++++++
 tests/test_workstream_endpoints.py            |  71 +++++
 turnstone/api/server_schemas.py               |  13 +
 turnstone/core/memory.py                      |   7 +
 turnstone/core/session.py                     |  25 +-
 turnstone/core/session_routes.py              |  95 +++++-
 turnstone/core/session_ui_base.py             |  68 +++++
 turnstone/core/storage/_postgresql.py         |  13 +
 turnstone/core/storage/_protocol.py           |  19 ++
 turnstone/core/storage/_schema.py             |  14 +
 turnstone/core/storage/_sqlite.py             |  13 +
 turnstone/core/storage/_utils.py              |  30 +-
 .../versions/059_conversations_event_id.py    |  51 ++++
 turnstone/ui/static/app.js                    |  36 ++-
 15 files changed, 765 insertions(+), 15 deletions(-)
 create mode 100644 tests/test_sse_cursor_resume.py
 create mode 100644 turnstone/core/storage/migrations/versions/059_conversations_event_id.py

diff --git a/tests/test_app_js.py b/tests/test_app_js.py
index e6b8c000..94849ae4 100644
--- a/tests/test_app_js.py
+++ b/tests/test_app_js.py
@@ -188,6 +188,61 @@ def test_replay_history_renders_persisted_verdict_badge() -> None:
     )
 
 
+def test_refetch_history_seeds_resume_cursor_only_on_initial_connect() -> None:
+    """``_refetchHistory`` must seed ``_lastEventId`` from a non-null
+    ``data.cursor`` so the initial ``connectSSE`` opens with
+    ``?last_event_id=`` and takes the ``replay_ok`` fast-forward —
+    rebuilding the executing in-flight turn that ``/history`` omitted
+    (the fresh-connect-during-parallel-batch fix).
+
+    Two load-bearing guards are pinned here:
+
+    1. The seed is gated on ``seedCursor`` so ONLY the initial-connect
+       caller (``_loadHistoryThenConnect``, which reconnects) seeds it;
+       the clear_ui / replay_truncated re-render callers (no reconnect)
+       must NOT rewind ``_lastEventId`` off the live stream position.
+    2. ``connectSSE`` gates the ``?last_event_id=`` param on
+       ``!= null`` (not truthiness) so a valid cursor of 0 — a brand-new
+       ws's first-turn boundary — isn't silently dropped to the fresh
+       snapshot path."""
+    body = _APP_JS.read_text(encoding="utf-8")
+    # ``_refetchHistory`` is an ``async`` method, which the shared
+    # ``_pane_method_offset`` header regex doesn't match — anchor on the
+    # definition directly and bound at the next method.
+    start = body.index("async _refetchHistory(")
+    end = body.index("_refetchWorkstreamsAndReassign(", start)
+    fn = body[start:end]
+    # (1a) seed is gated on BOTH seedCursor AND a non-null cursor.
+    seed_re = re.compile(
+        r"if\s*\(\s*seedCursor\s*&&\s*data\.cursor\s*!=\s*null\s*\)\s*"
+        r"this\._lastEventId\s*=\s*data\.cursor"
+    )
+    assert seed_re.search(fn), (
+        "_refetchHistory must seed this._lastEventId only when "
+        "seedCursor AND data.cursor != null — so re-render callers don't "
+        "rewind the live stream and a 0 cursor still fast-forwards."
+    )
+    assert "seedCursor = false" in fn, (
+        "seedCursor must default false so the clear_ui / replay_truncated "
+        "re-render callers (which pass only 2 args) never seed the cursor."
+    )
+    # (1b) the initial-connect path opts in with seedCursor=true.
+    assert "_refetchHistory(wsId, token, true)" in body, (
+        "_loadHistoryThenConnect must call _refetchHistory(..., true) so "
+        "the reconnecting initial-connect path is the only seeder."
+    )
+    # (2) connectSSE gates the last_event_id param on != null, not truthiness.
+    assert re.search(
+        r"if\s*\(\s*this\._lastEventId\s*!=\s*null\s*\)\s*\{\s*"
+        r"evtUrl\s*\+=\s*\"\?last_event_id=\"",
+        body,
+    ), (
+        "connectSSE must gate the ?last_event_id= param on "
+        "this._lastEventId != null (not truthiness) — else a cursor of 0 "
+        "(brand-new ws first turn) is dropped to the fresh snapshot path."
+    )
+
+
 def test_shared_utils_defines_replay_advisories_after_tool() -> None:
     """The shared ``replayAdvisoriesAfterTool`` helper in
     ``shared_static/utils.js`` is the single source of advisory-walk +
diff --git a/tests/test_sse_cursor_resume.py b/tests/test_sse_cursor_resume.py
new file mode 100644
index 00000000..cfa5351b
--- /dev/null
+++ b/tests/test_sse_cursor_resume.py
@@ -0,0 +1,270 @@
+"""Tests for the SSE event-id cursor-resume model.
+
+The fresh-connect fast-forward (issue: completed siblings of a parallel
+tool batch render empty until refresh).  ``/history`` returns the
+committed snapshot up to a cursor and OMITS the trailing executing
+in-flight turn; the client opens its initial SSE with that cursor so the
+existing ``replay_ok`` delta replays the in-flight turn whole.
+
+Covers the four seams that decide correctness:
+  - ``_resume_cursor_and_trim`` — the cut decision + the resolved-boundary
+    cursor (the property that makes out-of-order result saves safe).
+  - ``SessionUIBase.can_replay_from`` — the buffer-liveness gate.
+  - ``save_message(event_id=)`` round-trip + ``get_max_event_id`` +
+    ``_event_id`` reseed on UI construction.
+  - the in-flight delta actually flows through ``register_listener_with_replay``
+    from a cursor, and the orphan's content is in /history not the snapshot.
+"""
+
+from __future__ import annotations
+
+import collections
+import os
+import tempfile
+from typing import Any
+
+os.environ.setdefault("TURNSTONE_JWT_SECRET", "x" * 32)
+
+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
+
+
+class _ConcreteUI(SessionUIBase):
+    pass
+
+
+class _FakeUI:
+    """Minimal stand-in exposing only ``can_replay_from`` for the helper."""
+
+    def __init__(self, can_replay: bool = True) -> None:
+        self._can = can_replay
+        self.seen_cursor: int | None = None
+
+    def can_replay_from(self, cursor: int) -> bool:
+        self.seen_cursor = cursor
+        return self._can
+
+
+def _assistant(event_id: int, *call_ids: str) -> dict[str, Any]:
+    return {
+        "role": "assistant",
+        "content": "fetching",
+        "tool_calls": [
+            {"id": c, "function": {"name": "web_fetch", "arguments": "{}"}} for c in call_ids
+        ],
+        "_event_id": event_id,
+    }
+
+
+def _user(event_id: int | None) -> dict[str, Any]:
+    m: dict[str, Any] = {"role": "user", "content": "go"}
+    if event_id is not None:
+        m["_event_id"] = event_id
+    return m
+
+
+def _tool(call_id: str, event_id: int) -> dict[str, Any]:
+    return {"role": "tool", "tool_call_id": call_id, "content": "result", "_event_id": event_id}
+
+
+# ---------------------------------------------------------------------------
+# _resume_cursor_and_trim — the cut decision + resolved-boundary cursor
+# ---------------------------------------------------------------------------
+
+
+def test_trim_live_executing_orphan_returns_resolved_boundary_cursor() -> None:
+    """The bug case: a trailing assistant tool-call turn with no results
+    saved, ws executing (not awaiting), buffer replayable → drop the
+    orphan turn, cursor = the last resolved message's event_id."""
+    msgs = [_user(10), _assistant(12, "A", "B", "C")]
+    ui = _FakeUI(can_replay=True)
+    trimmed, cursor = _resume_cursor_and_trim(msgs, ui, awaiting_approval=False)
+    assert cursor == 10
+    assert trimmed == [msgs[0]]  # orphan assistant dropped
+    assert ui.seen_cursor == 10  # gate consulted with the resolved boundary
+
+
+def test_trim_unaffected_by_out_of_order_partial_result_saves() -> None:
+    """THE (B') property: while the post-batch loop saves results in input
+    order (here B landed first, with a HIGH event_id), the cursor stays
+    pinned at the resolved boundary — so a fresh connect mid-save-loop
+    never drops the not-yet-saved siblings (they fast-forward via the
+    delta).  A max(saved-event_id) cursor would jump to 15 and strip
+    A/C; the resolved-boundary cursor does not."""
+    msgs = [
+        _user(10),
+        _assistant(12, "A", "B", "C"),
+        _tool("B", 15),  # B saved out of order with a high stamp; A, C pending
+    ]
+    trimmed, cursor = _resume_cursor_and_trim(msgs, _FakeUI(True), awaiting_approval=False)
+    assert cursor == 10  # NOT 15 — the race-saved sibling can't move the cut
+    assert trimmed == [msgs[0]]  # whole in-flight turn (assistant + B) dropped
+
+
+def test_no_trim_when_awaiting_approval() -> None:
+    """Awaiting-approval orphans stay on the _pending_approval re-emit
+    path — no cursor, full messages."""
+    msgs = [_user(10), _assistant(12, "A")]
+    trimmed, cursor = _resume_cursor_and_trim(msgs, _FakeUI(True), awaiting_approval=True)
+    assert cursor is None
+    assert trimmed is msgs
+
+
+def test_no_trim_when_buffer_cannot_replay() -> None:
+    """Reloaded / evicted (buffer can't fast-forward) → keep the orphan in
+    /history (#610 block), no cursor."""
+    msgs = [_user(10), _assistant(12, "A")]
+    trimmed, cursor = _resume_cursor_and_trim(
+        msgs, _FakeUI(can_replay=False), awaiting_approval=False
+    )
+    assert cursor is None
+    assert trimmed is msgs
+
+
+def test_no_trim_when_no_orphan() -> None:
+    """Fully-resolved trailing turn → nothing in-flight, no cursor."""
+    msgs = [_user(10), _assistant(12, "A", "B"), _tool("A", 13), _tool("B", 14)]
+    trimmed, cursor = _resume_cursor_and_trim(msgs, _FakeUI(True), awaiting_approval=False)
+    assert cursor is None
+    assert trimmed is msgs
+
+
+def test_no_trim_without_resolved_boundary_event_id() -> None:
+    """Orphan present but the resolved prefix carries no event_id (old /
+    bulk-saved NULL rows) → no cursor to hand back → snapshot floor."""
+    msgs = [_user(None), _assistant(12, "A")]
+    trimmed, cursor = _resume_cursor_and_trim(msgs, _FakeUI(True), awaiting_approval=False)
+    assert cursor is None
+    assert trimmed is msgs
+
+
+def test_no_trim_when_orphan_is_first_message() -> None:
+    """Orphan at index 0 has no resolved boundary before it → no cursor."""
+    msgs = [_assistant(12, "A")]
+    trimmed, cursor = _resume_cursor_and_trim(msgs, _FakeUI(True), awaiting_approval=False)
+    assert cursor is None
+    assert trimmed is msgs
+
+
+def test_trim_cuts_before_orphan_across_prior_resolved_turns() -> None:
+    """Multi-turn: prior turn fully resolved, trailing turn in-flight →
+    cursor = the prior turn's last event_id; only the trailing turn drops."""
+    msgs = [
+        _user(10),
+        _assistant(12, "X"),
+        _tool("X", 14),  # prior turn resolved (event_id 14)
+        _assistant(16, "A", "B"),  # trailing in-flight orphan
+    ]
+    trimmed, cursor = _resume_cursor_and_trim(msgs, _FakeUI(True), awaiting_approval=False)
+    assert cursor == 14
+    assert trimmed == msgs[:3]
+
+
+# ---------------------------------------------------------------------------
+# SessionUIBase.can_replay_from — buffer-liveness gate
+# ---------------------------------------------------------------------------
+
+
+def test_can_replay_from_empty_buffer_false() -> None:
+    ui = _ConcreteUI(ws_id="ws", user_id="u")
+    assert ui.can_replay_from(0) is False
+
+
+def test_can_replay_from_within_buffer_true() -> None:
+    ui = _ConcreteUI(ws_id="ws", user_id="u")
+    for _ in range(5):
+        ui._enqueue({"type": "t"})  # ids 1..5
+    assert ui.can_replay_from(2) is True
+    assert ui.can_replay_from(0) is True
+
+
+def test_can_replay_from_no_events_past_cursor_false() -> None:
+    ui = _ConcreteUI(ws_id="ws", user_id="u")
+    for _ in range(5):
+        ui._enqueue({"type": "t"})
+    assert ui.can_replay_from(5) is False  # nothing in-flight to fast-forward
+
+
+def test_can_replay_from_truncated_false() -> None:
+    ui = _ConcreteUI(ws_id="ws", user_id="u")
+    ui._event_buffer = collections.deque(maxlen=3)
+    for _ in range(20):
+        ui._enqueue({"type": "t"})  # buffer holds ids 18,19,20
+    assert ui.can_replay_from(2) is False  # cursor evicted → would be truncated
+
+
+# ---------------------------------------------------------------------------
+# Storage: event_id round-trip, get_max_event_id, _event_id reseed
+# ---------------------------------------------------------------------------
+
+
+def _backend() -> SQLiteBackend:
+    return SQLiteBackend(os.path.join(tempfile.mkdtemp(), "t.db"))
+
+
+def test_event_id_round_trip_and_null() -> None:
+    s = _backend()
+    s.save_message("ws1", "assistant", "hi", tool_calls='[{"id":"A"}]', event_id=46)
+    s.save_message("ws1", "user", "next")  # no event_id → NULL
+    msgs = s.load_messages("ws1", repair=False)
+    assert [m.get("_event_id") for m in msgs] == [46, None]
+
+
+def test_get_max_event_id() -> None:
+    s = _backend()
+    assert s.get_max_event_id("ws1") is None  # no rows
+    s.save_message("ws1", "user", "a", event_id=5)
+    s.save_message("ws1", "assistant", "b", event_id=9)
+    s.save_message("ws1", "user", "c")  # NULL doesn't lower the max
+    assert s.get_max_event_id("ws1") == 9
+    assert s.get_max_event_id("other") is None
+
+
+def test_event_id_seeded_on_ui_construction(monkeypatch: Any) -> None:
+    """A rebuilt UI reseeds _event_id from the persisted high-water so the
+    cursor space stays monotonic across process restarts."""
+
+    class _Stub:
+        def get_max_event_id(self, ws_id: str) -> int | None:
+            return 99
+
+    monkeypatch.setattr(
+        "turnstone.core.storage._registry.get_storage", lambda: _Stub(), raising=True
+    )
+    ui = _ConcreteUI(ws_id="ws-reopen", user_id="u")
+    assert ui._event_id == 99
+    # Next emitted event continues strictly above the seed (no collision).
+    ui._enqueue({"type": "t"})
+    assert ui._event_buffer[-1][0] == 100
+
+
+# ---------------------------------------------------------------------------
+# The in-flight delta flows from the cursor; orphan content is in /history
+# ---------------------------------------------------------------------------
+
+
+def test_cursor_replays_inflight_discrete_events_not_content() -> None:
+    """With cursor = the resolved boundary, register_listener_with_replay
+    yields exactly the in-flight turn's events (here: the tool_result the
+    fresh connect was missing), and the content tokens that streamed
+    BEFORE the assistant committed are <= cursor (carried by /history,
+    not re-streamed)."""
+    ui = _ConcreteUI(ws_id="ws", user_id="u")
+    # prior resolved turn ends at event 10 (the cursor)
+    for _ in range(10):
+        ui._enqueue({"type": "noise"})
+    cursor = ui._event_id  # 10
+    # in-flight turn: assistant content streamed + committed, then tools
+    ui.on_content_token("Let me fetch")
+    ui.on_turn_committed()  # resets inflight buffers BEFORE tools run
+    ui._enqueue({"type": "tool_info", "items": [{"call_id": "A"}]})
+    ui.on_tool_result("A", "web_fetch", "result-A")
+    _lq, replay, status, *_ = ui.register_listener_with_replay(cursor)
+    assert status == "replay_ok"
+    types = [e.get("type") for e in replay]
+    assert "tool_info" in types and "tool_result" in types
+    # Snapshot is EMPTY during the tool-execution window (committed reset it),
+    # so the orphan's content must come from /history — confirmed here.
+    _lq2, snap = ui.register_listener_with_in_progress_snapshot()
+    assert snap["content"] == ""
diff --git a/tests/test_workstream_endpoints.py b/tests/test_workstream_endpoints.py
index 178401df..1af21f90 100644
--- a/tests/test_workstream_endpoints.py
+++ b/tests/test_workstream_endpoints.py
@@ -1295,6 +1295,77 @@ class TestHistoryInteractive:
         assistant_turn = next(m for m in messages if m.get("role") == "assistant")
         assert assistant_turn.get("pending") is True
 
+    def test_history_returns_cursor_and_trims_inflight_orphan_when_replayable(
+        self, _inject_storage
+    ):
+        """Fresh-connect fast-forward, end to end: an executing in-flight
+        orphan (assistant tool_calls saved, no results) whose live ring
+        buffer can replay → /history OMITS that turn and returns
+        ``cursor`` = the resolved boundary's event_id.  The client opens
+        its initial SSE with that cursor so the delta rebuilds the turn.
+        """
+        import json
+
+        ws_id = "ws-cursor"
+        _inject_storage.register_workstream(ws_id, kind="interactive", user_id="test-user")
+        _inject_storage.save_message(ws_id, "user", "kick off", event_id=10)
+        tc_json = json.dumps(
+            [{"id": "call_1", "type": "function", "function": {"name": "bash", "arguments": "{}"}}]
+        )
+        _inject_storage.save_message(ws_id, "assistant", "Working", tool_calls=tc_json, event_id=12)
+        # call_1 result not yet persisted — executing in-flight orphan.
+        mock_ws = MagicMock()
+        mock_ws.id = ws_id
+        mock_ws.ui._pending_approval = None  # executing, not awaiting
+        mock_ws.ui.can_replay_from.return_value = True  # buffer can fast-forward
+        mock_mgr = MagicMock()
+        mock_mgr.get.return_value = mock_ws
+        client = _build_history_app(mock_mgr, _inject_storage)
+
+        r = client.get(f"/v1/api/workstreams/{ws_id}/history")
+        assert r.status_code == 200
+        body = r.json()
+        # Cursor = the resolved boundary (the user row's event_id), NOT the
+        # orphan assistant's stamp.
+        assert body["cursor"] == 10
+        # The executing orphan turn is OMITTED — it fast-forwards via the
+        # SSE delta, disjoint from this committed snapshot.
+        assert [m.get("role") for m in body["messages"]] == ["user"]
+        # The gate was consulted with the resolved-boundary cursor.
+        mock_ws.ui.can_replay_from.assert_called_once_with(10)
+
+    def test_history_keeps_orphan_and_nulls_cursor_when_not_replayable(self, _inject_storage):
+        """Counterpart: when the live buffer can't fast-forward (reloaded /
+        evicted), /history keeps the in-flight turn (the #610 history-
+        rendered block) and returns ``cursor: null`` — the client connects
+        fresh to the synthetic-snapshot floor, never leaving the turn
+        unrenderable."""
+        import json
+
+        ws_id = "ws-cursor-reload"
+        _inject_storage.register_workstream(ws_id, kind="interactive", user_id="test-user")
+        _inject_storage.save_message(ws_id, "user", "kick off", event_id=10)
+        tc_json = json.dumps(
+            [{"id": "call_1", "type": "function", "function": {"name": "bash", "arguments": "{}"}}]
+        )
+        _inject_storage.save_message(ws_id, "assistant", "Working", tool_calls=tc_json, event_id=12)
+        mock_ws = MagicMock()
+        mock_ws.id = ws_id
+        mock_ws.ui._pending_approval = None
+        mock_ws.ui.can_replay_from.return_value = False  # empty/evicted buffer
+        mock_mgr = MagicMock()
+        mock_mgr.get.return_value = mock_ws
+        client = _build_history_app(mock_mgr, _inject_storage)
+
+        r = client.get(f"/v1/api/workstreams/{ws_id}/history")
+        assert r.status_code == 200
+        body = r.json()
+        assert body["cursor"] is None
+        # Orphan turn stays in /history (renders its #610 block); not pending.
+        assert [m.get("role") for m in body["messages"]] == ["user", "assistant"]
+        assistant_turn = next(m for m in body["messages"] if m.get("role") == "assistant")
+        assert assistant_turn.get("pending") is not True
+
     def test_history_does_not_synthesize_orphan_results(self, _inject_storage):
         """``repair=False`` via ``/history`` must NOT splice synthetic
         ``"Tool execution was cancelled."`` rows for mid-conversation
diff --git a/turnstone/api/server_schemas.py b/turnstone/api/server_schemas.py
index 34558c06..52c9a6d3 100644
--- a/turnstone/api/server_schemas.py
+++ b/turnstone/api/server_schemas.py
@@ -485,6 +485,19 @@ class WorkstreamHistoryResponse(BaseModel):
             "by the ``limit`` query parameter (default 100, max 500)."
         ),
     )
+    cursor: int | None = Field(
+        default=None,
+        description=(
+            "SSE resume cursor (a ``Last-Event-ID`` value). Non-null only "
+            "when the trailing turn is an executing in-flight tool batch "
+            "that the live ring buffer can replay: ``messages`` then omits "
+            "that turn and the client opens its initial SSE with this "
+            "cursor so the existing delta replay fast-forwards the "
+            "in-flight turn (tool calls, results, prompts) instead of the "
+            "lossy synthetic snapshot. Null on every other read — the "
+            "client connects fresh."
+        ),
+    )
 
 
 # ---------------------------------------------------------------------------
diff --git a/turnstone/core/memory.py b/turnstone/core/memory.py
index 65374c1e..739efccc 100644
--- a/turnstone/core/memory.py
+++ b/turnstone/core/memory.py
@@ -43,6 +43,7 @@ def save_message(
     tool_calls: str | None = None,
     source: str | None = None,
     reminders: str | None = None,
+    event_id: int | None = None,
 ) -> int:
     """Log a message to the conversations table.
 
@@ -53,6 +54,11 @@ def save_message(
     ``_reminders`` side-channels (``reminders`` JSON-encoded).  Both
     default to ``None`` for the common case where no metacog payload
     rides the row.
+
+    ``event_id`` is the per-ws SSE ring-buffer high-water mark at save
+    time (``SessionUIBase._event_id``); the caller in ``session.py``
+    passes ``self.ui._event_id`` so ``/history`` can return it as the
+    ``Last-Event-ID`` resume cursor.  ``None`` for offline / bulk saves.
     """
     try:
         return get_storage().save_message(
@@ -65,6 +71,7 @@ def save_message(
             tool_calls=tool_calls,
             source=source,
             reminders=reminders,
+            event_id=event_id,
         )
     except Exception:
         log.warning("Failed to save message for ws=%s role=%s", ws_id, role, exc_info=True)
diff --git a/turnstone/core/session.py b/turnstone/core/session.py
index 14a13d64..b3205dc6 100644
--- a/turnstone/core/session.py
+++ b/turnstone/core/session.py
@@ -2130,6 +2130,17 @@ class ChatSession:
             self._tool_error_flags[call_id] = True
         self.ui.on_tool_result(call_id, name, output, is_error=is_error)
 
+    def _ui_event_id(self) -> int | None:
+        """Current per-ws SSE ring-buffer high-water mark for stamping
+        saved messages with the ``Last-Event-ID`` resume cursor.
+
+        Returns ``getattr(self.ui, "_event_id", None)`` — ``None`` for
+        UIs without the counter (CLI / eval / placeholder), whose rows
+        then stay NULL and are treated by ``/history`` as "no
+        fast-forward cursor available" (the synthetic-snapshot floor).
+        """
+        return getattr(self.ui, "_event_id", None)
+
     def _remaining_token_budget(self) -> int:
         """Estimate how many tokens are available for new content.
 
@@ -3548,6 +3559,7 @@ class ChatSession:
             user_input,
             source=source if isinstance(source, str) and source else None,
             reminders=reminders_json,
+            event_id=self._ui_event_id(),
         )
         if attachments and message_id:
             mark_attachments_consumed(
@@ -3734,6 +3746,7 @@ class ChatSession:
                         content,
                         provider_data=provider_data,
                         tool_calls=tool_calls_json,
+                        event_id=self._ui_event_id(),
                     )
 
                 tool_calls = assistant_msg.get("tool_calls")
@@ -3980,6 +3993,7 @@ class ChatSession:
                         _tname,
                         tool_call_id=tc_id,
                         reminders=tool_reminders_json,
+                        event_id=self._ui_event_id(),
                     )
                 # Fold ``user_feedback`` (text typed alongside an approval,
                 # e.g. "y, use full path") and any queued messages that
@@ -4040,7 +4054,7 @@ class ChatSession:
                     msg["content"] = content + "\n\n[generation cancelled before completion]"
                 else:
                     msg["content"] = "[generation cancelled before completion]"
-                save_message(self._ws_id, "assistant", msg["content"])
+                save_message(self._ws_id, "assistant", msg["content"], event_id=self._ui_event_id())
                 self.messages.append(msg)
                 tok_est = max(
                     1,
@@ -4127,7 +4141,14 @@ class ChatSession:
                     }
                 )
                 self._msg_tokens.append(1)
-                save_message(self._ws_id, "tool", reason, func_name, tool_call_id=tc_id)
+                save_message(
+                    self._ws_id,
+                    "tool",
+                    reason,
+                    func_name,
+                    tool_call_id=tc_id,
+                    event_id=self._ui_event_id(),
+                )
                 # Emit synthetic tool_result so live SSE listeners can
                 # complete the in-DOM tool batch — without this the
                 # coord ``--running`` indicator (added by SSE
diff --git a/turnstone/core/session_routes.py b/turnstone/core/session_routes.py
index 9b84ad8a..068c9cf4 100644
--- a/turnstone/core/session_routes.py
+++ b/turnstone/core/session_routes.py
@@ -2681,6 +2681,74 @@ def make_saved_handler(cfg: SessionEndpointConfig) -> Handler:
     return saved_workstreams_handler
 
 
+def _resume_cursor_and_trim(
+    messages: list[dict[str, Any]],
+    ui: Any,
+    awaiting_approval: bool,
+) -> tuple[list[dict[str, Any]], int | None]:
+    """Decide the fresh-connect resume cursor and trim the in-flight turn.
+
+    Returns ``(messages_to_project, cursor)``.
+
+    When the trailing turn is an *executing* in-flight orphan — an
+    assistant ``tool_calls`` message whose results aren't all saved yet,
+    with the ws NOT awaiting approval — AND the live ring buffer can
+    replay the delta past the last resolved-turn boundary, this DROPS the
+    orphan turn from ``/history`` and returns ``cursor`` = the resolved
+    boundary's ``_event_id``.  The client opens its initial SSE with that
+    cursor and the existing ``replay_ok`` path fast-forwards the orphan
+    turn whole (content tokens, ``tool_info``, ``tool_result``, …), so the
+    committed snapshot and the live delta are disjoint — no double-render,
+    no lost siblings (the cursor sits *below* all the orphan's events, so
+    out-of-order result saves can't move it).
+
+    Otherwise returns ``(messages, None)`` unchanged, so the connect takes
+    the synthetic-snapshot floor:
+      - awaiting approval → the ``_pending_approval`` re-emit paints it;
+      - reloaded / evicted (empty or truncated buffer) → the orphan keeps
+        its #610 history-rendered block (never left unrenderable);
+      - quiescent / cursorless history → plain fresh connect.
+
+    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).
+    """
+    if awaiting_approval or not messages:
+        return messages, None
+    can_replay = getattr(ui, "can_replay_from", None)
+    if not callable(can_replay):
+        return messages, None
+    resulted: set[str] = {
+        str(m.get("tool_call_id"))
+        for m in messages
+        if m.get("role") == "tool" and m.get("tool_call_id")
+    }
+    # Locate the trailing assistant tool-call turn (break at the first
+    # one from the end — mirrors project_history_messages' #610 gate) and
+    # whether it still has an unresolved tool_call (an in-flight orphan).
+    orphan_idx: int | None = None
+    for i in range(len(messages) - 1, -1, -1):
+        tcs = messages[i].get("tool_calls")
+        if tcs:
+            has_orphan = any(
+                (tc.get("id") or "") and str(tc.get("id")) not in resulted for tc in tcs
+            )
+            orphan_idx = i if has_orphan else None
+            break
+    if not orphan_idx:  # None (no orphan) or 0 (no resolved boundary before it)
+        return messages, None
+    resolved_ids = [
+        m["_event_id"] for m in messages[:orphan_idx] if isinstance(m.get("_event_id"), int)
+    ]
+    if not resolved_ids:
+        return messages, None  # no committed cursor → snapshot floor
+    cursor = max(resolved_ids)
+    if not can_replay(cursor):
+        return messages, None  # buffer can't fast-forward → #610 floor
+    return messages[:orphan_idx], cursor
+
+
 def make_history_handler(cfg: SessionEndpointConfig) -> Handler:
     """Lifted body for ``GET {prefix}/{ws_id}/history`` — message history.
 
@@ -2796,6 +2864,13 @@ def make_history_handler(cfg: SessionEndpointConfig) -> Handler:
         limit = max(1, min(limit, 500))
 
         messages: list[dict[str, Any]] = []
+        # Fresh-connect resume cursor (the ``Last-Event-ID`` the client
+        # opens its initial SSE with).  Non-None only when the trailing
+        # turn is an executing in-flight orphan that the ring buffer can
+        # fast-forward — see :func:`_resume_cursor_and_trim`.  Stays None
+        # on every other path (and on any decoration failure below) so
+        # the client takes the synthetic-snapshot floor.
+        cursor: int | None = None
         if storage is not None:
             try:
                 # repair=False — display read; see reconstruct_messages docstring.
@@ -2897,6 +2972,17 @@ def make_history_handler(cfg: SessionEndpointConfig) -> Handler:
                     getattr(getattr(live_session, "ui", None), "_pending_approval", None),
                     dict,
                 )
+                # Fresh-connect fast-forward: when the trailing turn is an
+                # executing in-flight orphan the ring buffer can replay,
+                # drop it from the committed snapshot and hand back a
+                # resume cursor so the client's initial SSE rebuilds it via
+                # the existing delta replay (disjoint from /history — no
+                # double-render).  No-op on every other path (returns the
+                # list unchanged + cursor=None).  Runs on the pre-project
+                # list while the ``_event_id`` side-channel is still present.
+                to_project, cursor = _resume_cursor_and_trim(
+                    messages, getattr(live_session, "ui", None), awaiting_approval
+                )
                 # Final structural projection: flatten nested tool_calls,
                 # collapse multipart content, surface the
                 # ``_source`` / ``_reminders`` / ``_attachments_meta``
@@ -2908,7 +2994,7 @@ def make_history_handler(cfg: SessionEndpointConfig) -> Handler:
                 # interactive ``replayHistory`` and the coordinator history
                 # rebuild consume directly — no client-side normaliser.
                 messages = await asyncio.to_thread(
-                    project_history_messages, messages, awaiting_approval
+                    project_history_messages, to_project, awaiting_approval
                 )
             except Exception:
                 # Operationally interesting: a persistent decoration
@@ -2916,13 +3002,16 @@ def make_history_handler(cfg: SessionEndpointConfig) -> Handler:
                 # drift) silently strips verdict pills + output
                 # warnings from every reload of every workstream.
                 # Log at warning so it surfaces in normal log review
-                # rather than only when DEBUG is on.
+                # rather than only when DEBUG is on.  Reset the cursor so a
+                # mid-pipeline failure can't pair an un-trimmed orphan with
+                # a fast-forward cursor (which would double-render it).
+                cursor = None
                 log.warning(
                     "ws.history.decoration_failed ws=%s",
                     ws_id[:8],
                     exc_info=True,
                 )
-        return JSONResponse({"ws_id": ws_id, "messages": messages})
+        return JSONResponse({"ws_id": ws_id, "messages": messages, "cursor": cursor})
 
     return history
 
diff --git a/turnstone/core/session_ui_base.py b/turnstone/core/session_ui_base.py
index 5cf1bf63..cd95ea47 100644
--- a/turnstone/core/session_ui_base.py
+++ b/turnstone/core/session_ui_base.py
@@ -366,6 +366,13 @@ class SessionUIBase:
         # populate it).  Runs last in __init__ so all the lock + state
         # fields the replay touches are already initialised.
         self.replay_recent_auto_approvals_from_audit()
+        # Reseed the monotonic event-id counter from the persisted
+        # high-water so the ``Last-Event-ID`` cursor space stays
+        # monotonic across UI rebuilds (process restart / rehydrate /
+        # coord→node click-through).  Without this it would restart at
+        # 0 and re-issue ids the prior process already stamped onto
+        # ``conversations.event_id`` rows, corrupting cursor ordering.
+        self._seed_event_id_from_storage()
 
     # ------------------------------------------------------------------
     # Listener plumbing (SSE)
@@ -588,6 +595,37 @@ class SessionUIBase:
         replay_events = [ev for eid, ev in buffered if eid > last_event_id]
         return client_queue, replay_events, "replay_ok", 0, earliest_id, snapshot
 
+    def can_replay_from(self, cursor: int) -> bool:
+        """Would an SSE connect with ``Last-Event-ID = cursor`` replay a
+        non-empty delta through the ``replay_ok`` path?
+
+        This is the ``/history`` gate for the fresh-connect fast-forward:
+        the handler returns a resume cursor (and drops the in-flight turn
+        from the committed snapshot) ONLY when this is true, so the
+        in-flight turn is rebuilt from the ring buffer via the existing
+        delta replay.  When it is false — empty buffer (cold reload /
+        process restart), an evicted/truncated cursor, or simply no
+        events past ``cursor`` — the handler keeps the in-flight turn in
+        ``/history`` and returns no cursor, so the connect takes the
+        synthetic-snapshot floor (preserving the #610 in-flight render
+        and never leaving a turn unrenderable).
+
+        Mirrors :meth:`register_listener_with_replay`'s slice semantics
+        without registering a listener:
+          - empty buffer → False (nothing buffered to fast-forward);
+          - ``cursor < earliest_id - 1`` → False (would be ``truncated``);
+          - no event id strictly greater than ``cursor`` → False (the
+            delta would be empty — nothing in-flight to replay).
+        """
+        with self._listeners_lock:
+            if not self._event_buffer:
+                return False
+            earliest_id = self._event_buffer[0][0]
+            latest_id = self._event_buffer[-1][0]
+        if cursor < earliest_id - 1:
+            return False
+        return latest_id > cursor
+
     # ------------------------------------------------------------------
     # Approval / plan blocking gates
     # ------------------------------------------------------------------
@@ -1506,6 +1544,36 @@ class SessionUIBase:
         except Exception:
             log.debug("auto_approve.audit_failed ws=%s", self.ws_id, exc_info=True)
 
+    def _seed_event_id_from_storage(self) -> None:
+        """Reseed :attr:`_event_id` from the persisted high-water mark.
+
+        The per-ws event-id counter (the ``Last-Event-ID`` replay cursor)
+        is in-memory and restarts at 0 on every UI construction — process
+        restart, saved-workstream rehydrate, coord→node click-through.
+        Without reseeding, the new process would re-issue ids the prior
+        one already stamped onto ``conversations.event_id`` rows, so a
+        ``/history`` cursor would point into the wrong generation and the
+        fresh-connect fast-forward would mis-slice the ring buffer.
+
+        Best-effort: any storage error (or no persisted event_id yet)
+        leaves the counter at 0.  No-op without a ``ws_id`` (test
+        fixture).  Called from ``__init__`` before any listener can
+        register, so no lock is needed around the counter write.
+        """
+        if not self.ws_id:
+            return
+        try:
+            from turnstone.core.storage._registry import get_storage
+
+            storage = get_storage()
+            if storage is None:
+                return
+            mx = storage.get_max_event_id(self.ws_id)
+            if mx is not None and mx > self._event_id:
+                self._event_id = int(mx)
+        except Exception:
+            log.debug("ui.seed_event_id_failed ws=%s", self.ws_id[:8], exc_info=True)
+
     def replay_recent_auto_approvals_from_audit(self) -> None:
         """Seed :attr:`_recent_auto_approvals` from the audit log.
 
diff --git a/turnstone/core/storage/_postgresql.py b/turnstone/core/storage/_postgresql.py
index b04f56c7..e09cf5a0 100644
--- a/turnstone/core/storage/_postgresql.py
+++ b/turnstone/core/storage/_postgresql.py
@@ -291,6 +291,7 @@ class PostgreSQLBackend:
         tool_calls: str | None = None,
         source: str | None = None,
         reminders: str | None = None,
+        event_id: int | None = None,
     ) -> int:
         now = datetime.now(UTC).strftime("%Y-%m-%dT%H:%M:%S")
         content = sanitize_text(content)
@@ -311,6 +312,7 @@ class PostgreSQLBackend:
                     tool_calls=tool_calls,
                     _source=source,
                     _reminders=reminders,
+                    event_id=event_id,
                 )
                 .returning(conversations.c.id)
             )
@@ -368,6 +370,7 @@ class PostgreSQLBackend:
                         conversations.c.tool_calls,
                         conversations.c._source,
                         conversations.c._reminders,
+                        conversations.c.event_id,
                     )
                     .where(conversations.c.ws_id == ws_id)
                     .order_by(conversations.c.id.desc())
@@ -386,6 +389,7 @@ class PostgreSQLBackend:
                         conversations.c.tool_calls,
                         conversations.c._source,
                         conversations.c._reminders,
+                        conversations.c.event_id,
                     )
                     .where(conversations.c.ws_id == ws_id)
                     .order_by(conversations.c.id)
@@ -400,6 +404,15 @@ class PostgreSQLBackend:
         attachments = self.load_attachments_for_messages(ws_id, message_ids=message_ids)
         return _reconstruct_messages(list(rows), ws_id, attachments or None, repair=repair)
 
+    def get_max_event_id(self, ws_id: str) -> int | None:
+        with self._conn() as conn:
+            row = conn.execute(
+                sa.select(sa.func.max(conversations.c.event_id)).where(
+                    conversations.c.ws_id == ws_id
+                )
+            ).fetchone()
+        return int(row[0]) if row is not None and row[0] is not None else None
+
     def delete_messages_after(self, ws_id: str, keep_count: int) -> int:
         with self._conn() as conn:
             cutoff_row = conn.execute(
diff --git a/turnstone/core/storage/_protocol.py b/turnstone/core/storage/_protocol.py
index a217c8b5..745d576e 100644
--- a/turnstone/core/storage/_protocol.py
+++ b/turnstone/core/storage/_protocol.py
@@ -159,6 +159,7 @@ class StorageBackend(Protocol):
         tool_calls: str | None = None,
         source: str | None = None,
         reminders: str | None = None,
+        event_id: int | None = None,
     ) -> int:
         """Log a message to the conversations table.
 
@@ -171,6 +172,11 @@ class StorageBackend(Protocol):
         empty user turns).  ``reminders`` is a JSON-encoded list mirroring
         ``_reminders``; both are NULL for the common case where a
         message carries no metacog payload.
+
+        ``event_id`` is the per-ws SSE ring-buffer high-water mark at save
+        time (``SessionUIBase._event_id``) — the ``Last-Event-ID`` resume
+        cursor space, distinct from the returned ``id`` PK.  NULL when the
+        caller has no live UI counter (offline / bulk / fork re-saves).
         """
         ...
 
@@ -209,6 +215,19 @@ class StorageBackend(Protocol):
         """
         ...
 
+    def get_max_event_id(self, ws_id: str) -> int | None:
+        """Return the highest persisted ``event_id`` for ``ws_id``.
+
+        The SSE ``Last-Event-ID`` resume-cursor high-water mark across
+        the workstream's whole life.  ``None`` when no row carries one
+        (fresh ws, or only pre-migration-059 / bulk-saved NULL rows).
+        Used to reseed ``SessionUIBase._event_id`` on UI construction so
+        the per-ws event-id space stays monotonic across process
+        restarts / rehydrates (it resets to 0 otherwise, which would
+        re-issue ids the ring buffer already handed out).
+        """
+        ...
+
     # -- Workstream attachments -----------------------------------------------
 
     def save_attachment(
diff --git a/turnstone/core/storage/_schema.py b/turnstone/core/storage/_schema.py
index 4f9fa0e1..26f7153b 100644
--- a/turnstone/core/storage/_schema.py
+++ b/turnstone/core/storage/_schema.py
@@ -46,9 +46,23 @@ conversations = sa.Table(
     # replay sees the same bubble shape the originating tab saw live.
     sa.Column("_source", sa.Text),
     sa.Column("_reminders", sa.Text),
+    # SSE ``Last-Event-ID`` resume cursor: the per-ws ``_event_id``
+    # ring-buffer high-water mark at the moment this row was saved (see
+    # ``SessionUIBase._enqueue``).  Distinct id-space from the ``id`` PK
+    # (counts SSE events, not messages; per-ws, not table-global).
+    # ``/history`` returns ``max(event_id)`` of the resolved turns as a
+    # cursor so the client's initial SSE fast-forwards the in-flight turn
+    # through the existing delta replay.  Nullable: historical/bulk rows
+    # stay NULL → cursor logic falls back to the snapshot floor.  See
+    # migration 059.
+    sa.Column("event_id", sa.BigInteger),
 )
 
 sa.Index("idx_conversations_timestamp", conversations.c.timestamp)
+# Composite index serving the per-ws ``MAX(event_id)`` reseed (an index
+# seek, not a row scan) and per-ws event-cursor range queries.  See
+# migration 059.
+sa.Index("idx_conversations_ws_event", conversations.c.ws_id, conversations.c.event_id)
 
 workstreams = sa.Table(
     "workstreams",
diff --git a/turnstone/core/storage/_sqlite.py b/turnstone/core/storage/_sqlite.py
index af9244a7..d4d96284 100644
--- a/turnstone/core/storage/_sqlite.py
+++ b/turnstone/core/storage/_sqlite.py
@@ -325,6 +325,7 @@ class SQLiteBackend:
         tool_calls: str | None = None,
         source: str | None = None,
         reminders: str | None = None,
+        event_id: int | None = None,
     ) -> int:
         now = datetime.now(UTC).strftime("%Y-%m-%dT%H:%M:%S")
         content = sanitize_text(content)
@@ -345,6 +346,7 @@ class SQLiteBackend:
                     "tool_calls": tool_calls,
                     "_source": source,
                     "_reminders": reminders,
+                    "event_id": event_id,
                 },
             )
             if result.lastrowid is None:
@@ -429,6 +431,7 @@ class SQLiteBackend:
                         conversations.c.tool_calls,
                         conversations.c._source,
                         conversations.c._reminders,
+                        conversations.c.event_id,
                     )
                     .where(conversations.c.ws_id == ws_id)
                     .order_by(conversations.c.id.desc())
@@ -447,6 +450,7 @@ class SQLiteBackend:
                         conversations.c.tool_calls,
                         conversations.c._source,
                         conversations.c._reminders,
+                        conversations.c.event_id,
                     )
                     .where(conversations.c.ws_id == ws_id)
                     .order_by(conversations.c.id)
@@ -462,6 +466,15 @@ class SQLiteBackend:
         attachments = self.load_attachments_for_messages(ws_id, message_ids=message_ids)
         return _reconstruct_messages(list(rows), ws_id, attachments or None, repair=repair)
 
+    def get_max_event_id(self, ws_id: str) -> int | None:
+        with self._conn() as conn:
+            row = conn.execute(
+                sa.select(sa.func.max(conversations.c.event_id)).where(
+                    conversations.c.ws_id == ws_id
+                )
+            ).fetchone()
+        return int(row[0]) if row is not None and row[0] is not None else None
+
     def delete_messages_after(self, ws_id: str, keep_count: int) -> int:
         with self._conn() as conn:
             # Find the id of the first row to delete (the row at offset keep_count)
diff --git a/turnstone/core/storage/_utils.py b/turnstone/core/storage/_utils.py
index a7bf8e6b..8fe4efaf 100644
--- a/turnstone/core/storage/_utils.py
+++ b/turnstone/core/storage/_utils.py
@@ -336,12 +336,15 @@ def reconstruct_messages(
 ) -> list[dict[str, Any]]:
     """Reconstruct OpenAI message format from stored conversation rows.
 
-    Each *row* is a 9-tuple ``(id, role, content, tool_name,
-    tool_call_id, provider_data, tool_calls_json, source,
-    reminders_json)``, ordered chronologically by row id.  The trailing
-    two columns mirror the ``_source`` / ``_reminders`` in-memory side
-    channels so multi-tab / multi-device replay sees the same bubble
-    shape the originating tab saw live.
+    Each *row* is a 9- or 10-tuple ``(id, role, content, tool_name,
+    tool_call_id, provider_data, tool_calls_json, source, reminders_json
+    [, event_id])``, ordered chronologically by row id.  ``source`` /
+    ``reminders_json`` mirror the ``_source`` / ``_reminders`` in-memory
+    side channels so multi-tab / multi-device replay sees the same bubble
+    shape the originating tab saw live.  The optional 10th element
+    ``event_id`` (migration 059, the per-ws SSE ``Last-Event-ID`` resume
+    cursor) is surfaced as the ``_event_id`` side-channel; legacy 9-tuple
+    fixtures omit it (handled by the defensive unpack below).
 
     When ``attachments_by_msg`` is provided, any user row whose id has
     attachments is rebuilt with multipart list content (text +
@@ -370,7 +373,14 @@ def reconstruct_messages(
             tool_calls_json,
             source,
             reminders_json,
-        ) = row
+        ) = row[:9]
+        # ``event_id`` (10th column, migration 059) is the per-ws SSE
+        # ring-buffer high-water mark stamped at save time — the
+        # ``Last-Event-ID`` resume cursor space.  Surfaced as the
+        # ``_event_id`` side-channel so ``make_history_handler`` can
+        # compute the resume cursor + locate the in-flight-turn boundary.
+        # Defensive length check keeps pre-event_id 9-tuple fixtures valid.
+        event_id = row[9] if len(row) > 9 else None
 
         if role == "user":
             parts: list[dict[str, Any]] = []
@@ -406,6 +416,8 @@ def reconstruct_messages(
                 # swallowed silently rather than aborting load.
                 with contextlib.suppress(json.JSONDecodeError, TypeError):
                     umsg["_reminders"] = json.loads(reminders_json)
+            if event_id is not None:
+                umsg["_event_id"] = int(event_id)
             messages.append(umsg)
 
         elif role == "assistant":
@@ -416,6 +428,8 @@ def reconstruct_messages(
             if tool_calls_json:
                 with contextlib.suppress(json.JSONDecodeError, TypeError):
                     msg["tool_calls"] = json.loads(tool_calls_json)
+            if event_id is not None:
+                msg["_event_id"] = int(event_id)
             messages.append(msg)
 
         elif role == "tool":
@@ -430,6 +444,8 @@ def reconstruct_messages(
                 # the-tool bubble the originating tab rendered live.
                 with contextlib.suppress(json.JSONDecodeError, TypeError):
                     tmsg["_reminders"] = json.loads(reminders_json)
+            if event_id is not None:
+                tmsg["_event_id"] = int(event_id)
             messages.append(tmsg)
 
     if not repair:
diff --git a/turnstone/core/storage/migrations/versions/059_conversations_event_id.py b/turnstone/core/storage/migrations/versions/059_conversations_event_id.py
new file mode 100644
index 00000000..56dc1e8b
--- /dev/null
+++ b/turnstone/core/storage/migrations/versions/059_conversations_event_id.py
@@ -0,0 +1,51 @@
+"""Add ``event_id`` to ``conversations`` for SSE cursor-resume.
+
+The SSE layer stamps every fanned-out event with a per-workstream
+monotonic ``_event_id`` — the ``Last-Event-ID`` replay coordinate held
+in the in-memory ring buffer on ``SessionUIBase``.  Persisting that
+high-water mark on each saved message lets ``/history`` hand back a
+*resume cursor* in the same id-space the ring buffer slices on: the
+client opens its initial SSE with the cursor as ``Last-Event-ID`` and
+the existing delta-replay path fast-forwards the in-flight turn (tool
+blocks, results, approve/plan prompts) instead of the lossy synthetic
+snapshot.  It is also the durable anchor that reseeds the in-memory
+counter across process restarts (it resets to 0 otherwise, which would
+collide ids post-reopen).
+
+Nullable: historical rows (and bulk/fork re-saves) predate or omit the
+counter and stay NULL; the ``/history`` cursor logic treats a
+missing/old cursor as "no fast-forward available" and falls back to the
+synthetic snapshot floor.  ``BigInteger`` because the per-ws counter is
+monotonic across the workstream's whole life (reseeded from
+``MAX(event_id)`` on reopen), so a long-lived, high-throughput
+workstream can exceed 2**31 — distinct from the autoincrement ``id``
+PK, which counts messages, not events.
+
+Revision ID: 059
+Revises: 058
+Create Date: 2026-05-30
+"""
+
+import sqlalchemy as sa
+from alembic import op
+
+revision = "059"
+down_revision = "058"
+branch_labels = None
+depends_on = None
+
+
+def upgrade() -> None:
+    with op.batch_alter_table("conversations") as batch_op:
+        batch_op.add_column(sa.Column("event_id", sa.BigInteger(), nullable=True))
+    # Composite (ws_id, event_id) index: makes the per-ws
+    # ``SELECT MAX(event_id)`` reseed an index seek instead of a scan over
+    # the workstream's rows, and is the index the per-ws event-cursor
+    # queries want.  Cheap on this LLM-paced insert workload.
+    op.create_index("idx_conversations_ws_event", "conversations", ["ws_id", "event_id"])
+
+
+def downgrade() -> None:
+    op.drop_index("idx_conversations_ws_event", table_name="conversations")
+    with op.batch_alter_table("conversations") as batch_op:
+        batch_op.drop_column("event_id")
diff --git a/turnstone/ui/static/app.js b/turnstone/ui/static/app.js
index 10a37e00..496719aa 100644
--- a/turnstone/ui/static/app.js
+++ b/turnstone/ui/static/app.js
@@ -690,7 +690,13 @@ class Pane {
     // the server accepts both forms.  ``_lastEventId`` is captured
     // from the prior source's onmessage handler.
     let evtUrl = "/v1/api/workstreams/" + encodeURIComponent(wsId) + "/events";
-    if (this._lastEventId) {
+    // ``!= null`` (not truthiness): a resume cursor of 0 is valid — the
+    // ring buffer's first emitted event is id 1, so register_listener_with_replay(0)
+    // replays the whole in-flight turn. A brand-new ws seeds _event_id at 0,
+    // so its first user row (and thus a first-turn /history cursor) can be 0;
+    // a truthiness gate would silently drop it and fall back to the lossy
+    // fresh snapshot. Mirrors the ``data.cursor != null`` guard in _refetchHistory.
+    if (this._lastEventId != null) {
       evtUrl += "?last_event_id=" + encodeURIComponent(this._lastEventId);
     }
     this.evtSource = new EventSource(evtUrl);
@@ -768,17 +774,29 @@ class Pane {
     // the pane has switched to another ws. Newest load wins; older ones drop.
     const token = (this._historyLoadToken || 0) + 1;
     this._historyLoadToken = token;
-    this._refetchHistory(wsId, token).finally(() => {
+    // ``seedCursor=true``: this is the initial-connect path (the
+    // ``.finally`` reconnects), so a resume cursor from /history should
+    // seed _lastEventId for that connect. The clear_ui / replay_truncated
+    // re-render callers pass it false — they run on an already-live stream
+    // and must NOT rewind _lastEventId off the live position.
+    this._refetchHistory(wsId, token, true).finally(() => {
       if (token === this._historyLoadToken) this.connectSSE(wsId);
     });
   }
 
-  async _refetchHistory(wsId, token) {
+  async _refetchHistory(wsId, token, seedCursor = false) {
     // Fetch conversation history over REST. Used for first paint (before
     // connecting SSE) and to re-render after a clear_ui signal (rewind /
     // retry / resume / open). The FETCH is wrapped (network/parse failure
     // → empty pane); the render is deliberately OUTSIDE the catch so a
     // render bug surfaces loudly instead of being masked as an empty pane.
+    //
+    // ``seedCursor`` is true ONLY on the initial-connect path
+    // (_loadHistoryThenConnect, which reconnects via .finally). The
+    // re-render callers leave it false so a fast-forward cursor never
+    // rewinds the live stream's _lastEventId backward (which would
+    // double-render on a later transient reconnect, or — on a re-render
+    // that trims an orphan with no reconnect — strand the omitted turn).
     const id = wsId || this.wsId;
     let data = null;
     try {
@@ -794,6 +812,18 @@ class Pane {
     // paint the wrong ws's history into the pane.
     if (token !== undefined && token !== this._historyLoadToken) return;
     if (data) {
+      // Fresh-connect fast-forward: when the trailing turn is an
+      // executing in-flight tool batch the server can replay, /history
+      // returns a non-null ``cursor`` (a Last-Event-ID) and OMITS that
+      // turn from ``messages``. On the initial-connect path only
+      // (``seedCursor``), seed ``_lastEventId`` so the connectSSE below
+      // opens the initial stream with ?last_event_id=, taking the
+      // replay_ok delta path that rebuilds the in-flight turn (tool
+      // calls, results, prompts) through the live handlers — no synthetic
+      // snapshot, no /history-vs-delta double-render. Null cursor leaves
+      // _lastEventId untouched (fresh connect, already nulled by
+      // _loadHistoryThenConnect); the re-render callers never seed.
+      if (seedCursor && data.cursor != null) this._lastEventId = data.cursor;
       // The REST /history payload is already the canonical projected wire
       // shape (server-side projection in make_history_handler:
       // flat tool_calls, top-level source/reminders/attachments, collapsed