diff --git a/apps/sim/hooks/use-execution-stream.ts b/apps/sim/hooks/use-execution-stream.ts
index 024b4174a84..89efdd7100e 100644
--- a/apps/sim/hooks/use-execution-stream.ts
+++ b/apps/sim/hooks/use-execution-stream.ts
@@ -285,6 +285,11 @@ function reconnectStreamKey(workflowId: string, executionId: string): string {
   return `${workflowId}:reconnect:${executionId}`
 }
 
+/** Whether a reconnect stream for this execution is currently open in this tab. */
+export function isReconnectStreamOpen(workflowId: string, executionId: string): boolean {
+  return sharedAbortControllers.has(reconnectStreamKey(workflowId, executionId))
+}
+
 function abortStream(key: string): void {
   const controller = sharedAbortControllers.get(key)
   if (!controller) return
diff --git a/apps/sim/lib/mothership/tools/client/run-tool-execution.test.ts b/apps/sim/lib/mothership/tools/client/run-tool-execution.test.ts
index 591231efdf5..aa43f04fb1f 100644
--- a/apps/sim/lib/mothership/tools/client/run-tool-execution.test.ts
+++ b/apps/sim/lib/mothership/tools/client/run-tool-execution.test.ts
@@ -12,17 +12,21 @@ import {
   workflowRegistryStoreMock,
   workflowRegistryStoreMockFns,
 } from '@sim/testing/mocks/workflow-registry-store.mock'
-import { beforeEach, describe, expect, it, vi } from 'vitest'
+import { sleep } from '@sim/utils/helpers'
+import { beforeEach, describe, expect, it, onTestFinished, vi } from 'vitest'
+import { getWorkflowExecutionContract } from '@/lib/api/contracts/workflows'
 import type { WorkflowExecutionOptions } from '@/app/workspace/[workspaceId]/w/[workflowId]/utils/workflow-execution-utils'
 import { useExecutionStore } from '@/stores/execution/store'
 
 const {
   executeWorkflowWithFullLogging,
+  isReconnectStreamOpen,
   MockExecutionStreamHttpError,
   MockSSEEventHandlerError,
   MockSSEStreamInterruptedError,
 } = vi.hoisted(() => ({
   executeWorkflowWithFullLogging: vi.fn(),
+  isReconnectStreamOpen: vi.fn(() => false),
   MockExecutionStreamHttpError: class ExecutionStreamHttpError extends Error {
     constructor(
       message: string,
@@ -75,9 +79,16 @@ function requireAbortSignal(options: WorkflowExecutionOptions): AbortSignal {
   return options.abortSignal
 }
 
+/** Waits for the run tool to launch its client execution and returns the execution id. */
+async function launchedExecutionId(): Promise<string> {
+  await vi.waitFor(() => expect(executeWorkflowWithFullLogging).toHaveBeenCalled())
+  return executeWorkflowWithFullLogging.mock.calls[0][0].executionId
+}
+
 vi.mock('@/hooks/use-execution-stream', () => ({
   ExecutionStreamHttpError: MockExecutionStreamHttpError,
   isExecutionStreamHttpError: (error: unknown) => error instanceof MockExecutionStreamHttpError,
+  isReconnectStreamOpen,
   SSEEventHandlerError: MockSSEEventHandlerError,
   SSEStreamInterruptedError: MockSSEStreamInterruptedError,
 }))
@@ -180,22 +191,24 @@ describe('run tool execution cancellation', () => {
     ['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) => {
+    async (kind, interruption) => {
+      // A dropped run keeps its workflow until it settles, so each case gets its own.
+      const workflowId = `wf-stream-${kind}`
       const ownedAtRelease: boolean[] = []
-      const listener = vi.fn((workflowId: string) => {
-        ownedAtRelease.push(isRunToolActiveForWorkflow(workflowId))
+      const listener = vi.fn((releasedWorkflowId: string) => {
+        ownedAtRelease.push(isRunToolActiveForWorkflow(releasedWorkflowId))
       })
       const unsubscribe = subscribeToRunToolRelease(listener)
       executeWorkflowWithFullLogging.mockRejectedValueOnce(interruption)
 
       try {
-        executeRunToolOnClient('tool-1', 'run_workflow', { workflowId: 'wf-1' })
-        await vi.waitFor(() => expect(listener).toHaveBeenCalledWith('wf-1'))
+        executeRunToolOnClient('tool-1', 'run_workflow', { workflowId })
+        await vi.waitFor(() => expect(listener).toHaveBeenCalledWith(workflowId))
 
         expect(listener).toHaveBeenCalledTimes(1)
         expect(ownedAtRelease).toEqual([false])
-        expect(setIsExecuting).toHaveBeenCalledWith('wf-1', false)
-        expect(setCurrentExecutionId).toHaveBeenCalledWith('wf-1', null)
+        expect(setIsExecuting).toHaveBeenCalledWith(workflowId, false)
+        expect(setCurrentExecutionId).toHaveBeenCalledWith(workflowId, null)
         expect(setIsExecuting.mock.invocationCallOrder.at(-1)).toBeLessThan(
           listener.mock.invocationCallOrder[0]
         )
@@ -241,7 +254,7 @@ describe('run tool execution cancellation', () => {
         })
     )
     executeRunToolOnClient('tool-override', 'run_workflow', { workflowId: 'wf-1' })
-    const executionId = executeWorkflowWithFullLogging.mock.calls[0][0].executionId
+    const executionId = await launchedExecutionId()
     expect(stopRunToolForExecution('wf-1', 'newer-manual-execution')).toBe(false)
     expect(mockRequestJson).not.toHaveBeenCalled()
     getCurrentExecutionId.mockReturnValue('newer-manual-execution')
@@ -283,9 +296,7 @@ describe('run tool execution cancellation', () => {
       useDeployedState: true,
     })
 
-    await Promise.resolve()
-    await Promise.resolve()
-
+    await launchedExecutionId()
     expect(executeWorkflowWithFullLogging).toHaveBeenCalledWith(
       expect.objectContaining({
         workflowId: 'wf-1',
@@ -295,7 +306,7 @@ describe('run tool execution cancellation', () => {
         useDraftState: false,
       })
     )
-    const executionId = executeWorkflowWithFullLogging.mock.calls[0][0].executionId
+    const executionId = await launchedExecutionId()
     await vi.waitFor(() => {
       expect(fetch).toHaveBeenCalledWith(
         '/api/copilot/confirm',
@@ -320,8 +331,8 @@ describe('run tool execution cancellation', () => {
           })
       )
       const toolCallId = `owned-${outcome}`
-      executeRunToolOnClient(toolCallId, 'run_workflow', { workflowId: 'wf-1' })
-      const executionId = executeWorkflowWithFullLogging.mock.calls[0][0].executionId
+      executeRunToolOnClient(toolCallId, 'run_workflow', { workflowId: `wf-owned-${outcome}` })
+      const executionId = await launchedExecutionId()
       getCurrentExecutionId.mockReturnValue('new-manual-execution')
       setCurrentExecutionId.mockClear()
       setIsExecuting.mockClear()
@@ -513,7 +524,7 @@ describe('run tool execution cancellation', () => {
         })
       )
     })
-    const executionId = executeWorkflowWithFullLogging.mock.calls[0][0].executionId
+    const executionId = await launchedExecutionId()
     expect(fetchMock.mock.calls[0][1]?.body).toContain(`"executionId":"${executionId}"`)
     expect(fetchMock.mock.calls[0][1]?.body).not.toContain('workflow failed')
   })
@@ -668,4 +679,321 @@ describe('run tool execution cancellation', () => {
     expect(fetchMock).not.toHaveBeenCalled()
     expect(setIsExecuting).toHaveBeenCalledWith('wf-1', false)
   })
+
+  describe('same-workflow calls issued while an agent run is in flight', () => {
+    /** Holds every client execution open until the test settles it by tool call id. */
+    function holdExecutions() {
+      const settle = new Map<string, (result: { success: boolean }) => void>()
+      onTestFinished(() => {
+        executeWorkflowWithFullLogging.mockReset()
+      })
+      executeWorkflowWithFullLogging.mockImplementation(
+        (options: WorkflowExecutionOptions) =>
+          new Promise((resolve, reject) => {
+            settle.set(String(options.copilotToolCallId), resolve)
+            requireAbortSignal(options).addEventListener(
+              'abort',
+              () => reject(new DOMException('Aborted', 'AbortError')),
+              { once: true }
+            )
+          })
+      )
+      const launchedToolCallIds = () =>
+        executeWorkflowWithFullLogging.mock.calls.map(
+          ([options]: [WorkflowExecutionOptions]) => options.copilotToolCallId
+        )
+      const finish = (toolCallId: string) => {
+        const resolve = settle.get(toolCallId)
+        if (!resolve) throw new Error(`${toolCallId} was never launched`)
+        resolve({ success: true })
+      }
+      return { finish, launchedToolCallIds }
+    }
+
+    function confirmBodies(): Array<Record<string, unknown>> {
+      return vi
+        .mocked(fetch)
+        .mock.calls.filter(([url]) => url === '/api/copilot/confirm')
+        .map(([, init]) => JSON.parse(String(init?.body)))
+    }
+
+    function confirmStatuses(): Array<[unknown, unknown]> {
+      return confirmBodies().map((body) => [body.toolCallId, body.status])
+    }
+
+    function statusFetches(): number {
+      return mockRequestJson.mock.calls.filter(
+        ([contract]) => contract === getWorkflowExecutionContract
+      ).length
+    }
+
+    async function drain(workflowId = 'wf-1') {
+      await vi.waitFor(() => expect(isRunToolActiveForWorkflow(workflowId)).toBe(false))
+    }
+
+    /** Makes the next launched run lose its stream while the server keeps running it. */
+    function interruptFirstRun(executionId: string) {
+      let serverStatus = 'running'
+      mockRequestJson.mockImplementation(async (contract: unknown) =>
+        contract === getWorkflowExecutionContract ? { status: serverStatus } : { success: true }
+      )
+      executeWorkflowWithFullLogging.mockRejectedValueOnce(
+        new MockSSEStreamInterruptedError('Execution stream interrupted', executionId)
+      )
+      const released = vi.fn()
+      onTestFinished(subscribeToRunToolRelease(released))
+      return {
+        released,
+        setServerStatus: (status: string) => {
+          serverStatus = status
+        },
+      }
+    }
+
+    /** Controls whether the editor holds a reconnect stream open for the interrupted run. */
+    function reconnectStream(open: boolean) {
+      let isOpen = open
+      isReconnectStreamOpen.mockImplementation(() => isOpen)
+      onTestFinished(() => {
+        isReconnectStreamOpen.mockReset()
+        isReconnectStreamOpen.mockReturnValue(false)
+      })
+      return {
+        close: () => {
+          isOpen = false
+        },
+      }
+    }
+
+    it('runs them one at a time in the order they were issued instead of rejecting them as busy', async () => {
+      const { finish, launchedToolCallIds } = holdExecutions()
+
+      executeRunToolOnClient('tool-a', 'run_block', { workflowId: 'wf-1', blockId: 'block-1' })
+      executeRunToolOnClient('tool-b', 'run_block', { workflowId: 'wf-1', blockId: 'block-2' })
+      executeRunToolOnClient('tool-c', 'run_workflow', { workflowId: 'wf-1' })
+      await vi.waitFor(() => expect(launchedToolCallIds()).toEqual(['tool-a']))
+
+      finish('tool-a')
+      await vi.waitFor(() => expect(launchedToolCallIds()).toEqual(['tool-a', 'tool-b']))
+      finish('tool-b')
+      await vi.waitFor(() => expect(launchedToolCallIds()).toEqual(['tool-a', 'tool-b', 'tool-c']))
+      finish('tool-c')
+      await drain()
+
+      expect(confirmStatuses()).toEqual([
+        ['tool-a', 'success'],
+        ['tool-b', 'success'],
+        ['tool-c', 'success'],
+      ])
+    })
+
+    it('keeps a call issued while the previous run is releasing behind the calls already waiting', async () => {
+      const { finish, launchedToolCallIds } = holdExecutions()
+      let issuedLate = false
+      setIsExecuting.mockImplementation((workflowId: string, running: boolean) => {
+        if (workflowId === 'wf-1' && !running && !issuedLate) {
+          issuedLate = true
+          executeRunToolOnClient('tool-late', 'run_block', { workflowId: 'wf-1', blockId: 'b' })
+        }
+      })
+      onTestFinished(() => {
+        setIsExecuting.mockReset()
+      })
+
+      executeRunToolOnClient('tool-a', 'run_block', { workflowId: 'wf-1', blockId: 'block-1' })
+      executeRunToolOnClient('tool-b', 'run_block', { workflowId: 'wf-1', blockId: 'block-2' })
+      await vi.waitFor(() => expect(launchedToolCallIds()).toEqual(['tool-a']))
+
+      finish('tool-a')
+      await vi.waitFor(() => expect(launchedToolCallIds()).toEqual(['tool-a', 'tool-b']))
+      finish('tool-b')
+      await vi.waitFor(() =>
+        expect(launchedToolCallIds()).toEqual(['tool-a', 'tool-b', 'tool-late'])
+      )
+      finish('tool-late')
+      await drain()
+    })
+
+    it('does not hold up a run of a different workflow', async () => {
+      const { finish, launchedToolCallIds } = holdExecutions()
+
+      executeRunToolOnClient('tool-a', 'run_workflow', { workflowId: 'wf-1' })
+      executeRunToolOnClient('tool-other', 'run_workflow', { workflowId: 'wf-2' })
+      await vi.waitFor(() => expect(launchedToolCallIds()).toEqual(['tool-a', 'tool-other']))
+
+      finish('tool-a')
+      finish('tool-other')
+      await drain('wf-1')
+      await drain('wf-2')
+    })
+
+    it('cancels a stopped call that is still waiting without ever launching it', async () => {
+      const { finish, launchedToolCallIds } = holdExecutions()
+
+      executeRunToolOnClient('tool-a', 'run_block', { workflowId: 'wf-1', blockId: 'block-1' })
+      executeRunToolOnClient('tool-b', 'run_block', { workflowId: 'wf-1', blockId: 'block-2' })
+      executeRunToolOnClient('tool-other-chat', 'run_block', {
+        workflowId: 'wf-1',
+        blockId: 'block-3',
+      })
+      await vi.waitFor(() => expect(launchedToolCallIds()).toEqual(['tool-a']))
+
+      stopRunToolExecutions(new Set(['tool-a', 'tool-b']))
+
+      await vi.waitFor(() => expect(launchedToolCallIds()).toEqual(['tool-a', 'tool-other-chat']))
+      finish('tool-other-chat')
+      await drain()
+
+      const statusByToolCall = new Map(confirmStatuses())
+      expect(statusByToolCall.get('tool-a')).toBe('cancelled')
+      expect(statusByToolCall.get('tool-b')).toBe('cancelled')
+      expect(statusByToolCall.get('tool-other-chat')).toBe('success')
+    })
+
+    it('keeps the workflow for a run whose stream dropped until that run settles on the server', async () => {
+      const { finish, launchedToolCallIds } = holdExecutions()
+      const { released, setServerStatus } = interruptFirstRun('exec-interrupted')
+
+      executeRunToolOnClient('tool-a', 'run_block', { workflowId: 'wf-1', blockId: 'block-1' })
+      executeRunToolOnClient('tool-b', 'run_block', { workflowId: 'wf-1', blockId: 'block-2' })
+      await vi.waitFor(() => expect(released).toHaveBeenCalledWith('wf-1'))
+      await vi.waitFor(() => expect(statusFetches()).toBeGreaterThan(2))
+
+      expect(launchedToolCallIds()).toEqual(['tool-a'])
+      expect(saveExecutionPointer).toHaveBeenCalledTimes(1)
+      expect(isRunToolActiveForWorkflow('wf-1')).toBe(false)
+
+      setServerStatus('completed')
+      await vi.waitFor(() => expect(launchedToolCallIds()).toEqual(['tool-a', 'tool-b']))
+      finish('tool-b')
+      await drain()
+      expect(confirmStatuses()).toEqual([
+        ['tool-a', 'background'],
+        ['tool-b', 'success'],
+      ])
+    })
+
+    it('checks a dropped run only while a call waits on it and no reconnect is following it', async () => {
+      const { finish, launchedToolCallIds } = holdExecutions()
+      const { released, setServerStatus } = interruptFirstRun('exec-watched')
+      const reconnect = reconnectStream(true)
+
+      executeRunToolOnClient('tool-a', 'run_block', { workflowId: 'wf-1', blockId: 'block-1' })
+      await vi.waitFor(() => expect(released).toHaveBeenCalledWith('wf-1'))
+      await sleep(20)
+      expect(isReconnectStreamOpen).not.toHaveBeenCalled()
+
+      executeRunToolOnClient('tool-b', 'run_block', { workflowId: 'wf-1', blockId: 'block-2' })
+      await vi.waitFor(() => expect(isReconnectStreamOpen.mock.calls.length).toBeGreaterThan(2))
+      expect(statusFetches()).toBe(0)
+      expect(launchedToolCallIds()).toEqual(['tool-a'])
+
+      setServerStatus('completed')
+      reconnect.close()
+      await vi.waitFor(() => expect(launchedToolCallIds()).toEqual(['tool-a', 'tool-b']))
+      expect(statusFetches()).toBe(1)
+      finish('tool-b')
+      await drain()
+    })
+
+    it('lets waiting calls run once the dropped run settles, even if its reconnect was abandoned', async () => {
+      const { finish, launchedToolCallIds } = holdExecutions()
+      const { released, setServerStatus } = interruptFirstRun('exec-abandoned')
+      const reconnect = reconnectStream(true)
+      const running = new Map<string, boolean>()
+      setIsExecuting.mockImplementation((workflowId: string, value: boolean) => {
+        running.set(workflowId, value)
+      })
+      getWorkflowExecution.mockImplementation((workflowId: string) => ({
+        isExecuting: running.get(workflowId) ?? false,
+      }))
+      onTestFinished(() => {
+        setIsExecuting.mockReset()
+        getWorkflowExecution.mockReset()
+        getWorkflowExecution.mockImplementation(() => ({ isExecuting: false }))
+      })
+
+      loadExecutionPointer.mockResolvedValue({
+        workflowId: 'wf-1',
+        executionId: 'exec-abandoned',
+        lastEventId: 3,
+      })
+
+      executeRunToolOnClient('tool-a', 'run_block', { workflowId: 'wf-1', blockId: 'block-1' })
+      executeRunToolOnClient('tool-b', 'run_block', { workflowId: 'wf-1', blockId: 'block-2' })
+      await vi.waitFor(() => expect(released).toHaveBeenCalledWith('wf-1'))
+      // The editor's reconnect re-attaches, then the user navigates away and it is
+      // cancelled, leaving the execution marked current and running in the store.
+      setCurrentExecutionId('wf-1', 'exec-abandoned')
+      setIsExecuting('wf-1', true)
+      setServerStatus('completed')
+      await vi.waitFor(() =>
+        expect(isReconnectStreamOpen).toHaveBeenCalledWith('wf-1', 'exec-abandoned')
+      )
+      expect(launchedToolCallIds()).toEqual(['tool-a'])
+
+      reconnect.close()
+      await vi.waitFor(() => expect(launchedToolCallIds()).toEqual(['tool-a', 'tool-b']))
+      // The abandoned run's pointer is gone before the next run saves its own.
+      expect(clearExecutionPointer).toHaveBeenCalledWith('wf-1')
+      expect(clearExecutionPointer.mock.invocationCallOrder[0]).toBeLessThan(
+        saveExecutionPointer.mock.invocationCallOrder.at(-1) ?? 0
+      )
+      finish('tool-b')
+      await drain()
+      expect(confirmStatuses()).toEqual([
+        ['tool-a', 'background'],
+        ['tool-b', 'success'],
+      ])
+    })
+
+    it('re-sends an undelivered completion while a later call owns the workflow', async () => {
+      const { finish, launchedToolCallIds } = holdExecutions()
+      let confirmsFailFor: string | null = 'tool-a'
+      vi.stubGlobal(
+        'fetch',
+        vi.fn(async (_url: string, init?: RequestInit) =>
+          confirmsFailFor && String(init?.body).includes(`"toolCallId":"${confirmsFailFor}"`)
+            ? { ok: false, status: 500 }
+            : { ok: true }
+        )
+      )
+
+      executeRunToolOnClient('tool-a', 'run_block', { workflowId: 'wf-1', blockId: 'block-1' })
+      executeRunToolOnClient('tool-b', 'run_block', { workflowId: 'wf-1', blockId: 'block-2' })
+      await vi.waitFor(() => expect(launchedToolCallIds()).toEqual(['tool-a']))
+      const firstExecutionId = await launchedExecutionId()
+      finish('tool-a')
+      await vi.waitFor(() => expect(launchedToolCallIds()).toEqual(['tool-a', 'tool-b']))
+
+      confirmsFailFor = null
+      await expect(bindRunToolToExecution('tool-a', 'wf-1')).resolves.toBe(true)
+
+      const delivered = confirmBodies()
+        .filter((body) => body.toolCallId === 'tool-a')
+        .at(-1)
+      expect(delivered).toMatchObject({ status: 'success', executionId: firstExecutionId })
+      finish('tool-b')
+      await drain()
+    })
+
+    it('treats a waiting call as already owned by this tab when the chat recovers it', async () => {
+      const { finish, launchedToolCallIds } = holdExecutions()
+
+      executeRunToolOnClient('tool-a', 'run_block', { workflowId: 'wf-1', blockId: 'block-1' })
+      executeRunToolOnClient('tool-b', 'run_block', { workflowId: 'wf-1', blockId: 'block-2' })
+      await vi.waitFor(() => expect(launchedToolCallIds()).toEqual(['tool-a']))
+
+      await expect(bindRunToolToExecution('tool-b', 'wf-1')).resolves.toBe(true)
+
+      finish('tool-a')
+      await vi.waitFor(() => expect(launchedToolCallIds()).toEqual(['tool-a', 'tool-b']))
+      finish('tool-b')
+      await drain()
+      expect(confirmStatuses()).toEqual([
+        ['tool-a', 'success'],
+        ['tool-b', 'success'],
+      ])
+    })
+  })
 })
diff --git a/apps/sim/lib/mothership/tools/client/run-tool-execution.ts b/apps/sim/lib/mothership/tools/client/run-tool-execution.ts
index bf459388730..5b29355dc99 100644
--- a/apps/sim/lib/mothership/tools/client/run-tool-execution.ts
+++ b/apps/sim/lib/mothership/tools/client/run-tool-execution.ts
@@ -1,9 +1,15 @@
 import { createLogger } from '@sim/logger'
 import { toError } from '@sim/utils/errors'
+import { sleep } from '@sim/utils/helpers'
 import { generateId } from '@sim/utils/id'
 import { isPlainRecord } from '@sim/utils/object'
+import { backoffWithJitter } from '@sim/utils/retry'
+import { ApiClientError } from '@/lib/api/client/errors'
 import { requestJson } from '@/lib/api/client/request'
-import { cancelWorkflowExecutionContract } from '@/lib/api/contracts/workflows'
+import {
+  cancelWorkflowExecutionContract,
+  getWorkflowExecutionContract,
+} from '@/lib/api/contracts/workflows'
 import {
   ASYNC_TOOL_CONFIRMATION_STATUS,
   type AsyncConfirmationStatus,
@@ -32,6 +38,7 @@ import {
 import { executeWorkflowWithFullLogging } from '@/app/workspace/[workspaceId]/w/[workflowId]/utils/workflow-execution-utils'
 import {
   isExecutionStreamHttpError,
+  isReconnectStreamOpen,
   SSEEventHandlerError,
   SSEStreamInterruptedError,
 } from '@/hooks/use-execution-stream'
@@ -51,7 +58,29 @@ interface ActiveRunTool {
   execution?: { id: string; controller: AbortController }
 }
 
-const activeRunToolByWorkflowId = new Map<string, ActiveRunTool>()
+/** A run tool queued behind the workflow's owner; admitted with null when stopped first. */
+interface WaitingRunTool {
+  toolCallId: string
+  admit: (run: ActiveRunTool | null) => void
+}
+
+/**
+ * Who holds a workflow's single visible run in this tab: a run tool, or an execution
+ * whose run tool lost its stream while the server kept running it.
+ */
+type RunSlotOwner =
+  | { kind: 'runTool'; run: ActiveRunTool }
+  | { kind: 'interrupted'; executionId: string }
+
+interface WorkflowRunSlot {
+  owner: RunSlotOwner | null
+  waiters: WaitingRunTool[]
+  watchingInterrupted: boolean
+}
+
+const runSlotsByWorkflowId = new Map<string, WorkflowRunSlot>()
+const INTERRUPTED_RUN_POLL_MAX_MS = 15_000
+const INTERRUPTED_RUN_MAX_STATUS_FAILURES = 8
 const manuallyStoppedToolCallIds = new Set<string>()
 type RunToolReleaseListener = (workflowId: string) => void
 const runToolReleaseListeners = new Set<RunToolReleaseListener>()
@@ -73,6 +102,194 @@ interface PendingCompletionReport {
   clearExecutionPointerAfterReport?: boolean
 }
 
+function slotFor(workflowId: string): WorkflowRunSlot {
+  let slot = runSlotsByWorkflowId.get(workflowId)
+  if (!slot) {
+    slot = { owner: null, waiters: [], watchingInterrupted: false }
+    runSlotsByWorkflowId.set(workflowId, slot)
+  }
+  return slot
+}
+
+function pruneSlot(workflowId: string, slot: WorkflowRunSlot): void {
+  if (!slot.owner && slot.waiters.length === 0) runSlotsByWorkflowId.delete(workflowId)
+}
+
+/** Whether a new run tool for this workflow has to queue. */
+function isSlotHeld(workflowId: string): boolean {
+  return runSlotsByWorkflowId.has(workflowId)
+}
+
+function runToolOwner(workflowId: string): ActiveRunTool | undefined {
+  const owner = runSlotsByWorkflowId.get(workflowId)?.owner
+  return owner?.kind === 'runTool' ? owner.run : undefined
+}
+
+function* runToolOwners(): Generator<[string, ActiveRunTool]> {
+  for (const [workflowId, slot] of runSlotsByWorkflowId) {
+    if (slot.owner?.kind === 'runTool') yield [workflowId, slot.owner.run]
+  }
+}
+
+/**
+ * Takes the workflow's run slot, or waits in arrival order behind its owner.
+ * Resolves null when the call is stopped while waiting.
+ */
+function acquireRunToolSlot(workflowId: string, toolCallId: string): Promise<ActiveRunTool | null> {
+  if (!isSlotHeld(workflowId)) {
+    const run: ActiveRunTool = { toolCallId }
+    slotFor(workflowId).owner = { kind: 'runTool', run }
+    return Promise.resolve(run)
+  }
+  const slot = slotFor(workflowId)
+  logger.info("[RunTool] Waiting for this workflow's current run", {
+    toolCallId,
+    workflowId,
+    owner: slot.owner?.kind === 'runTool' ? slot.owner.run.toolCallId : slot.owner?.kind,
+  })
+  return new Promise((admit) => {
+    slot.waiters.push({ toolCallId, admit })
+    if (slot.owner?.kind === 'interrupted') watchInterruptedExecution(workflowId, slot)
+  })
+}
+
+/** Runs synchronously with the release that freed the slot, so a later call cannot jump ahead. */
+function admitNextRunTool(workflowId: string, slot: WorkflowRunSlot): void {
+  if (!slot.owner) {
+    const next = slot.waiters.shift()
+    if (next) {
+      const run: ActiveRunTool = { toolCallId: next.toolCallId }
+      slot.owner = { kind: 'runTool', run }
+      next.admit(run)
+    }
+  }
+  pruneSlot(workflowId, slot)
+}
+
+/**
+ * Gives up a run tool's slot. A run whose stream dropped while the server kept
+ * executing it hands the slot to that execution instead, until it settles.
+ */
+function releaseRunToolSlot(
+  workflowId: string,
+  run: ActiveRunTool,
+  interruptedExecutionId?: string
+): void {
+  const slot = runSlotsByWorkflowId.get(workflowId)
+  if (!slot || runToolOwner(workflowId) !== run) return
+  slot.owner = interruptedExecutionId
+    ? { kind: 'interrupted', executionId: interruptedExecutionId }
+    : null
+  admitNextRunTool(workflowId, slot)
+  if (interruptedExecutionId && slot.waiters.length > 0) {
+    watchInterruptedExecution(workflowId, slot)
+  }
+}
+
+async function releaseInterruptedHold(
+  workflowId: string,
+  slot: WorkflowRunSlot,
+  executionId: string
+): Promise<void> {
+  if (slot.owner?.kind !== 'interrupted' || slot.owner.executionId !== executionId) return
+  // A reconnect cancelled by navigating away leaves the settled run marked current and its
+  // pointer saved; both go before the next run can write its own.
+  clearVisibleExecution(workflowId, executionId)
+  const pointer = await loadExecutionPointer(workflowId).catch(() => null)
+  if (pointer?.executionId === executionId) await clearExecutionPointer(workflowId)
+  slot.owner = null
+  admitNextRunTool(workflowId, slot)
+}
+
+/**
+ * Polls an interrupted execution while calls wait behind it, releasing the hold once
+ * the server has settled it and no reconnect stream is still draining it.
+ */
+function watchInterruptedExecution(workflowId: string, slot: WorkflowRunSlot): void {
+  if (slot.watchingInterrupted || slot.owner?.kind !== 'interrupted') return
+  slot.watchingInterrupted = true
+  const { executionId } = slot.owner
+  const isHeldByThisExecution = () =>
+    slot.owner?.kind === 'interrupted' && slot.owner.executionId === executionId
+  void (async () => {
+    let serverSettled = false
+    let statusFailures = 0
+    for (let attempt = 1; ; attempt++) {
+      await sleep(backoffWithJitter(attempt, null, { maxMs: INTERRUPTED_RUN_POLL_MAX_MS }))
+      if (!isHeldByThisExecution() || slot.waiters.length === 0) break
+      if (isDocumentHidden() || isReconnectStreamOpen(workflowId, executionId)) continue
+      if (!serverSettled) {
+        const status = await readExecutionStatus(workflowId, executionId)
+        if (status === 'running') continue
+        if (status === 'unknown' && ++statusFailures < INTERRUPTED_RUN_MAX_STATUS_FAILURES) continue
+        serverSettled = true
+        if (isReconnectStreamOpen(workflowId, executionId)) continue
+      }
+      await releaseInterruptedHold(workflowId, slot, executionId)
+      break
+    }
+    slot.watchingInterrupted = false
+  })()
+}
+
+function isDocumentHidden(): boolean {
+  return typeof document !== 'undefined' && document.hidden
+}
+
+/** 'settled' also covers an execution that is gone or not ours (403/404). */
+async function readExecutionStatus(
+  workflowId: string,
+  executionId: string
+): Promise<'running' | 'settled' | 'unknown'> {
+  try {
+    const { status } = await requestJson(getWorkflowExecutionContract, {
+      params: { id: workflowId, executionId },
+      query: {},
+    })
+    return status === 'queued' || status === 'pending' || status === 'running'
+      ? 'running'
+      : 'settled'
+  } catch (error) {
+    if (error instanceof ApiClientError && (error.status === 403 || error.status === 404)) {
+      return 'settled'
+    }
+    logger.warn("[RunTool] Could not read an interrupted execution's status", {
+      workflowId,
+      executionId,
+      error: toError(error).message,
+    })
+    return 'unknown'
+  }
+}
+
+function isRunToolWaiting(workflowId: string, toolCallId: string): boolean {
+  return (
+    runSlotsByWorkflowId.get(workflowId)?.waiters.some((w) => w.toolCallId === toolCallId) ?? false
+  )
+}
+
+function dropWaitingRunTools(toolCallIds: ReadonlySet<string>): void {
+  for (const [workflowId, slot] of runSlotsByWorkflowId) {
+    const kept: WaitingRunTool[] = []
+    for (const waiter of slot.waiters) {
+      if (toolCallIds.has(waiter.toolCallId)) waiter.admit(null)
+      else kept.push(waiter)
+    }
+    slot.waiters = kept
+    pruneSlot(workflowId, slot)
+  }
+}
+
+/** Clears the editor's visible run when it still shows `executionId`; returns whether it did. */
+function clearVisibleExecution(workflowId: string, executionId: string): boolean {
+  const state = useExecutionStore.getState()
+  if (state.getCurrentExecutionId(workflowId) !== executionId) return false
+  state.setCurrentExecutionId(workflowId, null)
+  state.setIsExecuting(workflowId, false)
+  state.setActiveBlocks(workflowId, new Set())
+  return true
+}
+
 function resolveWorkflowInput(params: Record<string, unknown>): unknown {
   if (Object.hasOwn(params, 'workflow_input')) {
     return params.workflow_input
@@ -275,24 +492,20 @@ export async function bindRunToolToExecution(
   toolCallId: string,
   workflowId: string
 ): Promise<boolean> {
-  const existingToolCallId = activeRunToolByWorkflowId.get(workflowId)?.toolCallId
-  if (existingToolCallId === toolCallId) {
+  const owner = runSlotsByWorkflowId.get(workflowId)?.owner
+  const ownerToolCallId = owner?.kind === 'runTool' ? owner.run.toolCallId : undefined
+  if (ownerToolCallId === toolCallId || isRunToolWaiting(workflowId, toolCallId)) {
     logger.info('[RunTool] Recovery skipped: run tool is already active in this tab', {
       workflowId,
       toolCallId,
     })
     return true
   }
-  if (existingToolCallId && existingToolCallId !== toolCallId) {
-    logger.warn('[RunTool] Recovery skipped: another run tool is already active', {
-      workflowId,
-      toolCallId,
-      existingToolCallId,
-    })
-    return false
-  }
-
-  const pointer = await loadExecutionPointer(workflowId).catch(() => null)
+  // Another run's pointer says nothing about this call.
+  const otherRunOwnsWorkflow = owner != null
+  const pointer = otherRunOwnsWorkflow
+    ? null
+    : await loadExecutionPointer(workflowId).catch(() => null)
   const pendingCompletion = loadPendingCompletionReport(toolCallId)
   if (pendingCompletion) {
     const executionId = pendingCompletion.executionId ?? pointer?.executionId
@@ -313,7 +526,7 @@ export async function bindRunToolToExecution(
         executionId
       )
       clearPendingCompletionReport(toolCallId)
-      if (pendingCompletion.clearExecutionPointerAfterReport) {
+      if (pendingCompletion.clearExecutionPointerAfterReport && !otherRunOwnsWorkflow) {
         await clearExecutionPointer(workflowId)
       }
     } catch (error) {
@@ -327,6 +540,15 @@ export async function bindRunToolToExecution(
     return true
   }
 
+  if (otherRunOwnsWorkflow) {
+    logger.warn('[RunTool] Recovery skipped: another run owns this workflow', {
+      workflowId,
+      toolCallId,
+      owner: ownerToolCallId ?? owner.kind,
+    })
+    return false
+  }
+
   if (!pointer?.executionId) {
     logger.info('[RunTool] Recovery skipped: no tab-local execution pointer', {
       workflowId,
@@ -386,8 +608,8 @@ export function executeRunToolOnClient(
 }
 
 export function isRunToolActiveForId(toolCallId: string): boolean {
-  for (const active of activeRunToolByWorkflowId.values()) {
-    if (active.toolCallId === toolCallId) return true
+  for (const [, run] of runToolOwners()) {
+    if (run.toolCallId === toolCallId) return true
   }
   return false
 }
@@ -401,7 +623,7 @@ export function isRunToolActiveForId(toolCallId: string): boolean {
  * acknowledged the run.
  */
 export function isRunToolActiveForWorkflow(workflowId: string): boolean {
-  return activeRunToolByWorkflowId.has(workflowId)
+  return runToolOwner(workflowId) !== undefined
 }
 
 /**
@@ -426,7 +648,7 @@ function notifyRunToolReleased(workflowId: string): void {
 
 /** The resource's Stop button refers to its displayed execution, which may be a manual run. */
 export function stopRunToolForExecution(workflowId: string, executionId: string | null): boolean {
-  const active = activeRunToolByWorkflowId.get(workflowId)
+  const active = runToolOwner(workflowId)
   if (!executionId || active?.execution?.id !== executionId) return false
   stopRunToolExecutions(new Set([active.toolCallId]))
   return true
@@ -434,7 +656,8 @@ export function stopRunToolForExecution(workflowId: string, executionId: string
 
 /** Cancels exact tool-owned executions; a workflow's current UI pointer is not ownership. */
 export function stopRunToolExecutions(toolCallIds: ReadonlySet<string>): void {
-  for (const [workflowId, active] of activeRunToolByWorkflowId) {
+  dropWaitingRunTools(toolCallIds)
+  for (const [workflowId, active] of runToolOwners()) {
     const { toolCallId, execution } = active
     if (!toolCallIds.has(toolCallId) || !execution || manuallyStoppedToolCallIds.has(toolCallId)) {
       continue
@@ -470,13 +693,7 @@ export function stopRunToolExecutions(toolCallIds: ReadonlySet<string>): void {
       blockName: 'Run Cancelled',
       blockType: 'cancelled',
     })
-    const state = useExecutionStore.getState()
-    if (state.getCurrentExecutionId(workflowId) === execution.id) {
-      clearExecutionPointer(workflowId)
-      state.setCurrentExecutionId(workflowId, null)
-      state.setIsExecuting(workflowId, false)
-      state.setActiveBlocks(workflowId, new Set())
-    }
+    if (clearVisibleExecution(workflowId, execution.id)) clearExecutionPointer(workflowId)
   }
 }
 
@@ -501,32 +718,28 @@ async function doExecuteRunTool(
     return
   }
 
-  const existingToolCallId = activeRunToolByWorkflowId.get(targetWorkflowId)
-  if (existingToolCallId) {
-    logger.warn('[RunTool] Execution prevented: another run tool is already active', {
-      toolCallId,
-      toolName,
-      existingToolCallId: existingToolCallId.toolCallId,
-    })
+  const activeRun = await acquireRunToolSlot(targetWorkflowId, toolCallId)
+  if (!activeRun) {
+    logger.info('[RunTool] Stopped before its turn to run', { toolCallId, toolName })
     await reportCompletion(
       toolCallId,
-      MothershipStreamV1ToolOutcome.error,
-      WORKFLOW_EXECUTION_BUSY.message,
-      { code: WORKFLOW_EXECUTION_BUSY.code }
+      MothershipStreamV1ToolOutcome.cancelled,
+      getWorkflowToolCompletionMessage(MothershipStreamV1ToolOutcome.cancelled),
+      { reason: 'user_cancelled', cancelledByUser: true }
     )
     return
   }
 
   setActiveWorkflow(targetWorkflowId)
-  const activeRun: ActiveRunTool = { toolCallId }
-  activeRunToolByWorkflowId.set(targetWorkflowId, activeRun)
 
   const { getWorkflowExecution, setIsExecuting } = useExecutionStore.getState()
   const { isExecuting } = getWorkflowExecution(targetWorkflowId)
 
+  // The previous run tool has already released the workflow, so a run here is
+  // one no run tool drives: the user's, or a dropped stream being re-attached.
   if (isExecuting) {
     logger.warn('[RunTool] Execution prevented: already executing', { toolCallId, toolName })
-    activeRunToolByWorkflowId.delete(targetWorkflowId)
+    releaseRunToolSlot(targetWorkflowId, activeRun)
     await reportCompletion(
       toolCallId,
       MothershipStreamV1ToolOutcome.error,
@@ -551,9 +764,7 @@ async function doExecuteRunTool(
         triggerBlockId
       )
     } finally {
-      if (activeRunToolByWorkflowId.get(targetWorkflowId) === activeRun) {
-        activeRunToolByWorkflowId.delete(targetWorkflowId)
-      }
+      releaseRunToolSlot(targetWorkflowId, activeRun)
     }
     return
   }
@@ -602,15 +813,11 @@ async function doExecuteRunTool(
   setCurrentExecutionId(targetWorkflowId, executionId)
   saveExecutionPointer({ workflowId: targetWorkflowId, executionId, lastEventId: 0 })
   const releaseVisibleExecutionForBackground = () => {
-    const { setCurrentExecutionId: clearExecId, setActiveBlocks } = useExecutionStore.getState()
     if (
-      activeRunToolByWorkflowId.get(targetWorkflowId) === activeRun &&
-      useExecutionStore.getState().getCurrentExecutionId(targetWorkflowId) === executionId
+      runToolOwner(targetWorkflowId) === activeRun &&
+      clearVisibleExecution(targetWorkflowId, executionId)
     ) {
-      clearExecId(targetWorkflowId, null)
       consolePersistence.executionEnded(persistenceExecution)
-      setIsExecuting(targetWorkflowId, false)
-      setActiveBlocks(targetWorkflowId, new Set())
     }
   }
 
@@ -648,7 +855,7 @@ async function doExecuteRunTool(
   })
 
   let leaveExecutionRecoverable = false
-  let streamInterrupted = false
+  let interruptedExecutionId: string | undefined
 
   try {
     const result = await executeWorkflowWithFullLogging({
@@ -729,7 +936,7 @@ async function doExecuteRunTool(
       const msg = toError(err).message
       if (err instanceof SSEEventHandlerError || err instanceof SSEStreamInterruptedError) {
         leaveExecutionRecoverable = true
-        streamInterrupted = true
+        interruptedExecutionId = err.executionId ?? executionId
         logger.warn(
           '[RunTool] Execution stream interrupted; leaving workflow execution in background',
           {
@@ -782,24 +989,16 @@ async function doExecuteRunTool(
       window.removeEventListener('pagehide', onPageHide)
     }
     manuallyStoppedToolCallIds.delete(toolCallId)
-    const ownsRegistration = activeRunToolByWorkflowId.get(targetWorkflowId) === activeRun
-    if (ownsRegistration) {
-      activeRunToolByWorkflowId.delete(targetWorkflowId)
-    }
+    const ownsRegistration = runToolOwner(targetWorkflowId) === activeRun
     consolePersistence.executionEnded(persistenceExecution)
-    const { setCurrentExecutionId: clearExecId, setActiveBlocks } = useExecutionStore.getState()
     if (
-      !leaveExecutionRecoverable &&
       ownsRegistration &&
-      useExecutionStore.getState().getCurrentExecutionId(targetWorkflowId) === executionId
+      !leaveExecutionRecoverable &&
+      clearVisibleExecution(targetWorkflowId, executionId)
     ) {
-      clearExecId(targetWorkflowId, null)
       clearExecutionPointer(targetWorkflowId)
-      setIsExecuting(targetWorkflowId, false)
-      setActiveBlocks(targetWorkflowId, new Set())
-    }
-    if (streamInterrupted && ownsRegistration) {
-      notifyRunToolReleased(targetWorkflowId)
     }
+    releaseRunToolSlot(targetWorkflowId, activeRun, interruptedExecutionId)
+    if (ownsRegistration && interruptedExecutionId) notifyRunToolReleased(targetWorkflowId)
   }
 }