From 267fd853cdc1669e869d280eae687eb6cad89abe Mon Sep 17 00:00:00 2001 From: Riqwan Thamir Date: Mon, 20 Jul 2026 18:40:44 +0200 Subject: [PATCH] fix(core): Cancel Instance AI workspace tools promptly on stop (#34551) Co-authored-by: Cursor Co-authored-by: Robin Braumann <50590409+bjorger@users.noreply.github.com> --- .../workspace/base-filesystem.test.ts | 11 +- .../workspace/sandbox/daytona-sandbox.test.ts | 28 ++++ .../src/__tests__/workspace/test-utils.ts | 12 +- .../workspace/workspace-tools.test.ts | 80 +++++++-- packages/@n8n/agents/src/index.ts | 9 ++ .../agents/src/runtime/mcp/mcp-connection.ts | 7 +- .../src/runtime/mcp/mcp-tool-resolver.ts | 6 +- .../src/runtime/memory/episodic-memory.ts | 3 +- .../src/runtime/tools/tool-call-executor.ts | 25 ++- .../agents/src/sdk/__tests__/abort.test.ts | 92 +++++++++++ packages/@n8n/agents/src/sdk/abort.ts | 58 +++++++ .../workspace/filesystem/base-filesystem.ts | 11 +- .../filesystem/daytona-filesystem.ts | 52 +++--- .../filesystem/n8n-sandbox-filesystem.ts | 152 ++++++++++-------- packages/@n8n/agents/src/workspace/index.ts | 3 + .../src/workspace/sandbox/base-sandbox.ts | 20 ++- .../src/workspace/sandbox/daytona-sandbox.ts | 75 +++++---- .../agents/src/workspace/sandbox/index.ts | 6 +- .../workspace/sandbox/n8n-sandbox-sandbox.ts | 2 +- .../src/workspace/sandbox/run-in-sandbox.ts | 29 +++- .../agents/src/workspace/tools/append-file.ts | 6 +- .../workspace/tools/batch-str-replace-file.ts | 14 +- .../agents/src/workspace/tools/copy-file.ts | 7 +- .../agents/src/workspace/tools/delete-file.ts | 8 +- .../src/workspace/tools/execute-command.ts | 3 +- .../agents/src/workspace/tools/file-stat.ts | 6 +- .../agents/src/workspace/tools/list-files.ts | 7 +- .../@n8n/agents/src/workspace/tools/mkdir.ts | 7 +- .../agents/src/workspace/tools/move-file.ts | 7 +- .../src/workspace/tools/process-tools.ts | 12 +- .../agents/src/workspace/tools/read-file.ts | 3 +- .../@n8n/agents/src/workspace/tools/rmdir.ts | 8 +- .../src/workspace/tools/str-replace-file.ts | 14 +- .../agents/src/workspace/tools/write-file.ts | 7 +- packages/@n8n/agents/src/workspace/types.ts | 30 ++-- .../src/agent/mcp-tool-name-validation.ts | 3 +- .../@n8n/instance-ai/src/tool-registry.ts | 7 +- .../instance-ai/src/tools/executions.tool.ts | 4 +- .../create-tools-from-mcp-server.test.ts | 33 ++-- .../create-tools-from-mcp-server.ts | 16 +- .../instance-ai/src/tools/n8n-docs.tool.ts | 3 +- .../src/tools/n8n-docs/registry.ts | 3 +- .../verify-built-workflow.tool.ts | 1 + .../shared/__tests__/abortable-tool.test.ts | 114 ------------- .../src/tools/shared/abortable-tool.ts | 84 ---------- .../__tests__/build-workflow.tool.test.ts | 4 +- .../workflow-source-compiler.test.ts | 37 ++++- .../__tests__/write-sandbox-file.tool.test.ts | 3 + .../tools/workflows/build-workflow.tool.ts | 18 ++- .../workflows/materialize-node-type.tool.ts | 6 +- .../tools/workflows/workflow-file-bindings.ts | 2 + .../workflows/workflow-source-compiler.ts | 10 +- .../workflows/write-sandbox-file.tool.ts | 6 +- packages/@n8n/instance-ai/src/types.ts | 6 +- .../__tests__/lazy-runtime-workspace.test.ts | 21 +++ .../src/workspace/lazy-runtime-workspace.ts | 47 +++--- .../instance-ai/src/workspace/sandbox-fs.ts | 49 ++++-- .../src/workspace/scoped-workspace.ts | 17 +- .../src/workspace/workspace-files.ts | 20 ++- .../composite-local-mcp-server.test.ts | 14 +- .../instance-ai.adapter.service.test.ts | 44 ++++- .../__tests__/local-gateway.test.ts | 48 ++++++ .../browser/composite-local-mcp-server.ts | 7 +- .../instance-ai/filesystem/local-gateway.ts | 61 ++++++- .../instance-ai.adapter.service.ts | 46 +++++- packages/workflow/tsconfig.json | 1 + 66 files changed, 1049 insertions(+), 506 deletions(-) create mode 100644 packages/@n8n/agents/src/sdk/__tests__/abort.test.ts create mode 100644 packages/@n8n/agents/src/sdk/abort.ts delete mode 100644 packages/@n8n/instance-ai/src/tools/shared/__tests__/abortable-tool.test.ts delete mode 100644 packages/@n8n/instance-ai/src/tools/shared/abortable-tool.ts diff --git a/packages/@n8n/agents/src/__tests__/workspace/base-filesystem.test.ts b/packages/@n8n/agents/src/__tests__/workspace/base-filesystem.test.ts index 1cb71d4143f..6151f220013 100644 --- a/packages/@n8n/agents/src/__tests__/workspace/base-filesystem.test.ts +++ b/packages/@n8n/agents/src/__tests__/workspace/base-filesystem.test.ts @@ -1,9 +1,12 @@ import { BaseFilesystem } from '../../workspace/filesystem/base-filesystem'; import type { BaseFilesystemOptions } from '../../workspace/filesystem/base-filesystem'; import type { + AbortableOptions, + AppendOptions, FileContent, FileStat, FileEntry, + MkdirOptions, ReadOptions, WriteOptions, ListOptions, @@ -43,7 +46,7 @@ class TestFilesystem extends BaseFilesystem { await this.ensureReady(); } - async appendFile(_path: string, _content: FileContent): Promise { + async appendFile(_path: string, _content: FileContent, _options?: AppendOptions): Promise { await this.ensureReady(); } @@ -59,7 +62,7 @@ class TestFilesystem extends BaseFilesystem { await this.ensureReady(); } - async mkdir(_path: string, _options?: { recursive?: boolean }): Promise { + async mkdir(_path: string, _options?: MkdirOptions): Promise { await this.ensureReady(); } @@ -72,12 +75,12 @@ class TestFilesystem extends BaseFilesystem { return []; } - async exists(_path: string): Promise { + async exists(_path: string, _options?: AbortableOptions): Promise { await this.ensureReady(); return false; } - async stat(_path: string): Promise { + async stat(_path: string, _options?: AbortableOptions): Promise { await this.ensureReady(); return { name: 'test', diff --git a/packages/@n8n/agents/src/__tests__/workspace/sandbox/daytona-sandbox.test.ts b/packages/@n8n/agents/src/__tests__/workspace/sandbox/daytona-sandbox.test.ts index a8f496741a0..24e98b4d957 100644 --- a/packages/@n8n/agents/src/__tests__/workspace/sandbox/daytona-sandbox.test.ts +++ b/packages/@n8n/agents/src/__tests__/workspace/sandbox/daytona-sandbox.test.ts @@ -538,6 +538,34 @@ describe('DaytonaSandbox (remote sandbox gone during refetch)', () => { await expect(sandbox.executeCommand('echo', ['hi'])).rejects.toThrow(/create failed/i); }); + it('executeCommand() rejects without starting work when already aborted', async () => { + const sandbox = new DaytonaSandbox({ name: 'thread-1', apiKey: 'key' }); + const controller = new AbortController(); + controller.abort(); + + await expect( + sandbox.executeCommand('echo', ['hi'], { abortSignal: controller.signal }), + ).rejects.toMatchObject({ name: 'AbortError' }); + expect(clientLog).toHaveLength(0); + }); + + it('executeCommand() does not recover from AbortError', async () => { + const failing = makeMockSandbox('sb-abort', 'started'); + const abortError = new Error('This operation was aborted'); + abortError.name = 'AbortError'; + failing.process.executeCommand = vi.fn().mockRejectedValue(abortError); + queuedGetResults.push(failing); + + const sandbox = new DaytonaSandbox({ name: 'thread-1', apiKey: 'key' }); + + await expect(sandbox.executeCommand('echo', ['hi'])).rejects.toMatchObject({ + name: 'AbortError', + }); + expect(clientLog.every((c) => c.create.mock.calls.length === 0)).toBe(true); + // Only the initial start get — no isRecoverable probe get after AbortError. + expect(clientLog.reduce((n, c) => n + c.get.mock.calls.length, 0)).toBe(1); + }); + it('DaytonaFilesystem reuses the same recovery when the remote was deleted', async () => { const sandbox = await startAndStageRemoteGone(); // findExistingSandbox() lookup also 404s during recovery → create a fresh sandbox. diff --git a/packages/@n8n/agents/src/__tests__/workspace/test-utils.ts b/packages/@n8n/agents/src/__tests__/workspace/test-utils.ts index 3fb980333ce..ba03aac105e 100644 --- a/packages/@n8n/agents/src/__tests__/workspace/test-utils.ts +++ b/packages/@n8n/agents/src/__tests__/workspace/test-utils.ts @@ -2,6 +2,8 @@ import { BaseFilesystem } from '../../workspace/filesystem/base-filesystem'; import { BaseSandbox } from '../../workspace/sandbox/base-sandbox'; import { ProcessHandle, SandboxProcessManager } from '../../workspace/types'; import type { + AbortableOptions, + AppendOptions, CommandResult, FileContent, FileEntry, @@ -73,7 +75,11 @@ export class InMemoryFilesystem extends BaseFilesystem { this.files.set(p, Buffer.from(content)); } - async appendFile(filePath: string, content: FileContent): Promise { + async appendFile( + filePath: string, + content: FileContent, + _options?: AppendOptions, + ): Promise { await this.ensureReady(); const p = this.normalizePath(filePath); const existing = this.files.get(p) ?? Buffer.alloc(0); @@ -169,13 +175,13 @@ export class InMemoryFilesystem extends BaseFilesystem { return entries; } - async exists(filePath: string): Promise { + async exists(filePath: string, _options?: AbortableOptions): Promise { await this.ensureReady(); const p = this.normalizePath(filePath); return this.files.has(p) || this.dirs.has(p); } - async stat(filePath: string): Promise { + async stat(filePath: string, _options?: AbortableOptions): Promise { await this.ensureReady(); const p = this.normalizePath(filePath); if (this.dirs.has(p)) { diff --git a/packages/@n8n/agents/src/__tests__/workspace/workspace-tools.test.ts b/packages/@n8n/agents/src/__tests__/workspace/workspace-tools.test.ts index 2b74bd827a0..9795102bc31 100644 --- a/packages/@n8n/agents/src/__tests__/workspace/workspace-tools.test.ts +++ b/packages/@n8n/agents/src/__tests__/workspace/workspace-tools.test.ts @@ -114,7 +114,10 @@ describe('createWorkspaceTools', () => { const result = await readTool.handler!({ path: '/test.txt', encoding: 'utf-8' }, {} as never); - expect(fs.readFile).toHaveBeenCalledWith('/test.txt', { encoding: 'utf-8' }); + expect(fs.readFile).toHaveBeenCalledWith('/test.txt', { + encoding: 'utf-8', + abortSignal: undefined, + }); expect(result).toEqual({ content: 'file content' }); }); @@ -147,6 +150,7 @@ describe('createWorkspaceTools', () => { expect(fs.writeFile).toHaveBeenCalledWith('/test.txt', 'first\nchanged', { overwrite: true, + abortSignal: undefined, }); expect(result).toEqual({ success: true, result: 'Edit applied successfully.' }); }); @@ -174,6 +178,28 @@ describe('createWorkspaceTools', () => { }); }); + it('str_replace_file handler rethrows abort errors instead of soft-failing', async () => { + const abortError = new Error('This operation was aborted'); + abortError.name = 'AbortError'; + const fs = makeFakeFilesystem({ + readFile: vi.fn().mockRejectedValue(abortError), + }); + const tools = createWorkspaceTools({ filesystem: fs }); + const strReplaceTool = tools.find((t) => t.name === 'workspace_str_replace_file')!; + + await expect( + strReplaceTool.handler!( + { + path: '/test.txt', + old_str: 'a', + new_str: 'b', + }, + {} as never, + ), + ).rejects.toMatchObject({ name: 'AbortError' }); + expect(fs.writeFile).not.toHaveBeenCalled(); + }); + it('batch_str_replace_file handler applies all replacements atomically', async () => { const fs = makeFakeFilesystem({ readFile: vi.fn().mockResolvedValue('const a = 1;\nconst b = 2;'), @@ -194,6 +220,7 @@ describe('createWorkspaceTools', () => { expect(fs.writeFile).toHaveBeenCalledWith('/test.ts', 'const a = 10;\nconst b = 20;', { overwrite: true, + abortSignal: undefined, }); expect(result).toEqual({ success: true, @@ -240,13 +267,17 @@ describe('createWorkspaceTools', () => { const fs = makeFakeFilesystem(); const tools = createWorkspaceTools({ filesystem: fs }); const writeTool = tools.find((t) => t.name === 'workspace_write_file')!; + const abortController = new AbortController(); const result = await writeTool.handler!( { path: '/out.txt', content: 'hello', recursive: true }, - {} as never, + { abortSignal: abortController.signal } as never, ); - expect(fs.writeFile).toHaveBeenCalledWith('/out.txt', 'hello', { recursive: true }); + expect(fs.writeFile).toHaveBeenCalledWith('/out.txt', 'hello', { + recursive: true, + abortSignal: abortController.signal, + }); expect(result).toEqual({ success: true }); }); @@ -264,16 +295,18 @@ describe('createWorkspaceTools', () => { }); const tools = createWorkspaceTools({ sandbox }); const commandTool = tools.find((t) => t.name === 'workspace_execute_command')!; + const abortController = new AbortController(); const result = await commandTool.handler!( { command: 'node script.mjs', cwd: '/home/daytona/workspace' }, - {} as never, + { abortSignal: abortController.signal } as never, ); expect(executeCommand).toHaveBeenCalledWith('node script.mjs', undefined, { cwd: '/home/daytona/workspace', env: { CUSTOM_ENV: 'enabled' }, timeout: undefined, + abortSignal: abortController.signal, }); expect(result).toMatchObject({ success: true, stdout: 'ok' }); }); @@ -285,7 +318,10 @@ describe('createWorkspaceTools', () => { const result = await listTool.handler!({ path: '/', recursive: false }, {} as never); - expect(fs.readdir).toHaveBeenCalledWith('/', { recursive: false }); + expect(fs.readdir).toHaveBeenCalledWith('/', { + recursive: false, + abortSignal: undefined, + }); expect(result).toEqual({ entries: [ { name: 'file1.txt', type: 'file' }, @@ -301,7 +337,7 @@ describe('createWorkspaceTools', () => { const result = await statTool.handler!({ path: '/test.txt' }, {} as never); - expect(fs.stat).toHaveBeenCalledWith('/test.txt'); + expect(fs.stat).toHaveBeenCalledWith('/test.txt', { abortSignal: undefined }); expect(result).toEqual({ name: 'test.txt', path: '/test.txt', @@ -319,7 +355,10 @@ describe('createWorkspaceTools', () => { const result = await mkdirTool.handler!({ path: '/new-dir', recursive: true }, {} as never); - expect(fs.mkdir).toHaveBeenCalledWith('/new-dir', { recursive: true }); + expect(fs.mkdir).toHaveBeenCalledWith('/new-dir', { + recursive: true, + abortSignal: undefined, + }); expect(result).toEqual({ success: true }); }); @@ -333,7 +372,11 @@ describe('createWorkspaceTools', () => { {} as never, ); - expect(fs.deleteFile).toHaveBeenCalledWith('/old.txt', { recursive: false, force: true }); + expect(fs.deleteFile).toHaveBeenCalledWith('/old.txt', { + recursive: false, + force: true, + abortSignal: undefined, + }); expect(result).toEqual({ success: true }); }); @@ -347,7 +390,9 @@ describe('createWorkspaceTools', () => { {} as never, ); - expect(fs.appendFile).toHaveBeenCalledWith('/log.txt', 'new line'); + expect(fs.appendFile).toHaveBeenCalledWith('/log.txt', 'new line', { + abortSignal: undefined, + }); expect(result).toEqual({ success: true }); }); @@ -361,7 +406,10 @@ describe('createWorkspaceTools', () => { {} as never, ); - expect(fs.copyFile).toHaveBeenCalledWith('/a.txt', '/b.txt', { overwrite: true }); + expect(fs.copyFile).toHaveBeenCalledWith('/a.txt', '/b.txt', { + overwrite: true, + abortSignal: undefined, + }); expect(result).toEqual({ success: true }); }); @@ -375,7 +423,10 @@ describe('createWorkspaceTools', () => { {} as never, ); - expect(fs.moveFile).toHaveBeenCalledWith('/old.txt', '/new.txt', { overwrite: false }); + expect(fs.moveFile).toHaveBeenCalledWith('/old.txt', '/new.txt', { + overwrite: false, + abortSignal: undefined, + }); expect(result).toEqual({ success: true }); }); @@ -389,7 +440,11 @@ describe('createWorkspaceTools', () => { {} as never, ); - expect(fs.rmdir).toHaveBeenCalledWith('/old-dir', { recursive: true, force: false }); + expect(fs.rmdir).toHaveBeenCalledWith('/old-dir', { + recursive: true, + force: false, + abortSignal: undefined, + }); expect(result).toEqual({ success: true }); }); @@ -406,6 +461,7 @@ describe('createWorkspaceTools', () => { expect(sb.executeCommand).toHaveBeenCalledWith('echo hello', undefined, { cwd: '/tmp', timeout: 5000, + abortSignal: undefined, }); expect(result).toEqual({ success: true, diff --git a/packages/@n8n/agents/src/index.ts b/packages/@n8n/agents/src/index.ts index c2ec7103fd8..ac178263d84 100644 --- a/packages/@n8n/agents/src/index.ts +++ b/packages/@n8n/agents/src/index.ts @@ -107,6 +107,12 @@ export { export { createCancellation, isCancellation, CANCELLATION_TYPE } from './sdk/cancellation'; export type { Cancellation } from './sdk/cancellation'; +export { + createAbortError, + isAbortError, + raceWithAbort, + throwIfAborted, +} from './sdk/abort'; export { Tool, wrapToolForApproval, sanitizeToolName } from './sdk/tool'; export { Memory } from './sdk/memory'; export { VectorStore } from './sdk/vector-store'; @@ -396,11 +402,14 @@ export type { FileContent, FileStat, FileEntry, + AbortableOptions, + AppendOptions, ReadOptions, WriteOptions, ListOptions, RemoveOptions, CopyOptions, + MkdirOptions, ProviderStatus, SandboxInfo, LocalFilesystemOptions, diff --git a/packages/@n8n/agents/src/runtime/mcp/mcp-connection.ts b/packages/@n8n/agents/src/runtime/mcp/mcp-connection.ts index 80eafa26dda..f3b81ae6331 100644 --- a/packages/@n8n/agents/src/runtime/mcp/mcp-connection.ts +++ b/packages/@n8n/agents/src/runtime/mcp/mcp-connection.ts @@ -180,13 +180,18 @@ export class McpConnection { return false; } - async callTool(name: string, args: Record): Promise { + async callTool( + name: string, + args: Record, + options?: { abortSignal?: AbortSignal }, + ): Promise { if (!this.client) throw new Error('MCP client not initialized; connect() must be called first'); const { CallToolResultSchema } = await loadMcpSdk(); try { const result = (await this.client.callTool( { name, arguments: args }, CallToolResultSchema, + options?.abortSignal ? { signal: options.abortSignal } : undefined, )) as McpCallToolResult; await this.notifyToolCallSettled({ toolName: name, success: result.isError !== true }); return result; diff --git a/packages/@n8n/agents/src/runtime/mcp/mcp-tool-resolver.ts b/packages/@n8n/agents/src/runtime/mcp/mcp-tool-resolver.ts index 7f92a28e75e..6ec14c73143 100644 --- a/packages/@n8n/agents/src/runtime/mcp/mcp-tool-resolver.ts +++ b/packages/@n8n/agents/src/runtime/mcp/mcp-tool-resolver.ts @@ -23,10 +23,12 @@ export class McpToolResolver { const handler = async ( input: unknown, - _ctx: ToolContext | InterruptibleToolContext, + ctx: ToolContext | InterruptibleToolContext, ): Promise => { const args = (input ?? {}) as Record; - return await connection.callTool(originalName, args); + return await connection.callTool(originalName, args, { + abortSignal: ctx.abortSignal, + }); }; const toMessage = (output: unknown): AgentMessage | undefined => { diff --git a/packages/@n8n/agents/src/runtime/memory/episodic-memory.ts b/packages/@n8n/agents/src/runtime/memory/episodic-memory.ts index a0791c05223..b88ca308de0 100644 --- a/packages/@n8n/agents/src/runtime/memory/episodic-memory.ts +++ b/packages/@n8n/agents/src/runtime/memory/episodic-memory.ts @@ -188,11 +188,12 @@ export function createRecallMemoryTool(opts: { .systemInstruction(normalized.recallToolInstruction) .input(RecallMemoryInputSchema) .output(RecallMemoryOutputSchema) - .handler(async ({ query }): Promise => { + .handler(async ({ query }, ctx): Promise => { const { embed } = await import('ai'); const { embedding: queryEmbedding, usage } = await embed({ model: normalized.embedder, value: query, + abortSignal: ctx.abortSignal, }); incrementTokenCountFromUsage(opts.executionCounter, usage); const entries = await opts.memory.episodic.searchEntries(opts.scope, query, { diff --git a/packages/@n8n/agents/src/runtime/tools/tool-call-executor.ts b/packages/@n8n/agents/src/runtime/tools/tool-call-executor.ts index 1ade9121aa3..06a5643882c 100644 --- a/packages/@n8n/agents/src/runtime/tools/tool-call-executor.ts +++ b/packages/@n8n/agents/src/runtime/tools/tool-call-executor.ts @@ -7,6 +7,7 @@ import { import { toJsonValue } from '../json-value'; import { DEFAULT_SUB_AGENT_MAX_CHILDREN } from './sub-agent-task-path'; import { executeTool, isSuspendedToolResult, type SuspendedToolResult } from './tool-adapter'; +import { isAbortError, raceWithAbort } from '../../sdk/abort'; import { isCancellation } from '../../sdk/cancellation'; import { isLlmMessage } from '../../sdk/message'; import type { @@ -145,12 +146,6 @@ function isDeniedApprovalResumeData(value: unknown): boolean { return value !== null && typeof value === 'object' && Reflect.get(value, 'approved') === false; } -function isAbortError(error: unknown): boolean { - if (!(error instanceof Error)) return false; - if (error.name === 'AbortError') return true; - return error.message === 'Aborted' || error.message === 'This operation was aborted'; -} - function shouldEmitToolExecutionStart(tool: BuiltTool, resumeData: unknown): boolean { if (!tool.approval) return true; if (!tool.approval.required && tool.approval.conditional !== true) return true; @@ -800,14 +795,18 @@ export class ToolCallExecutor { input, resolvedTelemetry, async () => - await executeTool(input, builtTool, resumeData, resolvedTelemetry, toolCallId, { - runId, - persistence, - emitEvent: (event) => this.eventBus.emit(event), + await raceWithAbort( + async () => + await executeTool(input, builtTool, resumeData, resolvedTelemetry, toolCallId, { + runId, + persistence, + emitEvent: (event) => this.eventBus.emit(event), + abortSignal, + executionCounter, + suspendPayload, + }), abortSignal, - executionCounter, - suspendPayload, - }), + ), ); } diff --git a/packages/@n8n/agents/src/sdk/__tests__/abort.test.ts b/packages/@n8n/agents/src/sdk/__tests__/abort.test.ts new file mode 100644 index 00000000000..860eca8ed36 --- /dev/null +++ b/packages/@n8n/agents/src/sdk/__tests__/abort.test.ts @@ -0,0 +1,92 @@ +import { describe, expect, it, vi } from 'vitest'; + +import { createAbortError, isAbortError, raceWithAbort, throwIfAborted } from '../abort'; + +describe('abort helpers', () => { + describe('isAbortError', () => { + it('detects AbortError by name', () => { + const error = new Error('stopped'); + error.name = 'AbortError'; + expect(isAbortError(error)).toBe(true); + }); + + it('detects known abort messages', () => { + expect(isAbortError(new Error('Aborted'))).toBe(true); + expect(isAbortError(new Error('This operation was aborted'))).toBe(true); + }); + + it('rejects unrelated errors', () => { + expect(isAbortError(new Error('disk full'))).toBe(false); + expect(isAbortError('Aborted')).toBe(false); + }); + }); + + describe('throwIfAborted', () => { + it('throws when the signal is already aborted', () => { + const controller = new AbortController(); + controller.abort(); + expect(() => throwIfAborted(controller.signal)).toThrowError( + expect.objectContaining({ name: 'AbortError' }), + ); + }); + + it('no-ops when the signal is still open', () => { + expect(() => throwIfAborted(new AbortController().signal)).not.toThrow(); + expect(() => throwIfAborted(undefined)).not.toThrow(); + }); + }); + + describe('raceWithAbort', () => { + it('resolves the work promise when no signal is provided', async () => { + await expect(raceWithAbort(Promise.resolve('ok'))).resolves.toBe('ok'); + }); + + it('rejects immediately when the signal is already aborted', async () => { + const controller = new AbortController(); + controller.abort(); + + await expect( + raceWithAbort(new Promise(() => undefined), controller.signal), + ).rejects.toMatchObject({ name: 'AbortError' }); + }); + + it('rejects promptly when the signal aborts during execution', async () => { + const controller = new AbortController(); + const pending = raceWithAbort(new Promise(() => undefined), controller.signal); + controller.abort('Agent run was aborted'); + + await expect(pending).rejects.toMatchObject({ + name: 'AbortError', + message: 'Agent run was aborted', + }); + }); + + it('removes the abort listener when work wins the race', async () => { + const controller = new AbortController(); + const removeSpy = vi.spyOn(controller.signal, 'removeEventListener'); + + for (let i = 0; i < 12; i++) { + await expect(raceWithAbort(Promise.resolve(i), controller.signal)).resolves.toBe(i); + } + + expect(removeSpy).toHaveBeenCalledTimes(12); + expect(removeSpy.mock.calls.every(([type]) => type === 'abort')).toBe(true); + }); + + it('does not start factory work when the signal is already aborted', async () => { + const controller = new AbortController(); + controller.abort(); + const work = vi.fn().mockResolvedValue('started'); + + await expect(raceWithAbort(work, controller.signal)).rejects.toMatchObject({ + name: 'AbortError', + }); + expect(work).not.toHaveBeenCalled(); + }); + + it('returns createAbortError for string reasons', () => { + const error = createAbortError('stopped by user'); + expect(error).toMatchObject({ name: 'AbortError', message: 'stopped by user' }); + }); + }); +}); diff --git a/packages/@n8n/agents/src/sdk/abort.ts b/packages/@n8n/agents/src/sdk/abort.ts new file mode 100644 index 00000000000..dd518a7d6dc --- /dev/null +++ b/packages/@n8n/agents/src/sdk/abort.ts @@ -0,0 +1,58 @@ +/** + * Abort helpers for agent runs and long-running tool / sandbox work. + * Stop should unblock the executor even when underlying I/O does not cancel. + */ + +export function isAbortError(error: unknown): boolean { + if (!(error instanceof Error)) return false; + if (error.name === 'AbortError') return true; + return error.message === 'Aborted' || error.message === 'This operation was aborted'; +} + +export function createAbortError(reason?: unknown): Error { + if (reason instanceof Error) return reason; + const error = new Error(typeof reason === 'string' ? reason : 'This operation was aborted'); + error.name = 'AbortError'; + return error; +} + +/** Throw if the given signal has already fired. */ +export function throwIfAborted(signal?: AbortSignal): void { + if (signal?.aborted) { + throw createAbortError(signal.reason); + } +} + +/** + * Race work against an abort signal so Stop settles promptly even when the + * underlying work ignores cancellation. Pass a factory when work must not start + * until after the abort check (e.g. sandbox recover/retry). Cooperative callers + * should still forward `abortSignal` into I/O where the provider supports it. + * + * The abort listener is always removed when the race settles so run-scoped + * signals do not accumulate listeners across nested tool calls. + */ +export async function raceWithAbort( + work: Promise | (() => Promise), + signal?: AbortSignal, +): Promise { + const run = typeof work === 'function' ? work : async () => await work; + if (!signal) { + return await run(); + } + throwIfAborted(signal); + + let onAbort!: () => void; + const rejection = new Promise((_, reject) => { + onAbort = () => { + reject(createAbortError(signal.reason)); + }; + signal.addEventListener('abort', onAbort, { once: true }); + }); + + try { + return await Promise.race([run(), rejection]); + } finally { + signal.removeEventListener('abort', onAbort); + } +} diff --git a/packages/@n8n/agents/src/workspace/filesystem/base-filesystem.ts b/packages/@n8n/agents/src/workspace/filesystem/base-filesystem.ts index ac5b63eede7..aea6e8ff1c8 100644 --- a/packages/@n8n/agents/src/workspace/filesystem/base-filesystem.ts +++ b/packages/@n8n/agents/src/workspace/filesystem/base-filesystem.ts @@ -4,11 +4,14 @@ import type { FileContent, FileStat, FileEntry, + AbortableOptions, + AppendOptions, ReadOptions, WriteOptions, ListOptions, RemoveOptions, CopyOptions, + MkdirOptions, } from '../types'; export type FilesystemLifecycleHook = (args: { @@ -139,13 +142,13 @@ export abstract class BaseFilesystem implements WorkspaceFilesystem { abstract readFile(path: string, options?: ReadOptions): Promise; abstract writeFile(path: string, content: FileContent, options?: WriteOptions): Promise; - abstract appendFile(path: string, content: FileContent): Promise; + abstract appendFile(path: string, content: FileContent, options?: AppendOptions): Promise; abstract deleteFile(path: string, options?: RemoveOptions): Promise; abstract copyFile(src: string, dest: string, options?: CopyOptions): Promise; abstract moveFile(src: string, dest: string, options?: CopyOptions): Promise; - abstract mkdir(path: string, options?: { recursive?: boolean }): Promise; + abstract mkdir(path: string, options?: MkdirOptions): Promise; abstract rmdir(path: string, options?: RemoveOptions): Promise; abstract readdir(path: string, options?: ListOptions): Promise; - abstract exists(path: string): Promise; - abstract stat(path: string): Promise; + abstract exists(path: string, options?: AbortableOptions): Promise; + abstract stat(path: string, options?: AbortableOptions): Promise; } diff --git a/packages/@n8n/agents/src/workspace/filesystem/daytona-filesystem.ts b/packages/@n8n/agents/src/workspace/filesystem/daytona-filesystem.ts index 3b1b6e99d06..c2518cbd83d 100644 --- a/packages/@n8n/agents/src/workspace/filesystem/daytona-filesystem.ts +++ b/packages/@n8n/agents/src/workspace/filesystem/daytona-filesystem.ts @@ -8,11 +8,14 @@ * Without this adapter, Daytona workspaces only get sandbox tools (execute_command). */ import type { + AbortableOptions, + AppendOptions, CopyOptions, FileContent, FileEntry, FileStat, ListOptions, + MkdirOptions, ProviderStatus, ReadOptions, RemoveOptions, @@ -43,9 +46,12 @@ export class DaytonaFilesystem extends BaseFilesystem { * sandbox is running with fresh auth and recovers once if the remote was stopped or * deleted while idle, so callers never touch a stale `fs` handle directly. */ - private async withFs(op: (fs: DaytonaFsHandle) => Promise): Promise { + private async withFs( + op: (fs: DaytonaFsHandle) => Promise, + abortSignal?: AbortSignal, + ): Promise { await this.ensureReady(); - return await this.sandbox.withFilesystem(op); + return await this.sandbox.withFilesystem(op, { abortSignal }); } async readFile(path: string, options?: ReadOptions): Promise { @@ -55,7 +61,7 @@ export class DaytonaFilesystem extends BaseFilesystem { return buffer.toString(options.encoding); } return buffer; - }); + }, options?.abortSignal); } async writeFile(path: string, content: FileContent, options?: WriteOptions): Promise { @@ -69,10 +75,10 @@ export class DaytonaFilesystem extends BaseFilesystem { const buffer = typeof content === 'string' ? Buffer.from(content, 'utf-8') : Buffer.from(content); await fs.uploadFile(buffer, path); - }); + }, options?.abortSignal); } - async appendFile(path: string, content: FileContent): Promise { + async appendFile(path: string, content: FileContent, options?: AppendOptions): Promise { await this.withFs(async (fs) => { let existing: Buffer; try { @@ -86,34 +92,40 @@ export class DaytonaFilesystem extends BaseFilesystem { const append = typeof content === 'string' ? Buffer.from(content, 'utf-8') : Buffer.from(content); await fs.uploadFile(Buffer.concat([existing, append]), path); - }); + }, options?.abortSignal); } async deleteFile(path: string, options?: RemoveOptions): Promise { - await this.withFs(async (fs) => await fs.deleteFile(path, options?.recursive)); + await this.withFs( + async (fs) => await fs.deleteFile(path, options?.recursive), + options?.abortSignal, + ); } - async copyFile(src: string, dest: string, _options?: CopyOptions): Promise { + async copyFile(src: string, dest: string, options?: CopyOptions): Promise { await this.withFs(async (fs) => { const content = await fs.downloadFile(src); await fs.uploadFile(content, dest); - }); + }, options?.abortSignal); } - async moveFile(src: string, dest: string, _options?: CopyOptions): Promise { - await this.withFs(async (fs) => await fs.moveFiles(src, dest)); + async moveFile(src: string, dest: string, options?: CopyOptions): Promise { + await this.withFs(async (fs) => await fs.moveFiles(src, dest), options?.abortSignal); } - async mkdir(path: string, _options?: { recursive?: boolean }): Promise { + async mkdir(path: string, options?: MkdirOptions): Promise { // createFolder with mode '755' creates intermediate dirs - await this.withFs(async (fs) => await fs.createFolder(path, '755')); + await this.withFs(async (fs) => await fs.createFolder(path, '755'), options?.abortSignal); } async rmdir(path: string, options?: RemoveOptions): Promise { - await this.withFs(async (fs) => await fs.deleteFile(path, options?.recursive ?? false)); + await this.withFs( + async (fs) => await fs.deleteFile(path, options?.recursive ?? false), + options?.abortSignal, + ); } - async readdir(path: string, _options?: ListOptions): Promise { + async readdir(path: string, options?: ListOptions): Promise { return await this.withFs(async (fs) => { const files = await fs.listFiles(path); return files.map((f) => ({ @@ -121,10 +133,10 @@ export class DaytonaFilesystem extends BaseFilesystem { type: f.isDir ? ('directory' as const) : ('file' as const), size: f.size, })); - }); + }, options?.abortSignal); } - async exists(path: string): Promise { + async exists(path: string, options?: AbortableOptions): Promise { return await this.withFs(async (fs) => { try { await fs.getFileDetails(path); @@ -136,10 +148,10 @@ export class DaytonaFilesystem extends BaseFilesystem { if (isDaytona404(error)) return false; throw error; } - }); + }, options?.abortSignal); } - async stat(path: string): Promise { + async stat(path: string, options?: AbortableOptions): Promise { return await this.withFs(async (fs) => { let info; try { @@ -158,7 +170,7 @@ export class DaytonaFilesystem extends BaseFilesystem { createdAt: new Date(info.modTime ?? 0), modifiedAt: new Date(info.modTime ?? 0), }; - }); + }, options?.abortSignal); } } diff --git a/packages/@n8n/agents/src/workspace/filesystem/n8n-sandbox-filesystem.ts b/packages/@n8n/agents/src/workspace/filesystem/n8n-sandbox-filesystem.ts index 8fc412e2e1f..66e7c61adab 100644 --- a/packages/@n8n/agents/src/workspace/filesystem/n8n-sandbox-filesystem.ts +++ b/packages/@n8n/agents/src/workspace/filesystem/n8n-sandbox-filesystem.ts @@ -1,12 +1,16 @@ import { SandboxServiceError } from '@n8n/sandbox-client'; import { dirname } from 'node:path/posix'; +import { raceWithAbort } from '../../sdk/abort'; import type { + AbortableOptions, + AppendOptions, CopyOptions, FileContent, FileEntry, FileStat, ListOptions, + MkdirOptions, ProviderStatus, ReadOptions, RemoveOptions, @@ -35,98 +39,110 @@ export class N8nSandboxFilesystem extends BaseFilesystem { this.id = `n8n-sandbox-fs-${sandbox.id}`; } - private async getClientAndSandboxId() { - await this.sandbox.ensureRunning(); + private async getClientAndSandboxId(abortSignal?: AbortSignal) { + await this.sandbox.ensureRunning({ abortSignal }); return { client: this.sandbox.getClient(), sandboxId: this.sandbox.id, }; } - async readFile(path: string, options?: ReadOptions): Promise { + private async withSandbox( + abortSignal: AbortSignal | undefined, + op: ( + client: ReturnType, + sandboxId: string, + ) => Promise, + ): Promise { await this.ensureReady(); - const { client, sandboxId } = await this.getClientAndSandboxId(); - const content = await client.readFile(sandboxId, path); - if (options?.encoding) { - return content.toString(options.encoding); - } - return content; + return await raceWithAbort(async () => { + const { client, sandboxId } = await this.getClientAndSandboxId(abortSignal); + return await op(client, sandboxId); + }, abortSignal); + } + + async readFile(path: string, options?: ReadOptions): Promise { + return await this.withSandbox(options?.abortSignal, async (client, sandboxId) => { + const content = await client.readFile(sandboxId, path); + if (options?.encoding) { + return content.toString(options.encoding); + } + return content; + }); } async writeFile(path: string, content: FileContent, options?: WriteOptions): Promise { - await this.ensureReady(); - const { client, sandboxId } = await this.getClientAndSandboxId(); - if (options?.recursive) { - const parent = getParentDirectory(path); - if (parent) { - await client.mkdir(sandboxId, parent, true); + await this.withSandbox(options?.abortSignal, async (client, sandboxId) => { + if (options?.recursive) { + const parent = getParentDirectory(path); + if (parent) { + await client.mkdir(sandboxId, parent, true); + } } - } - await client.writeFile(sandboxId, path, content, options?.overwrite ?? true); + await client.writeFile(sandboxId, path, content, options?.overwrite ?? true); + }); } - async appendFile(path: string, content: FileContent): Promise { - await this.ensureReady(); - const { client, sandboxId } = await this.getClientAndSandboxId(); - await client.appendFile(sandboxId, path, content); + async appendFile(path: string, content: FileContent, options?: AppendOptions): Promise { + await this.withSandbox(options?.abortSignal, async (client, sandboxId) => { + await client.appendFile(sandboxId, path, content); + }); } async deleteFile(path: string, options?: RemoveOptions): Promise { - await this.ensureReady(); - const { client, sandboxId } = await this.getClientAndSandboxId(); - await client.deleteFile(sandboxId, path, { - recursive: options?.recursive, - force: options?.force, + await this.withSandbox(options?.abortSignal, async (client, sandboxId) => { + await client.deleteFile(sandboxId, path, { + recursive: options?.recursive, + force: options?.force, + }); }); } async copyFile(src: string, dest: string, options?: CopyOptions): Promise { - await this.ensureReady(); - const { client, sandboxId } = await this.getClientAndSandboxId(); - await client.copyFile(sandboxId, { - src, - dest, - recursive: options?.recursive, - overwrite: options?.overwrite, + await this.withSandbox(options?.abortSignal, async (client, sandboxId) => { + await client.copyFile(sandboxId, { + src, + dest, + recursive: options?.recursive, + overwrite: options?.overwrite, + }); }); } async moveFile(src: string, dest: string, options?: CopyOptions): Promise { - await this.ensureReady(); - const { client, sandboxId } = await this.getClientAndSandboxId(); - await client.moveFile(sandboxId, { - src, - dest, - overwrite: options?.overwrite, + await this.withSandbox(options?.abortSignal, async (client, sandboxId) => { + await client.moveFile(sandboxId, { + src, + dest, + overwrite: options?.overwrite, + }); }); } - async mkdir(path: string, options?: { recursive?: boolean }): Promise { - await this.ensureReady(); - const { client, sandboxId } = await this.getClientAndSandboxId(); - await client.mkdir(sandboxId, path, options?.recursive ?? false); + async mkdir(path: string, options?: MkdirOptions): Promise { + await this.withSandbox(options?.abortSignal, async (client, sandboxId) => { + await client.mkdir(sandboxId, path, options?.recursive ?? false); + }); } async rmdir(path: string, options?: RemoveOptions): Promise { await this.deleteFile(path, options); } - async readdir(path: string, _options?: ListOptions): Promise { - await this.ensureReady(); - const { client, sandboxId } = await this.getClientAndSandboxId(); - const files = await client.listFiles(sandboxId, { path }); - return files.map((entry) => ({ - name: entry.name, - type: entry.isDir ? 'directory' : 'file', - size: entry.size, - })); + async readdir(path: string, options?: ListOptions): Promise { + return await this.withSandbox(options?.abortSignal, async (client, sandboxId) => { + const files = await client.listFiles(sandboxId, { path }); + return files.map((entry) => ({ + name: entry.name, + type: entry.isDir ? 'directory' : 'file', + size: entry.size, + })); + }); } - async exists(path: string): Promise { - await this.ensureReady(); + async exists(path: string, options?: AbortableOptions): Promise { try { - const { client, sandboxId } = await this.getClientAndSandboxId(); - await client.stat(sandboxId, path); + await this.stat(path, options); return true; } catch (error) { if (error instanceof SandboxServiceError && error.status === 404) { @@ -136,17 +152,17 @@ export class N8nSandboxFilesystem extends BaseFilesystem { } } - async stat(path: string): Promise { - await this.ensureReady(); - const { client, sandboxId } = await this.getClientAndSandboxId(); - const stat = await client.stat(sandboxId, path); - return { - name: stat.name, - path: stat.path, - type: stat.type, - size: stat.size, - createdAt: new Date(stat.createdAt), - modifiedAt: new Date(stat.modifiedAt), - }; + async stat(path: string, options?: AbortableOptions): Promise { + return await this.withSandbox(options?.abortSignal, async (client, sandboxId) => { + const stat = await client.stat(sandboxId, path); + return { + name: stat.name, + path: stat.path, + type: stat.type, + size: stat.size, + createdAt: new Date(stat.createdAt), + modifiedAt: new Date(stat.modifiedAt), + }; + }); } } diff --git a/packages/@n8n/agents/src/workspace/index.ts b/packages/@n8n/agents/src/workspace/index.ts index de2e7b2a1e5..5105c23b3ab 100644 --- a/packages/@n8n/agents/src/workspace/index.ts +++ b/packages/@n8n/agents/src/workspace/index.ts @@ -19,11 +19,14 @@ export type { FileContent, FileStat, FileEntry, + AbortableOptions, + AppendOptions, ReadOptions, WriteOptions, ListOptions, RemoveOptions, CopyOptions, + MkdirOptions, ProviderStatus, SandboxInfo, LocalFilesystemOptions, diff --git a/packages/@n8n/agents/src/workspace/sandbox/base-sandbox.ts b/packages/@n8n/agents/src/workspace/sandbox/base-sandbox.ts index e2f6585954a..526a05f9ae9 100644 --- a/packages/@n8n/agents/src/workspace/sandbox/base-sandbox.ts +++ b/packages/@n8n/agents/src/workspace/sandbox/base-sandbox.ts @@ -1,4 +1,6 @@ +import { raceWithAbort } from '../../sdk/abort'; import type { + AbortableOptions, ProviderStatus, WorkspaceSandbox, BaseSandboxOptions, @@ -118,7 +120,7 @@ export abstract class BaseSandbox implements WorkspaceSandbox { this.status = 'pending'; } - async ensureRunning(): Promise { + async ensureRunning(options?: AbortableOptions): Promise { if (this.status === 'destroyed') { throw new Error(`Sandbox "${this.name}" has been destroyed`); } @@ -130,7 +132,7 @@ export abstract class BaseSandbox implements WorkspaceSandbox { if (this.stopPromise) await this.stopPromise.catch(() => {}); } if (this.status !== 'running') { - await this._start(); + await raceWithAbort(async () => await this._start(), options?.abortSignal); } if (this.status !== 'running') { throw new Error(`Sandbox "${this.name}" failed to start (status: ${this.status})`); @@ -142,16 +144,20 @@ export abstract class BaseSandbox implements WorkspaceSandbox { args?: string[], options?: ExecuteCommandOptions, ): Promise { - await this.ensureRunning(); + await this.ensureRunning({ abortSignal: options?.abortSignal }); if (!this.processes) { throw new Error(`Sandbox "${this.name}" has no process manager`); } const fullCommand = args?.length ? `${command} ${args.map(shellQuote).join(' ')}` : command; const handle = await this.processes.spawn(fullCommand, options); - return await handle.wait({ - onStdout: options?.onStdout, - onStderr: options?.onStderr, - }); + return await raceWithAbort( + async () => + await handle.wait({ + onStdout: options?.onStdout, + onStderr: options?.onStderr, + }), + options?.abortSignal, + ); } getInstructions(): string { diff --git a/packages/@n8n/agents/src/workspace/sandbox/daytona-sandbox.ts b/packages/@n8n/agents/src/workspace/sandbox/daytona-sandbox.ts index a219297760c..97324651135 100644 --- a/packages/@n8n/agents/src/workspace/sandbox/daytona-sandbox.ts +++ b/packages/@n8n/agents/src/workspace/sandbox/daytona-sandbox.ts @@ -10,7 +10,14 @@ import type { } from '@daytona/sdk'; import { randomUUID } from 'node:crypto'; -import type { CommandResult, ExecuteCommandOptions, ProviderStatus, SandboxInfo } from '../types'; +import { isAbortError, raceWithAbort } from '../../sdk/abort'; +import type { + AbortableOptions, + CommandResult, + ExecuteCommandOptions, + ProviderStatus, + SandboxInfo, +} from '../types'; import { BaseSandbox } from './base-sandbox'; import { DaytonaAuthManager } from './daytona-auth-manager'; import { loadDaytona } from './lazy-daytona'; @@ -195,30 +202,32 @@ export class DaytonaSandbox extends BaseSandbox { args: string[] = [], options?: ExecuteCommandOptions, ): Promise { - return await this.recoverAndRetry(async () => { - await this.ensureRunning(); - await this.ensureAuthFresh(); - const startedAt = Date.now(); - const fullCommand = toShellCommand(command, args); - const result = await this.instance.process.executeCommand( - fullCommand, - options?.cwd, - this.compactEnv(options?.env), - Math.ceil((options?.timeout ?? this.timeout) / 1000), - ); - const stdout = result.artifacts?.stdout ?? result.result ?? ''; - if (stdout) options?.onStdout?.(stdout); + return await raceWithAbort(async () => { + return await this.recoverAndRetry(async () => { + await this.ensureRunning({ abortSignal: options?.abortSignal }); + await this.ensureAuthFresh(); + const startedAt = Date.now(); + const fullCommand = toShellCommand(command, args); + const result = await this.instance.process.executeCommand( + fullCommand, + options?.cwd, + this.compactEnv(options?.env), + Math.ceil((options?.timeout ?? this.timeout) / 1000), + ); + const stdout = result.artifacts?.stdout ?? result.result ?? ''; + if (stdout) options?.onStdout?.(stdout); - return { - command, - args, - success: result.exitCode === 0, - exitCode: result.exitCode, - stdout, - stderr: '', - executionTimeMs: Date.now() - startedAt, - }; - }); + return { + command, + args, + success: result.exitCode === 0, + exitCode: result.exitCode, + stdout, + stderr: '', + executionTimeMs: Date.now() - startedAt, + }; + }); + }, options?.abortSignal); } /** @@ -227,12 +236,17 @@ export class DaytonaSandbox extends BaseSandbox { * stopped/deleted while idle. Lets `DaytonaFilesystem` reuse the same recovery as * `executeCommand` without reaching into private state. */ - async withFilesystem(op: (fs: Sandbox['fs']) => Promise): Promise { - return await this.recoverAndRetry(async () => { - await this.ensureRunning(); - await this.ensureAuthFresh(); - return await op(this.instance.fs); - }); + async withFilesystem( + op: (fs: Sandbox['fs']) => Promise, + options?: AbortableOptions, + ): Promise { + return await raceWithAbort(async () => { + return await this.recoverAndRetry(async () => { + await this.ensureRunning({ abortSignal: options?.abortSignal }); + await this.ensureAuthFresh(); + return await op(this.instance.fs); + }); + }, options?.abortSignal); } /** @@ -349,6 +363,7 @@ export class DaytonaSandbox extends BaseSandbox { try { return await op(); } catch (error) { + if (isAbortError(error)) throw error; if (!(await this.isRecoverable(error))) throw error; await this.recover(); return await op(); diff --git a/packages/@n8n/agents/src/workspace/sandbox/index.ts b/packages/@n8n/agents/src/workspace/sandbox/index.ts index 402138c7d81..e084b49ded3 100644 --- a/packages/@n8n/agents/src/workspace/sandbox/index.ts +++ b/packages/@n8n/agents/src/workspace/sandbox/index.ts @@ -14,7 +14,11 @@ export { getWorkspaceRoot, type SandboxWorkspace, } from './workspace-root'; -export { runInSandbox, type SandboxCommandTarget } from './run-in-sandbox'; +export { + runInSandbox, + type RunInSandboxOptions, + type SandboxCommandTarget, +} from './run-in-sandbox'; export { loadDaytona } from './lazy-daytona'; export { createFilesystem, createSandbox } from './create-workspace'; export type { diff --git a/packages/@n8n/agents/src/workspace/sandbox/n8n-sandbox-sandbox.ts b/packages/@n8n/agents/src/workspace/sandbox/n8n-sandbox-sandbox.ts index d40cb1363e9..83a79f39ecf 100644 --- a/packages/@n8n/agents/src/workspace/sandbox/n8n-sandbox-sandbox.ts +++ b/packages/@n8n/agents/src/workspace/sandbox/n8n-sandbox-sandbox.ts @@ -117,7 +117,7 @@ export class N8nSandboxServiceSandbox extends BaseSandbox { args: string[] = [], options?: ExecuteCommandOptions, ): Promise { - await this.ensureRunning(); + await this.ensureRunning({ abortSignal: options?.abortSignal }); const result = await this.client.exec(this.requireSandboxId(), { command: toShellCommand(command, args), env: this.compactEnv(options?.env), diff --git a/packages/@n8n/agents/src/workspace/sandbox/run-in-sandbox.ts b/packages/@n8n/agents/src/workspace/sandbox/run-in-sandbox.ts index 978457a19f3..aba7d13c47b 100644 --- a/packages/@n8n/agents/src/workspace/sandbox/run-in-sandbox.ts +++ b/packages/@n8n/agents/src/workspace/sandbox/run-in-sandbox.ts @@ -1,21 +1,28 @@ +import { raceWithAbort } from '../../sdk/abort'; + interface SandboxCommandResult { exitCode: number; stdout: string; stderr: string; } +export interface RunInSandboxOptions { + cwd?: string; + abortSignal?: AbortSignal; +} + export interface SandboxCommandTarget { sandbox?: { provider?: string; executeCommand?: ( command: string, args?: string[], - options?: { cwd?: string }, + options?: { cwd?: string; abortSignal?: AbortSignal }, ) => Promise; processes?: { spawn: ( command: string, - options?: { cwd?: string }, + options?: { cwd?: string; abortSignal?: AbortSignal }, ) => Promise<{ wait: () => Promise }>; }; }; @@ -24,23 +31,33 @@ export interface SandboxCommandTarget { /** * Execute a shell command in the sandbox and wait for completion. * Tries `executeCommand` first, falls back to `processes.spawn` + wait. + * + * The third argument may be a cwd string (legacy) or {@link RunInSandboxOptions}. */ export async function runInSandbox( workspace: SandboxCommandTarget, command: string, - cwd?: string, + cwdOrOptions?: string | RunInSandboxOptions, ): Promise<{ exitCode: number; stdout: string; stderr: string }> { + const options: RunInSandboxOptions = + typeof cwdOrOptions === 'string' ? { cwd: cwdOrOptions } : (cwdOrOptions ?? {}); const sandbox = workspace.sandbox; if (!sandbox) throw new Error('Workspace has no sandbox'); if (sandbox.executeCommand) { - const result = await sandbox.executeCommand(command, [], { cwd }); + const result = await sandbox.executeCommand(command, [], { + cwd: options.cwd, + abortSignal: options.abortSignal, + }); return { exitCode: result.exitCode, stdout: result.stdout, stderr: result.stderr }; } if (sandbox.processes) { - const handle = await sandbox.processes.spawn(command, { cwd }); - const result = await handle.wait(); + const handle = await sandbox.processes.spawn(command, { + cwd: options.cwd, + abortSignal: options.abortSignal, + }); + const result = await raceWithAbort(async () => await handle.wait(), options.abortSignal); return { exitCode: result.exitCode, stdout: result.stdout, stderr: result.stderr }; } diff --git a/packages/@n8n/agents/src/workspace/tools/append-file.ts b/packages/@n8n/agents/src/workspace/tools/append-file.ts index ffa946fa1da..da5a481dfc8 100644 --- a/packages/@n8n/agents/src/workspace/tools/append-file.ts +++ b/packages/@n8n/agents/src/workspace/tools/append-file.ts @@ -18,8 +18,10 @@ export function createAppendFileTool(filesystem: WorkspaceFilesystem): BuiltTool success: z.boolean().describe('Whether the append was successful'), }), ) - .handler(async (input) => { - await filesystem.appendFile(input.path, input.content); + .handler(async (input, ctx) => { + await filesystem.appendFile(input.path, input.content, { + abortSignal: ctx.abortSignal, + }); return { success: true }; }) .build(); diff --git a/packages/@n8n/agents/src/workspace/tools/batch-str-replace-file.ts b/packages/@n8n/agents/src/workspace/tools/batch-str-replace-file.ts index 92ea2c5dd5b..49e7ca97735 100644 --- a/packages/@n8n/agents/src/workspace/tools/batch-str-replace-file.ts +++ b/packages/@n8n/agents/src/workspace/tools/batch-str-replace-file.ts @@ -1,6 +1,7 @@ import { TextEditorDocument, type BatchReplaceResult } from '@n8n/ai-utilities/generic-text-editor'; import { z } from 'zod'; +import { isAbortError } from '../../sdk/abort'; import { Tool } from '../../sdk/tool'; import type { BuiltTool } from '../../types/sdk/tool'; import type { WorkspaceFilesystem } from '../types'; @@ -56,9 +57,12 @@ export function createBatchStrReplaceFileTool(filesystem: WorkspaceFilesystem): ) .input(inputSchema) .output(outputSchema) - .handler(async (input) => { + .handler(async (input, ctx) => { try { - const content = await filesystem.readFile(input.path, { encoding: 'utf-8' }); + const content = await filesystem.readFile(input.path, { + encoding: 'utf-8', + abortSignal: ctx.abortSignal, + }); const editor = new TextEditorDocument({ initialText: content.toString() }); const result = editor.executeBatch(input.replacements); @@ -71,9 +75,13 @@ export function createBatchStrReplaceFileTool(filesystem: WorkspaceFilesystem): throw new Error(`File "${input.path}" is not loaded.`); } - await filesystem.writeFile(input.path, editedContent, { overwrite: true }); + await filesystem.writeFile(input.path, editedContent, { + overwrite: true, + abortSignal: ctx.abortSignal, + }); return { success: true, result }; } catch (error) { + if (isAbortError(error)) throw error; return createErrorOutput(error); } }) diff --git a/packages/@n8n/agents/src/workspace/tools/copy-file.ts b/packages/@n8n/agents/src/workspace/tools/copy-file.ts index f80b12f26f8..7672d1736d4 100644 --- a/packages/@n8n/agents/src/workspace/tools/copy-file.ts +++ b/packages/@n8n/agents/src/workspace/tools/copy-file.ts @@ -22,8 +22,11 @@ export function createCopyFileTool(filesystem: WorkspaceFilesystem): BuiltTool { success: z.boolean().describe('Whether the copy was successful'), }), ) - .handler(async (input) => { - await filesystem.copyFile(input.src, input.dest, { overwrite: input.overwrite }); + .handler(async (input, ctx) => { + await filesystem.copyFile(input.src, input.dest, { + overwrite: input.overwrite, + abortSignal: ctx.abortSignal, + }); return { success: true }; }) .build(); diff --git a/packages/@n8n/agents/src/workspace/tools/delete-file.ts b/packages/@n8n/agents/src/workspace/tools/delete-file.ts index 940f838e216..f6036437066 100644 --- a/packages/@n8n/agents/src/workspace/tools/delete-file.ts +++ b/packages/@n8n/agents/src/workspace/tools/delete-file.ts @@ -22,8 +22,12 @@ export function createDeleteFileTool(filesystem: WorkspaceFilesystem): BuiltTool success: z.boolean().describe('Whether the deletion was successful'), }), ) - .handler(async (input) => { - await filesystem.deleteFile(input.path, { recursive: input.recursive, force: input.force }); + .handler(async (input, ctx) => { + await filesystem.deleteFile(input.path, { + recursive: input.recursive, + force: input.force, + abortSignal: ctx.abortSignal, + }); return { success: true }; }) .build(); diff --git a/packages/@n8n/agents/src/workspace/tools/execute-command.ts b/packages/@n8n/agents/src/workspace/tools/execute-command.ts index 690677eb90a..a3097438aeb 100644 --- a/packages/@n8n/agents/src/workspace/tools/execute-command.ts +++ b/packages/@n8n/agents/src/workspace/tools/execute-command.ts @@ -23,7 +23,7 @@ export function createExecuteCommandTool(sandbox: WorkspaceSandbox): BuiltTool { executionTimeMs: z.number(), }), ) - .handler(async (input) => { + .handler(async (input, ctx) => { if (!sandbox.executeCommand) { throw new Error('Sandbox does not support command execution'); } @@ -32,6 +32,7 @@ export function createExecuteCommandTool(sandbox: WorkspaceSandbox): BuiltTool { cwd: input.cwd, ...(env ? { env } : {}), timeout: input.timeout, + abortSignal: ctx.abortSignal, }); return { success: result.success, diff --git a/packages/@n8n/agents/src/workspace/tools/file-stat.ts b/packages/@n8n/agents/src/workspace/tools/file-stat.ts index 4066dfcc039..7e5f5178c49 100644 --- a/packages/@n8n/agents/src/workspace/tools/file-stat.ts +++ b/packages/@n8n/agents/src/workspace/tools/file-stat.ts @@ -22,8 +22,10 @@ export function createFileStatTool(filesystem: WorkspaceFilesystem): BuiltTool { modifiedAt: z.string(), }), ) - .handler(async (input) => { - const stat = await filesystem.stat(input.path); + .handler(async (input, ctx) => { + const stat = await filesystem.stat(input.path, { + abortSignal: ctx.abortSignal, + }); return { name: stat.name, path: stat.path, diff --git a/packages/@n8n/agents/src/workspace/tools/list-files.ts b/packages/@n8n/agents/src/workspace/tools/list-files.ts index ca3f42494c0..0f74e4ddaa9 100644 --- a/packages/@n8n/agents/src/workspace/tools/list-files.ts +++ b/packages/@n8n/agents/src/workspace/tools/list-files.ts @@ -26,8 +26,11 @@ export function createListFilesTool(filesystem: WorkspaceFilesystem): BuiltTool .describe('List of file entries'), }), ) - .handler(async (input) => { - const entries = await filesystem.readdir(input.path, { recursive: input.recursive }); + .handler(async (input, ctx) => { + const entries = await filesystem.readdir(input.path, { + recursive: input.recursive, + abortSignal: ctx.abortSignal, + }); return { entries }; }) .build(); diff --git a/packages/@n8n/agents/src/workspace/tools/mkdir.ts b/packages/@n8n/agents/src/workspace/tools/mkdir.ts index 7969585dfa4..980c7bf719c 100644 --- a/packages/@n8n/agents/src/workspace/tools/mkdir.ts +++ b/packages/@n8n/agents/src/workspace/tools/mkdir.ts @@ -18,8 +18,11 @@ export function createMkdirTool(filesystem: WorkspaceFilesystem): BuiltTool { success: z.boolean().describe('Whether the directory was created'), }), ) - .handler(async (input) => { - await filesystem.mkdir(input.path, { recursive: input.recursive }); + .handler(async (input, ctx) => { + await filesystem.mkdir(input.path, { + recursive: input.recursive, + abortSignal: ctx.abortSignal, + }); return { success: true }; }) .build(); diff --git a/packages/@n8n/agents/src/workspace/tools/move-file.ts b/packages/@n8n/agents/src/workspace/tools/move-file.ts index 8959dabad0c..0b45ec49187 100644 --- a/packages/@n8n/agents/src/workspace/tools/move-file.ts +++ b/packages/@n8n/agents/src/workspace/tools/move-file.ts @@ -22,8 +22,11 @@ export function createMoveFileTool(filesystem: WorkspaceFilesystem): BuiltTool { success: z.boolean().describe('Whether the move was successful'), }), ) - .handler(async (input) => { - await filesystem.moveFile(input.src, input.dest, { overwrite: input.overwrite }); + .handler(async (input, ctx) => { + await filesystem.moveFile(input.src, input.dest, { + overwrite: input.overwrite, + abortSignal: ctx.abortSignal, + }); return { success: true }; }) .build(); diff --git a/packages/@n8n/agents/src/workspace/tools/process-tools.ts b/packages/@n8n/agents/src/workspace/tools/process-tools.ts index d360a3ef68f..11219289944 100644 --- a/packages/@n8n/agents/src/workspace/tools/process-tools.ts +++ b/packages/@n8n/agents/src/workspace/tools/process-tools.ts @@ -1,5 +1,6 @@ import { z } from 'zod'; +import { raceWithAbort } from '../../sdk/abort'; import { Tool } from '../../sdk/tool'; import type { BuiltTool } from '../../types/sdk/tool'; import type { SandboxProcessManager } from '../types'; @@ -19,8 +20,8 @@ export function createListProcessesTool(processes: SandboxProcessManager): Built ), }), ) - .handler(async () => { - const list = await processes.list(); + .handler(async (_input, ctx) => { + const list = await raceWithAbort(async () => await processes.list(), ctx.abortSignal); return { processes: list }; }) .build(); @@ -35,8 +36,11 @@ export function createKillProcessTool(processes: SandboxProcessManager): BuiltTo }), ) .output(z.object({ killed: z.boolean() })) - .handler(async (input) => { - const killed = await processes.kill(input.pid); + .handler(async (input, ctx) => { + const killed = await raceWithAbort( + async () => await processes.kill(input.pid), + ctx.abortSignal, + ); return { killed }; }) .build(); diff --git a/packages/@n8n/agents/src/workspace/tools/read-file.ts b/packages/@n8n/agents/src/workspace/tools/read-file.ts index 4fce68a10ff..2b1ca0fec29 100644 --- a/packages/@n8n/agents/src/workspace/tools/read-file.ts +++ b/packages/@n8n/agents/src/workspace/tools/read-file.ts @@ -18,9 +18,10 @@ export function createReadFileTool(filesystem: WorkspaceFilesystem): BuiltTool { content: z.string().describe('File content'), }), ) - .handler(async (input) => { + .handler(async (input, ctx) => { const content = await filesystem.readFile(input.path, { encoding: (input.encoding ?? 'utf-8') as BufferEncoding, + abortSignal: ctx.abortSignal, }); return { content: content.toString() }; }) diff --git a/packages/@n8n/agents/src/workspace/tools/rmdir.ts b/packages/@n8n/agents/src/workspace/tools/rmdir.ts index 65c8a644d0f..cebd91f8c6d 100644 --- a/packages/@n8n/agents/src/workspace/tools/rmdir.ts +++ b/packages/@n8n/agents/src/workspace/tools/rmdir.ts @@ -22,8 +22,12 @@ export function createRmdirTool(filesystem: WorkspaceFilesystem): BuiltTool { success: z.boolean().describe('Whether the directory was removed'), }), ) - .handler(async (input) => { - await filesystem.rmdir(input.path, { recursive: input.recursive, force: input.force }); + .handler(async (input, ctx) => { + await filesystem.rmdir(input.path, { + recursive: input.recursive, + force: input.force, + abortSignal: ctx.abortSignal, + }); return { success: true }; }) .build(); diff --git a/packages/@n8n/agents/src/workspace/tools/str-replace-file.ts b/packages/@n8n/agents/src/workspace/tools/str-replace-file.ts index e0b0b251dab..0ffaf920c15 100644 --- a/packages/@n8n/agents/src/workspace/tools/str-replace-file.ts +++ b/packages/@n8n/agents/src/workspace/tools/str-replace-file.ts @@ -1,6 +1,7 @@ import { TextEditorDocument } from '@n8n/ai-utilities/generic-text-editor'; import { z } from 'zod'; +import { isAbortError } from '../../sdk/abort'; import { Tool } from '../../sdk/tool'; import type { BuiltTool } from '../../types/sdk/tool'; import type { WorkspaceFilesystem } from '../types'; @@ -33,9 +34,12 @@ export function createStrReplaceFileTool(filesystem: WorkspaceFilesystem): Built ) .input(inputSchema) .output(outputSchema) - .handler(async (input) => { + .handler(async (input, ctx) => { try { - const content = await filesystem.readFile(input.path, { encoding: 'utf-8' }); + const content = await filesystem.readFile(input.path, { + encoding: 'utf-8', + abortSignal: ctx.abortSignal, + }); const editor = new TextEditorDocument({ initialText: content.toString() }); const result = editor.execute({ command: 'str_replace', @@ -48,9 +52,13 @@ export function createStrReplaceFileTool(filesystem: WorkspaceFilesystem): Built throw new Error(`File "${input.path}" is not loaded.`); } - await filesystem.writeFile(input.path, editedContent, { overwrite: true }); + await filesystem.writeFile(input.path, editedContent, { + overwrite: true, + abortSignal: ctx.abortSignal, + }); return { success: true, result }; } catch (error) { + if (isAbortError(error)) throw error; return createErrorOutput(error); } }) diff --git a/packages/@n8n/agents/src/workspace/tools/write-file.ts b/packages/@n8n/agents/src/workspace/tools/write-file.ts index e0d85942496..7fc64cc1648 100644 --- a/packages/@n8n/agents/src/workspace/tools/write-file.ts +++ b/packages/@n8n/agents/src/workspace/tools/write-file.ts @@ -22,8 +22,11 @@ export function createWriteFileTool(filesystem: WorkspaceFilesystem): BuiltTool success: z.boolean().describe('Whether the write was successful'), }), ) - .handler(async (input) => { - await filesystem.writeFile(input.path, input.content, { recursive: input.recursive }); + .handler(async (input, ctx) => { + await filesystem.writeFile(input.path, input.content, { + recursive: input.recursive, + abortSignal: ctx.abortSignal, + }); return { success: true }; }) .build(); diff --git a/packages/@n8n/agents/src/workspace/types.ts b/packages/@n8n/agents/src/workspace/types.ts index d7529209553..4a32633799d 100644 --- a/packages/@n8n/agents/src/workspace/types.ts +++ b/packages/@n8n/agents/src/workspace/types.ts @@ -29,30 +29,42 @@ export interface FileStat { modifiedAt: Date; } -export interface ReadOptions { +/** Shared abort option for workspace filesystem / sandbox ops. */ +export interface AbortableOptions { + /** When aborted, in-flight workspace work should settle promptly. */ + abortSignal?: AbortSignal; +} + +export interface ReadOptions extends AbortableOptions { encoding?: BufferEncoding; } -export interface WriteOptions { +export interface WriteOptions extends AbortableOptions { recursive?: boolean; overwrite?: boolean; } -export interface ListOptions { +export interface ListOptions extends AbortableOptions { recursive?: boolean; extension?: string; } -export interface RemoveOptions { +export interface RemoveOptions extends AbortableOptions { recursive?: boolean; force?: boolean; } -export interface CopyOptions { +export interface CopyOptions extends AbortableOptions { overwrite?: boolean; recursive?: boolean; } +export interface MkdirOptions extends AbortableOptions { + recursive?: boolean; +} + +export type AppendOptions = AbortableOptions; + export interface MountConfig { type: 'local'; basePath: string; @@ -68,15 +80,15 @@ export interface WorkspaceFilesystem { readFile(path: string, options?: ReadOptions): Promise; writeFile(path: string, content: FileContent, options?: WriteOptions): Promise; - appendFile(path: string, content: FileContent): Promise; + appendFile(path: string, content: FileContent, options?: AppendOptions): Promise; deleteFile(path: string, options?: RemoveOptions): Promise; copyFile(src: string, dest: string, options?: CopyOptions): Promise; moveFile(src: string, dest: string, options?: CopyOptions): Promise; - mkdir(path: string, options?: { recursive?: boolean }): Promise; + mkdir(path: string, options?: MkdirOptions): Promise; rmdir(path: string, options?: RemoveOptions): Promise; readdir(path: string, options?: ListOptions): Promise; - exists(path: string): Promise; - stat(path: string): Promise; + exists(path: string, options?: AbortableOptions): Promise; + stat(path: string, options?: AbortableOptions): Promise; init?(): Promise; destroy?(): Promise; diff --git a/packages/@n8n/instance-ai/src/agent/mcp-tool-name-validation.ts b/packages/@n8n/instance-ai/src/agent/mcp-tool-name-validation.ts index 31f9e754f79..2ceccca673b 100644 --- a/packages/@n8n/instance-ai/src/agent/mcp-tool-name-validation.ts +++ b/packages/@n8n/instance-ai/src/agent/mcp-tool-name-validation.ts @@ -1,6 +1,5 @@ import { isSafeObjectKey } from '@n8n/api-types'; -import { makeToolAbortable } from '../tools/shared/abortable-tool'; import type { InstanceAiToolRegistry } from '../types'; type McpToolRegistry = InstanceAiToolRegistry; @@ -74,7 +73,7 @@ export function addSafeMcpTools( ); } options.claimedToolNames.set(normalizedName, name); - target.set(name, makeToolAbortable(tool)); + target.set(name, tool); } catch (error) { if (error instanceof McpToolNameValidationError) { options.warn?.(error); diff --git a/packages/@n8n/instance-ai/src/tool-registry.ts b/packages/@n8n/instance-ai/src/tool-registry.ts index 8983549861c..e93b504b480 100644 --- a/packages/@n8n/instance-ai/src/tool-registry.ts +++ b/packages/@n8n/instance-ai/src/tool-registry.ts @@ -1,6 +1,5 @@ import type { BuiltTool } from '@n8n/agents'; -import { makeToolAbortable } from './tools/shared/abortable-tool'; import type { InstanceAiToolRegistry } from './types'; export function createToolRegistry( @@ -8,7 +7,7 @@ export function createToolRegistry( ): InstanceAiToolRegistry { const registry = new Map(); for (const [name, tool] of entries) { - registry.set(name, makeToolAbortable(tool)); + registry.set(name, tool); } return registry; } @@ -16,7 +15,7 @@ export function createToolRegistry( export function createToolRegistryFromTools(tools: Iterable): InstanceAiToolRegistry { const registry = createToolRegistry(); for (const tool of tools) { - registry.set(tool.name, makeToolAbortable(tool)); + registry.set(tool.name, tool); } return registry; } @@ -28,7 +27,7 @@ export function mergeToolRegistries( for (const registry of registries) { if (!registry) continue; for (const [name, tool] of registry) { - merged.set(name, makeToolAbortable(tool)); + merged.set(name, tool); } } return merged; diff --git a/packages/@n8n/instance-ai/src/tools/executions.tool.ts b/packages/@n8n/instance-ai/src/tools/executions.tool.ts index 049c5d70974..b8cefb36cd4 100644 --- a/packages/@n8n/instance-ai/src/tools/executions.tool.ts +++ b/packages/@n8n/instance-ai/src/tools/executions.tool.ts @@ -202,6 +202,7 @@ async function handleRun( input: Extract, resumeData: z.infer | undefined, suspend: (payload: z.infer) => Promise, + abortSignal?: AbortSignal, ) { if (context.permissions?.runWorkflow === 'blocked') { return { @@ -292,6 +293,7 @@ async function handleRun( // Approved or always_allow — execute return await context.executionService.run(workflowId, input.inputData, { timeout: input.timeout, + abortSignal, }); } @@ -347,7 +349,7 @@ export function createExecutionsTool(context: InstanceAiContext) { case 'get': return await handleGet(context, input); case 'run': { - return await handleRun(context, input, ctx.resumeData, ctx.suspend); + return await handleRun(context, input, ctx.resumeData, ctx.suspend, ctx.abortSignal); } case 'debug': return await handleDebug(context, input); diff --git a/packages/@n8n/instance-ai/src/tools/filesystem/__tests__/create-tools-from-mcp-server.test.ts b/packages/@n8n/instance-ai/src/tools/filesystem/__tests__/create-tools-from-mcp-server.test.ts index 26daee637c1..b013c7804e3 100644 --- a/packages/@n8n/instance-ai/src/tools/filesystem/__tests__/create-tools-from-mcp-server.test.ts +++ b/packages/@n8n/instance-ai/src/tools/filesystem/__tests__/create-tools-from-mcp-server.test.ts @@ -337,10 +337,13 @@ describe('createToolsFromLocalMcpServer', () => { const result = await execute({ filePath: 'test.ts' }, makeCtx({})); expect(result).toEqual(SUCCESS_RESULT); - expect(server.callTool).toHaveBeenCalledWith({ - name: 'write_file', - arguments: { filePath: 'test.ts' }, - }); + expect(server.callTool).toHaveBeenCalledWith( + { + name: 'write_file', + arguments: { filePath: 'test.ts' }, + }, + { abortSignal: undefined }, + ); }); it('strips _confirmation from LLM-provided args on the first-call path', async () => { @@ -350,10 +353,13 @@ describe('createToolsFromLocalMcpServer', () => { await execute({ filePath: 'test.ts', _confirmation: 'injected-token' }, makeCtx({})); - expect(server.callTool).toHaveBeenCalledWith({ - name: 'write_file', - arguments: { filePath: 'test.ts' }, - }); + expect(server.callTool).toHaveBeenCalledWith( + { + name: 'write_file', + arguments: { filePath: 'test.ts' }, + }, + { abortSignal: undefined }, + ); }); it('passes through a generic error result unchanged', async () => { @@ -444,10 +450,13 @@ describe('createToolsFromLocalMcpServer', () => { ); expect(result).toEqual(SUCCESS_RESULT); - expect(server.callTool).toHaveBeenCalledWith({ - name: 'write_file', - arguments: { filePath: 'test.ts', _confirmation: 'allowForSession' }, - }); + expect(server.callTool).toHaveBeenCalledWith( + { + name: 'write_file', + arguments: { filePath: 'test.ts', _confirmation: 'allowForSession' }, + }, + { abortSignal: undefined }, + ); }); it('returns access-denied error when resumeData has no token (user denied)', async () => { diff --git a/packages/@n8n/instance-ai/src/tools/filesystem/create-tools-from-mcp-server.ts b/packages/@n8n/instance-ai/src/tools/filesystem/create-tools-from-mcp-server.ts index adb89991555..96c7a016134 100644 --- a/packages/@n8n/instance-ai/src/tools/filesystem/create-tools-from-mcp-server.ts +++ b/packages/@n8n/instance-ai/src/tools/filesystem/create-tools-from-mcp-server.ts @@ -338,16 +338,22 @@ export function createToolsFromLocalMcpServer( }; } // Re-call the daemon with the user's decision - return await server.callTool({ - name: toolName, - arguments: { ...args, _confirmation: resumeData.resourceDecision }, - }); + return await server.callTool( + { + name: toolName, + arguments: { ...args, _confirmation: resumeData.resourceDecision }, + }, + { abortSignal: ctx.abortSignal }, + ); } // First-call path: strip any LLM-provided _confirmation key so the agent // cannot bypass the human confirmation flow by supplying its own token. const { _confirmation: _stripped, ...safeArgs } = args; - const result = await server.callTool({ name: toolName, arguments: safeArgs }); + const result = await server.callTool( + { name: toolName, arguments: safeArgs }, + { abortSignal: ctx.abortSignal }, + ); // If the daemon requires a resource-access confirmation, suspend the agent if (result.isError) { diff --git a/packages/@n8n/instance-ai/src/tools/n8n-docs.tool.ts b/packages/@n8n/instance-ai/src/tools/n8n-docs.tool.ts index c8a147f52dc..49b19d69520 100644 --- a/packages/@n8n/instance-ai/src/tools/n8n-docs.tool.ts +++ b/packages/@n8n/instance-ai/src/tools/n8n-docs.tool.ts @@ -1,4 +1,4 @@ -import { Tool } from '@n8n/agents'; +import { isAbortError, Tool } from '@n8n/agents'; import type { InstanceAiContext } from '../types'; import { @@ -32,7 +32,6 @@ import { type N8nDocsReadInput, type N8nDocsSearchInput, } from './n8n-docs/schemas'; -import { isAbortError } from './shared/abortable-tool'; import { N8N_DOCS_TOOL_ID } from './tool-ids'; export { N8N_DOCS_TOOL_ID }; diff --git a/packages/@n8n/instance-ai/src/tools/n8n-docs/registry.ts b/packages/@n8n/instance-ai/src/tools/n8n-docs/registry.ts index b63fc2e6e0e..df935a890b7 100644 --- a/packages/@n8n/instance-ai/src/tools/n8n-docs/registry.ts +++ b/packages/@n8n/instance-ai/src/tools/n8n-docs/registry.ts @@ -1,5 +1,6 @@ +import { isAbortError } from '@n8n/agents'; + import type { Logger } from '../../logger'; -import { isAbortError } from '../shared/abortable-tool'; import { sanitizeWebContent, wrapUntrustedData } from '../web-research/sanitize-web-content'; const N8N_DOCS_ORIGIN = 'https://docs.n8n.io'; diff --git a/packages/@n8n/instance-ai/src/tools/orchestration/verify-built-workflow.tool.ts b/packages/@n8n/instance-ai/src/tools/orchestration/verify-built-workflow.tool.ts index 5e4b37c8f3f..b5f1b01d42f 100644 --- a/packages/@n8n/instance-ai/src/tools/orchestration/verify-built-workflow.tool.ts +++ b/packages/@n8n/instance-ai/src/tools/orchestration/verify-built-workflow.tool.ts @@ -162,6 +162,7 @@ export function createVerifyBuiltWorkflowTool(context: OrchestrationContext) { { timeout: resolvedInput.timeout, verificationPinData: prepared.verificationPinData, + abortSignal: context.abortSignal, }, ); diff --git a/packages/@n8n/instance-ai/src/tools/shared/__tests__/abortable-tool.test.ts b/packages/@n8n/instance-ai/src/tools/shared/__tests__/abortable-tool.test.ts deleted file mode 100644 index f3ead20e04b..00000000000 --- a/packages/@n8n/instance-ai/src/tools/shared/__tests__/abortable-tool.test.ts +++ /dev/null @@ -1,114 +0,0 @@ -import type { BuiltTool, ToolContext } from '@n8n/agents'; -import { describe, expect, it, vi } from 'vitest'; - -import { - createAbortError, - isAbortError, - makeToolAbortable, - throwIfAborted, - withAbortableToolHandler, -} from '../abortable-tool'; - -function makeCtx(signal?: AbortSignal): ToolContext { - return { abortSignal: signal }; -} - -describe('abortable-tool', () => { - describe('isAbortError', () => { - it('detects AbortError by name', () => { - expect(isAbortError(createAbortError())).toBe(true); - }); - - it('rejects unrelated errors', () => { - expect(isAbortError(new Error('boom'))).toBe(false); - }); - }); - - describe('throwIfAborted', () => { - it('throws when the signal is already aborted', () => { - const controller = new AbortController(); - controller.abort(); - expect(() => throwIfAborted(makeCtx(controller.signal))).toThrowError( - expect.objectContaining({ name: 'AbortError' }), - ); - }); - - it('no-ops when the signal is still open', () => { - expect(() => throwIfAborted(makeCtx(new AbortController().signal))).not.toThrow(); - }); - }); - - describe('withAbortableToolHandler', () => { - it('returns the handler result when not aborted', async () => { - const handler = withAbortableToolHandler(async () => await Promise.resolve({ ok: true })); - await expect(handler({}, makeCtx(new AbortController().signal))).resolves.toEqual({ - ok: true, - }); - }); - - it('rejects promptly when the signal aborts during execution', async () => { - const controller = new AbortController(); - let release!: () => void; - const hang = new Promise((resolve) => { - release = resolve; - }); - - const handler = withAbortableToolHandler(async () => { - await hang; - return { ok: true }; - }); - - const pending = handler({}, makeCtx(controller.signal)); - controller.abort(); - - await expect(pending).rejects.toMatchObject({ name: 'AbortError' }); - release(); - }); - - it('rejects before starting when already aborted', async () => { - const controller = new AbortController(); - controller.abort(); - const inner = vi.fn(async () => await Promise.resolve({ ok: true })); - const handler = withAbortableToolHandler(inner); - - await expect(handler({}, makeCtx(controller.signal))).rejects.toMatchObject({ - name: 'AbortError', - }); - expect(inner).not.toHaveBeenCalled(); - }); - - it('passes through when no abort signal is provided', async () => { - const handler = withAbortableToolHandler(async () => await Promise.resolve({ ok: true })); - await expect(handler({}, {})).resolves.toEqual({ ok: true }); - }); - }); - - describe('makeToolAbortable', () => { - it('wraps the handler and is idempotent', async () => { - const controller = new AbortController(); - let release!: () => void; - const hang = new Promise((resolve) => { - release = resolve; - }); - - const tool: BuiltTool = { - name: 'slow', - description: 'slow tool', - handler: async () => { - await hang; - return { done: true }; - }, - }; - - const once = makeToolAbortable(tool); - const twice = makeToolAbortable(once); - expect(twice).toBe(once); - expect(once.metadata?.abortableWrapped).toBe(true); - - const pending = once.handler?.({}, makeCtx(controller.signal)); - controller.abort(); - await expect(pending).rejects.toMatchObject({ name: 'AbortError' }); - release(); - }); - }); -}); diff --git a/packages/@n8n/instance-ai/src/tools/shared/abortable-tool.ts b/packages/@n8n/instance-ai/src/tools/shared/abortable-tool.ts deleted file mode 100644 index 7fd387bb386..00000000000 --- a/packages/@n8n/instance-ai/src/tools/shared/abortable-tool.ts +++ /dev/null @@ -1,84 +0,0 @@ -import type { BuiltTool, InterruptibleToolContext, ToolContext } from '@n8n/agents'; - -const ABORTABLE_WRAPPED_KEY = 'abortableWrapped'; - -type ToolHandler = NonNullable; - -export function isAbortError(error: unknown): boolean { - if (!(error instanceof Error)) return false; - if (error.name === 'AbortError') return true; - return error.message === 'Aborted' || error.message === 'This operation was aborted'; -} - -export function createAbortError(reason?: unknown): Error { - if (reason instanceof Error) return reason; - const error = new Error(typeof reason === 'string' ? reason : 'This operation was aborted'); - error.name = 'AbortError'; - return error; -} - -/** Throw if the run abort signal has already fired. */ -export function throwIfAborted(ctx: Pick): void { - const signal = ctx.abortSignal; - if (signal?.aborted) { - throw createAbortError(signal.reason); - } -} - -async function abortRejection(signal: AbortSignal): Promise { - return await new Promise((_, reject) => { - if (signal.aborted) { - reject(createAbortError(signal.reason)); - return; - } - signal.addEventListener( - 'abort', - () => { - reject(createAbortError(signal.reason)); - }, - { once: true }, - ); - }); -} - -/** - * Race a tool handler against the run abort signal so Stop unblocks the - * executor even when the underlying work does not cooperate. Cooperative - * tools should still forward `ctx.abortSignal` into I/O to stop real work. - */ -export function withAbortableToolHandler(handler: ToolHandler): ToolHandler { - return async (input, ctx) => { - const signal = ctx.abortSignal; - if (!signal) { - return await handler(input, ctx); - } - throwIfAborted(ctx); - return await Promise.race([handler(input, ctx), abortRejection(signal)]); - }; -} - -function isAlreadyWrapped(tool: BuiltTool): boolean { - return tool.metadata?.[ABORTABLE_WRAPPED_KEY] === true; -} - -/** - * Wrap a BuiltTool so its handler settles promptly when the run abort signal - * fires. Idempotent — safe to call on already-wrapped tools. - */ -export function makeToolAbortable(tool: BuiltTool): BuiltTool { - if (!tool.handler || isAlreadyWrapped(tool)) { - return tool; - } - - const wrappedHandler = withAbortableToolHandler(tool.handler); - - return { - ...tool, - handler: async (input, ctx: ToolContext | InterruptibleToolContext) => - await wrappedHandler(input, ctx), - metadata: { - ...tool.metadata, - [ABORTABLE_WRAPPED_KEY]: true, - }, - }; -} diff --git a/packages/@n8n/instance-ai/src/tools/workflows/__tests__/build-workflow.tool.test.ts b/packages/@n8n/instance-ai/src/tools/workflows/__tests__/build-workflow.tool.test.ts index 332be9a1ecf..1238505014d 100644 --- a/packages/@n8n/instance-ai/src/tools/workflows/__tests__/build-workflow.tool.test.ts +++ b/packages/@n8n/instance-ai/src/tools/workflows/__tests__/build-workflow.tool.test.ts @@ -244,7 +244,7 @@ describe('createBuildWorkflowTool', () => { expect(result.postBuildFlow?.guidance).toContain( 'Do not replace the error-workflow opt-in with a generic add-anything', ); - expect(compileWorkflowSource).toHaveBeenCalledWith(context, filePath, source); + expect(compileWorkflowSource).toHaveBeenCalledWith(context, filePath, source, undefined); expect(context.workflowService.createFromWorkflowJSON).toHaveBeenCalledWith( expect.objectContaining({ name: 'Daily Weather to Slack' }), { markAsAiTemporary: true }, @@ -669,7 +669,7 @@ describe('createBuildWorkflowTool', () => { workflowId: 'wf-existing', workflowName: 'Daily Slack Channel Digest', }); - expect(compileWorkflowSource).toHaveBeenCalledWith(context, filePath, source); + expect(compileWorkflowSource).toHaveBeenCalledWith(context, filePath, source, undefined); expect(context.workflowService.updateFromWorkflowJSON).toHaveBeenCalledWith( 'wf-existing', workflowJson, diff --git a/packages/@n8n/instance-ai/src/tools/workflows/__tests__/workflow-source-compiler.test.ts b/packages/@n8n/instance-ai/src/tools/workflows/__tests__/workflow-source-compiler.test.ts index 314d96c46a5..8779be72389 100644 --- a/packages/@n8n/instance-ai/src/tools/workflows/__tests__/workflow-source-compiler.test.ts +++ b/packages/@n8n/instance-ai/src/tools/workflows/__tests__/workflow-source-compiler.test.ts @@ -113,7 +113,7 @@ describe('compileWorkflowSource', () => { expect(runInSandbox).toHaveBeenCalledWith( context.workspace, "node --import tsx build.mjs '/home/daytona/workspace/src/workflows/main.workflow.ts'", - '/home/daytona/workspace', + { cwd: '/home/daytona/workspace', abortSignal: undefined }, ); expect(result).toMatchObject({ success: true, @@ -165,6 +165,41 @@ describe('compileWorkflowSource', () => { }); expect(runInSandbox).not.toHaveBeenCalled(); }); + + it('rethrows AbortError from sandbox execution instead of converting to a build failure', async () => { + const abortError = new Error('This operation was aborted'); + abortError.name = 'AbortError'; + vi.mocked(runInSandbox).mockRejectedValue(abortError); + + await expect( + compileWorkflowSource(makeContext(), 'src/workflows/main.workflow.ts', 'workflow source'), + ).rejects.toMatchObject({ name: 'AbortError' }); + }); + + it('forwards abortSignal into the sandbox runner', async () => { + const controller = new AbortController(); + vi.mocked(runInSandbox).mockResolvedValue({ + exitCode: 0, + stdout: JSON.stringify({ + success: true, + workflow: { name: 'TS workflow', nodes: [], connections: {} }, + warnings: [], + }), + stderr: '', + }); + + await compileWorkflowSource( + makeContext(), + 'src/workflows/main.workflow.ts', + 'workflow source', + controller.signal, + ); + + expect(runInSandbox).toHaveBeenCalledWith(expect.anything(), expect.any(String), { + cwd: '/home/daytona/workspace', + abortSignal: controller.signal, + }); + }); }); describe('compileWorkflowSource credential resolution', () => { diff --git a/packages/@n8n/instance-ai/src/tools/workflows/__tests__/write-sandbox-file.tool.test.ts b/packages/@n8n/instance-ai/src/tools/workflows/__tests__/write-sandbox-file.tool.test.ts index 2b99b798be7..d3a91b6c9be 100644 --- a/packages/@n8n/instance-ai/src/tools/workflows/__tests__/write-sandbox-file.tool.test.ts +++ b/packages/@n8n/instance-ai/src/tools/workflows/__tests__/write-sandbox-file.tool.test.ts @@ -72,6 +72,7 @@ describe('createWriteSandboxFileTool', () => { workspace, '/home/user/workspace/src/workflow.ts', 'export default {}', + { abortSignal: undefined }, ); }); }); @@ -98,6 +99,7 @@ describe('createWriteSandboxFileTool', () => { workspace, '/home/user/workspace/src/index.ts', 'console.log("hello")', + { abortSignal: undefined }, ); }); }); @@ -195,6 +197,7 @@ describe('createWriteSandboxFileTool', () => { workspace, '/home/user/workspace/chunks/helper.ts', 'export const x = 1;', + { abortSignal: undefined }, ); }); diff --git a/packages/@n8n/instance-ai/src/tools/workflows/build-workflow.tool.ts b/packages/@n8n/instance-ai/src/tools/workflows/build-workflow.tool.ts index 746c2275046..f6aeefef6dc 100644 --- a/packages/@n8n/instance-ai/src/tools/workflows/build-workflow.tool.ts +++ b/packages/@n8n/instance-ai/src/tools/workflows/build-workflow.tool.ts @@ -83,6 +83,7 @@ interface BuildCtx { toolCallId?: string; resumeData?: z.infer; suspend?: (payload: z.infer) => Promise; + abortSignal?: AbortSignal; } export const buildWorkflowInputSchema = z @@ -453,6 +454,7 @@ export function createBuildWorkflowTool(context: InstanceAiContext) { await writeWorkspaceFile(context.workspace, filePath, input.sourceCode, { logger: context.logger, resourceLabel: 'Workflow source file', + abortSignal: ctx.abortSignal, }); } catch (error) { const remediation = createCodeFixableRemediation({ @@ -482,7 +484,11 @@ export function createBuildWorkflowTool(context: InstanceAiContext) { let sourceCode: string; let sourceHash: string; try { - ({ source: sourceCode, sourceHash } = await readWorkflowSourceFile(context, filePath)); + ({ source: sourceCode, sourceHash } = await readWorkflowSourceFile( + context, + filePath, + ctx.abortSignal, + )); } catch (error) { const remediation = createCodeFixableRemediation({ reason: 'workflow_source_read_failed', @@ -545,7 +551,7 @@ export function createBuildWorkflowTool(context: InstanceAiContext) { let informational: ValidationWarning[] = []; - let compiled = await compileWorkflowSource(context, filePath, sourceCode); + let compiled = await compileWorkflowSource(context, filePath, sourceCode, ctx.abortSignal); if ( !compiled.success && compiled.reason === 'workflow_source_build_failed' && @@ -558,8 +564,14 @@ export function createBuildWorkflowTool(context: InstanceAiContext) { await writeWorkspaceFile(context.workspace, filePath, recovery.source, { logger: context.logger, resourceLabel: 'Workflow source file', + abortSignal: ctx.abortSignal, }); - const retried = await compileWorkflowSource(context, filePath, recovery.source); + const retried = await compileWorkflowSource( + context, + filePath, + recovery.source, + ctx.abortSignal, + ); // The corrected source is on disk; keep reported errors/hash in sync with it. sourceCode = recovery.source; sourceHash = hashWorkflowSource(recovery.source); diff --git a/packages/@n8n/instance-ai/src/tools/workflows/materialize-node-type.tool.ts b/packages/@n8n/instance-ai/src/tools/workflows/materialize-node-type.tool.ts index bea50c9514b..242f77d0105 100644 --- a/packages/@n8n/instance-ai/src/tools/workflows/materialize-node-type.tool.ts +++ b/packages/@n8n/instance-ai/src/tools/workflows/materialize-node-type.tool.ts @@ -74,7 +74,7 @@ export function createMaterializeNodeTypeTool( ), }), ) - .handler(async ({ nodeIds }: z.infer) => { + .handler(async ({ nodeIds }: z.infer, ctx) => { if (!context.nodeService.getNodeTypeDefinition) { return { definitions: nodeIds.map((req: z.infer) => ({ @@ -133,7 +133,9 @@ export function createMaterializeNodeTypeTool( const script = lines.join('\n'); const scriptB64 = Buffer.from(script, 'utf-8').toString('base64'); - const result = await runInSandbox(workspace, `echo '${scriptB64}' | base64 -d | bash`); + const result = await runInSandbox(workspace, `echo '${scriptB64}' | base64 -d | bash`, { + abortSignal: ctx.abortSignal, + }); if (result.exitCode !== 0) { // Mark all as failed but still return content (useful for the agent) diff --git a/packages/@n8n/instance-ai/src/tools/workflows/workflow-file-bindings.ts b/packages/@n8n/instance-ai/src/tools/workflows/workflow-file-bindings.ts index 4215089caaf..908c40f790a 100644 --- a/packages/@n8n/instance-ai/src/tools/workflows/workflow-file-bindings.ts +++ b/packages/@n8n/instance-ai/src/tools/workflows/workflow-file-bindings.ts @@ -176,6 +176,7 @@ export async function refreshWorkflowSourceFileBindingFromSave( export async function readWorkflowSourceFile( context: InstanceAiContext, filePath: string, + abortSignal?: AbortSignal, ): Promise<{ source: string; sourceHash: string }> { if (!context.workspace) { throw new Error('Runtime workspace is required for workflow source files.'); @@ -185,6 +186,7 @@ export async function readWorkflowSourceFile( const source = await readWorkspaceFile(context.workspace, normalizedFilePath, { logger: context.logger, resourceLabel: 'Workflow source file', + abortSignal, }); if (source === null) { diff --git a/packages/@n8n/instance-ai/src/tools/workflows/workflow-source-compiler.ts b/packages/@n8n/instance-ai/src/tools/workflows/workflow-source-compiler.ts index 297efb2f436..b62abe43dbb 100644 --- a/packages/@n8n/instance-ai/src/tools/workflows/workflow-source-compiler.ts +++ b/packages/@n8n/instance-ai/src/tools/workflows/workflow-source-compiler.ts @@ -1,3 +1,4 @@ +import { createAbortError, isAbortError } from '@n8n/agents'; import { getWorkspaceRoot } from '@n8n/agents/sandbox'; import { isRecord } from '@n8n/utils/is-record'; import { validateWorkflow, type WorkflowJSON } from '@n8n/workflow-sdk'; @@ -221,6 +222,7 @@ function enhanceBuildErrors(errors: string[]): string[] { async function compileTypeScriptWorkflowSource( context: InstanceAiContext, filePath: string, + abortSignal?: AbortSignal, ): Promise { if (!context.workspace) { return { @@ -242,9 +244,12 @@ async function compileTypeScriptWorkflowSource( buildResult = await runInSandbox( context.workspace, `node --import tsx build.mjs '${escapeSingleQuotes(sandboxFilePath)}'`, - root, + { cwd: root, abortSignal }, ); } catch (error) { + // Preserve Stop/cancel so callers do not record a sandbox build failure. + if (isAbortError(error)) throw error; + if (abortSignal?.aborted) throw createAbortError(abortSignal.reason); return { success: false, reason: 'workflow_source_sandbox_unavailable', @@ -336,12 +341,13 @@ export async function compileWorkflowSource( context: InstanceAiContext, filePath: string, source: string, + abortSignal?: AbortSignal, ): Promise { let result: WorkflowSourceCompileResult; if (isWorkflowJsonSourceFile(filePath)) { result = parseWorkflowJsonSource(source); } else if (isTypeScriptWorkflowSource(filePath)) { - result = await compileTypeScriptWorkflowSource(context, filePath); + result = await compileTypeScriptWorkflowSource(context, filePath, abortSignal); } else { result = { success: false, diff --git a/packages/@n8n/instance-ai/src/tools/workflows/write-sandbox-file.tool.ts b/packages/@n8n/instance-ai/src/tools/workflows/write-sandbox-file.tool.ts index e50ea23f67a..513b080a447 100644 --- a/packages/@n8n/instance-ai/src/tools/workflows/write-sandbox-file.tool.ts +++ b/packages/@n8n/instance-ai/src/tools/workflows/write-sandbox-file.tool.ts @@ -34,7 +34,7 @@ export function createWriteSandboxFileTool(workspace: SandboxWorkspace) { error: z.string().optional(), }), ) - .handler(async ({ filePath, content }: z.infer) => { + .handler(async ({ filePath, content }: z.infer, ctx) => { try { const root = await getWorkspaceRoot(workspace); @@ -51,7 +51,9 @@ export function createWriteSandboxFileTool(workspace: SandboxWorkspace) { }; } - await writeFileViaSandbox(workspace, normalized, content); + await writeFileViaSandbox(workspace, normalized, content, { + abortSignal: ctx.abortSignal, + }); return { success: true, path: normalized }; } catch (error) { return { diff --git a/packages/@n8n/instance-ai/src/types.ts b/packages/@n8n/instance-ai/src/types.ts index f6135eeacfc..328d299b5d2 100644 --- a/packages/@n8n/instance-ai/src/types.ts +++ b/packages/@n8n/instance-ai/src/types.ts @@ -386,6 +386,7 @@ export interface InstanceAiExecutionService { verificationPinData?: Record; /** When set, execute this specific trigger node instead of auto-detecting. */ triggerNodeName?: string; + abortSignal?: AbortSignal; }, ): Promise; getStatus(executionId: string): Promise; @@ -790,7 +791,10 @@ export interface LocalMcpServer { getAvailableTools(): McpTool[]; /** Return tools that belong to the given category (based on annotations.category). */ getToolsByCategory(category: string): McpTool[]; - callTool(req: McpToolCallRequest): Promise; + callTool( + req: McpToolCallRequest, + options?: { abortSignal?: AbortSignal }, + ): Promise; } // ── Workspace shapes ──────────────────────────────────────────────────────── diff --git a/packages/@n8n/instance-ai/src/workspace/__tests__/lazy-runtime-workspace.test.ts b/packages/@n8n/instance-ai/src/workspace/__tests__/lazy-runtime-workspace.test.ts index ae394aad926..38cbb38e9b7 100644 --- a/packages/@n8n/instance-ai/src/workspace/__tests__/lazy-runtime-workspace.test.ts +++ b/packages/@n8n/instance-ai/src/workspace/__tests__/lazy-runtime-workspace.test.ts @@ -215,4 +215,25 @@ describe('createLazyRuntimeWorkspace', () => { expect(lazyWorkspace.filesystem?.status).toBe('destroyed'); expect(lazyWorkspace.sandbox?.status).toBe('destroyed'); }); + + it('aborts writeFile while lazy workspace bring-up is still pending', async () => { + const abortController = new AbortController(); + const ensureWorkspace = vi.fn( + async () => + await new Promise(() => { + // Never resolves — simulates Daytona create/start hanging. + }), + ); + const lazyWorkspace = createLazyRuntimeWorkspace({ ensureWorkspace }); + const writePromise = lazyWorkspace.filesystem!.writeFile('/workflow.ts', 'export {}', { + abortSignal: abortController.signal, + }); + + abortController.abort('Agent run was aborted'); + + await expect(writePromise).rejects.toMatchObject({ + name: 'AbortError', + message: 'Agent run was aborted', + }); + }); }); diff --git a/packages/@n8n/instance-ai/src/workspace/lazy-runtime-workspace.ts b/packages/@n8n/instance-ai/src/workspace/lazy-runtime-workspace.ts index 8f320b1fa18..7517a8578f5 100644 --- a/packages/@n8n/instance-ai/src/workspace/lazy-runtime-workspace.ts +++ b/packages/@n8n/instance-ai/src/workspace/lazy-runtime-workspace.ts @@ -2,6 +2,9 @@ import { BaseFilesystem, BaseSandbox, Workspace, + raceWithAbort, + type AbortableOptions, + type AppendOptions, type CommandResult, type CopyOptions, type ExecuteCommandOptions, @@ -9,6 +12,7 @@ import { type FileEntry, type FileStat, type ListOptions, + type MkdirOptions, type ProviderStatus, type ReadOptions, type RemoveOptions, @@ -222,51 +226,54 @@ class LazyRuntimeFilesystem extends BaseFilesystem { } async readFile(path: string, options?: ReadOptions): Promise { - return await (await this.getFilesystem()).readFile(path, options); + return await (await this.getFilesystem(options?.abortSignal)).readFile(path, options); } async writeFile(path: string, content: FileContent, options?: WriteOptions): Promise { - await (await this.getFilesystem()).writeFile(path, content, options); + await (await this.getFilesystem(options?.abortSignal)).writeFile(path, content, options); } - async appendFile(path: string, content: FileContent): Promise { - await (await this.getFilesystem()).appendFile(path, content); + async appendFile(path: string, content: FileContent, options?: AppendOptions): Promise { + await (await this.getFilesystem(options?.abortSignal)).appendFile(path, content, options); } async deleteFile(path: string, options?: RemoveOptions): Promise { - await (await this.getFilesystem()).deleteFile(path, options); + await (await this.getFilesystem(options?.abortSignal)).deleteFile(path, options); } async copyFile(src: string, dest: string, options?: CopyOptions): Promise { - await (await this.getFilesystem()).copyFile(src, dest, options); + await (await this.getFilesystem(options?.abortSignal)).copyFile(src, dest, options); } async moveFile(src: string, dest: string, options?: CopyOptions): Promise { - await (await this.getFilesystem()).moveFile(src, dest, options); + await (await this.getFilesystem(options?.abortSignal)).moveFile(src, dest, options); } - async mkdir(path: string, options?: { recursive?: boolean }): Promise { - await (await this.getFilesystem()).mkdir(path, options); + async mkdir(path: string, options?: MkdirOptions): Promise { + await (await this.getFilesystem(options?.abortSignal)).mkdir(path, options); } async rmdir(path: string, options?: RemoveOptions): Promise { - await (await this.getFilesystem()).rmdir(path, options); + await (await this.getFilesystem(options?.abortSignal)).rmdir(path, options); } async readdir(path: string, options?: ListOptions): Promise { - return await (await this.getFilesystem()).readdir(path, options); + return await (await this.getFilesystem(options?.abortSignal)).readdir(path, options); } - async exists(path: string): Promise { - return await (await this.getFilesystem()).exists(path); + async exists(path: string, options?: AbortableOptions): Promise { + return await (await this.getFilesystem(options?.abortSignal)).exists(path, options); } - async stat(path: string): Promise { - return await (await this.getFilesystem()).stat(path); + async stat(path: string, options?: AbortableOptions): Promise { + return await (await this.getFilesystem(options?.abortSignal)).stat(path, options); } - private async getFilesystem(): Promise { - const filesystem = await this.resolver.getFilesystem(); + private async getFilesystem(abortSignal?: AbortSignal): Promise { + const filesystem = await raceWithAbort( + async () => await this.resolver.getFilesystem(), + abortSignal, + ); this.syncStatus(filesystem); return filesystem; } @@ -321,7 +328,7 @@ class LazyRuntimeSandbox extends BaseSandbox { args: string[] = [], options?: ExecuteCommandOptions, ): Promise { - const sandbox = await this.getSandbox(); + const sandbox = await this.getSandbox(options?.abortSignal); if (!sandbox.executeCommand) { throw new Error('Instance AI runtime sandbox does not support command execution.'); } @@ -351,8 +358,8 @@ class LazyRuntimeSandbox extends BaseSandbox { return 'Workspace command tools are available and create the runtime sandbox on first use.'; } - private async getSandbox(): Promise { - const sandbox = await this.resolver.getSandbox(); + private async getSandbox(abortSignal?: AbortSignal): Promise { + const sandbox = await raceWithAbort(async () => await this.resolver.getSandbox(), abortSignal); this.syncStatus(sandbox); return sandbox; } diff --git a/packages/@n8n/instance-ai/src/workspace/sandbox-fs.ts b/packages/@n8n/instance-ai/src/workspace/sandbox-fs.ts index a8d97bcfdba..6efcac05a41 100644 --- a/packages/@n8n/instance-ai/src/workspace/sandbox-fs.ts +++ b/packages/@n8n/instance-ai/src/workspace/sandbox-fs.ts @@ -8,8 +8,10 @@ * command fallback keeps setup compatible with command-only providers. */ +import { createAbortError, throwIfAborted } from '@n8n/agents'; import { runInSandbox as runInSharedSandbox, + type RunInSandboxOptions, type SandboxCommandTarget, type SandboxWorkspace as SharedSandboxWorkspace, } from '@n8n/agents/sandbox'; @@ -23,13 +25,19 @@ export interface SandboxWorkspace extends SharedSandboxWorkspace { provider?: string; basePath?: string; init?: () => Promise; - readFile?: (path: string, options?: { encoding?: BufferEncoding }) => Promise; + readFile?: ( + path: string, + options?: { encoding?: BufferEncoding; abortSignal?: AbortSignal }, + ) => Promise; writeFile: ( path: string, content: string | Buffer, - options?: { recursive?: boolean }, + options?: { recursive?: boolean; abortSignal?: AbortSignal }, + ) => Promise; + mkdir: ( + path: string, + options?: { recursive?: boolean; abortSignal?: AbortSignal }, ) => Promise; - mkdir: (path: string, options?: { recursive?: boolean }) => Promise; } & NonNullable; } @@ -42,14 +50,26 @@ export interface SandboxIoRetryOptions { logger?: Pick; resourceLabel?: string; retryBackoffBaseMs?: number; + abortSignal?: AbortSignal; } function ioResourceLabel(options?: SandboxIoRetryOptions): string { return options?.resourceLabel ?? 'Sandbox file'; } -async function sleep(ms: number): Promise { - await new Promise((resolve) => setTimeout(resolve, ms)); +async function sleep(ms: number, abortSignal?: AbortSignal): Promise { + throwIfAborted(abortSignal); + await new Promise((resolve, reject) => { + const timer = setTimeout(() => { + abortSignal?.removeEventListener('abort', onAbort); + resolve(); + }, ms); + const onAbort = () => { + clearTimeout(timer); + reject(createAbortError(abortSignal?.reason)); + }; + abortSignal?.addEventListener('abort', onAbort, { once: true }); + }); } // Daytona surfaces upstream gateway failures (e.g. Cloudflare 502/524) as errors with a numeric status. @@ -66,6 +86,7 @@ export async function retryTransientSandboxIo( ): Promise { const baseMs = options?.retryBackoffBaseMs ?? DEFAULT_SANDBOX_IO_RETRY_BACKOFF_BASE_MS; for (let attempt = 1; ; attempt++) { + throwIfAborted(options?.abortSignal); try { return await op(); } catch (error) { @@ -75,7 +96,10 @@ export async function retryTransientSandboxIo( attempt, error: formatErrorForLog(error), }); - await sleep(Math.min(baseMs * 2 ** (attempt - 1), SANDBOX_IO_RETRY_BACKOFF_CAP_MS)); + await sleep( + Math.min(baseMs * 2 ** (attempt - 1), SANDBOX_IO_RETRY_BACKOFF_CAP_MS), + options?.abortSignal, + ); } } } @@ -91,9 +115,9 @@ export async function retryTransientSandboxIo( export async function runInSandbox( workspace: SandboxCommandTarget, command: string, - cwd?: string, + cwdOrOptions?: string | RunInSandboxOptions, ): Promise<{ exitCode: number; stdout: string; stderr: string }> { - const result = await runInSharedSandbox(workspace, command, cwd); + const result = await runInSharedSandbox(workspace, command, cwdOrOptions); const session = getTemplateTelemetrySession(workspace); if (session) { @@ -122,7 +146,9 @@ export async function writeFileViaSandbox( await retryTransientSandboxIo( async () => { const runWriteCommand = async (command: string) => { - const result = await runInSandbox(workspace, command); + const result = await runInSandbox(workspace, command, { + abortSignal: options?.abortSignal, + }); if (result.exitCode !== 0) { throw new Error(`Failed to write file ${filePath}: ${result.stderr}`); } @@ -173,7 +199,10 @@ export async function readFileViaSandbox( options?: SandboxIoRetryOptions, ): Promise { const result = await retryTransientSandboxIo( - async () => await runInSandbox(workspace, `cat '${escapeSingleQuotes(filePath)}' 2>/dev/null`), + async () => + await runInSandbox(workspace, `cat '${escapeSingleQuotes(filePath)}' 2>/dev/null`, { + abortSignal: options?.abortSignal, + }), filePath, options, ); diff --git a/packages/@n8n/instance-ai/src/workspace/scoped-workspace.ts b/packages/@n8n/instance-ai/src/workspace/scoped-workspace.ts index 099ed5a9973..eb08d6ff613 100644 --- a/packages/@n8n/instance-ai/src/workspace/scoped-workspace.ts +++ b/packages/@n8n/instance-ai/src/workspace/scoped-workspace.ts @@ -1,10 +1,13 @@ import { Workspace, + type AbortableOptions, + type AppendOptions, type CopyOptions, type FileContent, type FileEntry, type FileStat, type ListOptions, + type MkdirOptions, type ProviderStatus, type ReadOptions, type RemoveOptions, @@ -79,8 +82,8 @@ class ScopedFilesystem implements WorkspaceFilesystem { await this.filesystem.writeFile(resolvePath(this.root, path), content, options); } - async appendFile(path: string, content: FileContent): Promise { - await this.filesystem.appendFile(resolvePath(this.root, path), content); + async appendFile(path: string, content: FileContent, options?: AppendOptions): Promise { + await this.filesystem.appendFile(resolvePath(this.root, path), content, options); } async deleteFile(path: string, options?: RemoveOptions): Promise { @@ -103,7 +106,7 @@ class ScopedFilesystem implements WorkspaceFilesystem { ); } - async mkdir(path: string, options?: { recursive?: boolean }): Promise { + async mkdir(path: string, options?: MkdirOptions): Promise { await this.filesystem.mkdir(resolvePath(this.root, path), options); } @@ -115,12 +118,12 @@ class ScopedFilesystem implements WorkspaceFilesystem { return await this.filesystem.readdir(resolvePath(this.root, path), options); } - async exists(path: string): Promise { - return await this.filesystem.exists(resolvePath(this.root, path)); + async exists(path: string, options?: AbortableOptions): Promise { + return await this.filesystem.exists(resolvePath(this.root, path), options); } - async stat(path: string): Promise { - return await this.filesystem.stat(resolvePath(this.root, path)); + async stat(path: string, options?: AbortableOptions): Promise { + return await this.filesystem.stat(resolvePath(this.root, path), options); } } diff --git a/packages/@n8n/instance-ai/src/workspace/workspace-files.ts b/packages/@n8n/instance-ai/src/workspace/workspace-files.ts index a3dcb2f2941..7110fd9d67b 100644 --- a/packages/@n8n/instance-ai/src/workspace/workspace-files.ts +++ b/packages/@n8n/instance-ai/src/workspace/workspace-files.ts @@ -10,11 +10,14 @@ import { export interface WorkspaceFileTarget { filesystem?: { - readFile?: (path: string, options?: { encoding?: 'utf-8' }) => Promise; + readFile?: ( + path: string, + options?: { encoding?: 'utf-8'; abortSignal?: AbortSignal }, + ) => Promise; writeFile: ( path: string, content: string | Buffer, - options?: { recursive?: boolean }, + options?: { recursive?: boolean; abortSignal?: AbortSignal }, ) => Promise; }; sandbox?: SandboxWorkspace['sandbox']; @@ -26,6 +29,7 @@ export interface WorkspaceFileOptions { resourceLabel?: string; /** Base for the exponential retry backoff on transient write errors. Default 1s. */ retryBackoffBaseMs?: number; + abortSignal?: AbortSignal; } function resourceLabel(options?: WorkspaceFileOptions): string { @@ -52,7 +56,11 @@ export async function readWorkspaceFile( return decodeWorkspaceFileContent( await retryTransientSandboxIo( // .call preserves the provider's `this` binding (e.g. LazyRuntimeFilesystem). - async () => await readFile.call(filesystem, filePath, { encoding: 'utf-8' }), + async () => + await readFile.call(filesystem, filePath, { + encoding: 'utf-8', + abortSignal: options?.abortSignal, + }), filePath, options, ), @@ -106,7 +114,11 @@ export async function writeWorkspaceFile( if (filesystem) { try { await retryTransientSandboxIo( - async () => await filesystem.writeFile(filePath, content, { recursive: true }), + async () => + await filesystem.writeFile(filePath, content, { + recursive: true, + abortSignal: options?.abortSignal, + }), filePath, options, ); diff --git a/packages/cli/src/modules/instance-ai/__tests__/composite-local-mcp-server.test.ts b/packages/cli/src/modules/instance-ai/__tests__/composite-local-mcp-server.test.ts index d2309ae06ec..256573458b1 100644 --- a/packages/cli/src/modules/instance-ai/__tests__/composite-local-mcp-server.test.ts +++ b/packages/cli/src/modules/instance-ai/__tests__/composite-local-mcp-server.test.ts @@ -20,7 +20,9 @@ function ok(text: string): McpToolCallResult { } function fakeServer(tools: McpTool[], result: McpToolCallResult) { - const callTool = vi.fn(async (_req: McpToolCallRequest) => result); + const callTool = vi.fn( + async (_req: McpToolCallRequest, _options?: { abortSignal?: AbortSignal }) => result, + ); const getToolsByCategory = vi.fn((category: string) => tools.filter((t) => t.annotations?.category === category), ); @@ -71,6 +73,16 @@ describe('CompositeLocalMcpServer', () => { expect(a.callTool).not.toHaveBeenCalled(); }); + it('forwards abortSignal options to the owning server', async () => { + const a = fakeServer([tool('x')], ok('from-a')); + const composite = new CompositeLocalMcpServer([a.server]); + const abortSignal = new AbortController().signal; + + await composite.callTool({ name: 'x', arguments: {} }, { abortSignal }); + + expect(a.callTool).toHaveBeenCalledWith({ name: 'x', arguments: {} }, { abortSignal }); + }); + it('routes a duplicated tool name to the last server that declared it', async () => { const a = fakeServer([tool('y')], ok('from-a')); const b = fakeServer([tool('y')], ok('from-b')); diff --git a/packages/cli/src/modules/instance-ai/__tests__/instance-ai.adapter.service.test.ts b/packages/cli/src/modules/instance-ai/__tests__/instance-ai.adapter.service.test.ts index d3c607eb537..abf0867d638 100644 --- a/packages/cli/src/modules/instance-ai/__tests__/instance-ai.adapter.service.test.ts +++ b/packages/cli/src/modules/instance-ai/__tests__/instance-ai.adapter.service.test.ts @@ -3134,7 +3134,10 @@ describe('createExecutionAdapter run()', () => { status: 'error', }); - expect(mockActiveExecutions.stopExecution).toHaveBeenCalled(); + expect(mockActiveExecutions.stopExecution).toHaveBeenCalledWith( + 'exec-1', + expect.objectContaining({ name: 'TimeoutExecutionCancelledError' }), + ); expect(mockTelemetry.track).toHaveBeenCalledWith( 'Builder executed workflow', expect.objectContaining({ @@ -3145,6 +3148,45 @@ describe('createExecutionAdapter run()', () => { ); }); + it('tracks abort cancellation as a manual cancel, not a timeout', async () => { + const { adapter, mockActiveExecutions, mockTelemetry } = createRunAdapterForTests( + { + id: 'wf-1', + nodes: [], + }, + { + activeExecution: true, + postExecutePromise: new Promise(() => {}), + threadId: 'thread-1', + }, + ); + const abortController = new AbortController(); + + const runPromise = adapter.run('wf-1', undefined, { + timeout: 60_000, + abortSignal: abortController.signal, + }); + abortController.abort(); + + await expect(runPromise).resolves.toMatchObject({ + status: 'error', + error: 'Execution was cancelled', + }); + + expect(mockActiveExecutions.stopExecution).toHaveBeenCalledWith( + 'exec-1', + expect.objectContaining({ name: 'ManualExecutionCancelledError' }), + ); + expect(mockTelemetry.track).toHaveBeenCalledWith( + 'Builder executed workflow', + expect.objectContaining({ + workflow_id: 'wf-1', + status: 'error', + error: 'Execution was cancelled', + }), + ); + }); + it('tracks error status when an execution fails to launch', async () => { const { adapter, mockTelemetry, mockWorkflowRunner } = createRunAdapterForTests( { diff --git a/packages/cli/src/modules/instance-ai/__tests__/local-gateway.test.ts b/packages/cli/src/modules/instance-ai/__tests__/local-gateway.test.ts index b384a34182d..d7d7806ff03 100644 --- a/packages/cli/src/modules/instance-ai/__tests__/local-gateway.test.ts +++ b/packages/cli/src/modules/instance-ai/__tests__/local-gateway.test.ts @@ -177,6 +177,54 @@ describe('LocalGateway', () => { vi.useRealTimers(); }); + it('should reject immediately when abortSignal is already aborted', async () => { + gateway.init(EMPTY_CAPABILITIES); + const controller = new AbortController(); + controller.abort('stopped by user'); + + const events: LocalGatewayRequestEvent[] = []; + gateway.onRequest((event) => events.push(event)); + + await expect( + gateway.callTool( + { name: 'read_file', arguments: { filePath: 'test.ts' } }, + { abortSignal: controller.signal }, + ), + ).rejects.toMatchObject({ name: 'AbortError', message: 'stopped by user' }); + + expect(events).toHaveLength(0); + }); + + it('should reject with AbortError when abortSignal fires mid-request and ignore later responses', async () => { + gateway.init(EMPTY_CAPABILITIES); + const controller = new AbortController(); + + const events: LocalGatewayRequestEvent[] = []; + gateway.onRequest((event) => events.push(event)); + + const callPromise = gateway.callTool( + { name: 'read_file', arguments: { filePath: 'test.ts' } }, + { abortSignal: controller.signal }, + ); + + expect(events).toHaveLength(1); + const requestId = events[0].payload.requestId; + + controller.abort('stopped by user'); + + await expect(callPromise).rejects.toMatchObject({ + name: 'AbortError', + message: 'stopped by user', + }); + + // A late gateway response must not settle a cleaned-up pending request. + expect( + gateway.resolveRequest(requestId, { + content: [{ type: 'text', text: 'late response' }], + }), + ).toBe(false); + }); + it('should dispatch different tool names correctly', async () => { gateway.init(EMPTY_CAPABILITIES); diff --git a/packages/cli/src/modules/instance-ai/browser/composite-local-mcp-server.ts b/packages/cli/src/modules/instance-ai/browser/composite-local-mcp-server.ts index e7ef8a14ee1..a6c453f4989 100644 --- a/packages/cli/src/modules/instance-ai/browser/composite-local-mcp-server.ts +++ b/packages/cli/src/modules/instance-ai/browser/composite-local-mcp-server.ts @@ -40,7 +40,10 @@ export class CompositeLocalMcpServer implements LocalMcpServer { return this.unwrapTools(this.availableToolsByCategory.get(category)!); } - async callTool(req: McpToolCallRequest): Promise { + async callTool( + req: McpToolCallRequest, + options?: { abortSignal?: AbortSignal }, + ): Promise { const serverTool = this.availableTools.get(req.name); if (!serverTool) { return { @@ -49,7 +52,7 @@ export class CompositeLocalMcpServer implements LocalMcpServer { }; } - return await serverTool.server.callTool(req); + return await serverTool.server.callTool(req, options); } private unwrapTools(tools: CompositeLocalMcpServerToolMap) { diff --git a/packages/cli/src/modules/instance-ai/filesystem/local-gateway.ts b/packages/cli/src/modules/instance-ai/filesystem/local-gateway.ts index 43ff1140efc..f3f65ae9066 100644 --- a/packages/cli/src/modules/instance-ai/filesystem/local-gateway.ts +++ b/packages/cli/src/modules/instance-ai/filesystem/local-gateway.ts @@ -185,9 +185,12 @@ export class LocalGateway { /** * Dispatch an MCP tool call to the remote client and await its result. - * Throws if not connected or if the request times out. + * Throws if not connected, if the request times out, or if aborted. */ - async callTool(toolCall: McpToolCallRequest): Promise { + async callTool( + toolCall: McpToolCallRequest, + options?: { abortSignal?: AbortSignal }, + ): Promise { if (!this._connected) { throw new Error('Local gateway is not connected'); } @@ -200,15 +203,63 @@ export class LocalGateway { }; } + const abortSignal = options?.abortSignal; + if (abortSignal?.aborted) { + const error = new Error( + typeof abortSignal.reason === 'string' ? abortSignal.reason : 'This operation was aborted', + ); + error.name = 'AbortError'; + throw error; + } + const requestId = `gw_${nanoid()}`; return await new Promise((resolve, reject) => { - const timer = setTimeout(() => { + let settled = false; + + const settle = (action: () => void) => { + if (settled) return; + settled = true; + clearTimeout(timer); + abortSignal?.removeEventListener('abort', onAbort); this.pendingRequests.delete(requestId); - reject(new Error(`Local gateway request timed out after ${REQUEST_TIMEOUT_MS}ms`)); + action(); + }; + + const onAbort = () => { + settle(() => { + const error = new Error( + typeof abortSignal?.reason === 'string' + ? abortSignal.reason + : 'This operation was aborted', + ); + error.name = 'AbortError'; + reject(error); + }); + }; + + const timer = setTimeout(() => { + settle(() => { + reject(new Error(`Local gateway request timed out after ${REQUEST_TIMEOUT_MS}ms`)); + }); }, REQUEST_TIMEOUT_MS); - this.pendingRequests.set(requestId, { resolve, reject, timer, toolCall }); + abortSignal?.addEventListener('abort', onAbort, { once: true }); + + this.pendingRequests.set(requestId, { + resolve: (result) => { + settle(() => { + resolve(result); + }); + }, + reject: (error) => { + settle(() => { + reject(error); + }); + }, + timer, + toolCall, + }); this.emitter.emit('filesystem-request', { type: 'filesystem-request', diff --git a/packages/cli/src/modules/instance-ai/instance-ai.adapter.service.ts b/packages/cli/src/modules/instance-ai/instance-ai.adapter.service.ts index c05649e3ade..3998e756368 100644 --- a/packages/cli/src/modules/instance-ai/instance-ai.adapter.service.ts +++ b/packages/cli/src/modules/instance-ai/instance-ai.adapter.service.ts @@ -110,6 +110,7 @@ import { FORM_TRIGGER_NODE_TYPE, WEBHOOK_NODE_TYPE, SCHEDULE_TRIGGER_NODE_TYPE, + ManualExecutionCancelledError, TimeoutExecutionCancelledError, UnexpectedError, UserError, @@ -1251,8 +1252,9 @@ export class InstanceAiAdapterService { } }; - // Wait for completion with timeout protection + // Wait for completion with timeout / abort protection const timeoutMs = Math.min(options?.timeout ?? DEFAULT_TIMEOUT_MS, MAX_TIMEOUT_MS); + const abortSignal = options?.abortSignal; if (activeExecutions.has(executionId)) { let timeoutId: NodeJS.Timeout | undefined; @@ -1262,28 +1264,60 @@ export class InstanceAiAdapterService { }, timeoutMs); }); + let onAbort: (() => void) | undefined; + const abortPromise = + abortSignal === undefined + ? undefined + : new Promise((_, reject) => { + onAbort = () => { + const error = new Error( + typeof abortSignal.reason === 'string' + ? abortSignal.reason + : 'This operation was aborted', + ); + error.name = 'AbortError'; + reject(error); + }; + if (abortSignal.aborted) { + onAbort(); + return; + } + abortSignal.addEventListener('abort', onAbort, { once: true }); + }); + try { await Promise.race([ activeExecutions.getPostExecutePromise(executionId), timeoutPromise, + ...(abortPromise ? [abortPromise] : []), ]); clearTimeout(timeoutId); + if (onAbort) abortSignal?.removeEventListener('abort', onAbort); } catch (error) { clearTimeout(timeoutId); - // On timeout, cancel the execution - if (error instanceof Error && error.message.includes('timed out')) { + if (onAbort) abortSignal?.removeEventListener('abort', onAbort); + const isTimeout = error instanceof Error && error.message.includes('timed out'); + const isAbort = + error instanceof Error && + (error.name === 'AbortError' || abortSignal?.aborted === true); + // On timeout or abort, cancel the execution with the matching reason + if (isTimeout || isAbort) { try { activeExecutions.stopExecution( executionId, - new TimeoutExecutionCancelledError(executionId), + isAbort + ? new ManualExecutionCancelledError(executionId) + : new TimeoutExecutionCancelledError(executionId), ); } catch { - // Execution may have completed between timeout and cancel + // Execution may have completed between timeout/abort and cancel } const result = { executionId, status: 'error', - error: `Execution timed out after ${timeoutMs}ms and was cancelled`, + error: isAbort + ? 'Execution was cancelled' + : `Execution timed out after ${timeoutMs}ms and was cancelled`, } satisfies ExecutionResult; await pruneVerificationPins(); trackBuilderExecutedWorkflow(result.status, result.error); diff --git a/packages/workflow/tsconfig.json b/packages/workflow/tsconfig.json index b38624ab915..8460a3ea842 100644 --- a/packages/workflow/tsconfig.json +++ b/packages/workflow/tsconfig.json @@ -5,6 +5,7 @@ "noUncheckedIndexedAccess": false, "types": ["vite/client", "vitest/globals"], "paths": { + "n8n-workflow": ["./src/index.ts"], "esprima-next": ["./node_modules/esprima-next/dist/esm/esprima"] } },