fix(execution): classify Chat stream drops and stop async launches writing a terminal pointer

Only useExecutionStream.execute wrapped a transport failure as
SSEStreamInterruptedError; executeWorkflowWithFullLogging rethrew the raw
TypeError, so a mid-run network drop on the Chat run-tool path took the
generic branch: the tool reported "error" to Sim, the confirm route marked
the row failed, and the pointer was cleared while the server kept running
the workflow. The classifier is now one exported helper
(toStreamInterruptedError) used by both execute paths and by the shared
executor's post-acknowledgement catch, so the run tool reaches its
recoverable branch, reports "background", keeps the pointer, and releases
the run to the terminal reconnect.

Async launches wrote the terminal execution pointer only so
bindRunToolToExecution would find something after a reload, but an async
run has no reconnectable stream, so any reconnect against that pointer
404'd into the same synthetic Run Error. The tab-local pending completion
report already carries the execution id, so async launches no longer touch
the pointer and recovery answers from the pending report first, falling
back to the pointer only for a live run this tab was observing. The legacy
clearExecutionPointerAfterReport flag is still honoured for pointers older
clients left behind.

Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com>

Author: icecrasher321

apps/sim/app/workspace/[workspaceId]/w/[workflowId]/utils/workflow-execution-utils.test.ts

@@ -13,7 +13,10 @@ import {
   reconcileFinalBlockLogs,
 } from '@/app/workspace/[workspaceId]/w/[workflowId]/utils/workflow-execution-utils'
 import type { BlockLog } from '@/executor/types'
-import type { ExecutionStreamHttpError } from '@/hooks/use-execution-stream'
+import {
+  type ExecutionStreamHttpError,
+  SSEStreamInterruptedError,
+} from '@/hooks/use-execution-stream'
 import { useExecutionStore } from '@/stores/execution'
 
 describe('workflow-execution-utils', () => {
@@ -61,6 +64,88 @@ describe('workflow-execution-utils', () => {
     expect(terminalConsoleMockFns.mockAddConsole).not.toHaveBeenCalled()
   })
 
+  describe('executeWorkflowWithFullLogging stream interruption', () => {
+    /** A response whose server acknowledged the run and whose body then fails with `readError`. */
+    function stubAcknowledgedStream(readError: unknown) {
+      vi.stubGlobal(
+        'fetch',
+        vi.fn().mockResolvedValue({
+          ok: true,
+          status: 200,
+          headers: { get: (name: string) => (name === 'X-Execution-Id' ? 'exec-server' : null) },
+          body: {
+            getReader: () => ({
+              read: vi.fn().mockRejectedValue(readError),
+              releaseLock: vi.fn(),
+            }),
+          },
+        })
+      )
+    }
+
+    function stubExecutionStore() {
+      const store = {
+        getCurrentExecutionId: vi.fn(() => 'exec-server'),
+        setActiveBlocks: vi.fn(),
+        setBlockRunStatus: vi.fn(),
+        setCurrentExecutionId: vi.fn(),
+        setEdgeRunStatus: vi.fn(),
+        setIsExecuting: vi.fn(),
+      }
+      vi.mocked(useExecutionStore.getState).mockReturnValue(store as any)
+      return store
+    }
+
+    it('classifies a transport drop after the server acknowledged the run as an interruption', async () => {
+      /*
+       * The Chat run tool only preserves a run for reconnect when it sees
+       * SSEStreamInterruptedError; a raw TypeError from the body reader used to
+       * fall through as a plain failure, reporting an error to Sim and tearing
+       * the run down while the server kept executing it.
+       */
+      const store = stubExecutionStore()
+      stubAcknowledgedStream(new TypeError('network error'))
+
+      const promise = executeWorkflowWithFullLogging({
+        workflowId: 'wf-1',
+        executionId: 'exec-1',
+        copilotToolCallId: 'tool-1',
+        preserveExecutionOnTerminal: true,
+      })
+
+      await expect(promise).rejects.toBeInstanceOf(SSEStreamInterruptedError)
+      await expect(promise).rejects.toMatchObject({ executionId: 'exec-server' })
+      expect(store.setCurrentExecutionId).toHaveBeenCalledWith('wf-1', 'exec-server')
+      expect(store.setCurrentExecutionId).not.toHaveBeenCalledWith('wf-1', null)
+      expect(store.setIsExecuting).not.toHaveBeenCalled()
+    })
+
+    it.each([
+      ['a client abort', new DOMException('Aborted', 'AbortError')],
+      [
+        'the run tool stop reason, which aborts with a plain string',
+        'user_stop:cancelRunToolExecution',
+      ],
+      ['a non-transport failure', new Error('Unexpected token in JSON')],
+    ])('rethrows %s unclassified', async (_label, readError) => {
+      stubExecutionStore()
+      stubAcknowledgedStream(readError)
+
+      const rejection = await executeWorkflowWithFullLogging({
+        workflowId: 'wf-1',
+        executionId: 'exec-1',
+        preserveExecutionOnTerminal: true,
+      }).then(
+        () => {
+          throw new Error('expected the stream failure to reject')
+        },
+        (error: unknown) => error
+      )
+
+      expect(rejection).toBe(readError)
+    })
+  })
+
   describe('createBlockEventHandlers', () => {
     it('skips duplicate block start rows during reconnect replay', () => {
       terminalConsoleMockFns.mockAddConsole({

apps/sim/app/workspace/[workspaceId]/w/[workflowId]/utils/workflow-execution-utils.ts

@@ -1,5 +1,5 @@
 import { createLogger } from '@sim/logger'
-import { toError } from '@sim/utils/errors'
+import { getErrorMessage, toError } from '@sim/utils/errors'
 import { generateId } from '@sim/utils/id'
 import { isPlainRecord } from '@sim/utils/object'
 import { normalizeWorkflowEdgeSourceHandle } from '@sim/workflow-types/workflow'
@@ -19,6 +19,7 @@ import {
   processSSEStream,
   SSEEventHandlerError,
   SSEStreamInterruptedError,
+  toStreamInterruptedError,
 } from '@/hooks/use-execution-stream'
 import { useExecutionStore } from '@/stores/execution'
 import type { ConsoleEntry, ConsoleUpdate } from '@/stores/terminal'
@@ -1221,6 +1222,20 @@ export async function executeWorkflowWithFullLogging(
       'CopilotExecution'
     )
   } catch (error) {
+    const interrupted = toStreamInterruptedError(
+      error,
+      executionIdRef.current,
+      'Execution stream interrupted before a terminal event was received'
+    )
+    if (interrupted) {
+      logger.warn('Execution stream interrupted; preserving execution for reconnect', {
+        workflowId: wfId,
+        executionId: executionIdRef.current,
+        error: getErrorMessage(error),
+      })
+      preserveExecutionForRecovery = true
+      throw interrupted
+    }
     if (error instanceof SSEEventHandlerError || error instanceof SSEStreamInterruptedError) {
       preserveExecutionForRecovery = true
     }

apps/sim/hooks/use-execution-stream.ts

@@ -79,6 +79,21 @@ function isRecoverableStreamError(error: any): boolean {
   )
 }
 
+/**
+ * Wraps a transport failure that cut a live execution stream before its
+ * terminal event, so every consumer of a live stream classifies interruptions
+ * the same way and recovery code can rely on one error type. Returns null for
+ * client aborts and for anything that is not a transport failure.
+ */
+export function toStreamInterruptedError(
+  error: unknown,
+  executionId: string | undefined,
+  message: string
+): SSEStreamInterruptedError | null {
+  if (!isRecoverableStreamError(error)) return null
+  return new SSEStreamInterruptedError(message, executionId, error)
+}
+
 /**
  * Processes SSE events from a response body and invokes appropriate callbacks.
  * Exported for use by standalone (non-hook) execution paths like executeWorkflowWithFullLogging.
@@ -318,16 +333,17 @@ export function useExecutionStream() {
         logger.info('Execution stream disconnected (page unload or abort)')
         return
       }
-      if (isRecoverableStreamError(error)) {
+      const interrupted = toStreamInterruptedError(
+        error,
+        serverExecutionId,
+        'Execution stream interrupted before a terminal event was received'
+      )
+      if (interrupted) {
         logger.warn('Execution stream interrupted; preserving execution for reconnect', {
           executionId: serverExecutionId,
           error: error.message,
         })
-        throw new SSEStreamInterruptedError(
-          'Execution stream interrupted before a terminal event was received',
-          serverExecutionId,
-          error
-        )
+        throw interrupted
       }
       logger.error('Execution stream error:', error)
       if (!(error instanceof SSEEventHandlerError)) {
@@ -423,16 +439,17 @@ export function useExecutionStream() {
         logger.info('Run-from-block stream disconnected (page unload or abort)')
         return
       }
-      if (isRecoverableStreamError(error)) {
+      const interrupted = toStreamInterruptedError(
+        error,
+        serverExecutionId,
+        'Run-from-block stream interrupted before a terminal event was received'
+      )
+      if (interrupted) {
         logger.warn('Run-from-block stream interrupted; preserving execution for reconnect', {
           executionId: serverExecutionId,
           error: error.message,
         })
-        throw new SSEStreamInterruptedError(
-          'Run-from-block stream interrupted before a terminal event was received',
-          serverExecutionId,
-          error
-        )
+        throw interrupted
       }
       logger.error('Run-from-block execution error:', error)
       if (!(error instanceof SSEEventHandlerError)) {

apps/sim/lib/copilot/tools/client/run-tool-execution.test.ts

@@ -185,32 +185,43 @@ describe('run tool execution cancellation', () => {
     expect(isRunToolActiveForWorkflow('wf-1')).toBe(false)
   })
 
-  it('releases an interrupted run to reconnect subscribers only after giving up ownership', async () => {
-    const ownedAtRelease: boolean[] = []
-    const listener = vi.fn((workflowId: string) => {
-      ownedAtRelease.push(isRunToolActiveForWorkflow(workflowId))
-    })
-    const unsubscribe = subscribeToRunToolRelease(listener)
-    executeWorkflowWithFullLogging.mockRejectedValueOnce(
-      new MockSSEEventHandlerError('Block handler failed on event 7', 'exec-1')
-    )
-
-    try {
-      executeRunToolOnClient('tool-1', 'run_workflow', { workflowId: 'wf-1' })
-      await vi.waitFor(() => expect(listener).toHaveBeenCalledWith('wf-1'))
-
-      expect(listener).toHaveBeenCalledTimes(1)
-      expect(ownedAtRelease).toEqual([false])
-      expect(setIsExecuting).toHaveBeenCalledWith('wf-1', false)
-      expect(setCurrentExecutionId).toHaveBeenCalledWith('wf-1', null)
-      expect(setIsExecuting.mock.invocationCallOrder.at(-1)).toBeLessThan(
-        listener.mock.invocationCallOrder[0]
-      )
-      expect(clearExecutionPointer).not.toHaveBeenCalled()
-    } finally {
-      unsubscribe()
+  it.each([
+    ['handler', new MockSSEEventHandlerError('Block handler failed on event 7', 'exec-1')],
+    ['transport', new MockSSEStreamInterruptedError('Execution stream interrupted', 'exec-1')],
+  ])(
+    'releases a run whose stream was cut by a %s failure only after giving up ownership',
+    async (_kind, interruption) => {
+      const ownedAtRelease: boolean[] = []
+      const listener = vi.fn((workflowId: string) => {
+        ownedAtRelease.push(isRunToolActiveForWorkflow(workflowId))
+      })
+      const unsubscribe = subscribeToRunToolRelease(listener)
+      executeWorkflowWithFullLogging.mockRejectedValueOnce(interruption)
+
+      try {
+        executeRunToolOnClient('tool-1', 'run_workflow', { workflowId: 'wf-1' })
+        await vi.waitFor(() => expect(listener).toHaveBeenCalledWith('wf-1'))
+
+        expect(listener).toHaveBeenCalledTimes(1)
+        expect(ownedAtRelease).toEqual([false])
+        expect(setIsExecuting).toHaveBeenCalledWith('wf-1', false)
+        expect(setCurrentExecutionId).toHaveBeenCalledWith('wf-1', null)
+        expect(setIsExecuting.mock.invocationCallOrder.at(-1)).toBeLessThan(
+          listener.mock.invocationCallOrder[0]
+        )
+        expect(clearExecutionPointer).not.toHaveBeenCalled()
+        expect(fetch).toHaveBeenCalledWith(
+          '/api/copilot/confirm',
+          expect.objectContaining({
+            body: expect.stringContaining('"status":"background"'),
+          })
+        )
+        expect(vi.mocked(fetch).mock.calls[0][1]?.body).toContain('"executionId":"exec-1"')
+      } finally {
+        unsubscribe()
+      }
     }
-  })
+  )
 
   it('does not release a run it observed to completion, even when the report fails', async () => {
     const listener = vi.fn()
@@ -331,12 +342,11 @@ describe('run tool execution cancellation', () => {
     expect(fetchMock.mock.calls[1][0]).toBe('/api/copilot/confirm')
     expect(fetchMock.mock.calls[1][1]?.body).toContain('"status":"background"')
     expect(fetchMock.mock.calls[1][1]?.body).toContain('"executionId":"exec-async"')
-    expect(saveExecutionPointer).toHaveBeenCalledWith({
-      workflowId: 'wf-1',
-      executionId: 'exec-async',
-      lastEventId: 0,
-    })
-    expect(clearExecutionPointer).toHaveBeenCalledWith('wf-1')
+    // An async run has no reconnectable stream, so it must never leave the
+    // terminal a pointer that a reconnect would 404 against.
+    expect(saveExecutionPointer).not.toHaveBeenCalled()
+    expect(clearExecutionPointer).not.toHaveBeenCalled()
+    expect(window.sessionStorage.getItem('sim:copilot:run-tool-completion:tool-async')).toBeNull()
   })
 
   it('recovers a queued async launch by re-reporting it without enqueueing again', async () => {
@@ -362,11 +372,10 @@ describe('run tool execution cancellation', () => {
 
     await vi.waitFor(() => expect(fetchMock).toHaveBeenCalledTimes(6))
     await vi.waitFor(() => expect(isRunToolActiveForId('tool-recover-async')).toBe(false))
-    loadExecutionPointer.mockResolvedValueOnce({
-      workflowId: 'wf-1',
-      executionId: 'exec-recover-async',
-      lastEventId: 0,
-    })
+    expect(saveExecutionPointer).not.toHaveBeenCalled()
+    expect(
+      window.sessionStorage.getItem('sim:copilot:run-tool-completion:tool-recover-async')
+    ).toContain('"executionId":"exec-recover-async"')
 
     await expect(bindRunToolToExecution('tool-recover-async', 'wf-1')).resolves.toBe(true)
 
@@ -377,6 +386,34 @@ describe('run tool execution cancellation', () => {
     expect(
       fetchMock.mock.calls.filter(([url]) => url === '/api/workflows/wf-1/execute')
     ).toHaveLength(1)
+    expect(clearExecutionPointer).not.toHaveBeenCalled()
+    expect(
+      window.sessionStorage.getItem('sim:copilot:run-tool-completion:tool-recover-async')
+    ).toBeNull()
+  })
+
+  it('cleans up the terminal pointer an earlier client left for an async launch', async () => {
+    const fetchMock = vi.fn().mockResolvedValue({ ok: true })
+    vi.stubGlobal('fetch', fetchMock)
+    loadExecutionPointer.mockResolvedValueOnce({
+      workflowId: 'wf-1',
+      executionId: 'exec-legacy-async',
+      lastEventId: 0,
+    })
+    window.sessionStorage.setItem(
+      'sim:copilot:run-tool-completion:tool-legacy-async',
+      JSON.stringify({
+        status: 'background',
+        executionId: 'exec-legacy-async',
+        clearExecutionPointerAfterReport: true,
+      })
+    )
+
+    await expect(bindRunToolToExecution('tool-legacy-async', 'wf-1')).resolves.toBe(true)
+
+    expect(fetchMock).toHaveBeenCalledTimes(1)
+    expect(fetchMock.mock.calls[0][1]?.body).toContain('"status":"background"')
+    expect(fetchMock.mock.calls[0][1]?.body).toContain('"executionId":"exec-legacy-async"')
     expect(clearExecutionPointer).toHaveBeenCalledWith('wf-1')
   })