fix(mothership): bound a silent replay, never skip past a trimmed cursor, and re-sync an expired ring

- The worker replay is bounded like a stream leg: no response headers, or no bytes
  including keepalives, for the idle timeout ends it so the reader re-attaches.
- A ring read that starts after the reader's next event (the ring trimmed its head
  between the gap check and the read) is never delivered; the reader re-attaches and
  re-syncs from the log, in both the live tail and batch reads.
- An empty ring serves only a reader starting from cursor 0; a live run whose buffer
  expired under a reader re-syncs from the log, and a finished one answers its
  terminal since its transcript is persisted.
- A leg that delivers events after a failure starts a fresh 30 s reachable window;
  its three retries still refill only after five minutes of delivered events.
- The two new integration suites close their worker server and restore env even
  when they are skipped.

Author: waleedlatif1

apps/sim/app/api/copilot/chat/stream/route.test.ts

@@ -219,4 +219,54 @@ describe('copilot chat stream replay route', () => {
     )
     trace.disable()
   })
+
+  it('never delivers a ring read that starts past the reader cursor, and ends without a terminal', async () => {
+    getLatestRunForStream.mockResolvedValue({
+      status: 'active',
+      executionId: 'exec-1',
+      id: 'run-1',
+    })
+    readEvents.mockResolvedValue([
+      {
+        stream: { streamId: 'stream-1', cursor: '5' },
+        seq: 5,
+        trace: { requestId: 'req-1' },
+        type: MothershipStreamV1EventType.text,
+        payload: { channel: 'assistant', text: 'the middle of the turn' },
+      },
+    ])
+
+    const response = await GET(
+      new NextRequest('http://localhost:3000/api/copilot/chat/stream?streamId=stream-1&after=0')
+    )
+    const text = (await readAllChunks(response)).join('')
+
+    expect(text).not.toContain('the middle of the turn')
+    expect(text).not.toContain(`"type":"${MothershipStreamV1EventType.complete}"`)
+  })
+
+  it('serves a batch read that starts past the reader cursor no events', async () => {
+    getLatestRunForStream.mockResolvedValue({
+      status: 'active',
+      executionId: 'exec-1',
+      id: 'run-1',
+    })
+    readEvents.mockResolvedValue([
+      {
+        stream: { streamId: 'stream-1', cursor: '5' },
+        seq: 5,
+        trace: { requestId: 'req-1' },
+        type: MothershipStreamV1EventType.text,
+        payload: { channel: 'assistant', text: 'the middle of the turn' },
+      },
+    ])
+
+    const response = await GET(
+      new NextRequest(
+        'http://localhost:3000/api/copilot/chat/stream?streamId=stream-1&after=0&batch=true'
+      )
+    )
+
+    await expect(response.json()).resolves.toMatchObject({ success: true, events: [] })
+  })
 })

apps/sim/app/api/copilot/chat/stream/route.ts

@@ -59,6 +59,15 @@ const RING_CHECK_EVERY_BUSY_POLLS = 8
  */
 const MAX_STREAM_MS = 60 * 60 * 1000 - 60_000
 
+/**
+ * Whether ring events read after `cursor` start right after it. The ring can trim its
+ * head between a gap check and the read, and a read that starts later would silently
+ * skip part of the turn.
+ */
+function startsAfterCursor(events: readonly { seq: number }[], cursor: string): boolean {
+  return events.length === 0 || events[0].seq <= Number(cursor || '0') + 1
+}
+
 function extractCanonicalRequestId(value: unknown): string {
   return typeof value === 'string' && value.length > 0 ? value : ''
 }
@@ -253,8 +262,10 @@ async function handleResumeRequestBody({
         return []
       }),
     ])
-    // A reader the ring cannot serve is re-synced from the worker log by the live tail.
-    const batchEvents = fromLog || gap ? [] : events.map(toStreamBatchEvent)
+    // A reader the ring cannot serve, or whose next event it trimmed after the gap check,
+    // is re-synced from the worker log by the live tail.
+    const batchEvents =
+      fromLog || gap || !startsAfterCursor(events, afterSeq) ? [] : events.map(toStreamBatchEvent)
     logger.info('[Resume] Batch response', {
       streamId,
       afterCursor: afterSeq,
@@ -287,9 +298,12 @@ async function handleResumeRequestBody({
    * no shared position to join on. The header tells the client to rebuild the turn
    * from an empty response, since the replay's cursors restart at 1.
    */
-  const gap = fromLog
+  const ringGap = fromLog
     ? null
     : await findReplayGap(streamId, afterCursor || '0', extractRunRequestId(run))
+  // A finished run whose buffer expired answers its terminal; its transcript is persisted.
+  const gap =
+    ringGap && !(ringGap.latestSeq <= 0 && isTerminalStreamStatus(run.status)) ? ringGap : null
   const resyncFromLog = fromLog || gap !== null
   let replayBody: ReadableStream<Uint8Array> | null = null
   /** Releases the worker's replay once this response ends; the request signal may never fire. */
@@ -380,8 +394,13 @@ async function handleResumeRequestBody({
     }
     request.signal.addEventListener('abort', abortListener, { once: true })
 
-    const flushEvents = async (): Promise<number> => {
+    /** Delivers the ring's events after the cursor, or returns null if it trimmed the next one. */
+    const flushEvents = async (): Promise<number | null> => {
       const events = await readEvents(streamId, cursor)
+      if (!startsAfterCursor(events, cursor)) {
+        logger.warn('Replay ring trimmed past a reader cursor', { streamId, cursor })
+        return null
+      }
       if (events.length > 0) {
         logger.debug('[Resume] Flushing events', {
           streamId,
@@ -482,6 +501,7 @@ async function handleResumeRequestBody({
       }
 
       let lastFlushed = await flushEvents()
+      if (lastFlushed === null) return
 
       let pollDelayMs = POLL_INTERVAL_MS
       while (!controllerClosed && Date.now() - startTime < MAX_STREAM_MS) {
@@ -493,14 +513,6 @@ async function handleResumeRequestBody({
           })
           return null
         })
-        // The ring lost its head or restarted under this tail; the re-attach re-syncs. Only a
-        // quiet ring can have restarted or have a dead controller that recovery re-sends
-        // under, so a busy tail checks every few polls rather than on each one.
-        const checkRing = lastFlushed === 0 || pollIterations % RING_CHECK_EVERY_BUSY_POLLS === 0
-        if (checkRing && !ringCanServe(await readRingPosition(streamId, cursor))) {
-          logger.warn('Replay ring can no longer serve a live tail', { streamId, cursor })
-          break
-        }
         if (!currentRun) {
           emitTerminalIfMissing(MothershipStreamV1CompletionStatus.error, {
             message: 'The stream could not be recovered because its run metadata is unavailable.',
@@ -509,10 +521,23 @@ async function handleResumeRequestBody({
           })
           break
         }
+        // The ring lost its head, restarted or expired under this live tail; the re-attach
+        // re-syncs, and a finished run answers its terminal instead. Only a quiet ring can
+        // restart or be re-sent into by a recovery, so a busy tail checks every few polls.
+        const checkRing = lastFlushed === 0 || pollIterations % RING_CHECK_EVERY_BUSY_POLLS === 0
+        if (
+          checkRing &&
+          !isTerminalStreamStatus(currentRun.status) &&
+          !ringCanServe(await readRingPosition(streamId, cursor))
+        ) {
+          logger.warn('Replay ring can no longer serve a live tail', { streamId, cursor })
+          break
+        }
 
         currentRequestId = extractRunRequestId(currentRun) || currentRequestId
 
         const flushed = await flushEvents()
+        if (flushed === null) break
         lastFlushed = flushed
         /* Adaptive tail: 4 Hz only while events are actually flowing; a quiet stream
            decays toward the cap so an attached client doesn't hammer Postgres + Redis

apps/sim/lib/mothership/request/lifecycle/stream-retry.test.ts

@@ -216,4 +216,18 @@ describe('stream recovery budget', () => {
     vi.advanceTimersByTime(30 * 60_000)
     expect(retry.nextDelay(error)).toBeNull()
   })
+
+  it('retries a leg that fails again minutes after it re-attached and delivered events', () => {
+    vi.useFakeTimers()
+    const error = new WorkerStreamInterruptedError(new Error('socket closed'))
+    const retry = new StreamRetryWindow()
+    const first = retry.nextDelay(error)
+    expect(first).not.toBeNull()
+    vi.advanceTimersByTime(first ?? 0)
+    for (let second = 0; second < 120; second += 10) {
+      retry.recovered()
+      vi.advanceTimersByTime(10_000)
+    }
+    expect(retry.nextDelay(error)).not.toBeNull()
+  })
 })

apps/sim/lib/mothership/request/lifecycle/stream-retry.ts

@@ -32,7 +32,8 @@ const HEALTHY_STREAM_REPLENISH_MS = 5 * 60_000
  * Recovery is bounded independently of the healthy leg's lifetime, by two
  * budgets that never share state: an unreachable worker gets a two-minute
  * window from the moment it stopped answering, and any failure of a worker that
- * did answer gets three retries within 30 s, replenished only after
+ * did answer gets three retries, each burst of them within 30 s of its first
+ * failure, replenished only after
  * {@link HEALTHY_STREAM_REPLENISH_MS} of healthy streaming. A leg has no deadline
  * unless the caller sets one.
  */
@@ -63,9 +64,13 @@ export class StreamRetryWindow {
     return remaining
   }
 
-  /** The worker delivered an event, so a later loss of it starts a fresh unreachable window. */
+  /**
+   * The worker delivered an event: a later loss starts a fresh unreachable window, and
+   * a fresh 30 s reachable window. Only the three reachable retries carry over.
+   */
   recovered(): void {
     this.resetUnreachable()
+    this.firstFailureAt = undefined
     this.lastEventAt = Date.now()
     this.streamingSince ??= this.lastEventAt
   }

apps/sim/lib/mothership/request/session/recovery.test.ts

@@ -12,7 +12,11 @@ vi.mock('./buffer', () => ({
   readEvents,
 }))
 
-import { findReplayGap, replayGapTerminal } from '@/lib/mothership/request/session/recovery'
+import {
+  findReplayGap,
+  replayGapTerminal,
+  ringCanServe,
+} from '@/lib/mothership/request/session/recovery'
 
 describe('replay gap', () => {
   it('uses the latest buffered request id when run metadata is missing it', async () => {
@@ -33,4 +37,9 @@ describe('replay gap', () => {
     expect(result?.envelopes[0].trace.requestId).toBe('req-live-123')
     expect(result?.envelopes[1].trace.requestId).toBe('req-live-123')
   })
+
+  it('cannot serve a reader that holds a cursor from an empty ring', () => {
+    expect(ringCanServe({ requestedAfterSeq: 12, oldestSeq: 0, latestSeq: 0 })).toBe(false)
+    expect(ringCanServe({ requestedAfterSeq: 0, oldestSeq: 0, latestSeq: 0 })).toBe(true)
+  })
 })

apps/sim/lib/mothership/request/session/recovery.ts

@@ -52,10 +52,11 @@ export async function readRingPosition(
  * head: the events before its oldest are gone, and a cursor that was served from the
  * worker's log instead is not a position in the ring, so no cursor is trusted. Nor can
  * it serve a cursor ahead of its latest event, which only a buffer whose numbering
- * restarted after it expired produces.
+ * restarted after it expired produces, nor any cursor from a buffer that expired.
  */
 export function ringCanServe({ requestedAfterSeq, oldestSeq, latestSeq }: RingPosition): boolean {
-  return latestSeq <= 0 || (startsAtReplayHead(oldestSeq) && requestedAfterSeq <= latestSeq)
+  if (latestSeq <= 0) return requestedAfterSeq <= 0
+  return startsAtReplayHead(oldestSeq) && requestedAfterSeq <= latestSeq
 }
 
 /** The ring's position when it cannot serve `afterCursor` (see {@link ringCanServe}). */

apps/sim/lib/mothership/request/session/replay-gap.integration.ts

@@ -134,6 +134,15 @@ function fullResponse(streamId: string) {
   ]
 }
 
+/** Runs whether or not the suite does, so a skipped suite never leaks the worker or env. */
+afterAll(async () => {
+  await new Promise<void>((resolve) => worker.server.close(() => resolve()))
+  for (const [key, value] of Object.entries(inheritedEnv)) {
+    if (value === undefined) delete process.env[key]
+    else process.env[key] = value
+  }
+})
+
 describe.runIf(Boolean(redisUrl))('reconnects past the replay ring', () => {
   beforeAll(async () => {
     const now = new Date()
@@ -177,11 +186,6 @@ describe.runIf(Boolean(redisUrl))('reconnects past the replay ring', () => {
     await db.delete(workspace).where(eq(workspace.id, workspaceId))
     await db.delete(user).where(eq(user.id, userId))
     await closeRedisConnection()
-    await new Promise<void>((resolve) => worker.server.close(() => resolve()))
-    for (const [key, value] of Object.entries(inheritedEnv)) {
-      if (value === undefined) delete process.env[key]
-      else process.env[key] = value
-    }
   })
 
   it.each([
@@ -375,6 +379,47 @@ describe.runIf(Boolean(redisUrl))('reconnects past the replay ring', () => {
     expect(response.headers.get(MOTHERSHIP_STREAM_REPLAY_HEADER)).toBeNull()
   })
 
+  it('re-syncs a live run whose buffer expired under a reader cursor', async () => {
+    const streamId = generateId()
+    await db.insert(copilotRuns).values({
+      id: generateId(),
+      executionId: generateId(),
+      chatId,
+      userId,
+      workspaceId,
+      streamId,
+    })
+    worker.reply.frames = fullResponse(streamId)
+
+    const response = await reconnect(streamId, '6')
+    const frames = dataFrames(await response.text())
+
+    expect(response.headers.get(MOTHERSHIP_STREAM_REPLAY_HEADER)).toBe('log')
+    expect(frames.map((frame) => frame.type)).toEqual(['session', 'text', 'complete'])
+  })
+
+  it('answers a finished run whose buffer expired with its terminal, not a replay', async () => {
+    const streamId = generateId()
+    await db.insert(copilotRuns).values({
+      id: generateId(),
+      executionId: generateId(),
+      chatId,
+      userId,
+      workspaceId,
+      streamId,
+      status: 'complete',
+    })
+
+    const response = await reconnect(streamId, '6')
+    const frames = dataFrames(await response.text())
+
+    expect(response.headers.get(MOTHERSHIP_STREAM_REPLAY_HEADER)).toBeNull()
+    expect(frames.map((frame) => [frame.type, frame.payload.status])).toEqual([
+      ['complete', 'complete'],
+    ])
+    expect(worker.requests).toEqual([])
+  })
+
   it('serves no ring events to a batch read the ring can no longer serve', async () => {
     const { streamId } = await liveRunWithTrimmedRing()