fix(execution): keep terminal reconnect off runs a Sim run tool owns

apps/sim/app/workspace/[workspaceId]/w/[workflowId]/hooks/use-workflow-execution.test.tsx

@@ -19,12 +19,15 @@ const {
   mockFetch,
   mockHandleExecutionCancelledConsole,
   mockHandleExecutionErrorConsole,
+  mockIsExecutionStreamHttpError,
+  mockIsRunToolActiveForWorkflow,
   mockLoadExecutionPointer,
   mockReconnect,
   mockRequestJson,
   mockResolveStartCandidates,
   mockSelectBestTrigger,
   mockUploadInternalFileSession,
+  runToolReleaseListeners,
   terminalStoreState,
   workflowBlocks,
   workflowStoreState,
@@ -101,12 +104,15 @@ const {
     mockFetch: vi.fn(),
     mockHandleExecutionCancelledConsole: vi.fn(),
     mockHandleExecutionErrorConsole: vi.fn(),
+    mockIsExecutionStreamHttpError: vi.fn(() => false),
+    mockIsRunToolActiveForWorkflow: vi.fn(() => false),
     mockLoadExecutionPointer: vi.fn(),
     mockReconnect: vi.fn(),
     mockRequestJson: vi.fn(),
     mockResolveStartCandidates: vi.fn(),
     mockSelectBestTrigger: vi.fn(),
     mockUploadInternalFileSession: vi.fn(),
+    runToolReleaseListeners: new Set<(workflowId: string) => void>(),
     terminalStoreState,
     workflowBlocks,
     workflowStoreState,
@@ -125,6 +131,16 @@ vi.mock('@/lib/api/client/request', () => ({
   requestJson: mockRequestJson,
 }))
 
+vi.mock('@/lib/copilot/tools/client/run-tool-execution', () => ({
+  isRunToolActiveForWorkflow: mockIsRunToolActiveForWorkflow,
+  subscribeToRunToolRelease: (listener: (workflowId: string) => void) => {
+    runToolReleaseListeners.add(listener)
+    return () => {
+      runToolReleaseListeners.delete(listener)
+    }
+  },
+}))
+
 vi.mock('@/lib/api/contracts/workflows', () => ({
   cancelWorkflowExecutionContract: {},
   workflowLogContract: {},
@@ -214,7 +230,7 @@ vi.mock('@/hooks/use-execution-stream', () => {
   class SSEStreamInterruptedError extends Error {}
 
   return {
-    isExecutionStreamHttpError: () => false,
+    isExecutionStreamHttpError: mockIsExecutionStreamHttpError,
     SSEEventHandlerError,
     SSEStreamInterruptedError,
     useExecutionStream: () => ({
@@ -419,6 +435,8 @@ function resetWorkflowExecutionTestState() {
   mockBeginScopedExecution.mockReset().mockReturnValue({})
   mockAdoptScopedExecution.mockReset().mockReturnValue(undefined)
   mockEndScopedExecution.mockReset().mockReturnValue(true)
+  mockIsExecutionStreamHttpError.mockReset().mockReturnValue(false)
+  mockIsRunToolActiveForWorkflow.mockReset().mockReturnValue(false)
   mockLoadExecutionPointer.mockReset().mockResolvedValue(null)
   mockReconnect.mockReset().mockResolvedValue(undefined)
   mockResolveStartCandidates.mockReset().mockReturnValue([])
@@ -430,6 +448,36 @@ function resetWorkflowExecutionTestState() {
   executionStoreState.getWorkflowExecution.mockReturnValue(idleExecution)
   executionStoreState.getCurrentExecutionId.mockReturnValue(null)
   workflowStoreState.edges.length = 0
+  runToolReleaseListeners.clear()
+}
+
+/**
+ * The store and pointer state a Sim run tool leaves behind the moment it starts
+ * a run, before the server has acknowledged it: this is what the reconnect
+ * flow reads as an orphaned run.
+ */
+function primeRunToolOwnedExecution() {
+  terminalStoreState._hasHydrated = true
+  executionStoreState.getWorkflowExecution.mockReturnValue({
+    ...executionStoreState.getWorkflowExecution(),
+    status: 'running',
+    isExecuting: true,
+    currentExecutionId: 'execution-1',
+  })
+  executionStoreState.getCurrentExecutionId.mockReturnValue('execution-1')
+  mockLoadExecutionPointer.mockResolvedValue({
+    workflowId: 'workflow-1',
+    executionId: 'execution-1',
+    lastEventId: 0,
+  })
+}
+
+/** The reconnect endpoint's answer while the run's buffer does not exist yet. */
+function rejectReconnectWithMissingRunBuffer() {
+  mockIsExecutionStreamHttpError.mockReturnValue(true)
+  mockReconnect.mockRejectedValue(
+    Object.assign(new Error('Reconnect failed (404)'), { httpStatus: 404 })
+  )
 }
 
 describe('useWorkflowExecution lifecycle ownership', () => {
@@ -594,6 +642,118 @@ describe('useWorkflowExecution lifecycle ownership', () => {
     unmount()
   })
 
+  it('logs a Run Error when a reconnect for an unowned pointer finds no run buffer', async () => {
+    primeRunToolOwnedExecution()
+    rejectReconnectWithMissingRunBuffer()
+
+    const { unmount } = renderWorkflowExecutionHook()
+    await act(async () => {
+      await Promise.resolve()
+      await Promise.resolve()
+    })
+
+    expect(mockReconnect).toHaveBeenCalledTimes(1)
+    expect(mockHandleExecutionErrorConsole.mock.calls[0]).toEqual(
+      expect.arrayContaining([
+        expect.objectContaining({
+          executionId: 'execution-1',
+          error: 'Execution state is no longer available after reconnect',
+        }),
+      ])
+    )
+    expect(executionStoreState.setCurrentExecutionId).toHaveBeenCalledWith('workflow-1', null)
+    expect(executionStoreState.setIsExecuting).toHaveBeenCalledWith('workflow-1', false)
+    expect(mockClearExecutionPointer).toHaveBeenCalledWith('workflow-1')
+
+    unmount()
+  })
+
+  it('leaves a run owned by a client run tool to its live stream instead of reconnecting', async () => {
+    /*
+     * Same state as above, but a Sim run tool in this tab still owns the run.
+     * Its live stream is the source of truth, so reconnecting here would race
+     * the run's own start (the 404 above, logged as a Run Error mid-run), tear
+     * down the live run's store state, and clear the pointer the tool keeps
+     * for reload recovery.
+     */
+    primeRunToolOwnedExecution()
+    rejectReconnectWithMissingRunBuffer()
+    mockIsRunToolActiveForWorkflow.mockReturnValue(true)
+
+    const { unmount } = renderWorkflowExecutionHook()
+    await act(async () => {
+      await Promise.resolve()
+      await Promise.resolve()
+    })
+
+    expect(mockIsRunToolActiveForWorkflow).toHaveBeenCalledWith('workflow-1')
+    expect(mockReconnect).not.toHaveBeenCalled()
+    expect(mockHandleExecutionErrorConsole).not.toHaveBeenCalled()
+    expect(mockClearExecutionPointer).not.toHaveBeenCalled()
+    expect(executionStoreState.setCurrentExecutionId).not.toHaveBeenCalled()
+    expect(executionStoreState.setIsExecuting).not.toHaveBeenCalled()
+    expect(executionStoreState.setActiveBlocks).not.toHaveBeenCalled()
+
+    unmount()
+  })
+
+  it('reconnects once the client run tool releases a run whose stream dropped', async () => {
+    primeRunToolOwnedExecution()
+    mockIsRunToolActiveForWorkflow.mockReturnValue(true)
+
+    const { unmount } = renderWorkflowExecutionHook()
+    await act(async () => {
+      await Promise.resolve()
+      await Promise.resolve()
+    })
+    expect(mockReconnect).not.toHaveBeenCalled()
+    expect(runToolReleaseListeners.size).toBeGreaterThan(0)
+
+    /*
+     * What the run tool leaves behind when it gives the run up: no current
+     * execution, not executing, ownership released, and the pointer still
+     * carrying the last event it persisted.
+     */
+    executionStoreState.getWorkflowExecution.mockReturnValue({
+      ...executionStoreState.getWorkflowExecution(),
+      status: 'idle',
+      isExecuting: false,
+      currentExecutionId: null,
+    })
+    executionStoreState.getCurrentExecutionId.mockReturnValue(null)
+    mockLoadExecutionPointer.mockResolvedValue({
+      workflowId: 'workflow-1',
+      executionId: 'execution-1',
+      lastEventId: 5,
+    })
+    mockIsRunToolActiveForWorkflow.mockReturnValue(false)
+    await act(async () => {
+      for (const listener of runToolReleaseListeners) listener('workflow-2')
+      await Promise.resolve()
+      await Promise.resolve()
+    })
+    expect(mockReconnect).not.toHaveBeenCalled()
+
+    await act(async () => {
+      for (const listener of runToolReleaseListeners) listener('workflow-1')
+      await Promise.resolve()
+      await Promise.resolve()
+    })
+
+    expect(mockReconnect).toHaveBeenCalledTimes(1)
+    expect(mockReconnect).toHaveBeenCalledWith(
+      expect.objectContaining({
+        workflowId: 'workflow-1',
+        executionId: 'execution-1',
+        fromEventId: 5,
+      })
+    )
+    expect(mockClearExecutionPointer).not.toHaveBeenCalled()
+
+    unmount()
+    expect(runToolReleaseListeners.size).toBe(0)
+  })
+
   it('does not let delayed debug completion reset a replacement execution', async () => {
     const debugPersistenceExecution = {}
     const replacementPersistenceExecution = {}

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

@@ -21,6 +21,10 @@ import {
   workflowLogContract,
   workflowStateSchema,
 } from '@/lib/api/contracts/workflows'
+import {
+  isRunToolActiveForWorkflow,
+  subscribeToRunToolRelease,
+} from '@/lib/copilot/tools/client/run-tool-execution'
 import type { SecretSafeBlockLog } from '@/lib/logs/execution/display-types'
 import { buildTraceSpans } from '@/lib/logs/execution/trace-spans/trace-spans'
 import { processStreamingBlockLogs } from '@/lib/tokenization'
@@ -2386,6 +2390,14 @@ export function useWorkflowExecution() {
     [activeWorkflowId, setExecutionResult, tryStartExecution]
   )
 
+  useEffect(() => {
+    if (!activeWorkflowId) return
+    return subscribeToRunToolRelease((workflowId) => {
+      if (workflowId !== activeWorkflowId) return
+      setReconnectAttemptNonce((nonce) => nonce + 1)
+    })
+  }, [activeWorkflowId])
+
   useEffect(() => {
     if (!activeWorkflowId || !hasHydrated) return
     if (activeReconnections.has(activeWorkflowId)) return
@@ -2404,6 +2416,13 @@ export function useWorkflowExecution() {
     }
 
     const runReconnect = async () => {
+      if (isRunToolActiveForWorkflow(reconnectWorkflowId)) {
+        logger.info('Reconnection skipped; a client run tool owns this workflow run', {
+          workflowId: reconnectWorkflowId,
+        })
+        return
+      }
+
       let executionId: string | undefined
       let fromEventId = 0
 

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

@@ -13,7 +13,11 @@ 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 {
+  ExecutionStreamHttpError,
+  SSEEventHandlerError,
+  SSEStreamInterruptedError,
+} from '@/hooks/use-execution-stream'
 import { useExecutionStore } from '@/stores/execution'
 
 describe('workflow-execution-utils', () => {
@@ -61,6 +65,111 @@ 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.each([
+      ['Chrome', 'network error'],
+      ['Chrome before headers', 'Failed to fetch'],
+      ['Firefox', 'NetworkError when attempting to fetch resource.'],
+      ['Safari', 'Load failed'],
+    ])(
+      'classifies a %s transport drop after the server acknowledged the run as an interruption',
+      async (_browser, message) => {
+        /*
+         * 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(message))
+
+        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 nullish rejection', null],
+      ['a client abort', new DOMException('Aborted', 'AbortError')],
+      [
+        'an HTTP rejection whose message mentions a transport phrase',
+        new ExecutionStreamHttpError('Failed to fetch workflow state', 500),
+      ],
+      [
+        'a handler failure whose message mentions a transport phrase',
+        new SSEEventHandlerError(
+          'network error while persisting console rows',
+          'block:completed',
+          3,
+          'exec-server',
+          new Error('persist failed')
+        ),
+      ],
+      [
+        '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({