fix(mothership): shed oversized replay events in one pass and retry mid-body stream cuts

- Replace the replay-compaction omit loop with one bounded pass over memoized
  exact serialized sizes: recurse into the smallest child that covers the
  overage, otherwise drop the largest children whole, keeping small siblings
- Map a non-abort body-read failure from the worker stream to a retryable
  WorkerStreamInterruptedError on the reachable budget, with the generic
  message and the original error on cause
- Log the cause message on retry warnings and orchestration failures
- Count replayed file preview content toward a restored turn's preview budget

Author: waleedlatif1

apps/sim/lib/mothership/request/context/restore.ts

@@ -13,6 +13,12 @@ export async function restoreStreamingContext(
   execContext: ExecutionContext
 ): Promise<void> {
   for (const saved of events) {
+    // Preview content already in the replay counts toward this turn's preview budget.
+    if (saved.type === 'tool' && 'previewPhase' in saved.payload) {
+      if (saved.payload.previewPhase === 'file_preview_content') {
+        context.filePreviewBudget.contentBytes += Buffer.byteLength(saved.payload.content, 'utf8')
+      }
+    }
     const event = reconcileTextEvent(saved, context.accumulatedContent)
     if (!event) continue
     const replay: StreamEvent =

apps/sim/lib/mothership/request/go/stream.test.ts

@@ -785,6 +785,53 @@ describe('copilot go stream helpers', () => {
     expect(fetch).toHaveBeenCalledTimes(1)
   })
 
+  it('reports a stream cut mid-body without the raw socket error', async () => {
+    const socketError = Object.assign(
+      new Error(
+        'The socket connection was closed unexpectedly. For more information, pass `verbose: true`'
+      ),
+      { code: 'ECONNRESET' }
+    )
+    const first = createEvent({
+      streamId: 'cut-stream',
+      cursor: '1',
+      seq: 1,
+      requestId: 'req-cut',
+      type: 'text',
+      payload: { channel: 'assistant', text: 'partial' },
+    })
+    vi.mocked(fetch).mockResolvedValueOnce(
+      new Response(
+        new ReadableStream<Uint8Array>({
+          start(controller) {
+            controller.enqueue(new TextEncoder().encode(`data: ${JSON.stringify(first)}\n\n`))
+          },
+          pull(controller) {
+            controller.error(socketError)
+          },
+        }),
+        { status: 200, headers: { 'Content-Type': 'text/event-stream' } }
+      )
+    )
+
+    await expect(
+      runStreamLoop(
+        'https://example.com/mothership/stream',
+        {},
+        createStreamingContext(),
+        turnScopedExecContext(),
+        {
+          timeout: 1000,
+          flushAfterEvent: false,
+        }
+      )
+    ).rejects.toMatchObject({
+      name: 'WorkerStreamInterruptedError',
+      message: 'The agent service is temporarily unavailable. Please try again.',
+      cause: socketError,
+    })
+  })
+
   it('reports a worker it could not reach without the raw network error', async () => {
     const networkError = new TypeError('fetch failed')
     vi.mocked(fetch).mockRejectedValueOnce(networkError)

apps/sim/lib/mothership/request/go/stream.ts

@@ -81,6 +81,18 @@ export class WorkerUnreachableError extends Error {
   }
 }
 
+/**
+ * The worker's response body failed mid-stream (the connection was reset or
+ * closed). The worker answered, so a retry reattaches under the short budget.
+ * The read error stays on `cause` for logs.
+ */
+export class WorkerStreamInterruptedError extends Error {
+  constructor(cause: unknown) {
+    super(BACKEND_UNAVAILABLE_MESSAGE, { cause })
+    this.name = 'WorkerStreamInterruptedError'
+  }
+}
+
 /**
  * A worker rejection message the user can act on: short, one line, plain text,
  * and free of identifiers (`userId`, `protocol_version_mismatch`) that only mean
@@ -293,7 +305,13 @@ export async function runStreamLoop(
   const rawReader = response.body.getReader()
   const reader: ReadableStreamDefaultReader<Uint8Array> = {
     async read() {
-      const result = await rawReader.read()
+      let result: ReadableStreamReadResult<Uint8Array>
+      try {
+        result = await rawReader.read()
+      } catch (error) {
+        if (requestSignal.aborted) throw error
+        throw new WorkerStreamInterruptedError(error)
+      }
       if (!result.done && result.value) {
         const now = performance.now()
         const gap = now - counters.lastChunkMs

apps/sim/lib/mothership/request/handlers/handlers.test.ts

@@ -215,6 +215,29 @@ describe('sse-handlers tool lifecycle', () => {
     expect(context.subAgentTraceSpans?.size).toBe(0)
   })
 
+  it('counts the preview content already in the replay toward the restored turn budget', async () => {
+    const preview = (content: string): StreamEvent => ({
+      type: 'tool',
+      payload: {
+        toolCallId: 'file-edit',
+        toolName: 'prepare_file_edit',
+        previewPhase: 'file_preview_content',
+        content,
+        contentMode: 'delta',
+        previewVersion: 1,
+        fileName: 'notes.md',
+      },
+    })
+
+    await restoreStreamingContext(
+      [preview('é'.repeat(1_000)), preview('x'.repeat(500))],
+      context,
+      execContext
+    )
+
+    expect(context.filePreviewBudget.contentBytes).toBe(2_500)
+  })
+
   it('restores a delivered prefix without executing its tools or consuming the next handoff', async () => {
     const call: StreamEvent = {
       type: 'tool',

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

@@ -128,9 +128,17 @@ vi.mock('@/lib/mothership/request/go/stream', () => {
     }
   }
 
+  class WorkerStreamInterruptedError extends Error {
+    constructor(cause: unknown) {
+      super('The agent service is temporarily unavailable. Please try again.', { cause })
+      this.name = 'WorkerStreamInterruptedError'
+    }
+  }
+
   return {
     BillingLimitError,
     CopilotBackendError,
+    WorkerStreamInterruptedError,
     WorkerUnreachableError,
     STREAM_ENDED_WITHOUT_TERMINAL_MESSAGE,
     StreamEndedWithoutTerminalError,

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

@@ -593,6 +593,7 @@ export async function runCopilotLifecycle(
       // explained, not just reduced to a message string.
       logger.error('Copilot orchestration failed', {
         error: err.message,
+        ...causeForLog(err),
         name: err.name,
         ...(error instanceof CopilotBackendError
           ? { backendStatus: error.status, backendBody: error.body?.slice(0, 2000) }
@@ -878,6 +879,7 @@ async function runResumeLegWithRetry(
           attempt: retry.attempts + 1,
           backoffMs: backoff,
           error: toError(error).message,
+          ...causeForLog(error),
         })
         await interruptibleSleep(backoff, options.abortSignal)
         continue
@@ -1240,6 +1242,7 @@ async function runCheckpointLoop(
             attempt: (retry?.attempts ?? 0) + 1,
             backoffMs: backoff,
             error: toError(streamError).message,
+            ...causeForLog(streamError),
           }
         )
         await interruptibleSleep(backoff, options.abortSignal)
@@ -1678,6 +1681,12 @@ async function withEnterpriseByokKey(
   return byokApiKey ? { ...refreshed, byokApiKey } : refreshed
 }
 
+/** The underlying failure behind a generic user-facing error, for logs. */
+function causeForLog(error: unknown): { cause?: string } {
+  const cause = error instanceof Error ? error.cause : undefined
+  return cause === undefined ? {} : { cause: getErrorMessage(cause) }
+}
+
 function isAborted(options: CopilotLifecycleOptions, context: StreamingContext): boolean {
   return !!(options.abortSignal?.aborted || context.wasAborted)
 }

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

@@ -2,6 +2,7 @@ import { afterEach, describe, expect, it, vi } from 'vitest'
 import {
   CopilotBackendError,
   StreamEndedWithoutTerminalError,
+  WorkerStreamInterruptedError,
   WorkerUnreachableError,
 } from '@/lib/mothership/request/go/stream'
 import { StreamRetryWindow } from '@/lib/mothership/request/lifecycle/stream-retry'
@@ -150,4 +151,16 @@ describe('stream recovery budget', () => {
     expect(retry.nextDelay(new Error('Invalid operation'))).toBeNull()
     expect(retry.attempt).toBe(0)
   })
+
+  it('gives a stream cut mid-body the reachable budget, not the unreachable window', () => {
+    vi.useFakeTimers()
+    const error = new WorkerStreamInterruptedError(new Error('socket closed'))
+    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.nextDelay(error)).toBeNull()
+  })
 })

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

@@ -4,6 +4,7 @@ import { StreamContinuityError } from '@/lib/mothership/request/go/parser'
 import {
   CopilotBackendError,
   StreamEndedWithoutTerminalError,
+  WorkerStreamInterruptedError,
   WorkerUnreachableError,
 } from '@/lib/mothership/request/go/stream'
 
@@ -108,7 +109,11 @@ function isJson(body: string | undefined): boolean {
 /** Initial sends and resumes both replay one durable identity after an ambiguous response. */
 function isRetryableStreamError(error: unknown): boolean {
   if (error instanceof Error && error.name === 'AbortError') return false
-  if (error instanceof StreamEndedWithoutTerminalError || error instanceof StreamContinuityError) {
+  if (
+    error instanceof StreamEndedWithoutTerminalError ||
+    error instanceof StreamContinuityError ||
+    error instanceof WorkerStreamInterruptedError
+  ) {
     return true
   }
   if (error instanceof CopilotBackendError) {

apps/sim/lib/mothership/request/session/replay-compaction.test.ts

@@ -16,6 +16,7 @@ import {
   STREAM_EVENT_COMPACTION_THRESHOLD_BYTES,
   STREAM_EVENT_MAX_PAYLOAD_BYTES,
   STREAM_STRING_PREVIEW_UNITS,
+  serializedBytes,
 } from '@/lib/mothership/request/session/replay-compaction'
 import type { StreamEvent } from '@/lib/mothership/request/session/types'
 
@@ -343,4 +344,129 @@ describe('compactStreamEvent', () => {
 
     expect(compactStreamEvent(event)).toBe(event)
   })
+
+  it('omits only the smallest sufficient bulk when no single child dominates, keeping its siblings', () => {
+    const manyKeys = (prefix: string) =>
+      Object.fromEntries(
+        Array.from({ length: 2_300 }, (_, index) => [`${prefix}-${index}`, 'v'.repeat(300)])
+      )
+    const event: StreamEvent = {
+      type: 'tool',
+      payload: {
+        toolCallId: 'c',
+        toolName: 'cli_tables_rows_query',
+        executor: 'sim',
+        mode: 'async',
+        phase: 'result',
+        success: true,
+        output: { fileId: 'file-1', rows: manyKeys('r'), cols: manyKeys('c') },
+      },
+    }
+
+    const compacted = payloadOf(compactStreamEvent(event))
+    const output = toRecord(compacted.output)
+
+    expect(Buffer.byteLength(JSON.stringify(compacted))).toBeLessThanOrEqual(
+      STREAM_EVENT_MAX_PAYLOAD_BYTES
+    )
+    expect(output.fileId).toBe('file-1')
+    expect([output.rows, output.cols].filter((value) => typeof value === 'string')).toHaveLength(1)
+  })
+
+  it('omits the smaller of two bulks when either alone would make the event fit', () => {
+    const manyKeys = (prefix: string, count: number) =>
+      Object.fromEntries(
+        Array.from({ length: count }, (_, index) => [`${prefix}-${index}`, 'v'.repeat(300)])
+      )
+    const event: StreamEvent = {
+      type: 'tool',
+      payload: {
+        toolCallId: 'c',
+        toolName: 'cli_tables_rows_query',
+        executor: 'sim',
+        mode: 'async',
+        phase: 'result',
+        success: true,
+        output: { fileId: 'file-1', rows: manyKeys('r', 3_000), cols: manyKeys('c', 1_900) },
+      },
+    }
+
+    const output = toRecord(payloadOf(compactStreamEvent(event)).output)
+
+    expect(output.fileId).toBe('file-1')
+    expect(Object.keys(toRecord(output.rows))).toHaveLength(3_000)
+    expect(output.cols).toMatch(/^…\[omitted, [\d.]+ KB total\]$/)
+  })
+
+  it('omits a sufficient bulk whole when its own large parts cannot cover the overage', () => {
+    const manyKeys = (prefix: string, count: number) =>
+      Object.fromEntries(
+        Array.from({ length: count }, (_, index) => [`${prefix}-${index}`, 'v'.repeat(3_000)])
+      )
+    const event: StreamEvent = {
+      type: 'tool',
+      payload: {
+        toolCallId: 'c',
+        toolName: 'cli_workflows_state_get',
+        executor: 'sim',
+        mode: 'async',
+        phase: 'result',
+        success: true,
+        output: { workflowId: 'wf-1', state: { nested: manyKeys('n', 35), ...manyKeys('k', 400) } },
+      },
+    }
+
+    const compacted = payloadOf(compactStreamEvent(event))
+    const output = toRecord(compacted.output)
+
+    expect(Buffer.byteLength(JSON.stringify(compacted))).toBeLessThanOrEqual(
+      STREAM_EVENT_MAX_PAYLOAD_BYTES
+    )
+    expect(output.workflowId).toBe('wf-1')
+    expect(output.state).toMatch(/^…\[omitted, [\d.]+ MB total\]$/)
+  })
+
+  it('bounds a balanced 16 MB tree in one pass', () => {
+    const tree = (depth: number, leaf: number): unknown =>
+      depth === 0
+        ? 'x'.repeat(leaf)
+        : { l: tree(depth - 1, leaf + 100), r: tree(depth - 1, Math.max(leaf - 100, 1)) }
+    const event: StreamEvent = {
+      type: 'tool',
+      payload: {
+        toolCallId: 'c',
+        toolName: 'cli_blocks_get',
+        executor: 'sim',
+        mode: 'async',
+        phase: 'result',
+        success: true,
+        output: tree(12, 3_900),
+      },
+    }
+
+    const started = performance.now()
+    const compacted = compactStreamEvent(event)
+    const elapsedMs = performance.now() - started
+
+    expect(Buffer.byteLength(JSON.stringify(compacted.payload))).toBeLessThanOrEqual(
+      STREAM_EVENT_MAX_PAYLOAD_BYTES
+    )
+    expect(elapsedMs).toBeLessThan(2_000)
+  })
+
+  it('measures serialized size exactly without serializing', () => {
+    const values: unknown[] = [
+      { a: 'é😀"\\\n', b: [1, null, true, { c: 'd' }], e: undefined, f: -1.5e-7 },
+      [undefined, 'x', { 'kéy "q"': 0 }],
+      'plain',
+      42,
+      null,
+      {},
+      [],
+    ]
+
+    for (const value of values) {
+      expect(serializedBytes(value)).toBe(Buffer.byteLength(JSON.stringify(value)))
+    }
+  })
 })