diff --git a/apps/sim/executor/handlers/agent/handler.test.ts b/apps/sim/executor/handlers/agent/handler.test.ts index 7b796c72202..b33da2edba5 100644 --- a/apps/sim/executor/handlers/agent/handler.test.ts +++ b/apps/sim/executor/handlers/agent/handler.test.ts @@ -42,7 +42,8 @@ import { AGENT, BlockType } from '@/executor/constants' import { AgentBlockHandler } from '@/executor/handlers/agent/handler' import * as agentMemory from '@/executor/handlers/agent/memory' import type { AgentInputs, Message } from '@/executor/handlers/agent/types' -import type { ExecutionContext, StreamingExecution } from '@/executor/types' +import type { BlockLog, ExecutionContext, StreamingExecution } from '@/executor/types' +import { projectResolvedSecretDiagnosticError } from '@/executor/utils/resolved-secret-content-projection' import { ResolvedSecretTraceRegistry } from '@/executor/utils/resolved-secret-trace-registry' import { getEncryptedConversationMessage, @@ -839,7 +840,7 @@ describe('AgentBlockHandler', () => { timing: { total: 1 }, }) - const openLog = (blockId = mockBlock.id, endedAt = '') => ({ + const openLog = (blockId = mockBlock.id, endedAt = ''): BlockLog => ({ blockId, startedAt: '2026-01-01T00:00:00.000Z', endedAt, @@ -1303,6 +1304,111 @@ describe('AgentBlockHandler', () => { expect(blockLog).toMatchObject({ modelFallbacks: ['gpt-4o'] }) }) + describe('a fallback whose provider cannot take the attached file', () => { + const attachment = { + id: 'file-1', + name: 'example.png', + key: 'execution/test-workspace/test-workflow/exec-1/example.png', + url: 'https://storage.example.com/expired', + size: 8, + type: 'image/png', + context: 'execution', + base64: 'iVBORw0KGgo=', + } + + /** A secret only the primary's tool call activates, so only its failed attempt's registries know it. */ + const contextWithToolSecretCatalog = (blockLog: ReturnType) => ({ + ...mockContext, + executionId: 'exec-1', + blockLogs: [blockLog], + resolvedSecretTraceRegistry: new ResolvedSecretTraceRegistry([ + { name: 'TOKEN', plaintext: 'tool-secret', encryptedValue: 'encrypted-tool-secret' }, + ]), + }) + + const inputsWithAttachment = (fallbackModels: Array<{ model: string }>) => ({ + ...baseInputs, + files: [attachment], + fallbackModels, + }) + + let hydrate: ReturnType + const primaryError = new Error('overloaded after reading tool-secret') + + beforeEach(() => { + mockGetProviderFromModel.mockImplementation((model: string) => + model.startsWith('deepseek') ? 'deepseek' : providerFor(model) + ) + hydrate = vi + .spyOn(userFileBase64, 'hydrateUserFilesWithBase64') + .mockImplementation(async (files) => files) + mockExecuteProviderRequest.mockImplementationOnce(async (_provider, _request, options) => { + const runtimeRegistry = options.resolvedSecretTraceRegistry as ResolvedSecretTraceRegistry + runtimeRegistry.recordResolved('TOKEN', 'tool-secret', { propagated: true }) + throw primaryError + }) + }) + + afterEach(() => { + hydrate.mockRestore() + }) + + it("is skipped, and the primary's error rethrows against its failed attempt's registries", async () => { + const blockLog = openLog() + blockLog.modelFallbacks = ['stale-from-earlier-run'] + const ctx = contextWithToolSecretCatalog(blockLog) + + await expect( + handler.execute(ctx, mockBlock, inputsWithAttachment([{ model: 'deepseek-chat' }])) + ).rejects.toBe(primaryError) + + expect(mockExecuteProviderRequest).toHaveBeenCalledTimes(1) + /** Only the final error failed, and a skip is not a failed try, so the log records nothing. */ + expect(blockLog.modelFallbacks).toBeUndefined() + for (const registry of [ + ctx.errorResolvedSecretTraceRegistry, + ctx.resolvedSecretTraceRegistry, + ]) { + expect(projectResolvedSecretDiagnosticError(primaryError, registry)).toMatchObject({ + error: 'overloaded after reading {{TOKEN}}', + }) + } + }) + + it('is skipped for a later fallback that starts from the settled registries and answers', async () => { + const blockLog = openLog() + const ctx = contextWithToolSecretCatalog(blockLog) + const fallbackRegistries: Array = [] + mockExecuteProviderRequest.mockImplementationOnce(async (_provider, _request, options) => { + const executionContext = options.executionContext as ExecutionContext + fallbackRegistries.push( + executionContext.errorResolvedSecretTraceRegistry, + executionContext.resolvedSecretTraceRegistry + ) + return providerResponse('claude-sonnet-5', 'from fallback') + }) + + const result = await handler.execute( + ctx, + mockBlock, + inputsWithAttachment([{ model: 'deepseek-chat' }, { model: 'claude-sonnet-5' }]) + ) + + expect(mockExecuteProviderRequest.mock.calls.map(([provider]) => provider)).toEqual([ + 'openai', + 'anthropic', + ]) + expect((result as { model: string }).model).toBe('claude-sonnet-5') + /** Only the primary's tool call activated the secret, so registries from the settled inputs leave it unredacted. */ + expect(fallbackRegistries).toHaveLength(2) + for (const registry of fallbackRegistries) { + expect(projectResolvedSecretDiagnosticError(primaryError, registry)).toMatchObject({ + error: 'overloaded after reading tool-secret', + }) + } + }) + }) + it('skips sim-auto, duplicates, the primary itself, and unusable providers', async () => { mockExecuteProviderRequest .mockRejectedValueOnce(new Error('down')) diff --git a/apps/sim/executor/handlers/agent/handler.ts b/apps/sim/executor/handlers/agent/handler.ts index bf1e876c31f..e5d9e7d50be 100644 --- a/apps/sim/executor/handlers/agent/handler.ts +++ b/apps/sim/executor/handlers/agent/handler.ts @@ -64,7 +64,6 @@ import { validateModelProvider, } from '@/ee/access-control/utils/permission-check' import { AGENT, BlockType, DEFAULTS, stripCustomToolPrefix } from '@/executor/constants' -import { isRetryableBlockError } from '@/executor/execution/block-retry' import { getMemoryMessageAppendKey, getMemoryMessageTurnId, @@ -87,6 +86,7 @@ import { parseResponseFormat } from '@/executor/handlers/shared/response-format' import type { BlockHandler, BlockNodeMetadata, + BlockRetryAttempt, ExecutionContext, StreamingExecution, UserFile, @@ -96,14 +96,13 @@ import { stringifyJSON } from '@/executor/utils/json' import { BLOCK_MODEL_CREDENTIAL_FIELDS, type BlockModelCredentials, - resolveBlockModelCredentials, } from '@/executor/utils/model-credentials' import { - assertFallbackFamilyCredentials, + areFallbacksHeld, getModelFallbacks, PROVIDER_FAMILY_CREDENTIAL_FIELDS, - recordModelFallbacks, resolveFallbackApiKey, + walkModelCandidates, } from '@/executor/utils/model-fallbacks' import { projectResolvedSecretDiagnosticContent } from '@/executor/utils/resolved-secret-content-projection' import { prepareResolvedSecretProjectedInputs } from '@/executor/utils/resolved-secret-input-projection' @@ -196,21 +195,11 @@ function stripAutoPreamble(messages: Message[] | undefined): Message[] | undefin }) } -/** One model in the order the block tries them; the primary carries the block's own key. */ -interface ModelCandidate extends FallbackModelCandidate { - isPrimary: boolean - /** - * What the trace calls this model when it fails. A routed sim-auto primary - * shows as the auto identity, since naming the pool model is the leak that - * `applyAutoModelLabel` exists to close. - */ - traceName?: string -} - interface ExecuteAcrossModelsConfig { - candidates: ModelCandidate[] + primary: { model: string; traceName?: string } + fallbacks: FallbackModelCandidate[] + retry: BlockRetryAttempt | undefined retryPrimaryOnStreamStart: boolean - primaryModel: string /** * The model the builder configured, which is what the editor showed the * per-row tuning fields against. Under sim-auto that is the auto id, not the @@ -604,13 +593,7 @@ export class AgentBlockHandler implements BlockHandler { throw new NonRetryableExecutionError('Evaluation models cannot serve as chat fallbacks') } const retry = nodeMetadata?.retry - const fallbacksHeld = retry !== undefined && !retry.isFinalTry - const fallbackCandidates = - modelInputs.previousInteractionId || fallbacksHeld - ? [] - : configuredFallbacks.filter( - (candidate) => candidate.model.toLowerCase() !== model.toLowerCase() - ) + const fallbacksHeld = areFallbacksHeld(retry) if (configuredFallbacks.length > 0 && modelInputs.previousInteractionId) { logger.info('Fallback models skipped for a deep-research follow-up turn', { blockId: block.id, @@ -618,28 +601,20 @@ export class AgentBlockHandler implements BlockHandler { } else if (configuredFallbacks.length > 0 && fallbacksHeld) { logger.info('Fallback models held for the final try', { blockId: block.id, - attempt: retry.attempt, - maxTries: retry.maxTries, + attempt: retry?.attempt, + maxTries: retry?.maxTries, }) } - const candidates: ModelCandidate[] = [ - { - model, - apiKey: modelInputs.apiKey, - isPrimary: true, - ...(autoRouting ? { traceName: SIM_AUTO_MODEL_ID } : {}), - }, - ...fallbackCandidates.map((candidate) => ({ ...candidate, isPrimary: false })), - ] const { result, servedModel, resultRegistry: servedRegistry, } = await this.executeAcrossModels(ctx, block, { - candidates, + primary: { model, ...(autoRouting ? { traceName: SIM_AUTO_MODEL_ID } : {}) }, + fallbacks: modelInputs.previousInteractionId ? [] : configuredFallbacks, + retry, retryPrimaryOnStreamStart: fallbacksHeld && configuredFallbacks.length > 0 && !modelInputs.previousInteractionId, - primaryModel: model, configuredModel: autoRouting ? SIM_AUTO_MODEL_ID : model, primaryProviderId: providerId, messages: messagesWithInputFiles, @@ -2587,46 +2562,21 @@ export class AgentBlockHandler implements BlockHandler { } /** - * Runs the provider request against each candidate in order until one answers. - * - * Which model serves the request is decided here, inside one handler - * invocation. How many invocations the block gets is the executor's retry - * policy, and the caller keeps the fallbacks out of the candidate list until - * the final try, so with retry on the block runs the primary alone on every - * earlier try and walks the whole chain once. - * - * Falling through is deliberately as indiscriminate as block retry - * (`isRetryableBlockError`): a provider error carries no status, so an - * overloaded upstream cannot be told from any other failure, and a builder who - * lists fallbacks wants the block to answer. Only a stop and an explicitly - * non-retryable failure end the chain early. The abort signal is checked - * before the error itself because `handleExecutionError` rewrites a timeout - * into a plain `Error` without a cause, which the predicate would then treat - * as replayable. + * Runs the Agent's provider request over `walkModelCandidates`, which owns the + * candidate order, the retry hold, skips, stops, saved credentials, and the + * block-log record. What is left here is Agent-only: * - * Messages are built once by the caller, since building them appends to - * memory. Hydration runs per provider, because attachment support and the - * inline budget differ, and is cached so two candidates on one provider do - * not download the same files twice. A fallback whose provider is - * blacklisted, not permitted, or cannot take the attachments is skipped - * rather than counted as a failed try. - * - * The block's saved Vertex or Bedrock credential resolves at most once per - * call and is shared by every candidate on the primary's provider; it is - * awaited inside the attempt, so a failure fails those attempts and other - * providers' fallbacks still run. It never reaches another provider. - * - * A streaming candidate is accepted only once its first chunk has arrived - * (`primeStreamingExecution`), so a startup failure inside the stream still - * falls through; that wait is skipped when no candidate follows, which keeps - * blocks without fallbacks on today's path. - * - * When every candidate fails, the last attempted candidate's error is thrown - * exactly as it escaped `executeProviderRequest`, with the registries that - * call installed for its projection, so error ports and the block-level - * error handling see the shapes they see today. The models that failed are - * written to the block log on every exit, cleared as well as set, because - * block retry reuses one log entry across tries. + * - Hydration runs per provider, because attachment support and the inline + * budget differ, and is cached per provider. A fallback whose provider cannot + * take the attachments is skipped in `prepare`, not counted as a failed try. + * - A streaming candidate is accepted only once its first chunk has arrived + * (`primeStreamingExecution`), so a startup failure inside the stream still + * falls through; the wait is skipped when no candidate follows. + * - `executeProviderRequest` installs a failed attempt's registries, the only + * ones that know secrets a tool call activated. They stay on `ctx` until + * another candidate actually runs, from the settled inputs again, so the + * fallback warn, the rethrown error, and the block-log record project + * against the last failed attempt. */ private async executeAcrossModels( ctx: ExecutionContext, @@ -2639,139 +2589,101 @@ export class AgentBlockHandler implements BlockHandler { }> { const { hydratedByProvider } = config let resultRegistry = config.resultRegistry - const failedModels: string[] = [] - let lastError: unknown - let savedCredentials: Promise | undefined - let lastErrorRegistries: - | { - error: ResolvedSecretTraceRegistry | undefined - resolved: ResolvedSecretTraceRegistry | undefined - } - | undefined - - for (let index = 0; index < config.candidates.length; index++) { - const candidate = config.candidates[index] - const hasNext = index < config.candidates.length - 1 - - /** A run stopped while a candidate was being skipped must not start another. */ - if (!candidate.isPrimary && ctx.abortSignal?.aborted) break - - let candidateProviderId: string - if (candidate.isPrimary) { - candidateProviderId = config.primaryProviderId - } else { - try { - candidateProviderId = getProviderFromModel(candidate.model) - await validateModelProvider(ctx.userId, ctx.workspaceId, candidate.model, ctx) - assertFallbackFamilyCredentials(candidateProviderId, config.primaryProviderId) - } catch (error) { - this.warnFallbackSkipped(ctx, block, candidate.model, 'unusable', error) - continue - } - } + /** The settled inputs, for skips and hydration, leaving the failed attempt's registries on `ctx`. */ + const settledCtx = (): ExecutionContext => ({ + ...ctx, + resolvedSecretTraceRegistry: config.settledInputRegistry, + errorResolvedSecretTraceRegistry: config.providerErrorRegistry, + }) - let messages: Message[] | undefined - if (hydratedByProvider.has(candidateProviderId)) { - messages = hydratedByProvider.get(candidateProviderId) - } else { - try { - messages = await this.hydrateMessageFilesForProvider( - ctx, + return walkModelCandidates({ + ctx, + block, + primary: config.primary, + primaryProviderId: config.primaryProviderId, + fallbacks: config.fallbacks, + retry: config.retry, + credentials: { inputs: config.modelInputs, callerLabel: 'vertex-agent' }, + prepare: async (_candidate, { providerId }) => { + if (hydratedByProvider.has(providerId)) return + hydratedByProvider.set( + providerId, + await this.hydrateMessageFilesForProvider( + settledCtx(), config.messages, - candidateProviderId, + providerId, config.fileProjection.projectedNameByFile, config.fileProjection.modelBoundInputPaths ) - } catch (error) { - if (candidate.isPrimary) throw error - this.warnFallbackSkipped( - ctx, - block, - candidate.model, - 'cannot take the attached files', - error - ) - continue - } - hydratedByProvider.set(candidateProviderId, messages) + ) /** Hydration imported this provider's file provenance; the result fork must carry it. */ resultRegistry = config.settledInputRegistry?.forkForInputPaths([]) - } + }, + attempt: async ( + candidate, + { isPrimary, hasNext, providerId, sameProvider, credentials: savedCredentials } + ) => { + if (!isPrimary) { + ctx.errorResolvedSecretTraceRegistry = config.providerErrorRegistry + ctx.resolvedSecretTraceRegistry = config.settledInputRegistry + } + const messages = hydratedByProvider.get(providerId) - /** - * A fallback's own key applies only while its key field is visible. On the - * primary's provider it reuses the block's key; otherwise the provider layer resolves BYOK or - * the platform key, or reports that a key is required, which counts as - * this candidate failing. A previous interaction id belongs to the primary's - * provider alone. Tuning is re-resolved against the fallback's own - * capabilities so the request is one its provider accepts. - */ - let inputs: AgentInputs = config.modelInputs - if (!candidate.isPrimary) { - const { adjustments, ...tuning } = resolveFallbackTuning( - candidate, - config.configuredModel, - config.modelInputs - ) - const sameProvider = candidateProviderId === config.primaryProviderId - inputs = { - ...(sameProvider - ? config.modelInputs - : omit(config.modelInputs, [ - ...PROVIDER_FAMILY_CREDENTIAL_FIELDS, - ...BLOCK_MODEL_CREDENTIAL_FIELDS, - ])), - apiKey: resolveFallbackApiKey({ + let inputs: AgentInputs = config.modelInputs + if (!isPrimary) { + const { adjustments, ...tuning } = resolveFallbackTuning( candidate, - configuredModel: config.configuredModel, - sameProvider, - primaryApiKey: config.modelInputs.apiKey, - blockId: block.id, - logger, - }), - previousInteractionId: undefined, - ...(config.fallbackSystemPrompt !== undefined - ? { systemPrompt: config.fallbackSystemPrompt } - : {}), - ...tuning, - } - if (adjustments.length > 0) { - logger.info( - 'Fallback model tuning adjusted', - projectAgentDiagnosticMetadata( - ctx, - { blockId: block.id, model: candidate.model, adjustments }, - { blockId: block.id, adjustmentCount: adjustments.length } - ) + config.configuredModel, + config.modelInputs ) + inputs = { + ...(sameProvider + ? config.modelInputs + : omit(config.modelInputs, [ + ...PROVIDER_FAMILY_CREDENTIAL_FIELDS, + ...BLOCK_MODEL_CREDENTIAL_FIELDS, + ])), + apiKey: resolveFallbackApiKey({ + candidate, + configuredModel: config.configuredModel, + sameProvider, + primaryApiKey: config.modelInputs.apiKey, + blockId: block.id, + logger, + }), + /** A previous interaction id belongs to the primary's provider alone. */ + previousInteractionId: undefined, + ...(config.fallbackSystemPrompt !== undefined + ? { systemPrompt: config.fallbackSystemPrompt } + : {}), + ...tuning, + } + if (adjustments.length > 0) { + logger.info( + 'Fallback model tuning adjusted', + projectAgentDiagnosticMetadata( + ctx, + { blockId: block.id, model: candidate.model, adjustments }, + { blockId: block.id, adjustmentCount: adjustments.length } + ) + ) + } } - } - const providerRequest = this.buildProviderRequest({ - ctx, - providerId: candidateProviderId, - model: candidate.model, - messages: - !candidate.isPrimary && config.fallbackSystemPrompt !== undefined - ? stripAutoPreamble(messages) - : messages, - inputs, - formattedTools: config.formattedTools, - responseFormat: config.responseFormat, - streaming: config.streaming, - }) - - const onPrimaryProvider = candidateProviderId === config.primaryProviderId - if (onPrimaryProvider) { - savedCredentials ??= resolveBlockModelCredentials( + const providerRequest = this.buildProviderRequest({ ctx, - config.primaryProviderId, - config.modelInputs, - 'vertex-agent' - ) - } + providerId, + model: candidate.model, + messages: + !isPrimary && config.fallbackSystemPrompt !== undefined + ? stripAutoPreamble(messages) + : messages, + inputs, + formattedTools: config.formattedTools, + responseFormat: config.responseFormat, + streaming: config.streaming, + }) - try { let result = await this.executeProviderRequest( ctx, providerRequest, @@ -2779,33 +2691,27 @@ export class AgentBlockHandler implements BlockHandler { config.responseFormat, resultRegistry, config.providerErrorRegistry, - onPrimaryProvider ? savedCredentials : undefined, + savedCredentials?.(), config.agentConversation ) if ((hasNext || config.retryPrimaryOnStreamStart) && this.isStreamingExecution(result)) { result = await this.primeStreamingExecution(result as StreamingExecution) } - recordModelFallbacks(ctx, block, failedModels) return { result, servedModel: config.agentConversation?.getFinalResponse()?.model ?? candidate.model, resultRegistry, } - } catch (error) { - lastError = error - failedModels.push(candidate.traceName ?? candidate.model) - if (!hasNext || ctx.abortSignal?.aborted || !isRetryableBlockError(error)) { - recordModelFallbacks(ctx, block, failedModels.slice(0, -1)) - throw error - } - - /** - * `executeProviderRequest` installed the failed attempt's error - * registry, which is the only one that knows secrets a tool call - * activated; the warn is projected against it before the next candidate - * starts from the settled inputs again. The pair is kept so a rethrow - * after every remaining candidate was skipped projects the same way. - */ + }, + onSkip: (model, error, stage) => + this.warnFallbackSkipped( + settledCtx(), + block, + model, + stage === 'prepare' ? 'cannot take the attached files' : 'unusable', + error + ), + onFallback: ({ error, failedModel, nextModel, position }) => { const errorRegistry = ctx.errorResolvedSecretTraceRegistry const diagnosticCtx = errorRegistry ? { ...ctx, resolvedSecretTraceRegistry: errorRegistry } @@ -2816,27 +2722,16 @@ export class AgentBlockHandler implements BlockHandler { diagnosticCtx, { blockId: block.id, - failedModel: candidate.model, - nextModel: config.candidates[index + 1].model, - candidate: index + 1, + failedModel, + nextModel, + candidate: position, error: getErrorMessage(error), }, - { blockId: block.id, candidate: index + 1 } + { blockId: block.id, candidate: position } ) ) - lastErrorRegistries = { error: errorRegistry, resolved: ctx.resolvedSecretTraceRegistry } - ctx.errorResolvedSecretTraceRegistry = config.providerErrorRegistry - ctx.resolvedSecretTraceRegistry = config.settledInputRegistry - } - } - - /** Reached only when every candidate after the last failure was skipped. */ - if (lastErrorRegistries) { - ctx.errorResolvedSecretTraceRegistry = lastErrorRegistries.error - ctx.resolvedSecretTraceRegistry = lastErrorRegistries.resolved - } - recordModelFallbacks(ctx, block, failedModels.slice(0, -1)) - throw lastError + }, + }) } /** diff --git a/apps/sim/executor/utils/model-fallback-request.ts b/apps/sim/executor/utils/model-fallback-request.ts index e4687e58c90..04317d69b44 100644 --- a/apps/sim/executor/utils/model-fallback-request.ts +++ b/apps/sim/executor/utils/model-fallback-request.ts @@ -1,27 +1,19 @@ import { createLogger } from '@sim/logger' import { omit } from '@sim/utils/object' import { resolveFallbackTuning } from '@/lib/workflows/blocks/fallback-models' -import { validateModelProvider } from '@/ee/access-control/utils/permission-check' -import { isRetryableBlockError } from '@/executor/execution/block-retry' import type { BlockRetryAttempt, ExecutionContext } from '@/executor/types' +import type { BlockModelCredentialInputs } from '@/executor/utils/model-credentials' import { - type BlockModelCredentialInputs, - type BlockModelCredentials, - resolveBlockModelCredentials, -} from '@/executor/utils/model-credentials' -import { - assertFallbackFamilyCredentials, getModelFallbacks, PROVIDER_FAMILY_CREDENTIAL_FIELDS, - recordModelFallbacks, resolveFallbackApiKey, + walkModelCandidates, } from '@/executor/utils/model-fallbacks' import { executeBlockProviderRequest } from '@/executor/utils/provider-request' import { projectResolvedSecretDiagnosticError } from '@/executor/utils/resolved-secret-content-projection' import type { ResolvedSecretTraceRegistry } from '@/executor/utils/resolved-secret-trace-registry' import { isAutoModel, SIM_AUTO_MODEL_ID } from '@/providers/models' import type { ProviderRequest, ProviderResponse } from '@/providers/types' -import { getProviderFromModel } from '@/providers/utils' import type { SerializedBlock } from '@/serializer/types' const logger = createLogger('BlockModelFallbacks') @@ -42,11 +34,10 @@ interface ModelFallbackRequestInput { } /** - * Shared non-streaming model execution for Router and Evaluator. Block retries - * exhaust the primary first; the final try walks the ordered fallback models. - * Saved credentials resolve inside the attempt, so one that cannot be resolved fails - * that attempt and the fallbacks still run. Only candidates on the primary's provider - * resolve them, once per call, and they never reach another provider. + * Shared non-streaming model execution for Router and Evaluator over + * `walkModelCandidates`. Saved credentials resolve inside the attempt, so one that + * cannot be resolved fails that attempt and the fallbacks still run; they never + * reach another provider. * Parsing a successful response remains the handler's job, so a routing decision * such as NO_MATCH is never silently replaced by another model's decision. */ @@ -62,77 +53,49 @@ export async function executeModelRequestWithFallbacks({ resolvedSecretTraceRegistry, credentials, }: ModelFallbackRequestInput): Promise<{ result: ProviderResponse; usedFallback: boolean }> { - const fallbacks = getModelFallbacks(ctx, block, fallbackModels, logger) - const candidates = [ - { model: request.model }, - ...(retry && !retry.isFinalTry - ? [] - : fallbacks.filter( - (candidate) => candidate.model.toLowerCase() !== request.model.toLowerCase() - )), - ] - const failedModels: string[] = [] - let lastError: unknown - let savedCredentials: Promise | undefined - - for (const [index, candidate] of candidates.entries()) { - const isPrimary = index === 0 - if (!isPrimary && ctx.abortSignal?.aborted) break - let candidateProviderId = providerId - if (!isPrimary) { - try { - candidateProviderId = getProviderFromModel(candidate.model) - await validateModelProvider(ctx.userId, ctx.workspaceId, candidate.model, ctx) - assertFallbackFamilyCredentials(candidateProviderId, providerId) - } catch (error) { - logger.warn( - 'Fallback model unusable; skipping', - projectResolvedSecretDiagnosticError(error, ctx.resolvedSecretTraceRegistry, { - blockId: block.id, - model: candidate.model, - }) - ) - continue - } - } - - let candidateRequest = request - if (!isPrimary) { - const sameProvider = candidateProviderId === providerId - const { adjustments: _adjustments, ...tuning } = resolveFallbackTuning( - candidate, - configuredModel, - request - ) - candidateRequest = { - ...(sameProvider - ? request - : omit(request, [...PROVIDER_FAMILY_CREDENTIAL_FIELDS, 'bedrockSession'])), - ...tuning, - temperature: tuning.temperature === undefined ? undefined : Number(tuning.temperature), - maxTokens: tuning.maxTokens === undefined ? undefined : Number(tuning.maxTokens), - model: candidate.model, - apiKey: resolveFallbackApiKey({ + return walkModelCandidates({ + ctx, + block, + primary: { + model: request.model, + ...(isAutoModel(configuredModel) ? { traceName: SIM_AUTO_MODEL_ID } : {}), + }, + primaryProviderId: providerId, + fallbacks: getModelFallbacks(ctx, block, fallbackModels, logger), + retry, + credentials, + attempt: async ( + candidate, + { isPrimary, providerId: candidateProviderId, sameProvider, credentials: savedCredentials } + ) => { + let candidateRequest = request + if (!isPrimary) { + const { adjustments: _adjustments, ...tuning } = resolveFallbackTuning( candidate, configuredModel, - sameProvider, - primaryApiKey: request.apiKey, - blockId: block.id, - logger, - }), - systemPrompt: fallbackSystemPrompt, - } - } - - try { - if (credentials && candidateProviderId === providerId) { - savedCredentials ??= resolveBlockModelCredentials( - ctx, - providerId, - credentials.inputs, - credentials.callerLabel + request ) - candidateRequest = { ...candidateRequest, ...(await savedCredentials) } + candidateRequest = { + ...(sameProvider + ? request + : omit(request, [...PROVIDER_FAMILY_CREDENTIAL_FIELDS, 'bedrockSession'])), + ...tuning, + temperature: tuning.temperature === undefined ? undefined : Number(tuning.temperature), + maxTokens: tuning.maxTokens === undefined ? undefined : Number(tuning.maxTokens), + model: candidate.model, + apiKey: resolveFallbackApiKey({ + candidate, + configuredModel, + sameProvider, + primaryApiKey: request.apiKey, + blockId: block.id, + logger, + }), + systemPrompt: fallbackSystemPrompt, + } + } + if (savedCredentials) { + candidateRequest = { ...candidateRequest, ...(await savedCredentials()) } } const result = await executeBlockProviderRequest({ ctx, @@ -140,29 +103,23 @@ export async function executeModelRequestWithFallbacks({ request: candidateRequest, resolvedSecretTraceRegistry, }) - recordModelFallbacks(ctx, block, failedModels) return { result, usedFallback: !isPrimary } - } catch (error) { - lastError = error - failedModels.push( - isPrimary && isAutoModel(configuredModel) ? SIM_AUTO_MODEL_ID : candidate.model - ) - if ( - index === candidates.length - 1 || - ctx.abortSignal?.aborted || - !isRetryableBlockError(error) - ) - break + }, + onSkip: (model, error) => + logger.warn( + 'Fallback model unusable; skipping', + projectResolvedSecretDiagnosticError(error, ctx.resolvedSecretTraceRegistry, { + blockId: block.id, + model, + }) + ), + onFallback: ({ error, failedModel }) => logger.warn( 'Model request failed; trying fallback', projectResolvedSecretDiagnosticError(error, ctx.resolvedSecretTraceRegistry, { blockId: block.id, - failedModel: candidate.model, + failedModel, }) - ) - } - } - - recordModelFallbacks(ctx, block, failedModels.slice(0, -1)) - throw lastError + ), + }) } diff --git a/apps/sim/executor/utils/model-fallbacks.test.ts b/apps/sim/executor/utils/model-fallbacks.test.ts new file mode 100644 index 00000000000..ab1cdf2aaba --- /dev/null +++ b/apps/sim/executor/utils/model-fallbacks.test.ts @@ -0,0 +1,66 @@ +import { llmBedrockCredentialMock } from '@sim/testing/mocks/llm-bedrock-credential.mock' +import { + permissionCheckMock, + permissionCheckMockFns, +} from '@sim/testing/mocks/permission-check.mock' +import { providersUtilsMock, providersUtilsMockFns } from '@sim/testing/mocks/providers-utils.mock' +import { beforeEach, describe, expect, it, vi } from 'vitest' +import type { ExecutionContext } from '@/executor/types' +import { walkModelCandidates } from '@/executor/utils/model-fallbacks' +import { ResolvedSecretTraceRegistry } from '@/executor/utils/resolved-secret-trace-registry' +import type { SerializedBlock } from '@/serializer/types' + +vi.mock('@/ee/access-control/utils/permission-check', () => permissionCheckMock) +vi.mock('@/providers/utils', () => providersUtilsMock) +vi.mock('@/lib/internal/llm/bedrock-credential', () => llmBedrockCredentialMock) + +const block = { id: 'block-1', config: { params: {} } } as SerializedBlock + +function context(): ExecutionContext { + return { + userId: 'user-1', + workspaceId: 'workspace-1', + resolvedSecretTraceRegistry: new ResolvedSecretTraceRegistry(), + blockLogs: [{ blockId: 'block-1', endedAt: '' }], + } as ExecutionContext +} + +/** The primary fails, the first fallback cannot be prepared, then `last` decides. */ +function walk(ctx: ExecutionContext, primaryError: Error, last: 'answers' | 'absent') { + return walkModelCandidates({ + ctx, + block, + primary: { model: 'gpt-4o' }, + primaryProviderId: 'openai', + fallbacks: [{ model: 'gpt-4o-mini' }, ...(last === 'answers' ? [{ model: 'gpt-4.1' }] : [])], + prepare: async (candidate) => { + if (candidate.model === 'gpt-4o-mini') throw new Error('cannot take the attachment') + }, + attempt: async (candidate) => { + if (candidate.model === 'gpt-4o') throw primaryError + return candidate.model + }, + onSkip: vi.fn(), + onFallback: vi.fn(), + }) +} + +beforeEach(() => { + providersUtilsMockFns.mockGetProviderFromModel.mockReturnValue('openai') + permissionCheckMockFns.mockValidateModelProvider.mockReset().mockResolvedValue(undefined) +}) + +describe('walkModelCandidates', () => { + it('passes over a candidate that cannot be prepared without recording it as a failed model', async () => { + const ctx = context() + await expect(walk(ctx, new Error('overloaded'), 'answers')).resolves.toBe('gpt-4.1') + expect(ctx.blockLogs[0].modelFallbacks).toEqual(['gpt-4o']) + }) + + it('rethrows the last failure when every remaining candidate is skipped', async () => { + const ctx = context() + const primaryError = new Error('overloaded') + await expect(walk(ctx, primaryError, 'absent')).rejects.toBe(primaryError) + expect(ctx.blockLogs[0].modelFallbacks).toBeUndefined() + }) +}) diff --git a/apps/sim/executor/utils/model-fallbacks.ts b/apps/sim/executor/utils/model-fallbacks.ts index 6bdcc3e3936..a33bb762aba 100644 --- a/apps/sim/executor/utils/model-fallbacks.ts +++ b/apps/sim/executor/utils/model-fallbacks.ts @@ -7,8 +7,16 @@ import { normalizeFallbackModels, } from '@/lib/workflows/blocks/fallback-models' import { providerRequiresFamilyCredentials } from '@/blocks/utils' -import type { ExecutionContext } from '@/executor/types' +import { validateModelProvider } from '@/ee/access-control/utils/permission-check' +import { isRetryableBlockError } from '@/executor/execution/block-retry' +import type { BlockRetryAttempt, ExecutionContext } from '@/executor/types' +import { + type BlockModelCredentialInputs, + type BlockModelCredentials, + resolveBlockModelCredentials, +} from '@/executor/utils/model-credentials' import { projectResolvedSecretDiagnosticContent } from '@/executor/utils/resolved-secret-content-projection' +import { getProviderFromModel } from '@/providers/utils' import type { SerializedBlock } from '@/serializer/types' type Logger = ReturnType @@ -28,7 +36,7 @@ export const PROVIDER_FAMILY_CREDENTIAL_FIELDS = [ * Refuses a fallback on another provider that needs a Vertex credential, Bedrock keys, or an * Azure endpoint: those fields exist only for the primary's provider family. */ -export function assertFallbackFamilyCredentials( +function assertFallbackFamilyCredentials( candidateProviderId: string, primaryProviderId: string ): void { @@ -100,7 +108,7 @@ export function resolveFallbackApiKey({ } /** Records failed candidates on the current attempt's log, projecting secret-derived model names. */ -export function recordModelFallbacks( +function recordModelFallbacks( ctx: ExecutionContext, block: SerializedBlock, failedModels: string[] @@ -124,3 +132,170 @@ export function recordModelFallbacks( return } } + +/** + * Retry on fail retries the selected model; the fallbacks join only on the try + * after which the executor promises no other. + */ +export function areFallbacksHeld(retry: BlockRetryAttempt | undefined): boolean { + return retry !== undefined && !retry.isFinalTry +} + +/** What the walk tells each attempt about the candidate it is running. */ +interface ModelCandidateAttempt { + isPrimary: boolean + hasNext: boolean + providerId: string + /** Whether the candidate runs on the primary's provider, and so may reuse its key and family credentials. */ + sameProvider: boolean + /** + * The block's saved Vertex or Bedrock credentials, resolved at most once per walk and + * shared by every candidate on the primary's provider. Absent on any other provider. + */ + credentials?: () => Promise +} + +interface ModelCandidateWalk { + ctx: ExecutionContext + block: SerializedBlock + /** + * The configured primary. `traceName` is what the block log calls it when it fails: + * a routed Auto primary shows as the Auto identity, never as the pool model. + */ + primary: { model: string; traceName?: string } + primaryProviderId: string + /** Validated fallback rows (`getModelFallbacks`), before the hold and de-duplication. */ + fallbacks: FallbackModelCandidate[] + retry?: BlockRetryAttempt + /** The block's saved credential selection, resolved only for candidates on the primary's provider. */ + credentials?: { inputs: BlockModelCredentialInputs; callerLabel: string } + /** + * Readies a fallback that passed the provider checks; a throw means it cannot take this + * request, and it is skipped like an unusable fallback. + */ + prepare?: (candidate: FallbackModelCandidate, info: ModelCandidateAttempt) => Promise + /** Builds and sends one candidate's request; a throw is that candidate failing. */ + attempt: (candidate: FallbackModelCandidate, info: ModelCandidateAttempt) => Promise + /** + * A fallback passed over without counting as a failed try: its provider is unknown, not + * permitted, or needs another family's credentials (`unusable`), or `prepare` threw. + */ + onSkip: (model: string, error: unknown, stage: 'unusable' | 'prepare') => void + /** A retryable failure with another candidate after it; `position` is the failed candidate's 1-based place. */ + onFallback: (failure: { + error: unknown + failedModel: string + nextModel: string + position: number + }) => void +} + +/** + * Runs a block's model request against the primary and then each fallback until one + * answers. Retries exhaust the primary first; the fallbacks, de-duplicated against the + * primary, join only on the final try. A fallback that cannot be used is skipped, not + * counted as a failure. The chain stops at the last candidate, on a stop, or on a + * non-retryable error, and then rethrows the last candidate's error unchanged. + * + * Falling through is deliberately as indiscriminate as block retry + * (`isRetryableBlockError`): a provider error carries no status, so an overloaded + * upstream cannot be told from any other failure, and a builder who lists fallbacks + * wants the block to answer. Only a stop and an explicitly non-retryable failure end + * the chain early. + * + * The models that failed are written to the block log on every exit, cleared as well + * as set, because block retry reuses one log entry across tries. + */ +export async function walkModelCandidates({ + ctx, + block, + primary, + primaryProviderId, + fallbacks, + retry, + credentials, + prepare, + attempt, + onSkip, + onFallback, +}: ModelCandidateWalk): Promise { + const primaryModel = primary.model.toLowerCase() + const candidates: FallbackModelCandidate[] = [ + primary, + ...(areFallbacksHeld(retry) + ? [] + : fallbacks.filter((candidate) => candidate.model.toLowerCase() !== primaryModel)), + ] + let savedCredentials: Promise | undefined + const resolveSavedCredentials = credentials + ? () => + (savedCredentials ??= resolveBlockModelCredentials( + ctx, + primaryProviderId, + credentials.inputs, + credentials.callerLabel + )) + : undefined + const failedModels: string[] = [] + let lastError: unknown + + for (const [index, candidate] of candidates.entries()) { + const isPrimary = index === 0 + const hasNext = index < candidates.length - 1 + /** A run stopped while a candidate was being skipped must not start another. */ + if (!isPrimary && ctx.abortSignal?.aborted) break + + let providerId = primaryProviderId + if (!isPrimary) { + try { + providerId = getProviderFromModel(candidate.model) + await validateModelProvider(ctx.userId, ctx.workspaceId, candidate.model, ctx) + assertFallbackFamilyCredentials(providerId, primaryProviderId) + } catch (error) { + onSkip(candidate.model, error, 'unusable') + continue + } + } + + const sameProvider = providerId === primaryProviderId + const info: ModelCandidateAttempt = { + isPrimary, + hasNext, + providerId, + sameProvider, + credentials: sameProvider ? resolveSavedCredentials : undefined, + } + if (!isPrimary && prepare) { + try { + await prepare(candidate, info) + } catch (error) { + onSkip(candidate.model, error, 'prepare') + continue + } + } + + try { + const result = await attempt(candidate, info) + recordModelFallbacks(ctx, block, failedModels) + return result + } catch (error) { + lastError = error + failedModels.push(isPrimary ? (primary.traceName ?? primary.model) : candidate.model) + /** + * The abort signal is checked before the error itself because a handler may + * rewrite a timeout into a plain `Error` without a cause, which the predicate + * would then treat as replayable. + */ + if (!hasNext || ctx.abortSignal?.aborted || !isRetryableBlockError(error)) break + onFallback({ + error, + failedModel: candidate.model, + nextModel: candidates[index + 1].model, + position: index + 1, + }) + } + } + + recordModelFallbacks(ctx, block, failedModels.slice(0, -1)) + throw lastError +} diff --git a/scripts/check-test-types.baseline.json b/scripts/check-test-types.baseline.json index b4defe1f4e3..94d455e53de 100644 --- a/scripts/check-test-types.baseline.json +++ b/scripts/check-test-types.baseline.json @@ -795,7 +795,7 @@ }, "apps/sim/executor/handlers/agent/handler.test.ts": { "TS18046": 2, - "TS2339": 3, + "TS2339": 1, "TS2345": 1, "TS2352": 1, "TS7006": 1