fix(fleet): lossless, honest live event stream across the fleet (#1770)

packages/app-bundle/overlay/packages/app/src/context/server-sdk.tsx

@@ -14,8 +14,8 @@ import { createRefCountMap } from "@/utils/refcount"
 import { useGlobal } from "./global"
 import { ServerScope } from "@/utils/server-scope"
 import { attachOwnerHeaderIfArmed } from "@/components/remote-create-arm"
-import { parseCursorNamespaces, resolveEventOrigin } from "./sse-origin"
-import { shouldReconnectIdleStream } from "./stream-liveness"
+import { createStreamEndResync, shouldReconnectIdleStream } from "./stream-liveness"
+import { applyStreamCursor, createStreamCursor } from "./stream-cursor"
 import { detectServerProtocol, type ServerProtocol } from "@/utils/server-protocol"
 import { createCompatibleApi, type CompatibleApi } from "@/utils/server-compat"
 
@@ -357,55 +357,43 @@ function createServerSdkContextBase(server: ServerConnection.Any, scope: ServerS
     }
   })()
 
-  // #1264 lossless reconnect: the frontdoor buffers recent events and
-  // replays the gap when the stream is re-opened with ?lastEventID=<id>.
-  // Track the last payload id seen (the server ids every event; id-less
-  // events just don't advance the cursor — replay may over-deliver, which
-  // the reconcile-based reducers tolerate by design).
+  // #1264 lossless reconnect: the engine (#1776) sequences its global stream
+  // and replays the gap when the stream is re-opened with ?lastEventID=<id>.
+  // #1777: the cursor is the SSE `id:` of the last applied frame — the engine's
+  // `<epoch>:<seq>`, or the fan-in's composite of every origin's position —
+  // never the JSON payload id (the engine rejects that as malformed, so every
+  // reconnect used to be a reset). It also checks contiguity per origin: a
+  // hole, an engine reset, or an unannounced restart forces a resync, and a
+  // replayed frame already applied is dropped (stream-cursor.ts).
   // #1264: the cursor must survive RELOADS — module state died with every
   // document, so a drop right after a reload reconnected cursor-less and
   // the gap's events were lost (the harness caught it: the sse-gap flow
   // failed only in the post-reload context). sessionStorage is per-tab,
   // cheap, and exactly the lifetime the cursor wants.
   const CURSOR_KEY = "amicode.sse.lastEventID"
-  let lastEventID: string | undefined
+  let storedCursor: string | undefined
   try {
-    lastEventID = sessionStorage.getItem(CURSOR_KEY) ?? undefined
+    storedCursor = sessionStorage.getItem(CURSOR_KEY) ?? undefined
   } catch {
     /* private mode etc. — degrade to module state */
   }
-  const trackEventID = (payload: unknown) => {
-    const id = (payload as { id?: unknown } | undefined)?.id
-    if (typeof id === "string" && id) {
-      lastEventID = id
-      try {
-        sessionStorage.setItem(CURSOR_KEY, id)
-      } catch {
-        /* best-effort */
-      }
+  const cursor = createStreamCursor(storedCursor)
+  const persistCursor = () => {
+    const current = cursor.current()
+    if (current === undefined || current === storedCursor) return
+    storedCursor = current
+    try {
+      sessionStorage.setItem(CURSOR_KEY, current)
+    } catch {
+      /* best-effort */
     }
   }
-  // #1539: track the parsed composite cursor to detect event origin (local vs
-  // remote peer). Initialized from the persisted cursor so the first event
-  // after reconnect diffs correctly.
-  let previousParsedCursor = parseCursorNamespaces(lastEventID)
   const sseFetch: FetchLike = (input, init) => {
     const base = eventFetch ?? globalThis.fetch
-    try {
-      if (lastEventID) {
-        const url = typeof input === "string" ? input : input instanceof URL ? input.href : input.url
-        const u = new URL(url, server.http.url)
-        if (u.pathname === "/event" || u.pathname === "/global/event") {
-          u.searchParams.set("lastEventID", lastEventID)
-          const modified = u.toString()
-          if (typeof input === "string" || input instanceof URL) return base(modified, init)
-          return base(new Request(modified, input), init)
-        }
-      }
-    } catch {
-      /* fall through unmodified */
-    }
-    return base(input as Parameters<typeof fetch>[0], init as Parameters<typeof fetch>[1])
+    // #1777: the ONE carrier is `?lastEventID=` (the SDK's own Last-Event-ID
+    // header is stripped so two cursors can never disagree).
+    const applied = applyStreamCursor(input, init, server.http.url, cursor.current())
+    return base(applied.input as Parameters<typeof fetch>[0], applied.init as Parameters<typeof fetch>[1])
   }
   const eventApi = createApiForServer({ server: server.http, fetch: sseFetch as typeof fetch })
   const eventSdk = createSdkForServer({
@@ -475,6 +463,28 @@ function createServerSdkContextBase(server: ServerConnection.Any, scope: ServerS
   // nominally connected, so the disconnect→connect edge never fires and the rail
   // would otherwise never self-heal a stranded view).
   const [resyncCount, setResyncCount] = createSignal(0)
+  // #1617 / #1777: a FULL resync of the viewed state — the forced-resync signal
+  // (the transcript and the entity rail refetch, #1646) plus a synthetic GLOBAL
+  // `amicode.sync.gap` event, on which server-sync refetches the bootstrap
+  // unconditionally (past the #1289 `server.connected` debounce, which keys on
+  // a different type). Driven by a relay's gap frame, and by the cursor's
+  // verdicts: an engine reset, a sequence hole, an unannounced restart.
+  const forceResync = (namespaces: string[]) => {
+    setResyncCount((n) => n + 1)
+    if (
+      enqueueServerEvent(queue, {
+        directory: "global",
+        payload: { type: SYNC_GAP_TYPE, properties: { namespaces } } as unknown as ServerEvent,
+      })
+    )
+      schedule()
+  }
+  // #1773: an unexpected NORMAL end of the global stream is a loss (the relays
+  // never end a listener cleanly except on shutdown). The loop arms this latch
+  // on such an end; the reconnected stream's first frame advances resyncCount,
+  // so the viewed transcript and the entity rail refetch once the stream is live
+  // again. A user stop, a page hide, or the client's own abort arm nothing.
+  const streamEndResync = createStreamEndResync()
 
   // #1584 client-side liveness watchdog. The v1 event iterator neither throws
   // nor completes when the socket goes silently half-open (sleep/wake, Wi-Fi
@@ -545,9 +555,21 @@ function createServerSdkContextBase(server: ServerConnection.Any, scope: ServerS
               error,
             })
           }
+          // #1777: the SSE id of the frame about to be yielded. The SDK reader
+          // calls onSseEvent with each parsed frame right before it yields that
+          // frame's data, so this is the id of the event the loop reads next.
+          let frameSseId: string | undefined
           const events =
             kind === "v1"
-              ? (await eventSdk.global.event({ signal: currentAttempt.signal, onSseError })).stream
+              ? (
+                  await eventSdk.global.event({
+                    signal: currentAttempt.signal,
+                    onSseError,
+                    onSseEvent: (frame: { id?: string }) => {
+                      frameSseId = frame.id
+                    },
+                  })
+                ).stream
               : eventApi.event.subscribe({ signal: currentAttempt.signal })
           if (!streamGeneration.accepts(active)) return
           setStreamStatus("connected")
@@ -560,8 +582,11 @@ function createServerSdkContextBase(server: ServerConnection.Any, scope: ServerS
             if (!streamGeneration.accepts(active)) return
             streamErrorLogged = false
             lastFrameAt = Date.now()
+            // #1773: the first frame after an unexpected end — the stream is live
+            // again, so force the resync now (the refetch cannot miss the events
+            // between the end and this re-subscription).
+            if (streamEndResync.frame()) setResyncCount((n) => n + 1)
             const legacy = "payload" in event
-            if (legacy && event.payload.type === "sync") continue
             const directory = legacy ? (event.directory ?? "global") : (event.location?.directory ?? "global")
             const payload = legacy ? (event.payload as Event) : adaptServerEvent(event)
             // #1522 (ADR 0033 decision A): the SSE fan-in aggregator emits
@@ -602,33 +627,45 @@ function createServerSdkContextBase(server: ServerConnection.Any, scope: ServerS
                 ? (payload as Record<string, unknown>)
                 : ((payload as { properties?: Record<string, unknown> }).properties ?? {})
               const gap = parseSyncGapFrame({ type: SYNC_GAP_TYPE, ...gapData })
+              // #1777: the named origins resync, so their next positions are new
+              // baselines (not a second "hole" against the lost stretch).
+              cursor.gap(gap?.namespaces ?? [])
               // Bump the forced-resync signal so the entity rail self-heals even
               // though the socket never crossed a disconnect→connect edge.
-              setResyncCount((n) => n + 1)
-              if (enqueueServerEvent(queue, {
-                directory: "global",
-                payload: { type: SYNC_GAP_TYPE, properties: { namespaces: gap?.namespaces ?? [] } } as unknown as ServerEvent,
-              }))
-                schedule()
+              forceResync(gap?.namespaces ?? [])
               continue
             }
-            trackEventID(legacy ? event.payload : (event as { id?: string }))
-            // #1539: resolve origin from the composite cursor diff. Fleet-of-one
-            // (bare scalar id) returns undefined immediately — zero overhead.
-            const sseId: string | undefined = legacy
-              ? (event.payload as { id?: string }).id
-              : (event as { id?: string }).id
-            const resolved = resolveEventOrigin(sseId, previousParsedCursor)
-            previousParsedCursor = resolved.cursor
-            const taggedPayload: ServerEvent = resolved.origin !== undefined
-              ? { ...payload, origin: resolved.origin }
+            // #1777: the cursor and the contiguity check, on the frame's SSE id
+            // (v1; the v2 stream at /api/event carries no resume cursor). A hole,
+            // a reset or an unannounced engine restart owes a full resync; a
+            // replayed frame already applied is dropped, never applied twice.
+            const verdict = cursor.frame({
+              id: kind === "v1" ? frameSseId : undefined,
+              type: legacy ? event.payload.type : payload.type,
+            })
+            if (verdict.resync) forceResync(verdict.resync.namespaces)
+            if (!verdict.apply) continue
+            persistCursor()
+            // a sync copy still moves the cursor (it holds a seq), but is no event
+            if (legacy && event.payload.type === "sync") continue
+            // #1539: the origin is the composite cursor's moving namespace.
+            // Fleet-of-one (a bare engine id) is always local.
+            const taggedPayload: ServerEvent = verdict.origin !== undefined
+              ? { ...payload, origin: verdict.origin }
               : payload
             if (enqueueServerEvent(queue, { directory, payload: taggedPayload })) schedule()
 
             if (Date.now() - yielded < STREAM_YIELD_MS) continue
             yielded = Date.now()
             await wait(0)
           }
+          // #1773: the iterator COMPLETED — a clean end of the global stream. If
+          // the client neither aborted this attempt nor stopped the loop, it is
+          // an unexpected end: a loss. Arm the resync for the reconnected stream.
+          streamEndResync.ended({
+            attemptAborted: currentAttempt.signal.aborted,
+            stopped: abort.signal.aborted || !started || !streamGeneration.accepts(active),
+          })
         } catch (error) {
           if (streamGeneration.accepts(active) && !isStreamClosed(error, currentAttempt.signal))
             setStreamStatus("disconnected")
@@ -658,8 +695,8 @@ function createServerSdkContextBase(server: ServerConnection.Any, scope: ServerS
   }
 
   const clearGlobalStreamCursor = () => {
-    lastEventID = undefined
-    previousParsedCursor = {}
+    cursor.clear()
+    storedCursor = undefined
     try {
       sessionStorage.removeItem(CURSOR_KEY)
     } catch {

packages/app-bundle/overlay/packages/app/src/context/sse-gap-frame.test.ts

@@ -60,3 +60,37 @@ describe("#1617 — gap interception is wired at both client seams", () => {
     expect(syncSource).toMatch(/server\.connected"\s*&&\s*Date\.now\(\)\s*-\s*lastConnectedQueueAt/)
   })
 })
+
+// #1773 — an unexpected NORMAL end of the global stream is a gap too (defense in
+// depth: kept even once replay exists). The relay chain now error-closes every
+// loss, but a clean end the client did not ask for still means events may have
+// been dropped. The decision (`createStreamEndResync`, stream-liveness.ts) is
+// unit-tested there; these pins assert the loop actually wires it — falsifiable:
+// deleting either call, or reporting the end without the abort/stop state, reds.
+describe("#1773 — an unexpected end of the global stream advances the forced-resync signal", () => {
+  const sdkSource = readFileSync(join(import.meta.dir, "server-sdk.tsx"), "utf8")
+  const loop = sdkSource.indexOf("for await (const event of events)")
+  const loopCatch = sdkSource.indexOf("} catch (error) {", loop)
+
+  test("the loop's NORMAL completion (the iterator ended without throwing) reports the end", () => {
+    expect(loop).toBeGreaterThan(0)
+    const body = sdkSource.slice(loop, loopCatch)
+    // after the for-await body's last statement and before the catch: the normal-end path
+    const afterLoop = body.slice(body.lastIndexOf("await wait(0)"))
+    expect(afterLoop).toContain("streamEndResync.ended(")
+  })
+
+  test("a user stop / page hide / the client's own abort are NOT a loss: the end carries both states", () => {
+    const body = sdkSource.slice(loop, loopCatch)
+    expect(body).toMatch(/attemptAborted:\s*currentAttempt\.signal\.aborted/)
+    // stopped = the root abort (teardown), the started latch (stop()/pagehide), the generation fence (reset)
+    expect(body).toMatch(/stopped:\s*abort\.signal\.aborted\s*\|\|\s*!started\s*\|\|\s*!streamGeneration\.accepts\(active\)/)
+  })
+
+  test("the reconnected stream's first frame advances the forced-resync signal (the transcript + entity rail refetch)", () => {
+    const body = sdkSource.slice(loop, loopCatch)
+    expect(body).toMatch(/if \(streamEndResync\.frame\(\)\) setResyncCount\(\(n\) => n \+ 1\)/)
+    // the signal the transcript (pages/session.tsx) and the entity rail read
+    expect(sdkSource).toMatch(/resyncCount,\s*\n/)
+  })
+})