fix(mothership): bound stream reconnection attempts

Author: Sg312

apps/sim/lib/mothership/request/lifecycle/run.test.ts

@@ -1992,12 +1992,12 @@ describe('runCopilotLifecycle', () => {
     }
   })
 
-  it('keeps the same request and partial response across a longer worker outage', async () => {
+  it('keeps the same request and partial response through the allowed reconnects', async () => {
     vi.useFakeTimers()
     try {
       const bodies: Record<string, unknown>[] = []
       const headers: Headers[] = []
-      for (let attempt = 0; attempt < 4; attempt++) {
+      for (let attempt = 0; attempt < 3; attempt++) {
         mockRunStreamLoop.mockImplementationOnce(
           async (_url, request, context: StreamingContext) => {
             bodies.push(JSON.parse(String(request.body)))
@@ -2041,19 +2041,58 @@ describe('runCopilotLifecycle', () => {
         content: 'Saved partial answer recovered',
       })
       expect(result.errors).toBeUndefined()
-      expect(bodies).toHaveLength(5)
-      expect(bodies.map((body) => body.messageId)).toEqual(Array(5).fill('outage-stream'))
+      expect(bodies).toHaveLength(4)
+      expect(bodies.map((body) => body.messageId)).toEqual(Array(4).fill('outage-stream'))
       expect(headers.map((header) => header.get('X-Sim-Request-ID'))).toEqual(
-        Array(5).fill('outage-request')
+        Array(4).fill('outage-request')
       )
       expect(bodies.slice(1).map((body) => body.receivedTextChars)).toEqual(
-        Array(4).fill('Saved partial answer'.length)
+        Array(3).fill('Saved partial answer'.length)
       )
     } finally {
       vi.useRealTimers()
     }
   })
 
+  it('ends an empty-stream outage after the initial attempt and three reconnects', async () => {
+    vi.useFakeTimers()
+    try {
+      for (let attempt = 0; attempt < 4; attempt++) {
+        mockRunStreamLoop.mockImplementationOnce(
+          async (_url, _request, context: StreamingContext) => {
+            context.errors.push(STREAM_ENDED_WITHOUT_TERMINAL_MESSAGE)
+            throw new StreamEndedWithoutTerminalError('/api/mothership')
+          }
+        )
+      }
+      const pending = runCopilotLifecycle(
+        { message: 'hello', messageId: 'bounded-outage' },
+        {
+          userId: 'user-1',
+          workspaceId: 'ws-1',
+          chatId: 'chat-1',
+          executionId: 'exec-1',
+          runId: 'run-1',
+          executionContext: {
+            userId: 'user-1',
+            workflowId: '',
+            workspaceId: 'ws-1',
+            chatId: 'chat-1',
+          },
+        }
+      )
+      await vi.advanceTimersByTimeAsync(30_000)
+      expect(await pending).toMatchObject({
+        success: false,
+        cancelled: false,
+        errors: [STREAM_ENDED_WITHOUT_TERMINAL_MESSAGE],
+      })
+      expect(mockRunStreamLoop).toHaveBeenCalledTimes(4)
+    } finally {
+      vi.useRealTimers()
+    }
+  })
+
   it('honors Stop while waiting to reconnect to the worker', async () => {
     vi.useFakeTimers()
     try {

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

@@ -1,20 +1,45 @@
 import { afterEach, describe, expect, it, vi } from 'vitest'
-import { CopilotBackendError } from '@/lib/mothership/request/go/stream'
+import {
+  CopilotBackendError,
+  StreamEndedWithoutTerminalError,
+} from '@/lib/mothership/request/go/stream'
 import { StreamRetryWindow } from '@/lib/mothership/request/lifecycle/stream-retry'
 
 afterEach(() => vi.useRealTimers())
 
 describe('stream recovery budget', () => {
-  it('survives multiple unavailable connections without resetting its deadline', () => {
+  it.each([
+    new TypeError('fetch failed'),
+    new StreamEndedWithoutTerminalError('/api/mothership'),
+    new CopilotBackendError('Unavailable', { status: 503 }),
+  ])('stops after three retries despite a long task budget: %s', (error) => {
     vi.useFakeTimers()
-    const retry = new StreamRetryWindow(120_000)
-    for (let index = 0; index < 6; index++) {
-      const delay = retry.nextDelay(new TypeError('fetch failed'))
+    const retry = new StreamRetryWindow()
+    for (let index = 0; index < 3; index++) {
+      const delay = retry.nextDelay(error)
       expect(delay).not.toBeNull()
       vi.advanceTimersByTime(delay ?? 0)
     }
-    expect(retry.attempt).toBe(6)
-    expect(retry.remainingMs()).toBeLessThan(120_000)
+    expect(retry.nextDelay(error)).toBeNull()
+    expect(retry.attempt).toBe(3)
+    expect(retry.remainingMs()).toBeGreaterThan(3_500_000)
+  })
+
+  it('bounds the recovery period from the first failure without shortening healthy work', () => {
+    vi.useFakeTimers()
+    const retry = new StreamRetryWindow()
+    vi.advanceTimersByTime(600_000)
+    expect(retry.nextDelay(new TypeError('fetch failed'))).not.toBeNull()
+    vi.advanceTimersByTime(30_000)
+    expect(retry.nextDelay(new TypeError('fetch failed'))).toBeNull()
+    expect(retry.remainingMs()).toBe(2_970_000)
+  })
+
+  it('never extends the original execution deadline', () => {
+    vi.useFakeTimers()
+    const retry = new StreamRetryWindow(120_000)
+    vi.advanceTimersByTime(119_999)
+    expect(retry.nextDelay(new TypeError('fetch failed'))).toBeNull()
     vi.advanceTimersByTime(120_000)
     expect(retry.nextDelay(new TypeError('fetch failed'))).toBeNull()
     expect(() => retry.remainingMs()).toThrow('could not be restored')

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

@@ -6,9 +6,13 @@ import {
   StreamEndedWithoutTerminalError,
 } from '@/lib/mothership/request/go/stream'
 
-/** A connection failure leaves the durable run unresolved; its leg keeps the existing time budget. */
+const MAX_STREAM_RETRIES = 3
+const STREAM_RECOVERY_WINDOW_MS = 30_000
+
+/** Recovery is bounded independently of the healthy run's execution budget. */
 export class StreamRetryWindow {
   private readonly deadline: number
+  private recoveryDeadline?: number
   attempt = 0
 
   constructor(timeoutMs = ORCHESTRATION_TIMEOUT_MS) {
@@ -24,8 +28,10 @@ export class StreamRetryWindow {
 
   nextDelay(error: unknown, signal?: AbortSignal): number | null {
     if (signal?.aborted || !isRetryableStreamError(error)) return null
+    this.recoveryDeadline ??= Date.now() + STREAM_RECOVERY_WINDOW_MS
+    if (this.attempt >= MAX_STREAM_RETRIES) return null
     const delay = backoffWithJitter(this.attempt + 1, null, { baseMs: 250, maxMs: 5_000 })
-    if (Date.now() + delay >= this.deadline) return null
+    if (Date.now() + delay >= Math.min(this.deadline, this.recoveryDeadline)) return null
     this.attempt++
     return delay
   }