diff --git a/apps/sim/app/api/resume/poll/route.ts b/apps/sim/app/api/resume/poll/route.ts index c43fb682c0b..d16de246b13 100644 --- a/apps/sim/app/api/resume/poll/route.ts +++ b/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 { 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, + }, }) }) } diff --git a/apps/sim/background/resume-execution.test.ts b/apps/sim/background/resume-execution.test.ts index 572046b551e..a7e2db666e1 100644 --- a/apps/sim/background/resume-execution.test.ts +++ b/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, diff --git a/apps/sim/background/resume-execution.ts b/apps/sim/background/resume-execution.ts index 73f86e52988..7b7d3e7cdf1 100644 --- a/apps/sim/background/resume-execution.ts +++ b/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() diff --git a/apps/sim/executor/execution/block-executor.test.ts b/apps/sim/executor/execution/block-executor.test.ts index fa63c98e71e..aba897999ee 100644 --- a/apps/sim/executor/execution/block-executor.test.ts +++ b/apps/sim/executor/execution/block-executor.test.ts @@ -1,3 +1,4 @@ +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' @@ -5,6 +6,8 @@ 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((_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({ diff --git a/apps/sim/executor/execution/block-executor.ts b/apps/sim/executor/execution/block-executor.ts index fdc51cbdb29..1259f1507b5 100644 --- a/apps/sim/executor/execution/block-executor.ts +++ b/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', diff --git a/apps/sim/executor/handlers/agent/handler.test.ts b/apps/sim/executor/handlers/agent/handler.test.ts index 85f342ef16e..e1b1a3cabc4 100644 --- a/apps/sim/executor/handlers/agent/handler.test.ts +++ b/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', diff --git a/apps/sim/executor/handlers/agent/handler.ts b/apps/sim/executor/handlers/agent/handler.ts index 057f9b6d217..0de3b244c88 100644 --- a/apps/sim/executor/handlers/agent/handler.ts +++ b/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') diff --git a/apps/sim/lib/media/falai-audio.ts b/apps/sim/lib/media/falai-audio.ts index 6754794cb05..5b611a73204 100644 --- a/apps/sim/lib/media/falai-audio.ts +++ b/apps/sim/lib/media/falai-audio.ts @@ -21,6 +21,52 @@ export const DEFAULT_AUDIO_MODELS: Record = { // Zero-shot voice cloning from a reference sample (F5-TTS: ref_audio_url + gen_text). export const DEFAULT_CLONE_MODEL = 'fal-ai/f5-tts' +/** The prebuilt voices fal's Gemini TTS endpoints accept; any other `voice` is a 422. */ +const GEMINI_TTS_VOICES = [ + 'Achernar', + 'Achird', + 'Algenib', + 'Algieba', + 'Alnilam', + 'Aoede', + 'Autonoe', + 'Callirrhoe', + 'Charon', + 'Despina', + 'Enceladus', + 'Erinome', + 'Fenrir', + 'Gacrux', + 'Iapetus', + 'Kore', + 'Laomedeia', + 'Leda', + 'Orus', + 'Pulcherrima', + 'Puck', + 'Rasalgethi', + 'Sadachbia', + 'Sadaltager', + 'Schedar', + 'Sulafat', + 'Umbriel', + 'Vindemiatrix', + 'Zephyr', + 'Zubenelgenubi', +] as const + +/** + * Refuses a voice the speech model does not offer before anything is submitted, naming the + * voices it does: an ElevenLabs name sent to Gemini TTS is otherwise a bare 422 from fal. + */ +function assertSpeechVoiceSupported(model: string, voice: string | undefined): void { + if (!voice || !/gemini.*tts/.test(model)) return + if ((GEMINI_TTS_VOICES as readonly string[]).includes(voice)) return + throw new Error( + `Voice "${voice}" is not available for ${model}. Use one of: ${GEMINI_TTS_VOICES.join(', ')}.` + ) +} + export interface GenerateFalAudioParams { prompt: string type?: AudioType @@ -122,6 +168,7 @@ export async function generateFalAudio(params: GenerateFalAudioParams): Promise< } const model = params.model || DEFAULT_AUDIO_MODELS[type] + if (type === 'speech') assertSpeechVoiceSupported(model, params.voice) const input = buildInput(type, params, model) // For fal audio models the model ID is the queue endpoint. diff --git a/apps/sim/lib/media/falai.test.ts b/apps/sim/lib/media/falai.test.ts index abb5c3c231f..a9dd83f2da7 100644 --- a/apps/sim/lib/media/falai.test.ts +++ b/apps/sim/lib/media/falai.test.ts @@ -421,12 +421,17 @@ describe('the speech request fal receives', () => { it.each([ ['fal-ai/elevenlabs/tts/eleven-v3', { text: 'Hello there', voice: 'Aria' }], ['fal-ai/elevenlabs/tts/multilingual-v2', { text: 'Hello there', voice: 'Aria' }], - [undefined, { prompt: 'Hello there', voice: 'Aria' }], + [undefined, { prompt: 'Hello there', voice: 'Kore' }], ])('submits the text %s requires', async (model, expectedBody) => { const submitted = stubRejectingFalSubmit() await expect( - generateFalAudio({ prompt: 'Hello there', voice: 'Aria', model, onCost: vi.fn() }) + generateFalAudio({ + prompt: 'Hello there', + voice: expectedBody.voice, + model, + onCost: vi.fn(), + }) ).rejects.toThrow() expect(submitted).toEqual([ @@ -436,4 +441,25 @@ describe('the speech request fal receives', () => { }, ]) }) + + it.each([ + ['an explicit Gemini TTS', 'fal-ai/gemini-3.1-flash-tts'], + ['the default speech', undefined], + ])( + 'refuses a voice %s model does not offer without submitting, naming the voices it does', + async (_label, model) => { + const submitted = stubRejectingFalSubmit() + + const failure = generateFalAudio({ + prompt: 'Hello there', + voice: 'Brian', + model, + onCost: vi.fn(), + }) + + await expect(failure).rejects.toThrow('Voice "Brian" is not available') + await expect(failure).rejects.toThrow('Kore') + expect(submitted).toEqual([]) + } + ) }) diff --git a/apps/sim/lib/mothership/tools/server/image/generate-image.ts b/apps/sim/lib/mothership/tools/server/image/generate-image.ts index bb7f4614619..1e686b50d68 100644 --- a/apps/sim/lib/mothership/tools/server/image/generate-image.ts +++ b/apps/sim/lib/mothership/tools/server/image/generate-image.ts @@ -15,7 +15,10 @@ import { type BaseServerTool, type ServerToolContext, } from '@/lib/mothership/tools/server/base-tool' -import { writeCopilotWorkspaceFileByPath } from '@/lib/mothership/vfs/resource-writer' +import { + validateCopilotWorkspaceFileWriteTarget, + writeCopilotWorkspaceFileByPath, +} from '@/lib/mothership/vfs/resource-writer' import type { WorkspaceFileRecord } from '@/lib/uploads/contexts/workspace/file-manager' import { createWorkspaceFileSecretProvenanceFromRegistry } from '@/lib/uploads/contexts/workspace/file-secret-provenance' import { fileOperations } from '@/lib/workspace-files/application/operations' @@ -123,6 +126,14 @@ export const generateImageServerTool: BaseServerTool recordServiceCost('falai_audio', cost.costDollars), }) - const outputFile = params.outputs?.files?.[0] const ext = audioExtFromContentType(result.contentType) const outputPath = outputFile?.path || `files/generated-audio.${ext}` const mode = outputFile?.mode ?? 'create' diff --git a/apps/sim/lib/mothership/tools/server/media/generate-video.ts b/apps/sim/lib/mothership/tools/server/media/generate-video.ts index e563fa702b6..680518350fe 100644 --- a/apps/sim/lib/mothership/tools/server/media/generate-video.ts +++ b/apps/sim/lib/mothership/tools/server/media/generate-video.ts @@ -13,7 +13,10 @@ import { type BaseServerTool, type ServerToolContext, } from '@/lib/mothership/tools/server/base-tool' -import { writeCopilotWorkspaceFileByPath } from '@/lib/mothership/vfs/resource-writer' +import { + validateCopilotWorkspaceFileWriteTarget, + writeCopilotWorkspaceFileByPath, +} from '@/lib/mothership/vfs/resource-writer' import { createWorkspaceFileSecretProvenanceFromRegistry } from '@/lib/uploads/contexts/workspace/file-secret-provenance' import { fileOperations } from '@/lib/workspace-files/application/operations' import { readWorkspaceFileContent } from '@/lib/workspace-files/application/read-workspace-file-content' @@ -64,6 +67,14 @@ export const generateVideoServerTool: BaseServerTool recordServiceCost('falai_video', cost.costDollars), }) - const outputFile = params.outputs?.files?.[0] const outputPath = outputFile?.path || 'files/generated-video.mp4' const mode = outputFile?.mode ?? 'create' diff --git a/apps/sim/lib/mothership/tools/server/media/write-target-precheck.integration.ts b/apps/sim/lib/mothership/tools/server/media/write-target-precheck.integration.ts new file mode 100644 index 00000000000..2456b795998 --- /dev/null +++ b/apps/sim/lib/mothership/tools/server/media/write-target-precheck.integration.ts @@ -0,0 +1,184 @@ +/** + * Real-Postgres proof that the paid media tools refuse a stale output path before they generate: + * the folder-moved refusal used to fire only at write time, after the provider had billed, so + * the model retried and paid again. + */ +import { mkdtempSync } from 'node:fs' +import { rm } from 'node:fs/promises' +import { tmpdir } from 'node:os' +import path from 'node:path' +import { db } from '@sim/db' +import { auditLog, organization, user, workspace, workspaceFiles } from '@sim/db/schema' +import { deleteWorkspaceFixture } from '@sim/db/testing/workspace-fixtures' +import { and, eq, inArray, isNull } from 'drizzle-orm' +import { afterAll, beforeAll, beforeEach, describe, expect, it, vi } from 'vitest' + +const fixtureStorage = vi.hoisted(() => ({ root: '' })) +const { mockGenerateContent, mockGenerateFalAudio, mockGenerateFalVideo } = vi.hoisted(() => ({ + mockGenerateContent: vi.fn(), + mockGenerateFalAudio: vi.fn(), + mockGenerateFalVideo: vi.fn(), +})) + +vi.mock('@/lib/uploads/core/setup.server', () => ({ + get UPLOAD_DIR_SERVER() { + return fixtureStorage.root + }, +})) +vi.mock('@google/genai', () => ({ + GoogleGenAI: class GoogleGenAI { + models = { generateContent: mockGenerateContent } + }, +})) +vi.mock('@/lib/core/config/api-keys', () => ({ getRotatingApiKey: vi.fn(() => 'api-key') })) +vi.mock('@/lib/media/falai-audio', () => ({ generateFalAudio: mockGenerateFalAudio })) +vi.mock('@/lib/media/falai-video', () => ({ generateFalVideo: mockGenerateFalVideo })) + +import { + createKnowledgeAclFixtureIds, + seedKnowledgeAclFixture, +} from '@/lib/knowledge/__integration__/seed-source-access-fixture' +import { observeServiceCosts } from '@/lib/mothership/billing/service-observer' +import type { ServerToolContext } from '@/lib/mothership/tools/server/base-tool' +import { generateImageServerTool } from '@/lib/mothership/tools/server/image/generate-image' +import { generateAudioServerTool } from '@/lib/mothership/tools/server/media/generate-audio' +import { generateVideoServerTool } from '@/lib/mothership/tools/server/media/generate-video' +import { createWorkspaceFileFolderAtPath } from '@/lib/uploads/contexts/workspace/file-folder-manager' +import { ResolvedSecretTraceRegistry } from '@/executor/utils/resolved-secret-trace-registry' + +const fixtures: ReturnType[] = [] + +beforeAll(() => { + fixtureStorage.root = mkdtempSync(path.join(tmpdir(), 'sim-media-write-target-')) +}) + +afterAll(async () => { + for (const ids of fixtures) { + await db.delete(auditLog).where(eq(auditLog.workspaceId, ids.workspaceId)) + await deleteWorkspaceFixture(db, eq(workspace.id, ids.workspaceId)) + await db.delete(organization).where(eq(organization.id, ids.organizationId)) + await db.delete(user).where(inArray(user.id, [ids.aliceId, ids.bobId])) + } + await rm(fixtureStorage.root, { recursive: true, force: true }) +}) + +/** A fal generation that bills through `onCost` before returning, as the real client does. */ +function billsThen(media: Record) { + return async ({ onCost }: { onCost: (cost: { costDollars: number }) => Promise }) => { + const cost = { costDollars: 0.1 } + await onCost(cost) + return { ...media, cost } + } +} + +beforeEach(() => { + mockGenerateContent.mockReset().mockResolvedValue({ + candidates: [ + { content: { parts: [{ inlineData: { data: 'aW1hZ2U=', mimeType: 'image/png' } }] } }, + ], + }) + mockGenerateFalAudio + .mockReset() + .mockImplementation( + billsThen({ buffer: Buffer.from('audio'), contentType: 'audio/mpeg', model: 'tts' }) + ) + mockGenerateFalVideo + .mockReset() + .mockImplementation( + billsThen({ buffer: Buffer.from('video'), contentType: 'video/mp4', model: 'veo' }) + ) +}) + +/** A workspace whose `media` folder was moved under `archive`, so `files/media/…` is stale. */ +async function seedMovedFolder() { + const ids = createKnowledgeAclFixtureIds() + fixtures.push(ids) + await seedKnowledgeAclFixture(ids) + await createWorkspaceFileFolderAtPath({ + workspaceId: ids.workspaceId, + userId: ids.aliceId, + path: '/archive', + }) + const moved = await createWorkspaceFileFolderAtPath({ + workspaceId: ids.workspaceId, + userId: ids.aliceId, + path: '/archive/media', + }) + const context: ServerToolContext = { + userId: ids.aliceId, + workspaceId: ids.workspaceId, + toolCallId: 'tool-1', + copilotToolExecution: true, + resolvedSecretTraceRegistry: new ResolvedSecretTraceRegistry([]), + } + return { ids, context, movedFolderId: moved.folder.id } +} + +const tools = [ + { + kind: 'audio', + tool: generateAudioServerTool, + provider: () => mockGenerateFalAudio, + file: 'clip.mp3', + }, + { + kind: 'video', + tool: generateVideoServerTool, + provider: () => mockGenerateFalVideo, + file: 'clip.mp4', + }, + { + kind: 'image', + tool: generateImageServerTool, + provider: () => mockGenerateContent, + file: 'clip.png', + }, +] as const + +describe('media generation output target precheck', () => { + it.each(tools)( + 'refuses a moved-folder $kind target before generating or billing', + async ({ tool, provider, file }) => { + const { context } = await seedMovedFolder() + const charges = vi.fn().mockResolvedValue(undefined) + + const result = await observeServiceCosts(charges, () => + tool.execute( + { prompt: 'p', outputs: { files: [{ path: `files/media/${file}` }] } }, + context + ) + ) + + expect(result.success).toBe(false) + expect(result.message).toContain('now lives at /archive/media') + expect(provider()).not.toHaveBeenCalled() + expect(charges).not.toHaveBeenCalled() + } + ) + + it.each(tools)( + 'still generates and writes a $kind into the folder at its current path', + async ({ tool, provider, file }) => { + const { ids, context, movedFolderId } = await seedMovedFolder() + + const result = await tool.execute( + { prompt: 'p', outputs: { files: [{ path: `files/archive/media/${file}` }] } }, + context + ) + + expect(result.success, result.message).toBe(true) + expect(provider()).toHaveBeenCalledTimes(1) + const written = await db + .select({ name: workspaceFiles.originalName }) + .from(workspaceFiles) + .where( + and( + eq(workspaceFiles.workspaceId, ids.workspaceId), + eq(workspaceFiles.folderId, movedFolderId), + isNull(workspaceFiles.deletedAt) + ) + ) + expect(written).toEqual([{ name: file }]) + } + ) +}) diff --git a/apps/sim/lib/mothership/vfs/resource-writer.ts b/apps/sim/lib/mothership/vfs/resource-writer.ts index d516f94da10..a9c28284860 100644 --- a/apps/sim/lib/mothership/vfs/resource-writer.ts +++ b/apps/sim/lib/mothership/vfs/resource-writer.ts @@ -294,3 +294,18 @@ export function writeCopilotWorkspaceFileByPath( principal: resolveCopilotFilePrincipal(context), }) } + +/** + * Validates a Copilot write target before the tool produces its bytes, so a stale or taken path + * fails before a paid generation rather than after it. The write re-validates; this only moves + * the refusal earlier. + */ +export function validateCopilotWorkspaceFileWriteTarget( + context: CopilotFileDelegationContext | undefined, + args: Omit[0], 'principal'> +) { + return validateWorkspaceFileWriteTarget({ + ...args, + principal: resolveCopilotFilePrincipal(context), + }) +} diff --git a/apps/sim/lib/workflows/executor/human-in-the-loop-manager.test.ts b/apps/sim/lib/workflows/executor/human-in-the-loop-manager.test.ts index 9ab85e78bd2..5711a55b671 100644 --- a/apps/sim/lib/workflows/executor/human-in-the-loop-manager.test.ts +++ b/apps/sim/lib/workflows/executor/human-in-the-loop-manager.test.ts @@ -1,4 +1,5 @@ import { pausedExecutions, resumeQueue, workflowExecutionLogs } from '@sim/db/schema' +import { createLogger } from '@sim/logger' import { dbChainMockFns, flattenMockConditions, @@ -16,6 +17,7 @@ import { } from '@sim/testing/mocks/execution-preprocessing.mock' import { largeValueMetadataMock } from '@sim/testing/mocks/large-value-metadata.mock' import { afterEach, beforeEach, describe, expect, it, vi } from 'vitest' +import { classifyFailure, logFailureOnce } from '@/lib/core/errors/failure-log' import { createTimeoutAbortController, getExecutionDeadlineAt } from '@/lib/core/execution-limits' import { markExecutionCancelled } from '@/lib/execution/cancellation' import { abortManualExecution } from '@/lib/execution/manual-cancellation' @@ -485,7 +487,7 @@ describe('resume failure diagnostic projection', () => { .mockRejectedValueOnce(rawError) const markResumeFailedSpy = vi .spyOn(managerInternals, 'markResumeFailed') - .mockResolvedValueOnce() + .mockResolvedValueOnce(undefined) const processQueuedResumesSpy = vi .spyOn(PauseResumeManager, 'processQueuedResumes') .mockResolvedValueOnce() @@ -513,10 +515,15 @@ describe('resume failure diagnostic projection', () => { }) ).rejects.toBe(rawError) - expect(humanInTheLoopLogger.error).toHaveBeenCalledWith('Resume execution failed', { - errorType: 'error', - hasStack: true, - }) + expect(humanInTheLoopLogger.error).toHaveBeenCalledWith( + 'Resume execution failed', + expect.objectContaining({ + errorType: 'error', + hasStack: true, + executionId: 'parent-execution-1', + failureKind: 'internal', + }) + ) const loggerPayload = JSON.stringify(humanInTheLoopLogger.error.mock.calls) expect(loggerPayload).not.toContain(secret) expect(loggerPayload).not.toContain('__var_') @@ -529,6 +536,57 @@ describe('resume failure diagnostic projection', () => { } }) + /** The resumed engine and execution core log under the durable run id and rethrow the same value. */ + it('does not log again a failure the resumed run already logged', async () => { + const rawError = new Error('block failed inside the resumed run') + const managerInternals = PauseResumeManager as unknown as PauseResumeManagerInternals + const runResumeExecutionSpy = vi + .spyOn(managerInternals, 'runResumeExecution') + .mockImplementationOnce(async () => { + logFailureOnce(createLogger('ExecutionCore'), 'Execution failed', rawError, { + executionId: 'parent-execution-1', + }) + throw rawError + }) + const markResumeFailedSpy = vi + .spyOn(managerInternals, 'markResumeFailed') + .mockResolvedValueOnce(undefined) + const processQueuedResumesSpy = vi + .spyOn(PauseResumeManager, 'processQueuedResumes') + .mockResolvedValueOnce() + type StartResumeArgs = Parameters[0] + humanInTheLoopLogger.error.mockClear() + + try { + await expect( + PauseResumeManager.startResumeExecution({ + resumeEntryId: 'resume-entry-1', + resumeExecutionId: 'resume-execution-1', + pausedExecution: { + id: 'paused-execution-1', + workflowId: 'workflow-1', + executionId: 'parent-execution-1', + pausePoints: { 'context-1': { contextId: 'context-1', blockId: 'hitl-1' } }, + executionSnapshot: createSnapshotSeed(), + metadata: {}, + } as StartResumeArgs['pausedExecution'], + contextId: 'context-1', + resumeInput: {}, + userId: 'user-1', + }) + ).rejects.toBe(rawError) + + expect(humanInTheLoopLogger.error).not.toHaveBeenCalledWith( + 'Resume execution failed', + expect.anything() + ) + } finally { + runResumeExecutionSpy.mockRestore() + markResumeFailedSpy.mockRestore() + processQueuedResumesSpy.mockRestore() + } + }) + it('registers and removes the active resume under its exact execution id', async () => { const rawError = new Error('resume stopped') const managerInternals = PauseResumeManager as unknown as PauseResumeManagerInternals @@ -543,7 +601,7 @@ describe('resume failure diagnostic projection', () => { }) const markResumeFailedSpy = vi .spyOn(managerInternals, 'markResumeFailed') - .mockResolvedValueOnce() + .mockResolvedValueOnce(undefined) const processQueuedResumesSpy = vi .spyOn(PauseResumeManager, 'processQueuedResumes') .mockResolvedValueOnce() @@ -2019,12 +2077,12 @@ describe('PauseResumeManager resume log claims', () => { workflowId: 'workflow-1', executionDeadlineAt: new Date('2026-08-04T13:00:00.000Z'), }) - ).rejects.toMatchObject({ - name: 'ResumeAdmissionError', - message: 'Execution can no longer be resumed', - statusCode: 409, - retryable: false, - }) + ).rejects.toSatisfy( + (error) => + error instanceof Error && + error.message === 'Execution can no longer be resumed' && + classifyFailure(error) === 'user' + ) }) it('reuses the exact historical version claimed from a deployed run log', () => { @@ -2040,9 +2098,15 @@ describe('PauseResumeManager resume log claims', () => { ])( 'rejects an inconsistent paused mode/version binding', ({ useDraftState, deploymentVersionId }) => { - expect(() => requireResumeDeploymentVersion(useDraftState, deploymentVersionId)).toThrowError( - expect.objectContaining({ name: 'ResumeAdmissionError', statusCode: 409, retryable: false }) - ) + let thrown: unknown + try { + requireResumeDeploymentVersion(useDraftState, deploymentVersionId) + } catch (error) { + thrown = error + } + expect(thrown).toMatchObject({ name: 'ResumeAdmissionError', statusCode: 409 }) + /** A corrupt paused-run binding is Sim's fault and must keep logging at error. */ + expect(classifyFailure(thrown)).toBe('internal') } ) }) diff --git a/apps/sim/lib/workflows/executor/human-in-the-loop-manager.ts b/apps/sim/lib/workflows/executor/human-in-the-loop-manager.ts index 51a3a23d8d2..b8bfe1f744c 100644 --- a/apps/sim/lib/workflows/executor/human-in-the-loop-manager.ts +++ b/apps/sim/lib/workflows/executor/human-in-the-loop-manager.ts @@ -8,6 +8,7 @@ import type { Edge } from '@xyflow/react' import { and, asc, desc, eq, inArray, lt, type SQL, sql } from 'drizzle-orm' import { releaseExecutionSlot } from '@/lib/billing/calculations/usage-reservation' import { assertBillingAttributionSnapshot } from '@/lib/billing/core/attribution' +import { logFailureOnce, markFailureKind } from '@/lib/core/errors/failure-log' import { createTimeoutAbortController, getAsyncExecutionTimeoutForBillingAttribution, @@ -1063,14 +1064,16 @@ export class PauseResumeManager { ) }) } - logger.error( - 'Resume execution failed', - projectResolvedSecretDiagnosticError(error, undefined, { - parentExecutionId: pausedExecution.executionId, - resumeExecutionId, - contextId, - }) - ) + /** Keyed by the durable run id the resumed engine and execution core log under. */ + logFailureOnce(logger, 'Resume execution failed', error, { + executionId: pausedExecution.executionId, + metadata: () => + projectResolvedSecretDiagnosticError(error, undefined, { + parentExecutionId: pausedExecution.executionId, + resumeExecutionId, + contextId, + }), + }) if (!(error instanceof ResumeAdmissionError)) { await PauseResumeManager.processQueuedResumes( pausedExecution.executionId, @@ -1108,8 +1111,12 @@ export class PauseResumeManager { deploymentVersionId: workflowExecutionLogs.deploymentVersionId, }) + /** Another attempt or a cancel settled the run first: a lost race, not a Sim fault. */ if (!claimedExecution) { - throw new ResumeAdmissionError('Execution can no longer be resumed', 409, false) + throw markFailureKind( + new ResumeAdmissionError('Execution can no longer be resumed', 409, false), + 'user' + ) } return { deploymentVersionId: claimedExecution.deploymentVersionId } } diff --git a/apps/sim/lib/workspace-files/api/archive-presenter.test.ts b/apps/sim/lib/workspace-files/api/archive-presenter.test.ts index 7c526896b41..54db8725be2 100644 --- a/apps/sim/lib/workspace-files/api/archive-presenter.test.ts +++ b/apps/sim/lib/workspace-files/api/archive-presenter.test.ts @@ -2,6 +2,7 @@ import { Readable } from 'node:stream' import { flushMacrotask } from '@sim/testing/helpers' import { createDeferred } from '@sim/testing/helpers/deferred' +import { getMockLogger } from '@sim/testing/mocks/logger.mock' import { storageServiceMock, storageServiceMockFns } from '@sim/testing/mocks/storage-service.mock' import JSZip from 'jszip' import { describe, expect, it, vi } from 'vitest' @@ -11,6 +12,8 @@ vi.mock('@/lib/uploads/core/storage-service', () => storageServiceMock) import { presentWorkspaceFileArchive } from '@/lib/workspace-files/api/archive-presenter' +const deliveryLogger = getMockLogger('FileArchiveDelivery') + function plan() { const filesToZip: WorkspaceFileRecord[] = Array.from({ length: 3 }, (_, index) => ({ id: `file-${index}`, @@ -112,4 +115,51 @@ describe('file archive resource lifetime', () => { await reader.cancel() await vi.waitFor(() => expect(source.destroyed).toBe(true)) }) + + /** A storage response torn down mid-read fails with Node's bare `aborted`, not an AbortError. */ + function storageStreamThatFailsWhenDestroyed() { + return new Readable({ + read() { + this.push(Buffer.alloc(65536)) + }, + destroy(_error, callback) { + callback(new Error('aborted')) + }, + }) + } + + it('does not log a failure when the client disconnects mid-download', async () => { + deliveryLogger.error.mockClear() + const source = storageStreamThatFailsWhenDestroyed() + storageServiceMockFns.mockDownloadFileStream.mockResolvedValue(source) + const reader = presentWorkspaceFileArchive(plan()).body.getReader() + await reader.read() + const disconnect = new Error('') + disconnect.name = 'ResponseAborted' + await reader.cancel(disconnect) + await vi.waitFor(() => expect(source.destroyed).toBe(true)) + await flushMacrotask() + + expect(deliveryLogger.error).not.toHaveBeenCalled() + }) + + it('still logs a storage failure while the client is connected', async () => { + deliveryLogger.error.mockClear() + const source = storageStreamThatFailsWhenDestroyed() + storageServiceMockFns.mockDownloadFileStream.mockResolvedValue(source) + const result = presentWorkspaceFileArchive(plan()) + const reader = result.body.getReader() + await reader.read() + source.destroy() + + await expect( + (async () => { + while (!(await reader.read()).done) {} + })() + ).rejects.toThrow('aborted') + expect(deliveryLogger.error).toHaveBeenCalledWith( + 'Failed to build file archive', + expect.anything() + ) + }) }) diff --git a/apps/sim/lib/workspace-files/api/archive-presenter.ts b/apps/sim/lib/workspace-files/api/archive-presenter.ts index ce628926fa4..992414c42b8 100644 --- a/apps/sim/lib/workspace-files/api/archive-presenter.ts +++ b/apps/sim/lib/workspace-files/api/archive-presenter.ts @@ -78,7 +78,8 @@ export function presentWorkspaceFileArchive({ await archive.finalize() } appendEntries().catch((error: unknown) => { - if (toError(error).name === 'AbortError') return + /** A closed archive means the client went away; the entry's own error then says nothing. */ + if (closed.signal.aborted || toError(error).name === 'AbortError') return logger.error('Failed to build file archive', { error }) archive.destroy(toError(error)) }) diff --git a/apps/sim/providers/anthropic/core.test.ts b/apps/sim/providers/anthropic/core.test.ts index af8a8e9cef6..7ffc582c85d 100644 --- a/apps/sim/providers/anthropic/core.test.ts +++ b/apps/sim/providers/anthropic/core.test.ts @@ -1,4 +1,4 @@ -import type Anthropic from '@anthropic-ai/sdk' +import Anthropic from '@anthropic-ai/sdk' import { toolsMock, toolsMockFns } from '@sim/testing/mocks/tools.mock' import { beforeEach, describe, expect, it, vi } from 'vitest' import type { StreamingExecution } from '@/executor/types' @@ -981,3 +981,65 @@ describe('unsupported thinking level logging', () => { expect(describeModelLevel('high')).toBe('high') }) }) + +describe('executeAnthropicProviderRequest failure log severity', () => { + /** Never answers; rejects only when the SDK cancels it, so the SDK raises its own abort. */ + const hangingFetch = (_input: unknown, init?: RequestInit) => + new Promise((_resolve, reject) => { + init?.signal?.addEventListener('abort', () => reject(init.signal?.reason), { once: true }) + }) + + function runRequest(fetch: typeof hangingFetch, abortSignal: AbortSignal) { + const logger = { info: vi.fn(), warn: vi.fn(), error: vi.fn(), debug: vi.fn() } + const settled = executeAnthropicProviderRequest( + { + model: 'claude-sonnet-4-5', + apiKey: 'test-key', + maxTokens: 16, + messages: [{ role: 'user', content: 'hi' }], + abortSignal, + }, + { + providerId: 'anthropic', + providerLabel: 'Anthropic', + createClient: (apiKey) => new Anthropic({ apiKey, fetch, maxRetries: 0 }), + logger: logger as never, + } + ).catch((error: unknown) => error) + const failureLevels = () => + (['info', 'warn', 'error'] as const).filter((level) => + logger[level].mock.calls.some(([message]) => message === 'Error in Anthropic request:') + ) + return { settled, failureLevels } + } + + it('logs a request the user stopped at info', async () => { + const run = new AbortController() + const { settled, failureLevels } = runRequest(hangingFetch, run.signal) + run.abort(new DOMException('user', 'AbortError')) + await settled + expect(failureLevels()).toEqual(['info']) + }) + + it('logs a request the run time limit cut off at warn', async () => { + const run = new AbortController() + const { settled, failureLevels } = runRequest(hangingFetch, run.signal) + run.abort(new DOMException('timeout', 'AbortError')) + await settled + expect(failureLevels()).toEqual(['warn']) + }) + + it('logs a genuine provider failure on a live run at error', async () => { + const failingFetch = async () => + new Response( + JSON.stringify({ type: 'error', error: { type: 'api_error', message: 'boom' } }), + { + status: 500, + headers: { 'content-type': 'application/json' }, + } + ) + const { settled, failureLevels } = runRequest(failingFetch, new AbortController().signal) + await settled + expect(failureLevels()).toEqual(['error']) + }) +}) diff --git a/apps/sim/providers/anthropic/core.ts b/apps/sim/providers/anthropic/core.ts index fefadce0d01..6cf80c3d820 100644 --- a/apps/sim/providers/anthropic/core.ts +++ b/apps/sim/providers/anthropic/core.ts @@ -37,7 +37,7 @@ import { } from '@/providers/models' import { executeProviderTool } from '@/providers/runtime-context' import { createStreamingExecution } from '@/providers/streaming-execution' -import { isAbortError } from '@/providers/streaming-tool-loop-shared' +import { isAbortError, providerFailureLogLevel } from '@/providers/streaming-tool-loop-shared' import { adaptAnthropicToolSchema } from '@/providers/tool-schema-adapter' import { enrichLastModelSegment } from '@/providers/trace-enrichment' import type { ProviderRequest, ProviderResponse, TimeSegment } from '@/providers/types' @@ -952,7 +952,10 @@ export async function executeAnthropicProviderRequest( }) } } catch (error) { - logger.error(`Error in ${providerLabel} request:`, { error }) + logger[providerFailureLogLevel(error, request.abortSignal)]( + `Error in ${providerLabel} request:`, + { error } + ) throw error } @@ -995,10 +998,13 @@ export async function executeAnthropicProviderRequest( const providerEndTimeISO = new Date(providerEndTime).toISOString() const totalDuration = providerEndTime - providerStartTime - logger.error(`Error in ${providerLabel} request:`, { - error, - duration: totalDuration, - }) + logger[providerFailureLogLevel(error, request.abortSignal)]( + `Error in ${providerLabel} request:`, + { + error, + duration: totalDuration, + } + ) if (isAbortError(error) || request.abortSignal?.aborted || isConversationContextError(error)) { throw error diff --git a/apps/sim/providers/azure-openai/index.ts b/apps/sim/providers/azure-openai/index.ts index fde4f599d86..37db6e20761 100644 --- a/apps/sim/providers/azure-openai/index.ts +++ b/apps/sim/providers/azure-openai/index.ts @@ -45,7 +45,11 @@ import { buildJsonSchemaResponseFormat } from '@/providers/response-format' import { executeProviderTool } from '@/providers/runtime-context' import { createSettledAgentEventStream } from '@/providers/stream-events' import { createStreamingExecution } from '@/providers/streaming-execution' -import { isAbortError, parseToolArguments } from '@/providers/streaming-tool-loop-shared' +import { + isAbortError, + parseToolArguments, + providerFailureLogLevel, +} from '@/providers/streaming-tool-loop-shared' import { adaptOpenAIChatToolSchema } from '@/providers/tool-schema-adapter' import { enrichLastModelSegmentFromChatCompletions } from '@/providers/trace-enrichment' import { openAICompatTransport } from '@/providers/transport' @@ -704,10 +708,13 @@ async function executeChatCompletionsRequest( const providerEndTimeISO = new Date(providerEndTime).toISOString() const totalDuration = providerEndTime - providerStartTime - logger.error('Error in Azure OpenAI chat completions request:', { - error, - duration: totalDuration, - }) + logger[providerFailureLogLevel(error, request.abortSignal)]( + 'Error in Azure OpenAI chat completions request:', + { + error, + duration: totalDuration, + } + ) if (isAbortError(error) || request.abortSignal?.aborted || isConversationContextError(error)) { throw error diff --git a/apps/sim/providers/bedrock/index.ts b/apps/sim/providers/bedrock/index.ts index b472a60d42e..7b7f19585a6 100644 --- a/apps/sim/providers/bedrock/index.ts +++ b/apps/sim/providers/bedrock/index.ts @@ -48,7 +48,7 @@ import { import { executeProviderTool } from '@/providers/runtime-context' import { createSettledAgentEventStream } from '@/providers/stream-events' import { createStreamingExecution } from '@/providers/streaming-execution' -import { isAbortError } from '@/providers/streaming-tool-loop-shared' +import { isAbortError, providerFailureLogLevel } from '@/providers/streaming-tool-loop-shared' import { enrichLastModelSegment } from '@/providers/trace-enrichment' import type { FunctionCallResponse, @@ -1052,7 +1052,7 @@ export const bedrockProvider: ProviderConfig = { const providerEndTimeISO = new Date(providerEndTime).toISOString() const totalDuration = providerEndTime - providerStartTime - logger.error('Error in Bedrock request:', { + logger[providerFailureLogLevel(error, request.abortSignal)]('Error in Bedrock request:', { error, duration: totalDuration, }) diff --git a/apps/sim/providers/cerebras/index.ts b/apps/sim/providers/cerebras/index.ts index 2a957482b07..1a6d94f0db8 100644 --- a/apps/sim/providers/cerebras/index.ts +++ b/apps/sim/providers/cerebras/index.ts @@ -22,7 +22,11 @@ import { buildJsonSchemaResponseFormat } from '@/providers/response-format' import { executeProviderTool } from '@/providers/runtime-context' import { createSettledAgentEventStream } from '@/providers/stream-events' import { createStreamingExecution } from '@/providers/streaming-execution' -import { isAbortError, parseToolArguments } from '@/providers/streaming-tool-loop-shared' +import { + isAbortError, + parseToolArguments, + providerFailureLogLevel, +} from '@/providers/streaming-tool-loop-shared' import { adaptOpenAIChatToolSchema } from '@/providers/tool-schema-adapter' import { enrichLastModelSegmentFromChatCompletions } from '@/providers/trace-enrichment' import { openAICompatTransport } from '@/providers/transport' @@ -585,7 +589,10 @@ export const cerebrasProvider: ProviderConfig = { iterationCount++ } } catch (error) { - logger.error('Error in Cerebras tool processing:', { error }) + logger[providerFailureLogLevel(error, request.abortSignal)]( + 'Error in Cerebras tool processing:', + { error } + ) throw error } @@ -661,7 +668,7 @@ export const cerebrasProvider: ProviderConfig = { const providerEndTimeISO = new Date(providerEndTime).toISOString() const totalDuration = providerEndTime - providerStartTime - logger.error('Error in Cerebras request:', { + logger[providerFailureLogLevel(error, request.abortSignal)]('Error in Cerebras request:', { error, duration: totalDuration, }) diff --git a/apps/sim/providers/deepseek/index.ts b/apps/sim/providers/deepseek/index.ts index a0149482777..b2750a8090c 100644 --- a/apps/sim/providers/deepseek/index.ts +++ b/apps/sim/providers/deepseek/index.ts @@ -19,7 +19,11 @@ import { createOpenAICompatibleAgentEventStream } from '@/providers/openai-compa import { createOpenAICompatStreamingToolLoopStream } from '@/providers/openai-compat/streaming-tool-loop' import { executeProviderTool } from '@/providers/runtime-context' import { createStreamingExecution } from '@/providers/streaming-execution' -import { isAbortError, parseToolArguments } from '@/providers/streaming-tool-loop-shared' +import { + isAbortError, + parseToolArguments, + providerFailureLogLevel, +} from '@/providers/streaming-tool-loop-shared' import { adaptOpenAIChatToolSchema } from '@/providers/tool-schema-adapter' import { enrichLastModelSegmentFromChatCompletions } from '@/providers/trace-enrichment' import { openAICompatTransport } from '@/providers/transport' @@ -642,7 +646,9 @@ export const deepseekProvider: ProviderConfig = { ) } } catch (error) { - logger.error('Error in Deepseek request:', { error }) + logger[providerFailureLogLevel(error, request.abortSignal)]('Error in Deepseek request:', { + error, + }) throw error } @@ -672,7 +678,7 @@ export const deepseekProvider: ProviderConfig = { const providerEndTimeISO = new Date(providerEndTime).toISOString() const totalDuration = providerEndTime - providerStartTime - logger.error('Error in Deepseek request:', { + logger[providerFailureLogLevel(error, request.abortSignal)]('Error in Deepseek request:', { error, duration: totalDuration, }) diff --git a/apps/sim/providers/gemini/core.test.ts b/apps/sim/providers/gemini/core.test.ts index bb69686778f..f956d2a0eaf 100644 --- a/apps/sim/providers/gemini/core.test.ts +++ b/apps/sim/providers/gemini/core.test.ts @@ -2,6 +2,7 @@ import { ApiError, type GenerateContentParameters, type GenerateContentResponse, + GoogleGenAI, } from '@google/genai' import { providersMock } from '@sim/testing/mocks/providers.mock' import { @@ -10,6 +11,8 @@ import { } from '@sim/testing/mocks/providers-conversation-history.mock' import { toolsMock, toolsMockFns } from '@sim/testing/mocks/tools.mock' import { beforeEach, describe, expect, it, vi } from 'vitest' +import { classifyFailure } from '@/lib/core/errors/failure-log' +import { isNonRetryableExecutionError } from '@/lib/execution/non-retryable-error' import type { StreamingExecution } from '@/executor/types' import { executeGeminiRequest } from '@/providers/gemini/core' import type { ProviderRequest, ProviderResponse } from '@/providers/types' @@ -397,3 +400,37 @@ describe('Gemini block cost with implicit prompt caching', () => { expect(response.cost).toEqual(calculateCost(MODEL, 100_000, 1_000)) }) }) + +describe('Gemini request with no user content', () => { + function runAgainstSdk(request: Partial) { + return executeGeminiRequest({ + ai: new GoogleGenAI({ apiKey: 'test-key' }), + model: 'gemini-2.5-flash', + providerType: 'google', + request: { model: 'gemini-2.5-flash', apiKey: 'test-key', ...request }, + }).catch((caught: unknown) => caught) + } + + it.each([ + ['only a system prompt', { systemPrompt: 'You are terse.', messages: [] }], + [ + 'messages that all resolved to empty', + { + messages: [ + { role: 'system' as const, content: '' }, + { role: 'user' as const, content: '' }, + ], + }, + ], + ])( + 'refuses %s as a nonretryable author failure instead of the SDK opaque error', + async (_label, request) => { + const error = await runAgainstSdk(request) + + expect(error).toMatchObject({ name: 'EmptyProviderInputError' }) + expect((error as Error).message).toContain('no user message') + expect(isNonRetryableExecutionError(error)).toBe(true) + expect(classifyFailure(error)).toBe('user') + } + ) +}) diff --git a/apps/sim/providers/gemini/core.ts b/apps/sim/providers/gemini/core.ts index 94c10af2102..b1764a78d46 100644 --- a/apps/sim/providers/gemini/core.ts +++ b/apps/sim/providers/gemini/core.ts @@ -41,14 +41,15 @@ import { withProviderRetry } from '@/providers/retry' import { executeProviderTool } from '@/providers/runtime-context' import { createSettledAgentEventStream } from '@/providers/stream-events' import { createStreamingExecution } from '@/providers/streaming-execution' -import { isAbortError } from '@/providers/streaming-tool-loop-shared' +import { isAbortError, providerFailureLogLevel } from '@/providers/streaming-tool-loop-shared' import { ensureToolCallId } from '@/providers/tool-call-id' import { enrichLastModelSegment } from '@/providers/trace-enrichment' -import type { - FunctionCallResponse, - ProviderRequest, - ProviderResponse, - TimeSegment, +import { + EmptyProviderInputError, + type FunctionCallResponse, + type ProviderRequest, + type ProviderResponse, + type TimeSegment, } from '@/providers/types' import { isDeepResearchModel, @@ -924,7 +925,7 @@ export async function executeDeepResearchRequest( const providerEndTime = Date.now() const duration = providerEndTime - providerStartTime - logger.error('Error in deep research request:', { + logger[providerFailureLogLevel(error, request.abortSignal)]('Error in deep research request:', { error: toError(error).message, stack: error instanceof Error ? error.stack : undefined, }) @@ -980,6 +981,12 @@ export async function executeGeminiRequest( try { const { contents, tools, systemInstruction } = convertToGeminiFormat(request, providerType) + /** The SDK refuses empty `contents` with an opaque "contents are required"; a system prompt alone is not enough. */ + if (contents.length === 0) { + throw new EmptyProviderInputError( + 'The Agent has no user message to send, and Gemini needs one alongside any system prompt. Add a user message, or check that the value it references (for example ) is not empty.' + ) + } // Build configuration const geminiConfig: GenerateContentConfig = {} @@ -1444,7 +1451,7 @@ export async function executeGeminiRequest( const providerEndTime = Date.now() const duration = providerEndTime - providerStartTime - logger.error('Error in Gemini request:', { + logger[providerFailureLogLevel(error, request.abortSignal)]('Error in Gemini request:', { error: toError(error).message, stack: error instanceof Error ? error.stack : undefined, }) diff --git a/apps/sim/providers/groq/index.ts b/apps/sim/providers/groq/index.ts index e5cd22d9e26..1357a763d18 100644 --- a/apps/sim/providers/groq/index.ts +++ b/apps/sim/providers/groq/index.ts @@ -26,7 +26,11 @@ import { createOpenAICompatStreamingToolLoopStream } from '@/providers/openai-co import { buildJsonSchemaResponseFormat } from '@/providers/response-format' import { executeProviderTool } from '@/providers/runtime-context' import { createStreamingExecution } from '@/providers/streaming-execution' -import { isAbortError, parseToolArguments } from '@/providers/streaming-tool-loop-shared' +import { + isAbortError, + parseToolArguments, + providerFailureLogLevel, +} from '@/providers/streaming-tool-loop-shared' import { adaptOpenAIChatToolSchema } from '@/providers/tool-schema-adapter' import { enrichLastModelSegmentFromChatCompletions } from '@/providers/trace-enrichment' import { openAICompatTransport } from '@/providers/transport' @@ -586,7 +590,9 @@ export const groqProvider: ProviderConfig = { ) } } catch (error) { - logger.error('Error in Groq request:', { error }) + logger[providerFailureLogLevel(error, request.abortSignal)]('Error in Groq request:', { + error, + }) throw error } @@ -616,7 +622,7 @@ export const groqProvider: ProviderConfig = { const providerEndTimeISO = new Date(providerEndTime).toISOString() const totalDuration = providerEndTime - providerStartTime - logger.error('Error in Groq request:', { + logger[providerFailureLogLevel(error, request.abortSignal)]('Error in Groq request:', { error, duration: totalDuration, }) diff --git a/apps/sim/providers/kimi/index.ts b/apps/sim/providers/kimi/index.ts index 1aa9eb5ea68..0434dc047d5 100644 --- a/apps/sim/providers/kimi/index.ts +++ b/apps/sim/providers/kimi/index.ts @@ -25,7 +25,11 @@ import { createOpenAICompatibleAgentEventStream } from '@/providers/openai-compa import { executeProviderTool } from '@/providers/runtime-context' import { createSettledAgentEventStream } from '@/providers/stream-events' import { createStreamingExecution } from '@/providers/streaming-execution' -import { isAbortError, parseToolArguments } from '@/providers/streaming-tool-loop-shared' +import { + isAbortError, + parseToolArguments, + providerFailureLogLevel, +} from '@/providers/streaming-tool-loop-shared' import { adaptOpenAIChatToolSchema } from '@/providers/tool-schema-adapter' import { enrichLastModelSegmentFromChatCompletions } from '@/providers/trace-enrichment' import { openAICompatTransport } from '@/providers/transport' @@ -664,7 +668,9 @@ export const kimiProvider: ProviderConfig = { } } } catch (error) { - logger.error('Error in Kimi request:', { error }) + logger[providerFailureLogLevel(error, request.abortSignal)]('Error in Kimi request:', { + error, + }) throw error } @@ -740,7 +746,7 @@ export const kimiProvider: ProviderConfig = { const providerEndTimeISO = new Date(providerEndTime).toISOString() const totalDuration = providerEndTime - providerStartTime - logger.error('Error in Kimi request:', { + logger[providerFailureLogLevel(error, request.abortSignal)]('Error in Kimi request:', { error, duration: totalDuration, }) diff --git a/apps/sim/providers/litellm/index.ts b/apps/sim/providers/litellm/index.ts index 37539f6aa17..913816b5e60 100644 --- a/apps/sim/providers/litellm/index.ts +++ b/apps/sim/providers/litellm/index.ts @@ -22,7 +22,11 @@ import { createOpenAICompatibleAgentEventStream } from '@/providers/openai-compa import { executeProviderTool } from '@/providers/runtime-context' import { createSettledAgentEventStream } from '@/providers/stream-events' import { createStreamingExecution } from '@/providers/streaming-execution' -import { isAbortError, parseToolArguments } from '@/providers/streaming-tool-loop-shared' +import { + isAbortError, + parseToolArguments, + providerFailureLogLevel, +} from '@/providers/streaming-tool-loop-shared' import { adaptOpenAIChatToolSchema } from '@/providers/tool-schema-adapter' import { enrichLastModelSegmentFromChatCompletions } from '@/providers/trace-enrichment' import { openAICompatTransport } from '@/providers/transport' @@ -802,7 +806,7 @@ export const litellmProvider: ProviderConfig = { } } - logger.error('Error in LiteLLM request:', { + logger[providerFailureLogLevel(error, request.abortSignal)]('Error in LiteLLM request:', { error: errorMessage, errorType, errorCode, diff --git a/apps/sim/providers/meta/index.ts b/apps/sim/providers/meta/index.ts index d3366bf794c..25e33098d4f 100644 --- a/apps/sim/providers/meta/index.ts +++ b/apps/sim/providers/meta/index.ts @@ -21,7 +21,11 @@ import { buildJsonSchemaResponseFormat } from '@/providers/response-format' import { executeProviderTool } from '@/providers/runtime-context' import { createSettledAgentEventStream } from '@/providers/stream-events' import { createStreamingExecution } from '@/providers/streaming-execution' -import { isAbortError, parseToolArguments } from '@/providers/streaming-tool-loop-shared' +import { + isAbortError, + parseToolArguments, + providerFailureLogLevel, +} from '@/providers/streaming-tool-loop-shared' import { adaptOpenAIChatToolSchema } from '@/providers/tool-schema-adapter' import { enrichLastModelSegmentFromChatCompletions } from '@/providers/trace-enrichment' import { openAICompatTransport } from '@/providers/transport' @@ -550,7 +554,9 @@ export const metaProvider: ProviderConfig = { } } } catch (error) { - logger.error('Error in Meta request:', { error }) + logger[providerFailureLogLevel(error, request.abortSignal)]('Error in Meta request:', { + error, + }) throw error } @@ -682,7 +688,7 @@ export const metaProvider: ProviderConfig = { const providerEndTimeISO = new Date(providerEndTime).toISOString() const totalDuration = providerEndTime - providerStartTime - logger.error('Error in Meta request:', { + logger[providerFailureLogLevel(error, request.abortSignal)]('Error in Meta request:', { error, duration: totalDuration, }) diff --git a/apps/sim/providers/mistral/index.ts b/apps/sim/providers/mistral/index.ts index 055870622c9..03d225be391 100644 --- a/apps/sim/providers/mistral/index.ts +++ b/apps/sim/providers/mistral/index.ts @@ -22,7 +22,11 @@ import { buildJsonSchemaResponseFormat } from '@/providers/response-format' import { executeProviderTool } from '@/providers/runtime-context' import { createSettledAgentEventStream } from '@/providers/stream-events' import { createStreamingExecution } from '@/providers/streaming-execution' -import { isAbortError, parseToolArguments } from '@/providers/streaming-tool-loop-shared' +import { + isAbortError, + parseToolArguments, + providerFailureLogLevel, +} from '@/providers/streaming-tool-loop-shared' import { adaptOpenAIChatToolSchema } from '@/providers/tool-schema-adapter' import { enrichLastModelSegmentFromChatCompletions } from '@/providers/trace-enrichment' import { openAICompatTransport } from '@/providers/transport' @@ -647,7 +651,7 @@ export const mistralProvider: ProviderConfig = { const providerEndTimeISO = new Date(providerEndTime).toISOString() const totalDuration = providerEndTime - providerStartTime - logger.error('Error in Mistral request:', { + logger[providerFailureLogLevel(error, request.abortSignal)]('Error in Mistral request:', { error, duration: totalDuration, }) diff --git a/apps/sim/providers/nvidia/index.ts b/apps/sim/providers/nvidia/index.ts index eae48b4506c..956433308b9 100644 --- a/apps/sim/providers/nvidia/index.ts +++ b/apps/sim/providers/nvidia/index.ts @@ -26,7 +26,11 @@ import { buildJsonSchemaResponseFormat } from '@/providers/response-format' import { executeProviderTool } from '@/providers/runtime-context' import { createSettledAgentEventStream } from '@/providers/stream-events' import { createStreamingExecution } from '@/providers/streaming-execution' -import { isAbortError, parseToolArguments } from '@/providers/streaming-tool-loop-shared' +import { + isAbortError, + parseToolArguments, + providerFailureLogLevel, +} from '@/providers/streaming-tool-loop-shared' import { adaptOpenAIChatToolSchema } from '@/providers/tool-schema-adapter' import { enrichLastModelSegmentFromChatCompletions } from '@/providers/trace-enrichment' import { openAICompatTransport } from '@/providers/transport' @@ -601,7 +605,10 @@ export const nvidiaProvider: ProviderConfig = { } } } catch (error) { - logger.error('Error in NVIDIA NIM request:', { error }) + logger[providerFailureLogLevel(error, request.abortSignal)]( + 'Error in NVIDIA NIM request:', + { error } + ) throw error } @@ -731,7 +738,7 @@ export const nvidiaProvider: ProviderConfig = { const providerEndTimeISO = new Date(providerEndTime).toISOString() const totalDuration = providerEndTime - providerStartTime - logger.error('Error in NVIDIA NIM request:', { + logger[providerFailureLogLevel(error, request.abortSignal)]('Error in NVIDIA NIM request:', { error, duration: totalDuration, }) diff --git a/apps/sim/providers/ollama/core.ts b/apps/sim/providers/ollama/core.ts index f0d0baffbfb..1b2d0a621da 100644 --- a/apps/sim/providers/ollama/core.ts +++ b/apps/sim/providers/ollama/core.ts @@ -20,7 +20,11 @@ import { createOpenAICompatibleAgentEventStream } from '@/providers/openai-compa import { executeProviderTool } from '@/providers/runtime-context' import { createSettledAgentEventStream } from '@/providers/stream-events' import { createStreamingExecution } from '@/providers/streaming-execution' -import { isAbortError, parseToolArguments } from '@/providers/streaming-tool-loop-shared' +import { + isAbortError, + parseToolArguments, + providerFailureLogLevel, +} from '@/providers/streaming-tool-loop-shared' import { adaptOpenAIChatToolSchema } from '@/providers/tool-schema-adapter' import { enrichLastModelSegmentFromChatCompletions } from '@/providers/trace-enrichment' import type { Message, ProviderRequest, ProviderResponse, TimeSegment } from '@/providers/types' @@ -706,13 +710,16 @@ export async function executeOllamaProviderRequest( status = error.status } - logger.error(`Error in ${providerLabel} request:`, { - error: errorMessage, - errorType, - errorCode, - status, - duration: totalDuration, - }) + logger[providerFailureLogLevel(error, request.abortSignal)]( + `Error in ${providerLabel} request:`, + { + error: errorMessage, + errorType, + errorCode, + status, + duration: totalDuration, + } + ) if (isAbortError(error) || request.abortSignal?.aborted || isConversationContextError(error)) { throw error diff --git a/apps/sim/providers/openai-compat/chat-completions.ts b/apps/sim/providers/openai-compat/chat-completions.ts index 993dfb340b5..5e20ba45b75 100644 --- a/apps/sim/providers/openai-compat/chat-completions.ts +++ b/apps/sim/providers/openai-compat/chat-completions.ts @@ -30,7 +30,11 @@ import { createOpenAICompatibleAgentEventStream } from '@/providers/openai-compa import { executeProviderTool } from '@/providers/runtime-context' import { createSettledAgentEventStream } from '@/providers/stream-events' import { createStreamingExecution } from '@/providers/streaming-execution' -import { isAbortError, parseToolArguments } from '@/providers/streaming-tool-loop-shared' +import { + isAbortError, + parseToolArguments, + providerFailureLogLevel, +} from '@/providers/streaming-tool-loop-shared' import { adaptOpenAIChatToolSchema } from '@/providers/tool-schema-adapter' import { enrichLastModelSegmentFromChatCompletions } from '@/providers/trace-enrichment' import type { @@ -703,7 +707,10 @@ export async function executeChatCompletionRequest( } } - logger.error(`Error in ${providerName} request:`, errorDetails) + logger[providerFailureLogLevel(error, request.abortSignal)]( + `Error in ${providerName} request:`, + errorDetails + ) if (isAbortError(error) || request.abortSignal?.aborted || isConversationContextError(error)) { throw error } diff --git a/apps/sim/providers/openai/core.ts b/apps/sim/providers/openai/core.ts index be1bf695d8c..85d53ec8efe 100644 --- a/apps/sim/providers/openai/core.ts +++ b/apps/sim/providers/openai/core.ts @@ -27,7 +27,11 @@ import { import { fetchWithProviderRetry } from '@/providers/retry' import { executeProviderTool } from '@/providers/runtime-context' import { createStreamingExecution } from '@/providers/streaming-execution' -import { isAbortError, parseToolArguments } from '@/providers/streaming-tool-loop-shared' +import { + isAbortError, + parseToolArguments, + providerFailureLogLevel, +} from '@/providers/streaming-tool-loop-shared' import { adaptOpenAIChatToolSchema } from '@/providers/tool-schema-adapter' import type { Message, ProviderRequest, ProviderResponse, TimeSegment } from '@/providers/types' import { EmptyProviderInputError, ProviderError } from '@/providers/types' @@ -356,16 +360,19 @@ export async function executeResponsesProviderRequest( .map(([key, value]) => `${key}=${value}`) const context = [`phase=${phase}`, `elapsedMs=${elapsedMs}`, ...fields].join(' ') - logger.error(`${config.providerLabel} request failed in transport`, { - phase, - elapsedMs, - errorName: error.name, - model: config.modelName, - workflowId: request.workflowId, - blockId: request.blockId, - executionId: request.executionId, - ...detail, - }) + logger[providerFailureLogLevel(error, request.abortSignal)]( + `${config.providerLabel} request failed in transport`, + { + phase, + elapsedMs, + errorName: error.name, + model: config.modelName, + workflowId: request.workflowId, + blockId: request.blockId, + executionId: request.executionId, + ...detail, + } + ) /** * A new Error rather than a mutation: the runtime raises these as `DOMException`, @@ -978,10 +985,13 @@ export async function executeResponsesProviderRequest( const providerEndTimeISO = new Date(providerEndTime).toISOString() const totalDuration = providerEndTime - providerStartTime - logger.error(`Error in ${config.providerLabel} request:`, { - error, - duration: totalDuration, - }) + logger[providerFailureLogLevel(error, request.abortSignal)]( + `Error in ${config.providerLabel} request:`, + { + error, + duration: totalDuration, + } + ) if (isAbortError(error) || request.abortSignal?.aborted || isConversationContextError(error)) { throw error diff --git a/apps/sim/providers/runtime-context.ts b/apps/sim/providers/runtime-context.ts index 5b7459e35d6..d3fad7846b4 100644 --- a/apps/sim/providers/runtime-context.ts +++ b/apps/sim/providers/runtime-context.ts @@ -22,6 +22,7 @@ import type { ExecutionContext } from '@/executor/types' import { projectResolvedSecretModelContent } from '@/executor/utils/resolved-secret-content-projection' import type { ResolvedSecretTraceRegistry } from '@/executor/utils/resolved-secret-trace-registry' import { stripCloneSuffixes } from '@/executor/utils/subflow-node-id-codec' +import { isAbortError } from '@/providers/streaming-tool-loop-shared' import { getPreparedProviderToolInputProvenance } from '@/providers/tool-input-provenance' import type { ProviderId } from '@/providers/types' import { type ExecuteToolOptions, executeTool } from '@/tools' @@ -318,12 +319,8 @@ export async function executeProviderTool( return recordResult({ rawResponse: result, modelResponse }) } catch (error) { if (!registry || !toolCallRegistry) throw error - const errorName = - error && typeof error === 'object' && 'name' in error ? String(error.name) : undefined registry.mergeToolCallRegistry(toolCallRegistry) - if (errorName === 'AbortError' || errorName === 'APIUserAbortError') { - throw error - } + if (isAbortError(error)) throw error const rawResponse: ToolResponse = { success: false, output: {}, diff --git a/apps/sim/providers/sakana/index.ts b/apps/sim/providers/sakana/index.ts index d88521715ac..5ab637f4fde 100644 --- a/apps/sim/providers/sakana/index.ts +++ b/apps/sim/providers/sakana/index.ts @@ -22,7 +22,11 @@ import { buildJsonSchemaResponseFormat } from '@/providers/response-format' import { executeProviderTool } from '@/providers/runtime-context' import { createSettledAgentEventStream } from '@/providers/stream-events' import { createStreamingExecution } from '@/providers/streaming-execution' -import { isAbortError, parseToolArguments } from '@/providers/streaming-tool-loop-shared' +import { + isAbortError, + parseToolArguments, + providerFailureLogLevel, +} from '@/providers/streaming-tool-loop-shared' import { adaptOpenAIChatToolSchema } from '@/providers/tool-schema-adapter' import { enrichLastModelSegmentFromChatCompletions } from '@/providers/trace-enrichment' import { openAICompatTransport } from '@/providers/transport' @@ -585,7 +589,9 @@ export const sakanaProvider: ProviderConfig = { } } } catch (error) { - logger.error('Error in Sakana request:', { error }) + logger[providerFailureLogLevel(error, request.abortSignal)]('Error in Sakana request:', { + error, + }) throw error } @@ -717,7 +723,7 @@ export const sakanaProvider: ProviderConfig = { const providerEndTimeISO = new Date(providerEndTime).toISOString() const totalDuration = providerEndTime - providerStartTime - logger.error('Error in Sakana request:', { + logger[providerFailureLogLevel(error, request.abortSignal)]('Error in Sakana request:', { error, duration: totalDuration, }) diff --git a/apps/sim/providers/streaming-tool-loop-shared.ts b/apps/sim/providers/streaming-tool-loop-shared.ts index 50781f79e59..f76df2892cb 100644 --- a/apps/sim/providers/streaming-tool-loop-shared.ts +++ b/apps/sim/providers/streaming-tool-loop-shared.ts @@ -6,8 +6,13 @@ * provider-agnostic contract they share. */ -import { toError } from '@sim/utils/errors' +import { APIUserAbortError as AnthropicUserAbortError } from '@anthropic-ai/sdk/error' +import { findCause, toError } from '@sim/utils/errors' import { isRecordLike } from '@sim/utils/object' +import { APIUserAbortError as GroqUserAbortError } from 'groq-sdk/error' +import { APIUserAbortError as OpenAIUserAbortError } from 'openai/error' +import { UserFailure } from '@/lib/core/errors/user-failure' +import { isTimeoutAbortReason } from '@/lib/core/execution-limits/types' import type { NormalizedBlockOutput } from '@/executor/types' import type { AgentStreamEvent, ToolCallEndStatus } from '@/providers/stream-events' @@ -40,11 +45,46 @@ export interface StreamingToolLoopComplete { iterations: number } -/** True for user/SDK abort errors raised when a run is cancelled mid-stream. */ +/** + * True for the error a cancelled request rejects with: a platform `AbortError`, or the + * `APIUserAbortError` the vendor SDKs raise in its place. The SDK class leaves `name` as + * `'Error'`, so it is recognised by class, never by name. + */ export function isAbortError(error: unknown): boolean { - if (!error || typeof error !== 'object') return false - const name = (error as { name?: string }).name - return name === 'AbortError' || name === 'APIUserAbortError' + return ( + error instanceof AnthropicUserAbortError || + error instanceof OpenAIUserAbortError || + error instanceof GroqUserAbortError || + (isRecordLike(error) && error.name === 'AbortError') + ) +} + +/** + * True when `error` is the run's own abort surfacing: the run's signal fired, and the failure + * is its reason or an abort error anywhere in the cause chain. The signal decides; an abort + * error while the run is still live is a request-level deadline, a real failure. + */ +export function isRunAbortError(error: unknown, abortSignal: AbortSignal | undefined): boolean { + if (!abortSignal?.aborted) return false + return ( + error === abortSignal.reason || + findCause(error, (link): link is Error => isAbortError(link)) !== undefined + ) +} + +/** + * The severity a provider logs its own request failure at, on the `failure-log` scale: a run + * the user stopped or a request the author must fix ({@link UserFailure}) is routine (info), one + * the run's time limit cut off is expected but worth seeing (warn), and anything else is an + * unattributed failure (error). + */ +export function providerFailureLogLevel( + error: unknown, + abortSignal: AbortSignal | undefined +): 'info' | 'warn' | 'error' { + if (error instanceof UserFailure) return 'info' + if (!isRunAbortError(error, abortSignal)) return 'error' + return isTimeoutAbortReason(abortSignal?.reason) ? 'warn' : 'info' } /** Parse provider-supplied tool arguments without accepting non-object JSON values. */ diff --git a/apps/sim/providers/types.ts b/apps/sim/providers/types.ts index 57f0d3bbb32..b0c68a388c3 100644 --- a/apps/sim/providers/types.ts +++ b/apps/sim/providers/types.ts @@ -303,10 +303,10 @@ export class ProviderError extends Error { export class EmptyProviderInputError extends UserFailure { readonly retryable = false - constructor() { - super( - 'The Agent has no messages to send. Add a system prompt or user message, or check that the value it references (for example ) is not empty.' - ) + constructor( + message = 'The Agent has no messages to send. Add a system prompt or user message, or check that the value it references (for example ) is not empty.' + ) { + super(message) this.name = 'EmptyProviderInputError' } } diff --git a/apps/sim/providers/vllm/index.ts b/apps/sim/providers/vllm/index.ts index 3e5027351be..0fac82e9eaa 100644 --- a/apps/sim/providers/vllm/index.ts +++ b/apps/sim/providers/vllm/index.ts @@ -27,7 +27,11 @@ import { buildJsonSchemaResponseFormat } from '@/providers/response-format' import { executeProviderTool } from '@/providers/runtime-context' import { createSettledAgentEventStream } from '@/providers/stream-events' import { createStreamingExecution } from '@/providers/streaming-execution' -import { isAbortError, parseToolArguments } from '@/providers/streaming-tool-loop-shared' +import { + isAbortError, + parseToolArguments, + providerFailureLogLevel, +} from '@/providers/streaming-tool-loop-shared' import { adaptOpenAIChatToolSchema } from '@/providers/tool-schema-adapter' import { enrichLastModelSegmentFromChatCompletions } from '@/providers/trace-enrichment' import { openAICompatTransport } from '@/providers/transport' @@ -768,7 +772,7 @@ export const vllmProvider: ProviderConfig = { } } - logger.error('Error in vLLM request:', { + logger[providerFailureLogLevel(error, request.abortSignal)]('Error in vLLM request:', { error: errorMessage, errorType, errorCode, diff --git a/apps/sim/providers/xai/index.ts b/apps/sim/providers/xai/index.ts index 45d7945e21c..bf708f25058 100644 --- a/apps/sim/providers/xai/index.ts +++ b/apps/sim/providers/xai/index.ts @@ -24,7 +24,11 @@ import { createOpenAICompatibleAgentEventStream } from '@/providers/openai-compa import { executeProviderTool } from '@/providers/runtime-context' import { createSettledAgentEventStream } from '@/providers/stream-events' import { createStreamingExecution } from '@/providers/streaming-execution' -import { isAbortError, parseToolArguments } from '@/providers/streaming-tool-loop-shared' +import { + isAbortError, + parseToolArguments, + providerFailureLogLevel, +} from '@/providers/streaming-tool-loop-shared' import { adaptOpenAIChatToolSchema } from '@/providers/tool-schema-adapter' import { enrichLastModelSegmentFromChatCompletions } from '@/providers/trace-enrichment' import { openAICompatTransport } from '@/providers/transport' @@ -721,12 +725,15 @@ export const xAIProvider: ProviderConfig = { const providerEndTimeISO = new Date(providerEndTime).toISOString() const totalDuration = providerEndTime - providerStartTime - logger.error('XAI Provider - Request failed:', { - error: toError(error).message, - duration: totalDuration, - hasTools: !!tools?.length, - hasResponseFormat: !!request.responseFormat, - }) + logger[providerFailureLogLevel(error, request.abortSignal)]( + 'XAI Provider - Request failed:', + { + error: toError(error).message, + duration: totalDuration, + hasTools: !!tools?.length, + hasResponseFormat: !!request.responseFormat, + } + ) if ( isAbortError(error) || diff --git a/apps/sim/providers/zai/index.ts b/apps/sim/providers/zai/index.ts index 328195c2f48..10ff0aec053 100644 --- a/apps/sim/providers/zai/index.ts +++ b/apps/sim/providers/zai/index.ts @@ -21,7 +21,11 @@ import { createOpenAICompatibleAgentEventStream } from '@/providers/openai-compa import { executeProviderTool } from '@/providers/runtime-context' import { createSettledAgentEventStream } from '@/providers/stream-events' import { createStreamingExecution } from '@/providers/streaming-execution' -import { isAbortError, parseToolArguments } from '@/providers/streaming-tool-loop-shared' +import { + isAbortError, + parseToolArguments, + providerFailureLogLevel, +} from '@/providers/streaming-tool-loop-shared' import { adaptOpenAIChatToolSchema } from '@/providers/tool-schema-adapter' import { enrichLastModelSegmentFromChatCompletions } from '@/providers/trace-enrichment' import { openAICompatTransport } from '@/providers/transport' @@ -570,7 +574,9 @@ export const zaiProvider: ProviderConfig = { } } } catch (error) { - logger.error('Error in Z.ai request:', { error }) + logger[providerFailureLogLevel(error, request.abortSignal)]('Error in Z.ai request:', { + error, + }) throw error } @@ -703,7 +709,7 @@ export const zaiProvider: ProviderConfig = { const providerEndTimeISO = new Date(providerEndTime).toISOString() const totalDuration = providerEndTime - providerStartTime - logger.error('Error in Z.ai request:', { + logger[providerFailureLogLevel(error, request.abortSignal)]('Error in Z.ai request:', { error, duration: totalDuration, }) diff --git a/scripts/check-test-types.baseline.json b/scripts/check-test-types.baseline.json index cb2b56fbc55..fd41529af48 100644 --- a/scripts/check-test-types.baseline.json +++ b/scripts/check-test-types.baseline.json @@ -2355,9 +2355,6 @@ "apps/sim/lib/workflows/executor/execution-status.test.ts": { "TS2352": 1 }, - "apps/sim/lib/workflows/executor/human-in-the-loop-manager.test.ts": { - "TS2554": 2 - }, "apps/sim/lib/workflows/migrations/subblock-migrations.test.ts": { "TS2322": 1 },