fix(server): boot-epoch staleness signal on the global SSE stream (#881)

The global ring's counter is process-local and reboots at 0, so after a
node restart a pre-restart cursor was first invisibly ahead of the reborn
ring (replay_ok with an empty slice) and then aliased into the new id
space as the counter re-grew — both silently skipping the restart
boundary (ghost rosters).  Every global SSE id is now
"{boot_epoch}-{counter}" (per-process nonce); the browser echoes it
verbatim on native reconnect, so provenance rides every path with zero
client cooperation.  A cursor from any other epoch — prior boot, another
node, a pre-epoch bare-int client, garbage — draws replay_truncated
(reason=boot_epoch, loss unknowable so the numeric fields are omitted)
plus the node_snapshot recovery floor; in-epoch ring misses keep honest
lost_count under reason=ring_evicted.  Same-epoch cursors run the ring
logic unchanged.  Chokepoint log line added; per-ws ids deliberately stay
bare ints (storage-seeded counter — asymmetry documented at both sites);
collector audit ruling recorded at its cursorless connect.

tests/test_global_sse_boot_epoch.py

@@ -0,0 +1,391 @@
+"""Boot-epoch staleness signal on the node-global SSE stream (#881).
+
+The global ring buffer and its event counter are process-local: unlike the
+per-ws lane (whose counter ``_seed_event_id_from_storage`` seeds from
+``MAX(conversations.event_id)``), ``global_event_id_holder`` reboots at 0
+with the process.  A bare integer cursor therefore cannot prove which boot
+minted it — after a restart it is first "ahead" of the reborn ring (empty
+replay slice) and then aliases into the new id space as the counter
+re-grows, both of which used to draw ``replay_ok`` and silently skip the
+restart boundary.
+
+The fix stamps every global SSE ``id:`` as ``"{boot_epoch}-{counter}"``
+and treats any cursor that does not carry the live epoch as stale,
+answering ``replay_truncated`` (``reason="boot_epoch"``) plus the
+``node_snapshot`` recovery floor.  These tests walk the reconnect matrix
+at the :func:`turnstone.server.global_events_sse` boundary:
+
+  cursor ∈ {absent, same-epoch, stale-epoch, legacy bare-int, garbage,
+            negative, forged-against-empty-ring}
+  ×  ring ∈ {empty, covers cursor, evicted past cursor}
+
+The browser half (app.js capture/presentation) is pinned in
+``test_app_js.py``; the end-to-end restart loop is
+``scripts/recovery_e2e.py --scenario roster-restart``.
+"""
+
+from __future__ import annotations
+
+import asyncio
+import collections
+import json
+import queue
+import threading
+from types import SimpleNamespace as SimpleNS
+from typing import Any
+
+import pytest
+from starlette.requests import Request
+
+import turnstone.server as server_mod
+
+EPOCH = "0badf00d"  # test-pinned boot epoch (hex, like secrets.token_hex(4))
+
+
+def _make_app_state(
+    *,
+    buffered: list[tuple[int, dict[str, Any]]] | None = None,
+    epoch: str = EPOCH,
+) -> SimpleNS:
+    """Minimal ``app.state`` for the global SSE handler.
+
+    Real lock / deque / list so registration and slicing run the
+    production code paths; only the snapshot builder is stubbed (its
+    composition has its own coverage in ``test_console.py``).
+    """
+    buf: collections.deque[tuple[int, dict[str, Any]]] = collections.deque(maxlen=50)
+    for item in buffered or []:
+        buf.append(item)
+    return SimpleNS(
+        node_id="node-under-test",
+        global_listeners=[],
+        global_listeners_lock=threading.Lock(),
+        global_event_buffer=buf,
+        global_boot_epoch=epoch,
+        sse_executor=None,  # live loop is never entered by these tests
+    )
+
+
+def _fake_request(
+    app_state: SimpleNS,
+    *,
+    headers: dict[str, str] | None = None,
+    query: dict[str, str] | None = None,
+) -> Request:
+    """ASGI-scope-honest request carrying a service-scoped principal."""
+    header_list = [(k.lower().encode(), v.encode()) for k, v in (headers or {}).items()]
+    query_string = "&".join(f"{k}={v}" for k, v in query.items()).encode() if query else b""
+    scope = {
+        "type": "http",
+        "method": "GET",
+        "headers": header_list,
+        "path": "/v1/api/events/global",
+        "raw_path": b"/v1/api/events/global",
+        "query_string": query_string,
+        "app": SimpleNS(state=app_state),
+        "state": {"auth_result": SimpleNS(scopes=["service"])},
+    }
+
+    async def _recv() -> dict[str, Any]:  # noqa: RUF029 — async signature required
+        return {"type": "http.disconnect"}
+
+    return Request(scope, receive=_recv)
+
+
+_SNAPSHOT_STUB = {"type": "node_snapshot", "node_id": "node-under-test", "workstreams": []}
+
+
+def _drain(
+    monkeypatch: pytest.MonkeyPatch,
+    *,
+    buffered: list[tuple[int, dict[str, Any]]] | None = None,
+    headers: dict[str, str] | None = None,
+    query: dict[str, str] | None = None,
+    max_yields: int = 8,
+    epoch: str = EPOCH,
+) -> list[dict[str, Any]]:
+    """Run the handler and collect its pre-live yields as raw dicts.
+
+    ``max_yields`` must not exceed the pre-live yield count for the
+    branch under test + 1 — the live loop blocks on an executor draw,
+    so every test enumerates its expected frames and stops short.
+    """
+    monkeypatch.setattr(server_mod, "_build_node_snapshot", lambda _s: dict(_SNAPSHOT_STUB))
+    app_state = _make_app_state(buffered=buffered, epoch=epoch)
+    req = _fake_request(app_state, headers=headers, query=query)
+
+    async def _run() -> list[dict[str, Any]]:
+        # Guard against a miscounted ``max_yields`` reaching the live
+        # loop (which blocks on an executor draw and would hang the
+        # test + leak the draw thread) — fail fast and visibly instead.
+        async with asyncio.timeout(10):
+            # ``Any``: the endpoint is annotated ``-> Response``; the SSE
+            # subtype's ``body_iterator`` is what the drain consumes.
+            resp: Any = await server_mod.global_events_sse(req)
+            out: list[dict[str, Any]] = []
+            async for chunk in resp.body_iterator:
+                out.append(chunk)
+                if len(out) >= max_yields:
+                    break
+            await resp.body_iterator.aclose()
+            return out
+
+    return asyncio.run(_run())
+
+
+def _data_frames(yields: list[dict[str, Any]]) -> list[dict[str, Any]]:
+    """JSON-decoded ``data:`` payloads, in yield order."""
+    return [json.loads(y["data"]) for y in yields if "data" in y]
+
+
+def _types(yields: list[dict[str, Any]]) -> list[str]:
+    return [d.get("type", "") for d in _data_frames(yields)]
+
+
+def _ev(eid: int, **extra: Any) -> tuple[int, dict[str, Any]]:
+    """A buffered ring entry the way ``_global_fanout_thread`` stores it:
+    the event dict carries ``_event_id`` and the tuple repeats the id."""
+    return eid, {"type": "ws_state", "ws_id": f"ws-{eid}", "_event_id": eid, **extra}
+
+
+# ---------------------------------------------------------------------------
+# Fresh connect (no cursor)
+# ---------------------------------------------------------------------------
+
+
+def test_fresh_connect_gets_snapshot_no_envelope(monkeypatch: pytest.MonkeyPatch) -> None:
+    yields = _drain(monkeypatch, buffered=[_ev(1)], max_yields=2)
+    assert "retry" in yields[0]
+    assert _types(yields) == ["node_snapshot"]
+
+
+# ---------------------------------------------------------------------------
+# Same-epoch cursors — ring logic must behave exactly as before
+# ---------------------------------------------------------------------------
+
+
+def test_same_epoch_cursor_replays_slice_without_snapshot(
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    yields = _drain(
+        monkeypatch,
+        buffered=[_ev(1), _ev(2), _ev(3)],
+        headers={"Last-Event-ID": f"{EPOCH}-1"},
+        max_yields=3,
+    )
+    frames = _data_frames(yields)
+    assert [f["ws_id"] for f in frames] == ["ws-2", "ws-3"]
+    assert "node_snapshot" not in _types(yields)
+    assert "replay_truncated" not in _types(yields)
+
+
+def test_replayed_and_live_ids_are_epoch_tagged(monkeypatch: pytest.MonkeyPatch) -> None:
+    """Every ``id:`` on the wire carries the live epoch — the browser
+    echoes it verbatim on native reconnect, which is what lets the
+    server prove cursor provenance without any client cooperation."""
+    yields = _drain(
+        monkeypatch,
+        buffered=[_ev(1), _ev(2)],
+        headers={"Last-Event-ID": f"{EPOCH}-0"},
+        max_yields=3,
+    )
+    ids = [y["id"] for y in yields if "id" in y]
+    assert ids == [f"{EPOCH}-1", f"{EPOCH}-2"]
+    # And the internal ``_event_id`` never leaks onto the wire.
+    assert all("_event_id" not in f for f in _data_frames(yields))
+
+
+def test_same_epoch_cursor_at_head_is_caught_up_replay_ok(
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    """A cursor at the newest id is the normal caught-up reconnect:
+    nothing to replay, no snapshot, and crucially no false truncated."""
+    yields = _drain(
+        monkeypatch,
+        buffered=[_ev(4), _ev(5)],
+        headers={"Last-Event-ID": f"{EPOCH}-5"},
+        max_yields=1,
+    )
+    assert _types(yields) == []
+
+
+def test_same_epoch_ring_miss_is_truncated_with_honest_counts(
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    yields = _drain(
+        monkeypatch,
+        buffered=[_ev(10), _ev(11)],
+        headers={"Last-Event-ID": f"{EPOCH}-3"},
+        max_yields=3,
+    )
+    envelope, snapshot = _data_frames(yields)
+    assert envelope["type"] == "replay_truncated"
+    assert envelope["reason"] == "ring_evicted"
+    assert envelope["lost_count"] == 6  # ids 4..9 died with the ring
+    assert envelope["earliest_available_id"] == 10
+    assert snapshot["type"] == "node_snapshot"
+
+
+# ---------------------------------------------------------------------------
+# Stale / foreign cursors — the #881 class
+# ---------------------------------------------------------------------------
+
+
+def _assert_boot_epoch_truncated(yields: list[dict[str, Any]]) -> None:
+    """Envelope (reason=boot_epoch, no invented counts) then snapshot,
+    and no replay slice leaked around them."""
+    frames = _data_frames(yields)
+    assert [f["type"] for f in frames] == ["replay_truncated", "node_snapshot"]
+    envelope = frames[0]
+    assert envelope["reason"] == "boot_epoch"
+    assert "lost_count" not in envelope
+    assert "earliest_available_id" not in envelope
+
+
+def test_prior_boot_cursor_against_regrown_ring_is_truncated(
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    """THE restart-aliasing case: the reborn counter has re-grown past
+    the stale cursor, so pre-#881 the ring sliced from it as if the ids
+    were contiguous across the boot — silently skipping the restart
+    boundary.  The epoch mismatch must win over the plausible slice."""
+    yields = _drain(
+        monkeypatch,
+        buffered=[_ev(1), _ev(2), _ev(3)],
+        headers={"Last-Event-ID": "deadbeef-2"},
+        max_yields=3,
+    )
+    _assert_boot_epoch_truncated(yields)
+
+
+def test_legacy_bare_int_cursor_on_empty_ring_is_truncated(
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    """The original #881 lie: empty reborn ring + pre-restart cursor
+    used to answer ``replay_ok`` with nothing.  A bare-int cursor (no
+    epoch half — pre-#881 client or prior-boot native echo) must draw
+    the truncated floor instead of the silent gap."""
+    yields = _drain(
+        monkeypatch,
+        buffered=[],
+        headers={"Last-Event-ID": "42"},
+        max_yields=3,
+    )
+    _assert_boot_epoch_truncated(yields)
+
+
+def test_legacy_bare_int_cursor_on_live_ring_is_truncated(
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    yields = _drain(
+        monkeypatch,
+        buffered=[_ev(1), _ev(2)],
+        headers={"Last-Event-ID": "1"},
+        max_yields=3,
+    )
+    _assert_boot_epoch_truncated(yields)
+
+
+def test_same_epoch_cursor_against_empty_ring_fails_safe(
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    """Same-epoch ids imply a non-empty ring (single append site, first
+    id 1, never cleared) — a same-epoch cursor over an empty ring is
+    forged or a bug, and must fail to the snapshot floor rather than
+    resurrect the silent-gap shape."""
+    yields = _drain(
+        monkeypatch,
+        buffered=[],
+        headers={"Last-Event-ID": f"{EPOCH}-5"},
+        max_yields=3,
+    )
+    _assert_boot_epoch_truncated(yields)
+
+
+@pytest.mark.parametrize(
+    "cursor",
+    [
+        "not-a-cursor",  # wrong epoch, non-numeric counter
+        f"{EPOCH}-xyz",  # live epoch, garbage counter
+        f"{EPOCH}--5",  # live epoch, negative (forged) counter
+        "-",  # empty epoch, empty counter
+    ],
+)
+def test_unusable_cursor_shapes_all_draw_the_truncated_floor(
+    monkeypatch: pytest.MonkeyPatch, cursor: str
+) -> None:
+    """A present-but-unusable cursor must NOT fall back to ``fresh``:
+    fresh is the no-loss shape, and these callers provably lost events.
+    (Pre-#881 the unparseable arm silently became fresh — the semantic
+    shift is deliberate and this test pins it.)"""
+    yields = _drain(
+        monkeypatch,
+        buffered=[_ev(1)],
+        headers={"Last-Event-ID": cursor},
+        max_yields=3,
+    )
+    _assert_boot_epoch_truncated(yields)
+
+
+# ---------------------------------------------------------------------------
+# Transport details
+# ---------------------------------------------------------------------------
+
+
+def test_query_param_fallback_matches_header_semantics(
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    """Manual browser reconnects can't set headers on ``new
+    EventSource(url)`` — the ``?last_event_id=`` fallback must run the
+    same epoch logic (app.js presents its stored cursor this way)."""
+    yields = _drain(
+        monkeypatch,
+        buffered=[_ev(1), _ev(2)],
+        query={"last_event_id": f"{EPOCH}-1"},
+        max_yields=2,
+    )
+    frames = _data_frames(yields)
+    assert [f["ws_id"] for f in frames] == ["ws-2"]
+
+    yields = _drain(
+        monkeypatch,
+        buffered=[_ev(1), _ev(2)],
+        query={"last_event_id": "deadbeef-1"},
+        max_yields=3,
+    )
+    _assert_boot_epoch_truncated(yields)
+
+
+def test_epoch_is_hex_and_dashless_by_construction() -> None:
+    """The parse splits on the FIRST ``-``; the epoch half must never
+    contain one.  ``secrets.token_hex`` guarantees pure hex — this pin
+    exists so a future 'readable epoch' refactor (timestamps, uuids
+    with dashes) fails here instead of corrupting cursor parsing."""
+    import secrets
+
+    for _ in range(64):
+        assert "-" not in secrets.token_hex(4)
+    # And the production init uses token_hex — source-level pin.
+    import inspect
+
+    src = inspect.getsource(server_mod)
+    assert "app.state.global_boot_epoch = secrets.token_hex(" in src
+
+
+def test_listener_registered_exactly_once_per_connect(
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    """Registration is atomic with the replay decision for every branch
+    (stale cursors included) — the truncated floor must not skip or
+    double the listener append."""
+    monkeypatch.setattr(server_mod, "_build_node_snapshot", lambda _s: dict(_SNAPSHOT_STUB))
+    app_state = _make_app_state(buffered=[_ev(1)])
+    req = _fake_request(app_state, headers={"Last-Event-ID": "deadbeef-1"})
+
+    async def _run() -> None:
+        resp: Any = await server_mod.global_events_sse(req)
+        assert len(app_state.global_listeners) == 1
+        assert isinstance(app_state.global_listeners[0], queue.Queue)
+        await resp.body_iterator.aclose()
+
+    asyncio.run(_run())

turnstone/api/server_spec.py

@@ -186,7 +186,13 @@ SERVER_ENDPOINTS: list[EndpointSpec] = [
         "Emits a node_snapshot event on connect (workstreams, health, aggregate), "
         "followed by real-time delta events (ws_state, ws_activity, ws_created, "
         "ws_closed, ws_rename, health_changed, aggregate). "
-        "Pass ?expected_node_id=X for identity verification (returns 409 on mismatch).",
+        "Pass ?expected_node_id=X for identity verification (returns 409 on mismatch). "
+        "Every event's SSE id is an opaque '{boot_epoch}-{counter}' string; presenting "
+        "it on reconnect (Last-Event-ID header or ?last_event_id=) replays missed "
+        "events, or emits a replay_truncated event (reason: ring_evicted with "
+        "lost_count + earliest_available_id, or boot_epoch when the cursor predates "
+        "this server process) followed by a fresh node_snapshot. Treat the id as "
+        "opaque — its format may change.",
         tags=["Streaming"],
     ),
     EndpointSpec(

turnstone/console/collector.py

@@ -277,7 +277,20 @@ class ClusterCollector:
                 await task
 
     async def _node_sse_task(self, node_id: str, stop_event: asyncio.Event) -> None:
-        """Persistent SSE connection to a single server node."""
+        """Persistent SSE connection to a single server node.
+
+        DELIBERATELY cursorless (#881 audit ruling): every (re)connect is
+        fresh — no ``Last-Event-ID``, so the node answers with a full
+        ``node_snapshot`` and the collector rebuilds wholesale.  That is
+        this consumer's recovery model (idempotent state-of-world, not
+        append-only history), it side-steps cursor staleness across node
+        restarts entirely, and it means the epoch-tagged ids and the
+        ``replay_truncated`` envelope on this stream can never fire here.
+        If a cursor is ever adopted, present the SSE ``id:`` VERBATIM
+        (opaque ``"{boot_epoch}-{counter}"`` — never parse it) and handle
+        ``replay_truncated`` explicitly; today an unknown event type
+        no-ops in ``_apply_delta`` by design.
+        """
         backoff = 1.0
         while not stop_event.is_set() and self._running:
             url = self._get_node_url(node_id)

turnstone/server.py

@@ -22,6 +22,7 @@ import os
 import queue
 import random
 import re
+import secrets
 import sys
 import textwrap
 import threading
@@ -1009,6 +1010,19 @@ async def global_events_sse(request: Request) -> Response:
     On connect, emits a ``node_snapshot`` event with the full node state
     (workstreams, health, aggregate) followed by real-time delta events.
     The snapshot and listener registration are atomic — no events are lost.
+
+    Resume contract (#881): every event's SSE ``id:`` is
+    ``"{boot_epoch}-{counter}"``, where ``boot_epoch`` is a per-process
+    nonce and ``counter`` is the process-local monotonic event id.  A
+    reconnect presenting a cursor (``Last-Event-ID`` header or
+    ``?last_event_id=``) whose epoch matches gets the ring replay
+    (``replay_ok`` past the counter, or a ``replay_truncated`` envelope
+    with ``reason="ring_evicted"`` when the ring has moved on); a cursor
+    from any other epoch — a prior boot, another node, a pre-#881
+    bare-int client — gets ``replay_truncated`` with
+    ``reason="boot_epoch"`` followed by a fresh ``node_snapshot``, since
+    the events it missed died with the process that minted it.  Clients
+    treat the cursor as an opaque string; only this handler parses it.
     """
     # -- Service-scope gate ---------------------------------------------------
     # The global stream carries cluster-wide workstream inventory across
@@ -1036,14 +1050,44 @@ async def global_events_sse(request: Request) -> Response:
     # Native EventSource sets the header on auto-reconnect; the
     # manual-reconnect path (which can't set custom headers on
     # ``new EventSource(url)``) uses the query-param fallback.
+    #
+    # Global ids are ``"{boot_epoch}-{counter}"`` (#881): the epoch half
+    # proves the cursor was minted by THIS process.  The counter reboots
+    # at 0 with the process, so without the epoch a pre-restart cursor is
+    # first invisibly "ahead" of the reborn ring (empty replay slice) and
+    # then, as the counter re-grows past it, aliases into the new id
+    # space — both draw ``replay_ok`` and silently skip the restart
+    # boundary.  Parse outcomes:
+    #   - no cursor            → ``last_event_id = None`` (fresh);
+    #   - epoch matches        → ``last_event_id = counter`` (ring logic);
+    #   - anything else        → ``cursor_stale = True`` (a cursor was
+    #     presented but is unusable: prior-boot epoch, another node's
+    #     epoch, a bare-int cursor from a pre-#881 client, or garbage)
+    #     → the ``replay_truncated`` + node_snapshot floor below.  A
+    #     present-but-unusable cursor must NOT fall back to ``fresh``:
+    #     fresh is the no-loss shape, and this caller has provably
+    #     lost the events between its cursor and this boot.
+    boot_epoch: str = request.app.state.global_boot_epoch
     last_event_id_raw = request.headers.get("Last-Event-ID") or request.query_params.get(
         "last_event_id"
     )
-    last_event_id: int | None
-    try:
-        last_event_id = int(last_event_id_raw) if last_event_id_raw else None
-    except (TypeError, ValueError):
-        last_event_id = None
+    last_event_id: int | None = None
+    cursor_stale = False
+    if last_event_id_raw:
+        cursor_epoch, sep, counter_raw = last_event_id_raw.partition("-")
+        if sep and cursor_epoch == boot_epoch:
+            try:
+                last_event_id = int(counter_raw)
+            except ValueError:
+                cursor_stale = True
+            else:
+                if last_event_id < 0:
+                    # Our counters start at 1; a negative counter can
+                    # only be forged.  Same floor as garbage.
+                    last_event_id = None
+                    cursor_stale = True
+        else:
+            cursor_stale = True
 
     # -- Atomic snapshot / replay-slice + listener registration ---------------
     client_queue: queue.Queue[dict[str, Any]] = queue.Queue(maxsize=1000)
@@ -1053,40 +1097,54 @@ async def global_events_sse(request: Request) -> Response:
         request.app.state.global_event_buffer
     )
 
-    # Three replay shapes, matching :func:`make_events_handler`:
+    # Four replay shapes (the first three matching :func:`make_events_handler`):
     #   - ``last_event_id is None`` → ``"fresh"``: emit node_snapshot
     #     then live.
     #   - ``last_event_id`` + buffer covers gap → ``"replay_ok"``:
     #     emit buffered events past the id, SKIP node_snapshot, then
     #     live.
-    #   - ``last_event_id`` + buffer too short → ``"truncated"``: emit
-    #     a ``replay_truncated`` envelope then fall through to
-    #     ``"fresh"`` (node_snapshot is the recovery floor).
+    #   - ``last_event_id`` + buffer too short → ``"truncated"``
+    #     (``reason="ring_evicted"``, honest ``lost_count``): emit a
+    #     ``replay_truncated`` envelope then fall through to ``"fresh"``
+    #     (node_snapshot is the recovery floor).
+    #   - ``cursor_stale`` (#881) → ``"truncated"``
+    #     (``reason="boot_epoch"``): the cursor is from another boot —
+    #     the prior ring died with its process, so what was lost is
+    #     unknowable; the envelope omits ``lost_count`` /
+    #     ``earliest_available_id`` rather than guess.  The per-ws
+    #     stream never needs this arm: its counter is storage-seeded
+    #     across restarts, so plain counter comparison detects staleness
+    #     there (see ``register_listener_with_replay``); the global
+    #     counter is process-local and reboots at 0, hence the epoch.
     replay_status: str
+    truncated_reason = "ring_evicted"
     replay_events: list[dict[str, Any]] = []
     lost_count = 0
     earliest_available_id = 0
     snapshot: dict[str, Any] | None = None
 
     with listeners_lock:
-        if last_event_id is None:
+        if cursor_stale:
+            replay_status = "truncated"
+            truncated_reason = "boot_epoch"
+            snapshot = _build_node_snapshot(request.app.state)
+        elif last_event_id is None:
             replay_status = "fresh"
             snapshot = _build_node_snapshot(request.app.state)
         else:
             buffered = list(event_buffer)
             if not buffered:
-                # KNOWN GAP (#881): an empty ring after a node restart
-                # reports ``replay_ok`` even when the client's cursor is
-                # stale, silently skipping the events lost with the old
-                # process.  The per-ws stream fixes this by deriving
-                # staleness from its storage-seeded ``_event_id`` counter
-                # (see ``register_listener_with_replay``); the global
-                # counter (``global_event_id_holder``) resets to 0 at
-                # boot, so the same rule cannot apply — a boot-epoch
-                # staleness signal is the fix direction tracked in #881.
-                # Tolerable meanwhile: consumers are idempotent roster
-                # state refreshed periodically, not append-only history.
-                replay_status = "replay_ok"
+                # Same-epoch cursor against an empty ring is impossible
+                # by construction (ids are only minted by the fanout
+                # thread's single append site, first id 1, and the ring
+                # is never cleared — an id implies a non-empty ring), so
+                # a cursor landing here is forged or a bug.  Fail SAFE
+                # to the truncated floor: the snapshot rebuild is
+                # idempotent, whereas trusting the cursor would replay
+                # nothing and ghost the roster (#881's original lie).
+                replay_status = "truncated"
+                truncated_reason = "boot_epoch"
+                snapshot = _build_node_snapshot(request.app.state)
             else:
                 earliest_available_id = buffered[0][0]
                 if last_event_id < earliest_available_id - 1:
@@ -1098,16 +1156,40 @@ async def global_events_sse(request: Request) -> Response:
                     replay_events = [ev for eid, ev in buffered if eid > last_event_id]
         listeners.append(client_queue)
 
+    if replay_status == "truncated":
+        # Envelope chokepoint log, analog of ``ws.events.replay_truncated``
+        # on the per-ws lane — one line per gap detection, the operator
+        # signal that clients are being told to resync (a burst of
+        # ``reason=boot_epoch`` right after startup is the restart herd
+        # healing; sustained ``reason=ring_evicted`` means the ring cap
+        # is being outrun).
+        log.info(
+            "global.events.replay_truncated",
+            reason=truncated_reason,
+            lost_count=lost_count if truncated_reason == "ring_evicted" else None,
+            cursor=(last_event_id_raw or "")[:64],
+        )
+
     async def event_generator() -> AsyncGenerator[dict[str, Any], None]:
         _metrics.record_sse_connect()
 
         def _format_event(event: dict[str, Any]) -> dict[str, str]:
-            """Strip ``_event_id`` from the wire dict, attach SSE ``id:``."""
+            """Strip ``_event_id`` from the wire dict, attach SSE ``id:``.
+
+            The id is ``"{boot_epoch}-{counter}"`` — GLOBAL-LANE ONLY
+            (#881).  Do not propagate the epoch tag to the per-ws
+            formatter in ``make_events_handler``: per-ws ids must stay
+            bare ints — their counter is storage-seeded across restarts
+            (epoch unnecessary), clients seed cursors from ``/history``'s
+            integer ``cursor``, and coordinator.js's counter-reset
+            detector does ``Number(event.lastEventId)`` comparisons that
+            an epoch prefix would turn into ``NaN`` and silently kill.
+            """
             ev_copy = dict(event)
             eid = ev_copy.pop("_event_id", None)
             out: dict[str, str] = {"data": json.dumps(ev_copy)}
             if eid is not None:
-                out["id"] = str(eid)
+                out["id"] = f"{boot_epoch}-{eid}"
             return out
 
         try:
@@ -1117,15 +1199,23 @@ async def global_events_sse(request: Request) -> Response:
             yield {"retry": random.randint(2500, 4500)}
 
             if replay_status == "truncated":
-                yield {
-                    "data": json.dumps(
-                        {
-                            "type": "replay_truncated",
-                            "lost_count": lost_count,
-                            "earliest_available_id": earliest_available_id,
-                        }
-                    )
+                # ``reason`` distinguishes an in-epoch ring overrun
+                # (``ring_evicted`` — ``lost_count`` /
+                # ``earliest_available_id`` are honest) from a
+                # cross-boot cursor (``boot_epoch`` — the prior ring
+                # died with its process, so the loss is unknowable and
+                # the numeric fields are OMITTED rather than invented).
+                # No first-party consumer reads the numeric fields
+                # (app.js reads only ``type`` and resyncs), so the
+                # conditional shape is wire-safe.
+                envelope: dict[str, Any] = {
+                    "type": "replay_truncated",
+                    "reason": truncated_reason,
                 }
+                if truncated_reason == "ring_evicted":
+                    envelope["lost_count"] = lost_count
+                    envelope["earliest_available_id"] = earliest_available_id
+                yield {"data": json.dumps(envelope)}
             if replay_status == "replay_ok":
                 for ev in replay_events:
                     yield _format_event(ev)
@@ -4972,6 +5062,18 @@ def create_app(
     # Single-element list as a mutable int holder so the fanout
     # thread can ``counter_holder[0] += 1`` under the lock.
     app.state.global_event_id_holder = [0]
+    # Per-process boot epoch for the global SSE lane (#881).  The
+    # counter above reboots at 0 with the process (unlike the per-ws
+    # counters, which ``_seed_event_id_from_storage`` seeds from
+    # ``MAX(conversations.event_id)``), so a bare integer cursor from a
+    # prior boot is indistinguishable from — and eventually aliases
+    # into — the new id space.  Every global SSE ``id:`` is therefore
+    # stamped ``"{epoch}-{counter}"``; a reconnect cursor whose epoch
+    # differs from the live process is stale by construction and draws
+    # the ``replay_truncated`` + node_snapshot recovery floor in
+    # :func:`global_events_sse`.  Hex nonce (never contains ``-``), so
+    # ``partition("-")`` splits the id unambiguously.
+    app.state.global_boot_epoch = secrets.token_hex(4)
     app.state.skip_permissions = skip_permissions
     app.state.jwt_secret = jwt_secret
     app.state.auth_storage = auth_storage