refactor(mothership): tidy replay compaction and stream retry after review

- Share the replay write ceiling between compaction and file previews
- Scan for the smallest sufficient child only once one can cover the need,
  and reuse one size memo across every compaction step
- Stop retrying a raw TypeError; fetch and body-read failures are already
  wrapped as WorkerUnreachableError and WorkerStreamInterruptedError
- Use absolute imports, a private retry counter, a typed unsettled-state set,
  and hasUnsettledTool naming
- Cover a timed-out body read and the preview completion edge cases

Author: waleedlatif1

apps/sim/app/workspace/[workspaceId]/home/hooks/message-reconcile.ts

@@ -146,11 +146,11 @@ export function buildAssistantSnapshotMessage(params: {
 }
 
 export function markMessageStopped(message: PersistedMessage): PersistedMessage {
-  const hasExecutingTool = message.contentBlocks?.some((block) =>
+  const hasUnsettledTool = message.contentBlocks?.some((block) =>
     isUnsettledToolState(block.toolCall?.state)
   )
   const hasOpenBlock = message.contentBlocks?.some((block) => block.endedAt === undefined)
-  if (!hasExecutingTool && !hasOpenBlock) {
+  if (!hasUnsettledTool && !hasOpenBlock) {
     return message
   }
 

apps/sim/app/workspace/[workspaceId]/home/hooks/preview/apply-file-preview-phase.test.ts

@@ -1,6 +1,9 @@
 import { describe, expect, it } from 'vitest'
 import type { FilePreviewSession } from '@/lib/mothership/request/session'
-import { deriveFilePreviewSession, previewHoldsFinalContent } from './apply-file-preview-phase'
+import {
+  deriveFilePreviewSession,
+  previewHoldsFinalContent,
+} from '@/app/workspace/[workspaceId]/home/hooks/preview/apply-file-preview-phase'
 
 const NOW = '2026-06-08T00:00:00.000Z'
 
@@ -118,4 +121,16 @@ describe('previewHoldsFinalContent', () => {
   it('does not hold it when no content was received at all', () => {
     expect(previewHoldsFinalContent(undefined, complete(7))).toBe(false)
   })
+
+  it('does not hold it when the session exists but received no text', () => {
+    const prev = session({ previewText: '', previewVersion: 7 })
+
+    expect(previewHoldsFinalContent(prev, complete(7))).toBe(false)
+  })
+
+  it('holds the received text when the completion carries no version to compare', () => {
+    const prev = session({ previewText: 'final text', previewVersion: 3 })
+
+    expect(previewHoldsFinalContent(prev, complete())).toBe(true)
+  })
 })

apps/sim/app/workspace/[workspaceId]/home/hooks/use-chat.ts

@@ -4370,11 +4370,11 @@ export function useChat(
         } else {
           setPendingMessages((prev) =>
             prev.map((msg) => {
-              const hasExecutingTool = msg.contentBlocks?.some((block) =>
+              const hasUnsettledTool = msg.contentBlocks?.some((block) =>
                 isUnsettledToolState(block.toolCall?.status)
               )
               const hasOpenBlock = msg.contentBlocks?.some((block) => block.endedAt === undefined)
-              if (!hasExecutingTool && !hasOpenBlock) {
+              if (!hasUnsettledTool && !hasOpenBlock) {
                 return msg
               }
               const updatedBlocks: ContentBlock[] = (msg.contentBlocks ?? []).map((block) => ({

apps/sim/lib/mothership/chat/display-message.ts

@@ -1,4 +1,8 @@
 import type { PersistedContentBlock } from '@/lib/api/contracts/copilot-messages'
+import { getMothershipAttachmentPreviewUrl } from '@/lib/mothership/chat/attachment-preview'
+import { isLiveAssistantMessageId } from '@/lib/mothership/chat/live-message-id'
+import type { PersistedMessage } from '@/lib/mothership/chat/persisted-message'
+import { isUnsettledToolState, withBlockTiming } from '@/lib/mothership/chat/persisted-message'
 import {
   MothershipStreamV1CompletionStatus,
   MothershipStreamV1EventType,
@@ -17,10 +21,6 @@ import {
   type ToolCallInfo,
   ToolCallStatus,
 } from '@/app/workspace/[workspaceId]/home/types'
-import { getMothershipAttachmentPreviewUrl } from './attachment-preview'
-import { isLiveAssistantMessageId } from './live-message-id'
-import type { PersistedMessage } from './persisted-message'
-import { isUnsettledToolState, withBlockTiming } from './persisted-message'
 
 const STATE_TO_STATUS: Record<string, ToolCallStatus> = {
   [MothershipStreamV1ToolOutcome.success]: ToolCallStatus.success,

apps/sim/lib/mothership/chat/persisted-message.ts

@@ -21,7 +21,11 @@ import {
   MothershipStreamV1ToolOutcome,
   MothershipStreamV1ToolPhase,
 } from '@/lib/mothership/generated/mothership-stream-v1'
-import type { ContentBlock, OrchestratorResult } from '@/lib/mothership/request/types'
+import type {
+  ContentBlock,
+  LocalToolCallStatus,
+  OrchestratorResult,
+} from '@/lib/mothership/request/types'
 import { RETIRED_BROWSER_REQUEST_TAKEOVER_ID } from '@/lib/mothership/tools/retired-tools'
 import { normalizeToolActivityDescription } from '@/lib/mothership/tools/tool-display'
 import type { BrowserTextSelection, TerminalTextSelection } from '@/stores/panel/types'
@@ -362,15 +366,16 @@ export function buildPersistedAssistantMessage(
   return message
 }
 
-const UNSETTLED_TOOL_STATES: ReadonlySet<string> = new Set([
+const UNSETTLED_TOOL_STATES: ReadonlySet<LocalToolCallStatus> = new Set<LocalToolCallStatus>([
   'pending',
   'executing',
   'awaiting_approval',
 ])
 
 /** A tool row that has not finished: waiting to run, running, or awaiting a decision. */
 export function isUnsettledToolState(state: string | undefined): boolean {
-  return state !== undefined && UNSETTLED_TOOL_STATES.has(state)
+  const unsettled: ReadonlySet<string | undefined> = UNSETTLED_TOOL_STATES
+  return unsettled.has(state)
 }
 
 /** Settles every unfinished tool row at a turn terminal so none reloads as a spinner. */

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

@@ -14,10 +14,12 @@ export async function restoreStreamingContext(
 ): 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')
-      }
+    if (
+      saved.type === 'tool' &&
+      'previewPhase' in saved.payload &&
+      saved.payload.previewPhase === 'file_preview_content'
+    ) {
+      context.filePreviewBudget.contentBytes += Buffer.byteLength(saved.payload.content, 'utf8')
     }
     const event = reconcileTextEvent(saved, context.accumulatedContent)
     if (!event) continue

apps/sim/lib/mothership/request/go/file-preview-adapter.test.ts

@@ -25,11 +25,11 @@ import { createStreamingContext } from '@/lib/mothership/request/context/request
 import {
   createFilePreviewAdapterState,
   type FilePreviewAdapterState,
-  PREVIEW_FRAME_MAX_BYTES,
   PREVIEW_TURN_CONTENT_BYTES,
   processFilePreviewStreamEvent,
 } from '@/lib/mothership/request/go/file-preview-adapter'
 import { createEvent, eventToStreamEvent } from '@/lib/mothership/request/session'
+import { STREAM_EVENT_MAX_PAYLOAD_BYTES } from '@/lib/mothership/request/session/replay-compaction'
 import type {
   ActiveFileIntent,
   ExecutionContext,
@@ -262,7 +262,7 @@ describe('processFilePreviewStreamEvent — preview byte rate', () => {
 
     for (const payload of payloads) {
       expect(Buffer.byteLength(JSON.stringify(payload))).toBeLessThanOrEqual(
-        PREVIEW_FRAME_MAX_BYTES
+        STREAM_EVENT_MAX_PAYLOAD_BYTES
       )
     }
     expect(completed).toBe(true)

apps/sim/lib/mothership/request/go/file-preview-adapter.ts

@@ -1,7 +1,6 @@
 import { createLogger } from '@sim/logger'
 import { toError } from '@sim/utils/errors'
 import { isRecordLike } from '@sim/utils/object'
-import { getRedisBudgetLimits } from '@/lib/core/redis/byte-budget.server'
 import { executeCopilotFileUseCase } from '@/lib/mothership/application/execute-file-use-case'
 import { MothershipStreamV1EventType } from '@/lib/mothership/generated/mothership-stream-v1'
 import {
@@ -15,6 +14,7 @@ import {
   type SyntheticFilePreviewPayload,
   upsertFilePreviewSession,
 } from '@/lib/mothership/request/session'
+import { STREAM_EVENT_MAX_PAYLOAD_BYTES } from '@/lib/mothership/request/session/replay-compaction'
 import type {
   ActiveFileIntent,
   ExecutionContext,
@@ -70,13 +70,6 @@ const PREVIEW_SNAPSHOT_CHARS_PER_SECOND = 256 * 1024
  */
 export const PREVIEW_TURN_CONTENT_BYTES = 8 * 1024 * 1024
 
-/** Room left in a replay write for the envelope around a preview payload. */
-const PREVIEW_FRAME_ENVELOPE_HEADROOM_BYTES = 16 * 1024
-
-/** The largest serialized preview payload the replay buffer can persist in one write. */
-export const PREVIEW_FRAME_MAX_BYTES =
-  getRedisBudgetLimits('copilot_stream').maxSingleWriteBytes - PREVIEW_FRAME_ENVELOPE_HEADROOM_BYTES
-
 /** The minimum gap between full snapshots of a preview this long. */
 function snapshotIntervalMs(baseMs: number, previewText: string): number {
   return Math.max(baseMs, (previewText.length / PREVIEW_SNAPSHOT_CHARS_PER_SECOND) * 1000)
@@ -397,13 +390,13 @@ async function emitPreviewEvent(
   if (frame.previewPhase === 'file_preview_content') {
     const budget = context.filePreviewBudget
     const contentBytes = budget.contentBytes + Buffer.byteLength(frame.content, 'utf8')
-    if (frameBytes > PREVIEW_FRAME_MAX_BYTES || contentBytes > PREVIEW_TURN_CONTENT_BYTES) {
+    if (frameBytes > STREAM_EVENT_MAX_PAYLOAD_BYTES || contentBytes > PREVIEW_TURN_CONTENT_BYTES) {
       return false
     }
     budget.contentBytes = contentBytes
   } else if (
     frame.previewPhase === 'file_preview_complete' &&
-    frameBytes > PREVIEW_FRAME_MAX_BYTES
+    frameBytes > STREAM_EVENT_MAX_PAYLOAD_BYTES
   ) {
     const { output: _output, ...withoutOutput } = frame
     frame = withoutOutput

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

@@ -832,6 +832,39 @@ describe('copilot go stream helpers', () => {
     })
   })
 
+  it('keeps the timeout error when the body read fails after the request timed out', async () => {
+    const first = createEvent({
+      streamId: 'timed-out-stream',
+      cursor: '1',
+      seq: 1,
+      requestId: 'req-timed-out',
+      type: 'text',
+      payload: { channel: 'assistant', text: 'partial' },
+    })
+    vi.mocked(fetch).mockImplementationOnce(async (_url, init) => {
+      const signal = init?.signal
+      return new Response(
+        new ReadableStream<Uint8Array>({
+          start(controller) {
+            controller.enqueue(new TextEncoder().encode(`data: ${JSON.stringify(first)}\n\n`))
+            signal?.addEventListener('abort', () => controller.error(signal.reason))
+          },
+        }),
+        { status: 200, headers: { 'Content-Type': 'text/event-stream' } }
+      )
+    })
+
+    await expect(
+      runStreamLoop(
+        'https://example.com/mothership/stream',
+        {},
+        createStreamingContext(),
+        turnScopedExecContext(),
+        { timeout: 20, flushAfterEvent: false }
+      )
+    ).rejects.toMatchObject({ name: 'TimeoutError' })
+  })
+
   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/lifecycle/run.test.ts

@@ -221,6 +221,7 @@ import {
   CopilotBackendError,
   STREAM_ENDED_WITHOUT_TERMINAL_MESSAGE,
   StreamEndedWithoutTerminalError,
+  WorkerUnreachableError,
 } from '@/lib/mothership/request/go/stream'
 import { runCopilotLifecycle } from '@/lib/mothership/request/lifecycle/run'
 import {
@@ -2718,7 +2719,7 @@ describe('runCopilotLifecycle', () => {
             headers.push(new Headers(request.headers))
             context.accumulatedContent = 'Saved partial answer'
             context.errors.push('connection interrupted')
-            throw new TypeError('fetch failed')
+            throw new WorkerUnreachableError(new TypeError('fetch failed'))
           }
         )
       }
@@ -2811,7 +2812,9 @@ describe('runCopilotLifecycle', () => {
     vi.useFakeTimers()
     try {
       const controller = new AbortController()
-      mockRunStreamLoop.mockRejectedValueOnce(new TypeError('fetch failed'))
+      mockRunStreamLoop.mockRejectedValueOnce(
+        new WorkerUnreachableError(new TypeError('fetch failed'))
+      )
       const pending = runCopilotLifecycle(
         { message: 'hello', messageId: 'stopped-outage' },
         {
@@ -2929,50 +2932,6 @@ describe('runCopilotLifecycle', () => {
     expect(JSON.stringify(persisted)).not.toMatch(/<html|Bad Gateway|nginx/)
   })
 
-  it('stops a TypeError that repeats while handling each reattach at three retries', async () => {
-    vi.useFakeTimers()
-    try {
-      let attempts = 0
-      mockRunStreamLoop.mockImplementation(
-        async (
-          _url: string,
-          _init: RequestInit,
-          _context: StreamingContext,
-          _exec: ExecutionContext,
-          options: { onEvent?: (event: unknown) => Promise<void> }
-        ): Promise<void> => {
-          attempts++
-          await options.onEvent?.({ type: 'session', payload: { kind: 'start' } })
-          throw new TypeError("Cannot read properties of undefined (reading 'payload')")
-        }
-      )
-
-      const pending = runCopilotLifecycle(
-        { message: 'hello', messageId: 'stream-repeated-type-error' },
-        {
-          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(60_000)
-
-      expect(attempts).toBe(4)
-      expect(await pending).toEqual(expect.objectContaining({ success: false }))
-    } finally {
-      mockRunStreamLoop.mockReset()
-      vi.useRealTimers()
-    }
-  })
-
   it('stops a failure that repeats after every reattach at three retries', async () => {
     vi.useFakeTimers()
     try {