diff --git a/packages/cli/src/modules/agents/tools/__tests__/workflow-tool-factory.test.ts b/packages/cli/src/modules/agents/tools/__tests__/workflow-tool-factory.test.ts index f435c3fa899..cee906a33d7 100644 --- a/packages/cli/src/modules/agents/tools/__tests__/workflow-tool-factory.test.ts +++ b/packages/cli/src/modules/agents/tools/__tests__/workflow-tool-factory.test.ts @@ -3,7 +3,7 @@ import { N8N_CHAT_INTEGRATION_TYPE } from '@n8n/api-types'; import { GlobalConfig } from '@n8n/config'; import type { WorkflowEntity } from '@n8n/db'; import { Container } from '@n8n/di'; -import type { IDeferredPromise } from '@n8n/utils/promise/deferred-promise'; +import { createDeferredPromise, type IDeferredPromise } from '@n8n/utils/promise/deferred-promise'; import { sleep } from '@n8n/utils/sleep'; import type { IExecuteResponsePromiseData, @@ -16,6 +16,7 @@ import { mock } from 'vitest-mock-extended'; import type { ActiveExecutions } from '@/active-executions'; import { ExecutionPersistence } from '@/executions/execution-persistence'; +import { ExecutionService } from '@/executions/execution.service'; import type { SubworkflowPolicyChecker } from '@/executions/pre-execution-checks'; import type { EphemeralNodeExecutor } from '@/node-execution/ephemeral-node-executor'; import { WebhookResponseRelay } from '@/scaling/webhook-response-relay'; @@ -98,6 +99,96 @@ describe('executeWorkflow → execution classification', () => { }, ); + it('passes a foreground abort to workflow stopping and releases the wait', async () => { + const controller = new AbortController(); + const started = createDeferredPromise(); + const finished = createDeferredPromise(); + const stop = vi.fn().mockResolvedValue(undefined); + Container.set(ExecutionService, mock({ stop })); + const pending = executeWorkflow( + workflow, + triggerNode, + {}, + { + ...buildContext(vi.fn().mockResolvedValue('exec-1'), { + activeExecutions: mock({ + has: () => true, + getPostExecutePromise: async () => { + started.resolve(undefined); + return await finished.promise; + }, + }), + }), + abortSignal: controller.signal, + }, + ); + await started.promise; + controller.abort(new Error('Response stopped')); + await expect(pending).rejects.toThrow('Response stopped'); + expect(stop).toHaveBeenCalledExactlyOnceWith('exec-1', ['wf-1']); + finished.resolve(mock({ status: 'canceled' })); + }); + + it.each([false, true])( + 'waits for workflow stopping when its callback rejects (stop fails: %s)', + async (stopFails) => { + const controller = new AbortController(); + const started = createDeferredPromise(); + const finished = createDeferredPromise(); + const stopped = createDeferredPromise(); + const stop = vi.fn().mockReturnValue(stopped.promise); + Container.set(ExecutionService, mock({ stop })); + let settled = false; + const outcome = executeWorkflow( + workflow, + triggerNode, + {}, + { + ...buildContext(vi.fn().mockResolvedValue('exec-1'), { + activeExecutions: mock({ + has: () => true, + getPostExecutePromise: async () => { + started.resolve(undefined); + return await finished.promise; + }, + }), + }), + abortSignal: controller.signal, + }, + ).catch((error: unknown) => { + settled = true; + return error; + }); + await started.promise; + const reason = new Error('Response stopped'); + controller.abort(reason); + finished.reject(reason); + await vi.dynamicImportSettled(); + expect(stop).toHaveBeenCalledExactlyOnceWith('exec-1', ['wf-1']); + expect(settled).toBe(false); + + const stopError = new Error('Workflow stop failed'); + if (stopFails) stopped.reject(stopError); + else stopped.resolve(undefined); + expect(await outcome).toBe(stopFails ? stopError : reason); + }, + ); + + it('does not cancel a workflow after the foreground tool has returned', async () => { + const controller = new AbortController(); + const stop = vi.fn(); + Container.set(ExecutionService, mock({ stop })); + await executeWorkflow( + workflow, + triggerNode, + {}, + { ...buildContext(vi.fn().mockResolvedValue('exec-1')), abortSignal: controller.signal }, + ); + controller.abort(); + await vi.dynamicImportSettled(); + expect(stop).not.toHaveBeenCalled(); + }); + it('checks the caller policy before starting the workflow', async () => { const run = vi.fn().mockResolvedValue('exec-1'); const subworkflowPolicyChecker = mock(); @@ -958,6 +1049,35 @@ describe('workflow tool → background job handoff', () => { expect(findSingleExecution).toHaveBeenCalledTimes(1); }); + it('stops the workflow when the foreground run is aborted during registration', async () => { + setPersistence({ status: 'canceled' }); + const jobService = setJobService(); + const controller = new AbortController(); + const registering = createDeferredPromise(); + const registered = createDeferredPromise<{ status: 'started'; jobId: string }>(); + jobService.registerWorkflowJob.mockImplementation(async () => { + registering.resolve(undefined); + return await registered.promise; + }); + const stop = vi.fn().mockResolvedValue(undefined); + Container.set(ExecutionService, mock({ stop })); + const tool = await buildBackgroundTool(); + const { ctx, suspend } = makeParentCtx(); + const pending = tool.handler?.({}, { + ...(ctx as object), + abortSignal: controller.signal, + } as never); + await registering.promise; + + controller.abort(new Error('Response stopped')); + await expect(pending).rejects.toThrow('Response stopped'); + expect(stop).toHaveBeenCalledExactlyOnceWith('exec-1', ['wf-1']); + + registered.resolve({ status: 'started', jobId: 'job-1' }); + await vi.dynamicImportSettled(); + expect(suspend).not.toHaveBeenCalled(); + }); + it('polls a wait due soon to completion instead of backgrounding it', async () => { setPersistence(settledInDb()); const jobService = setJobService(); diff --git a/packages/cli/src/modules/agents/tools/workflow-tool-factory.ts b/packages/cli/src/modules/agents/tools/workflow-tool-factory.ts index 80e26174214..0c8f69aa722 100644 --- a/packages/cli/src/modules/agents/tools/workflow-tool-factory.ts +++ b/packages/cli/src/modules/agents/tools/workflow-tool-factory.ts @@ -33,6 +33,7 @@ import { EXECUTE_WORKFLOW_TRIGGER_NODE_TYPE, WORKFLOW_TOOL_LANGCHAIN_NODE_TYPE, TimeoutExecutionCancelledError, + OperationalError, isIndefiniteWait, } from 'n8n-workflow'; import { v4 as uuid } from 'uuid'; @@ -182,6 +183,7 @@ export interface WorkflowToolContext { /** {@link WorkflowToolContext} plus fields that only exist once a run does. */ export interface WorkflowToolRunContext extends WorkflowToolContext { + abortSignal?: AbortSignal; /** Stamped onto sub-executions so a Wait node finishing can wake this run. */ agentRun?: RelatedAgentRun; } @@ -463,7 +465,9 @@ export async function executeWorkflow( ): Promise { const { workflowRunner, subworkflowPolicyChecker, activeExecutions } = context; + context.abortSignal?.throwIfAborted(); await subworkflowPolicyChecker.checkForProject(workflow, context.projectId); + context.abortSignal?.throwIfAborted(); // Build pin data for the trigger const triggerPinData: IPinData = { @@ -524,60 +528,104 @@ export async function executeWorkflow( responsePromise, ); - // Wait for completion with timeout protection - const timeoutMs = DEFAULT_TIMEOUT_MS; + return await withWorkflowCancellation(executionId, workflow.id, context.abortSignal, async () => { + // Wait for completion with timeout protection + const timeoutMs = DEFAULT_TIMEOUT_MS; - let completedRun: IRun | undefined; - if (activeExecutions.has(executionId)) { - let timeoutId: NodeJS.Timeout | undefined; - const timeoutPromise = new Promise((_, reject) => { - timeoutId = setTimeout(() => { - reject(new Error(`Execution timed out after ${timeoutMs}ms`)); - }, timeoutMs); - }); + let completedRun: IRun | undefined; + if (activeExecutions.has(executionId)) { + let timeoutId: NodeJS.Timeout | undefined; + const timeoutPromise = new Promise((_, reject) => { + timeoutId = setTimeout(() => { + reject(new Error(`Execution timed out after ${timeoutMs}ms`)); + }, timeoutMs); + }); - try { - completedRun = await Promise.race([ - activeExecutions.getPostExecutePromise(executionId), - timeoutPromise, - ]); - clearTimeout(timeoutId); - } catch (error) { - clearTimeout(timeoutId); - if (error instanceof Error && error.message.includes('timed out')) { - try { - activeExecutions.stopExecution( + try { + completedRun = await Promise.race([ + activeExecutions.getPostExecutePromise(executionId), + timeoutPromise, + ]); + clearTimeout(timeoutId); + } catch (error) { + clearTimeout(timeoutId); + if (error instanceof Error && error.message.includes('timed out')) { + try { + activeExecutions.stopExecution( + executionId, + new TimeoutExecutionCancelledError(executionId), + ); + } catch { + // Execution may have completed between timeout and cancel + } + return { executionId, - new TimeoutExecutionCancelledError(executionId), - ); - } catch { - // Execution may have completed between timeout and cancel + status: 'error', + error: `Execution timed out after ${timeoutMs}ms and was cancelled`, + }; } - return { - executionId, - status: 'error', - error: `Execution timed out after ${timeoutMs}ms and was cancelled`, - }; + throw error; } - throw error; } - } - const result = - completedRun && context.executionMode === 'integrated' - ? formatResult(executionId, completedRun.status, completedRun.data, allOutputs) - : await extractResult(executionId, allOutputs); - if (isWorkflowToolResponse(webhookResponse)) { - const response = await Container.get(WebhookResponseRelay).restoreOffloadedBody( - webhookResponse, - { reclaim: true, context: { workflowId: workflow.id, executionId } }, - ); - result.data = { - ...(result.data ?? {}), - response: truncateWebhookResponse(response), - }; + const result = + completedRun && context.executionMode === 'integrated' + ? formatResult(executionId, completedRun.status, completedRun.data, allOutputs) + : await extractResult(executionId, allOutputs); + if (isWorkflowToolResponse(webhookResponse)) { + const response = await Container.get(WebhookResponseRelay).restoreOffloadedBody( + webhookResponse, + { reclaim: true, context: { workflowId: workflow.id, executionId } }, + ); + result.data = { + ...(result.data ?? {}), + response: truncateWebhookResponse(response), + }; + } + return result; + }); +} + +/** The foreground tool owns cancellation until it hands the workflow to a background job. */ +async function withWorkflowCancellation( + executionId: string, + workflowId: string | undefined, + signal: AbortSignal | undefined, + execute: () => Promise, +): Promise { + if (!signal) return await execute(); + const aborted = createDeferredPromise(); + const onAbort = () => { + void (async () => { + try { + const { ExecutionService } = await import('@/executions/execution.service.js'); + const id = + workflowId ?? + (await Container.get(ExecutionPersistence).findSingleExecution(executionId))?.workflowId; + if (!id) throw new OperationalError('The workflow execution is no longer available'); + await Container.get(ExecutionService).stop(executionId, [id]); + aborted.reject(signal.reason); + } catch (error) { + aborted.reject(error); + } + })(); + }; + signal.addEventListener('abort', onAbort, { once: true }); + try { + if (signal.aborted) { + onAbort(); + return await aborted.promise; + } + const result = await Promise.race([execute(), aborted.promise]); + if (signal.aborted) return await aborted.promise; + return result; + } catch (error) { + // The callback can observe the abort before workflow stopping completes. + if (signal.aborted) return await aborted.promise; + throw error; + } finally { + signal.removeEventListener('abort', onAbort); } - return result; } // --------------------------------------------------------------------------- @@ -1111,7 +1159,11 @@ function assembleWorkflowTool( current.workflow, current.triggerNode, parsedInput, - { ...context, agentRun: agentRunOf(context, ctx) }, + { + ...context, + agentRun: agentRunOf(context, ctx), + abortSignal: ctx.abortSignal, + }, allOutputs, toolName, ); @@ -1129,27 +1181,37 @@ function assembleWorkflowTool( // A wait due within the poll window resolves inline. Only a wait the poll // did not settle is handed to the background. - if (result.status === 'waiting') { - result = await pollIfDueSoon(result, allOutputs, ctx.abortSignal); - } + if (result.status !== 'waiting') return withoutWaitState(result); - if (result.status === 'waiting' && context.backgroundTasksEnabled) { - const backgrounded = await backgroundWaitingExecution( - result, - context, - ctx, - reference, - allOutputs, - ); - if (backgrounded) return backgrounded; - } + const waiting = result; + return await withWorkflowCancellation( + waiting.executionId, + reference.workflowId, + ctx.abortSignal, + async () => { + result = await pollIfDueSoon(waiting, allOutputs, ctx.abortSignal); + ctx.abortSignal?.throwIfAborted(); + if (result.status === 'waiting' && context.backgroundTasksEnabled) { + const backgrounded = await backgroundWaitingExecution( + result, + context, + ctx, + reference, + allOutputs, + ); + ctx.abortSignal?.throwIfAborted(); + if (backgrounded) return backgrounded; + } - if (result.status !== 'waiting' || !supportsHitl) return withoutWaitState(result); + if (result.status !== 'waiting' || !supportsHitl) return withoutWaitState(result); - current ??= await loadCurrentWorkflow(context, reference); - return await ctx.suspend(buildWaitCard(current.workflow.name, result.wait), { - continuation: { executionId: result.executionId }, - }); + current ??= await loadCurrentWorkflow(context, reference); + ctx.abortSignal?.throwIfAborted(); + return await ctx.suspend(buildWaitCard(current.workflow.name, result.wait), { + continuation: { executionId: result.executionId }, + }); + }, + ); }); const built = builder.build();