fix(execution): classify SDK run aborts and refuse doomed provider and media requests before sending

apps/sim/app/api/resume/poll/route.ts

@@ -8,6 +8,7 @@ import { and, asc, inArray, isNotNull, lte, sql } from 'drizzle-orm'
 import { type NextRequest, NextResponse } from 'next/server'
 import { verifyCronAuth } from '@/lib/auth/internal'
 import { acquireLock, releaseLock } from '@/lib/core/config/redis'
+import { logFailureOnce } from '@/lib/core/errors/failure-log'
 import { isRetryableInfrastructureError } from '@/lib/core/errors/retryable-infrastructure'
 import { mapWithConcurrency } from '@/lib/core/utils/concurrency'
 import { withRouteHandler } from '@/lib/core/utils/with-route-handler'
@@ -518,10 +519,13 @@ async function dispatchRow(row: PreparedDueRow, now: Date): Promise<RowResult> {
           workflowId: row.workflowId,
           parentExecutionId: row.executionId,
         }).catch((error) => {
-          logger.error('Background time-pause resume failed', {
+          logFailureOnce(logger, 'Background time-pause resume failed', error, {
             executionId: row.executionId,
-            contextId: point.contextId,
-            error: toError(error).message,
+            metadata: {
+              resumeExecutionId: enqueueResult.resumeExecutionId,
+              contextId: point.contextId,
+              error: toError(error).message,
+            },
           })
         })
       }

apps/sim/background/resume-execution.test.ts

@@ -1,3 +1,4 @@
+import { createLogger } from '@sim/logger'
 import { loggerMock } from '@sim/testing'
 import { billingAttributionMock } from '@sim/testing/mocks/billing-attribution.mock'
 import {
@@ -28,6 +29,7 @@ vi.mock('@/executor/execution/snapshot', () => ({
   ExecutionSnapshot: { fromJSON: mockSnapshotFromJson },
 }))
 
+import { logFailureOnce } from '@/lib/core/errors/failure-log'
 import { executeResumeJob, type ResumeExecutionPayload } from '@/background/resume-execution'
 
 const { mockFindCellContextByExecutionId } = tableWorkflowColumnsMockFns
@@ -94,16 +96,31 @@ describe('executeResumeJob terminal errors', () => {
 
     await expect(executeResumeJob(payload)).rejects.toBe(rawError)
 
-    expect(resumeExecutionLogger.error).toHaveBeenCalledWith('Background resume execution failed', {
-      errorType: 'error',
-      hasStack: true,
-    })
+    expect(resumeExecutionLogger.error).toHaveBeenCalledWith(
+      'Background resume execution failed',
+      expect.objectContaining({ errorType: 'error', hasStack: true, failureKind: 'internal' })
+    )
     const loggerPayload = JSON.stringify(resumeExecutionLogger.error.mock.calls)
     expect(loggerPayload).not.toContain(secret)
     expect(loggerPayload).not.toContain('__var_')
     expect(rawError.message).toContain(secret)
   })
 
+  it('does not log again a failure the resumed run already logged', async () => {
+    const rawError = new Error('resume blew up')
+    mockStartResumeExecution.mockImplementation(async () => {
+      logFailureOnce(createLogger('ResumeBoundary'), 'Resume execution failed', rawError, {
+        executionId: payload.parentExecutionId,
+      })
+      throw rawError
+    })
+    resumeExecutionLogger.error.mockClear()
+
+    await expect(executeResumeJob(payload)).rejects.toBe(rawError)
+
+    expect(resumeExecutionLogger.error).not.toHaveBeenCalled()
+  })
+
   it('starts a legacy attempt deadline before deserializing the full snapshot', async () => {
     mockStartResumeExecution.mockResolvedValue({
       success: true,

apps/sim/background/resume-execution.ts

@@ -7,6 +7,7 @@ import {
   type BillingAttributionSnapshot,
   billingAttributionsEqual,
 } from '@/lib/billing/core/attribution'
+import { logFailureOnce } from '@/lib/core/errors/failure-log'
 import {
   capExecutionTimeoutMs,
   createTimeoutAbortController,
@@ -248,13 +249,11 @@ export async function executeResumeJob(payload: ResumeExecutionPayload, signal?:
       executedAt: new Date().toISOString(),
     }
   } catch (error) {
-    logger.error(
-      'Background resume execution failed',
-      projectResolvedSecretDiagnosticError(error, undefined, {
-        resumeExecutionId,
-        workflowId,
-      })
-    )
+    logFailureOnce(logger, 'Background resume execution failed', error, {
+      executionId: parentExecutionId,
+      metadata: () =>
+        projectResolvedSecretDiagnosticError(error, undefined, { resumeExecutionId, workflowId }),
+    })
     throw error
   } finally {
     timeoutController?.cleanup()

apps/sim/executor/execution/block-executor.test.ts

@@ -1,10 +1,13 @@
+import Anthropic from '@anthropic-ai/sdk'
 import { createLogger } from '@sim/logger'
 import { loggerMock } from '@sim/testing'
 import { maskClientMock, maskClientMockFns } from '@sim/testing/mocks/mask-client.mock'
 import { permissionCheckMock } from '@sim/testing/mocks/permission-check.mock'
 import { storageServiceMockFns } from '@sim/testing/mocks/storage-service.mock'
 import { uploadsMock } from '@sim/testing/mocks/uploads.mock'
 import { DrizzleQueryError } from 'drizzle-orm/errors'
+import Groq from 'groq-sdk'
+import OpenAI from 'openai'
 import { beforeEach, describe, expect, it, vi } from 'vitest'
 import { classifyFailure, logFailureOnce } from '@/lib/core/errors/failure-log'
 import { clearLargeValueCacheForTests } from '@/lib/execution/payloads/cache'
@@ -1153,6 +1156,115 @@ describe('BlockExecutor streaming pump', () => {
     expect(rawError.message).toContain(secret)
   })
 
+  /**
+   * A fetch that never answers and rejects only when its request is cancelled, so each SDK
+   * raises its own abort error through its real cancellation path.
+   */
+  const hangingFetch = (_input: unknown, init?: RequestInit) =>
+    new Promise<Response>((_resolve, reject) => {
+      init?.signal?.addEventListener('abort', () => reject(init.signal?.reason), { once: true })
+    })
+
+  const vendorRequests = {
+    anthropic: (signal: AbortSignal) =>
+      new Anthropic({ apiKey: 'test', fetch: hangingFetch, maxRetries: 0 }).messages.create(
+        { model: 'claude-test', max_tokens: 1, messages: [{ role: 'user', content: 'hi' }] },
+        { signal }
+      ),
+    openai: (signal: AbortSignal) =>
+      new OpenAI({ apiKey: 'test', fetch: hangingFetch, maxRetries: 0 }).chat.completions.create(
+        { model: 'gpt-test', messages: [{ role: 'user', content: 'hi' }] },
+        { signal }
+      ),
+    groq: (signal: AbortSignal) =>
+      new Groq({ apiKey: 'test', fetch: hangingFetch, maxRetries: 0 }).chat.completions.create(
+        { model: 'groq-test', messages: [{ role: 'user', content: 'hi' }] },
+        { signal }
+      ),
+  }
+
+  /** An Agent handler that awaits a real SDK request, calling `abort` once it is in flight. */
+  function vendorAgentHandler(
+    vendor: keyof typeof vendorRequests,
+    signalFor: (ctx: ExecutionContext) => AbortSignal | undefined,
+    abort: () => void
+  ): BlockHandler {
+    return {
+      canHandle: () => true,
+      execute: async (blockContext) => {
+        const signal = signalFor(blockContext)
+        if (!signal) throw new Error('test handler needs a signal')
+        const request = vendorRequests[vendor](signal)
+        abort()
+        return request
+      },
+    }
+  }
+
+  it.each(Object.keys(vendorRequests) as (keyof typeof vendorRequests)[])(
+    'soft-completes an Agent block the user stopped mid-%s-request',
+    async (vendor) => {
+      const run = new AbortController()
+      const { executor, block, state } = createExecutor(
+        vendorAgentHandler(
+          vendor,
+          (blockContext) => blockContext.abortSignal,
+          () => run.abort(new DOMException('user', 'AbortError'))
+        )
+      )
+      const ctx = createContext(state)
+      ctx.abortSignal = run.signal
+
+      await executor.execute(ctx, createNode(block), block)
+
+      expect(state.getBlockOutput(block.id)).toEqual({ content: '' })
+      expect(ctx.blockLogs[0]).toMatchObject({ success: true })
+    }
+  )
+
+  it.each(Object.keys(vendorRequests) as (keyof typeof vendorRequests)[])(
+    'fails an Agent block the run time limit cut off mid-%s-request, attributed to the user',
+    async (vendor) => {
+      const run = new AbortController()
+      const { executor, block, state } = createExecutor(
+        vendorAgentHandler(
+          vendor,
+          (blockContext) => blockContext.abortSignal,
+          () => run.abort(new DOMException('timeout', 'AbortError'))
+        )
+      )
+      const ctx = createContext(state)
+      ctx.abortSignal = run.signal
+
+      const thrown = await executor.execute(ctx, createNode(block), block).catch((e) => e)
+
+      expect(thrown).toBeInstanceOf(Error)
+      expect(state.getBlockOutput(block.id)?.error).toBeTruthy()
+      expect(classifyFailure(thrown)).toBe('user')
+    }
+  )
+
+  it.each(Object.keys(vendorRequests) as (keyof typeof vendorRequests)[])(
+    'fails a %s request cancelled while the run is still live as an internal fault',
+    async (vendor) => {
+      const requestOnly = new AbortController()
+      const { executor, block, state } = createExecutor(
+        vendorAgentHandler(
+          vendor,
+          () => requestOnly.signal,
+          () => requestOnly.abort()
+        )
+      )
+      const ctx = createContext(state)
+      ctx.abortSignal = new AbortController().signal
+
+      const thrown = await executor.execute(ctx, createNode(block), block).catch((e) => e)
+
+      expect(state.getBlockOutput(block.id)?.error).toBeTruthy()
+      expect(classifyFailure(thrown)).toBe('internal')
+    }
+  )
+
   it('soft-completes on user abort with drained answer text (no failed block)', async () => {
     const abortController = new AbortController()
     const handler = createAgentEventsStreamingHandler({

apps/sim/executor/execution/block-executor.ts

@@ -83,6 +83,7 @@ import {
   type VariableResolver,
 } from '@/executor/variables/resolver'
 import { createAgentStreamPump } from '@/providers/stream-pump'
+import { isRunAbortError } from '@/providers/streaming-tool-loop-shared'
 import type { SerializedBlock } from '@/serializer/types'
 import { SYSTEM_SUBBLOCK_IDS } from '@/triggers/constants'
 
@@ -673,12 +674,10 @@ export class BlockExecutor {
     // (workflow is already cancelled). Timeouts abort with reason `'timeout'`.
     // Non-agent blocks (HTTP, Function, etc.) still fail normally on AbortError
     // so logs don't show a green empty success.
-    const isAbort =
-      (error instanceof DOMException && error.name === 'AbortError') ||
-      (error instanceof Error && error.name === 'AbortError')
+    const isRunAbort = isRunAbortError(error, ctx.abortSignal)
     const isTimeout = isTimeoutAbortReason(ctx.abortSignal?.reason)
     const isAgentBlock = block.metadata?.id === BlockType.AGENT
-    if (isAbort && !isTimeout && ctx.abortSignal?.aborted && isAgentBlock) {
+    if (isRunAbort && !isTimeout && isAgentBlock) {
       const softOutput: NormalizedBlockOutput = {
         content: '',
       }
@@ -817,7 +816,7 @@ export class BlockExecutor {
     }
 
     /** A user Stop or the run's own time limit aborted this block; neither is a Sim fault. */
-    if (isAbort && ctx.abortSignal?.aborted) markFailureKind(error, 'user')
+    if (isRunAbort) markFailureKind(error, 'user')
     logFailureOnce(
       this.execLogger,
       phase === 'input_resolution' ? 'Failed to resolve block inputs' : 'Block execution failed',

apps/sim/executor/handlers/agent/handler.test.ts

@@ -2759,6 +2759,38 @@ describe('AgentBlockHandler', () => {
       expect(error.message).toContain('status=200')
     })
 
+    /**
+     * A Stop rejects the provider call with the run's own abort, which is not a provider
+     * timeout; rewriting it would hide it from the block executor's Stop handling.
+     */
+    it('passes the abort of a run the user stopped through unrewritten', async () => {
+      const inputs = { model: 'gemini-2.5-flash', userPrompt: 'hi', apiKey: 'test-api-key' }
+      mockGetProviderFromModel.mockReturnValue('google')
+      const run = new AbortController()
+      run.abort(new DOMException('user', 'AbortError'))
+      mockExecuteProviderRequest.mockRejectedValueOnce(run.signal.reason)
+
+      const error = await handler
+        .execute({ ...mockContext, abortSignal: run.signal }, mockBlock, inputs)
+        .catch((e) => e)
+
+      expect(error).toBe(run.signal.reason)
+    })
+
+    it('still reports a run the time limit cut off as a provider timeout', async () => {
+      const inputs = { model: 'gemini-2.5-flash', userPrompt: 'hi', apiKey: 'test-api-key' }
+      mockGetProviderFromModel.mockReturnValue('google')
+      const run = new AbortController()
+      run.abort(new DOMException('timeout', 'AbortError'))
+      mockExecuteProviderRequest.mockRejectedValueOnce(run.signal.reason)
+
+      const error = await handler
+        .execute({ ...mockContext, abortSignal: run.signal }, mockBlock, inputs)
+        .catch((e) => e)
+
+      expect(error.message).toContain('Provider request timed out')
+    })
+
     it('should not duplicate system prompt if it exists in memories', async () => {
       const inputs = {
         model: 'gpt-4o',

apps/sim/executor/handlers/agent/handler.ts

@@ -3,6 +3,7 @@ import { getErrorMessage, toError } from '@sim/utils/errors'
 import { isPlainRecord, isRecordLike, omit } from '@sim/utils/object'
 import { truncate } from '@sim/utils/string'
 import { markFailureKind } from '@/lib/core/errors/failure-log'
+import { isTimeoutAbortReason } from '@/lib/core/execution-limits/types'
 import { normalizeStringRecord, normalizeWorkflowVariables } from '@/lib/core/utils/records'
 import {
   projectModelSchemaAnnotations,
@@ -125,6 +126,7 @@ import {
   getInlineHydrationMaxBytes,
 } from '@/providers/file-attachments.server'
 import { isAutoModel, isEvaluationModel, SIM_AUTO_MODEL_ID } from '@/providers/models'
+import { isRunAbortError } from '@/providers/streaming-tool-loop-shared'
 import {
   type ProviderToolInputProvenance,
   registerProviderToolInputProvenance,
@@ -3064,7 +3066,7 @@ export class AgentBlockHandler implements BlockHandler {
         modelRuntimeRegistry
       )
       try {
-        this.handleExecutionError(error)
+        this.handleExecutionError(error, ctx.abortSignal)
       } finally {
         if (modelRuntimeRegistry) {
           ctx.resolvedSecretTraceRegistry = modelRuntimeRegistry.forkForPropagatedEntries()
@@ -3088,8 +3090,12 @@ export class AgentBlockHandler implements BlockHandler {
   /**
    * Attributes a provider failure and rewrites a transport failure into a message the author can
    * act on. The block executor owns the log line, with the block's and run's identity.
+   *
+   * A run the user stopped passes through untouched: its abort is not a provider timeout, and
+   * the block executor recognises it only in its original shape.
    */
-  private handleExecutionError(error: unknown) {
+  private handleExecutionError(error: unknown, abortSignal: AbortSignal | undefined) {
+    if (isRunAbortError(error, abortSignal) && !isTimeoutAbortReason(abortSignal?.reason)) return
     const transportFailure = error instanceof Error ? describeProviderTransportFailure(error) : null
     if (transportFailure) {
       throw markFailureKind(new Error(transportFailure), 'third_party_server')