From 1d00c1ae806acef961708b4f96ee11a4730e57a9 Mon Sep 17 00:00:00 2001 From: Eugene Molodkin Date: Fri, 16 Oct 2026 14:05:24 +0200 Subject: [PATCH 1/5] fix(core): Separate Agent foreground stops from background work (no-changelog) --- .../runtime/__tests__/agent-runtime.test.ts | 56 ++++++++++++++++++ .../runtime/guardrails/guardrail-runner.ts | 8 ++- .../src/runtime/tools/tool-call-runner.ts | 14 ++++- .../@n8n/agents/src/types/sdk/guardrail.ts | 8 ++- packages/@n8n/api-types/src/agents/dto.ts | 4 ++ .../decorators/src/pubsub/pubsub-metadata.ts | 1 + .../agent-chat-execution.service.test.ts | 59 ++++++++++++++++++- ...ent-message-queue-consumer.service.test.ts | 46 +++++++++++++++ .../agent-turn-execution.service.test.ts | 52 ++++++++++++++++ .../__tests__/execution-recorder.test.ts | 36 +++++++++++ .../agents/agent-chat-execution.service.ts | 50 +++++++++++++--- .../modules/agents/agent-chat.controller.ts | 6 ++ .../agent-message-queue-consumer.service.ts | 9 +++ .../agents/agent-turn-execution.service.ts | 19 ++++++ .../src/modules/agents/execution-recorder.ts | 5 ++ .../utils/execution-to-message-mapper.ts | 1 + packages/cli/src/scaling/constants.ts | 1 + .../src/scaling/pubsub/pubsub.event-map.ts | 9 +++ .../cli/src/scaling/pubsub/pubsub.types.ts | 2 + 19 files changed, 370 insertions(+), 16 deletions(-) create mode 100644 packages/cli/src/modules/agents/__tests__/agent-turn-execution.service.test.ts diff --git a/packages/@n8n/agents/src/runtime/__tests__/agent-runtime.test.ts b/packages/@n8n/agents/src/runtime/__tests__/agent-runtime.test.ts index 5aa7cbc30ed..a2c7556728a 100644 --- a/packages/@n8n/agents/src/runtime/__tests__/agent-runtime.test.ts +++ b/packages/@n8n/agents/src/runtime/__tests__/agent-runtime.test.ts @@ -1815,6 +1815,62 @@ describe('AgentRuntime — guardrails', () => { expect(call.state === 'rejected' && call.error).toContain('tool.stop'); }); + it.each(['generate', 'stream'] as const)( + 'records canceled calls without errors and stops the next %s model turn', + async (mode) => { + const canceled: GuardrailDecision = { action: 'stop', code: 'user.cancel', canceled: true }; + const calls = ['tc-1', 'tc-2'].map((toolCallId) => ({ + toolCallId, + toolName: 'echo', + args: { v: 'x' }, + })); + generateText.mockResolvedValueOnce(makeGenerateWithToolCalls(calls)); + streamText.mockReturnValueOnce(makeStreamWithToolCalls(calls)); + const handler = vi.fn(async () => 'Must not run'); + const hook = makeGuardrail({ + before: vi.fn().mockResolvedValueOnce({ action: 'allow' }).mockResolvedValue(canceled), + beforeTool: vi.fn().mockResolvedValue(canceled), + }); + const runtime = createRuntimeWithEchoTool(handler); + const options = { guardrails: guardrailsOption(hook) }; + if (mode === 'generate') { + const result = await runtime.generate('go', options); + const blocks = result.messages.flatMap((message) => + isLlmMessage(message) + ? message.content.filter((block) => block.type === 'tool-call') + : [], + ); + expect(blocks).toHaveLength(2); + expect(blocks).toEqual([ + expect.objectContaining({ state: 'resolved', canceled: true }), + expect.objectContaining({ state: 'resolved', canceled: true }), + ]); + expect(result.error).toBeUndefined(); + expect(result.guardrail).toEqual({ code: 'user.cancel', canceled: true }); + expect(generateText).toHaveBeenCalledTimes(1); + } else { + const { stream } = await runtime.stream('go', options); + const chunks = await collectChunks(stream); + const results = chunks.filter((chunk) => chunk.type === 'tool-result'); + expect(results).toHaveLength(2); + for (const result of results) { + expect(result.canceled).toBe(true); + expect(result.output).toContain('user.cancel'); + expect(result.output).not.toContain('User said'); + expect(result.isError).not.toBe(true); + } + expect(chunks.some((chunk) => chunk.type === 'error')).toBe(false); + expect(chunks.at(-1)).toMatchObject({ + type: 'finish', + guardrail: { code: 'user.cancel', canceled: true }, + }); + expect(streamText).toHaveBeenCalledTimes(1); + } + expect(handler).not.toHaveBeenCalled(); + expect(hook.afterTool).not.toHaveBeenCalled(); + }, + ); + it('passes the handler result to afterTool() when the tool call is allowed', async () => { generateText .mockResolvedValueOnce(makeGenerateWithToolCall('tc-1', 'echo', { v: 'x' })) diff --git a/packages/@n8n/agents/src/runtime/guardrails/guardrail-runner.ts b/packages/@n8n/agents/src/runtime/guardrails/guardrail-runner.ts index c67b96204f6..fe290702bfb 100644 --- a/packages/@n8n/agents/src/runtime/guardrails/guardrail-runner.ts +++ b/packages/@n8n/agents/src/runtime/guardrails/guardrail-runner.ts @@ -46,7 +46,9 @@ export class GuardrailRunner { async before(ctx: GuardrailModelCallContext): Promise { for (const hook of this.hooks) { const decision = await hook.before?.(ctx); - if (decision?.action === 'stop') return { code: decision.code }; + if (decision?.action === 'stop') { + return { code: decision.code, ...(decision.canceled ? { canceled: true } : {}) }; + } } return undefined; } @@ -60,7 +62,9 @@ export class GuardrailRunner { async beforeTool(ctx: GuardrailToolCallContext): Promise { for (const hook of this.hooks) { const decision = await hook.beforeTool?.(ctx); - if (decision?.action === 'stop') return { code: decision.code }; + if (decision?.action === 'stop') { + return { code: decision.code, ...(decision.canceled ? { canceled: true } : {}) }; + } } return undefined; } diff --git a/packages/@n8n/agents/src/runtime/tools/tool-call-runner.ts b/packages/@n8n/agents/src/runtime/tools/tool-call-runner.ts index 89acc9f8488..84bc8948364 100644 --- a/packages/@n8n/agents/src/runtime/tools/tool-call-runner.ts +++ b/packages/@n8n/agents/src/runtime/tools/tool-call-runner.ts @@ -137,6 +137,12 @@ export class ToolCallRunner { if (guardrails && guardCtx && !params.previouslySuspended) { const stop = await guardrails.beforeTool(guardCtx); if (stop) { + if (stop.canceled) + return this.buildCancelledOutcome( + params, + 'Tool call canceled', + `[Tool call stopped by guardrail: ${stop.code}]`, + ); return await this.toolError( params, new Error(`Tool call stopped by guardrail: ${stop.code}`), @@ -183,7 +189,11 @@ export class ToolCallRunner { suspension.abortObserved = true; await this.cleanupAbortedTool(params, builtTool, input, suspension); this.deps.onCancelled(); - return this.buildCancelledOutcome(params, 'Run aborted'); + return this.buildCancelledOutcome( + params, + 'Run aborted', + '[Tool call canceled because the run stopped]', + ); } return await this.toolError(params, error, builtTool); } @@ -324,9 +334,9 @@ export class ToolCallRunner { private buildCancelledOutcome( params: ProcessToolCallParams, userMessage: string, + modelOutput = `[Tool call cancelled. User said: "${userMessage}"]`, ): ToolCallOutcome { const { toolCallId, toolName, input, list } = params; - const modelOutput = `[Tool call cancelled. User said: "${userMessage}"]`; this.eventBus.emit({ type: AgentEvent.ToolExecutionEnd, toolCallId, diff --git a/packages/@n8n/agents/src/types/sdk/guardrail.ts b/packages/@n8n/agents/src/types/sdk/guardrail.ts index 8512501e46e..cc0c93e51ef 100644 --- a/packages/@n8n/agents/src/types/sdk/guardrail.ts +++ b/packages/@n8n/agents/src/types/sdk/guardrail.ts @@ -12,9 +12,13 @@ export interface BuiltGuardrail { /** @internal */ readonly _config: Record; } -export type GuardrailDecision = { action: 'allow' } | { action: 'stop'; code: string }; +export type GuardrailDecision = { action: 'allow' } | ({ action: 'stop' } & GuardrailStop); -export type GuardrailStop = { code: string }; +export type GuardrailStop = { + code: string; + /** Record a stopped model or tool call as canceled. */ + canceled?: true; +}; export type GuardrailModelCallSource = 'turn' | 'title' | 'memory'; export interface GuardrailModelCallContext { diff --git a/packages/@n8n/api-types/src/agents/dto.ts b/packages/@n8n/api-types/src/agents/dto.ts index 30e2e8c1eb7..58d4a8e8ca6 100644 --- a/packages/@n8n/api-types/src/agents/dto.ts +++ b/packages/@n8n/api-types/src/agents/dto.ts @@ -453,3 +453,7 @@ export interface VectorStoreTestResult { message?: string; warning?: string; } + +export class AgentChatCancelDto extends Z.class({ + scope: z.enum(['foreground', 'all']).optional(), +}) {} diff --git a/packages/@n8n/decorators/src/pubsub/pubsub-metadata.ts b/packages/@n8n/decorators/src/pubsub/pubsub-metadata.ts index 31c331c50ee..22bf8afad11 100644 --- a/packages/@n8n/decorators/src/pubsub/pubsub-metadata.ts +++ b/packages/@n8n/decorators/src/pubsub/pubsub-metadata.ts @@ -29,6 +29,7 @@ export type PubSubEventName = | 'relay-agent-execution-update' | 'relay-agent-queued-chat' | 'cancel-agent-chat-execution' + | 'cancel-agent-chat-foreground' | 'relay-agent-message-queue-update' | 'relay-agent-background-tasks-update' | 'relay-agent-update' diff --git a/packages/cli/src/modules/agents/__tests__/agent-chat-execution.service.test.ts b/packages/cli/src/modules/agents/__tests__/agent-chat-execution.service.test.ts index d355f97a5ff..14167db7f5d 100644 --- a/packages/cli/src/modules/agents/__tests__/agent-chat-execution.service.test.ts +++ b/packages/cli/src/modules/agents/__tests__/agent-chat-execution.service.test.ts @@ -1,4 +1,3 @@ -import type { AgentMessageSteeringService } from '../agent-message-steering.service'; import type { SerializableAgentState } from '@n8n/agents'; import { LockService } from '@n8n/backend-common'; import { Container } from '@n8n/di'; @@ -7,6 +6,7 @@ import type { InstanceSettings } from 'n8n-core'; import { mock } from 'vitest-mock-extended'; import { NotFoundError } from '@n8n/errors'; +import type { AgentMessageSteeringService } from '../agent-message-steering.service'; import type { Publisher } from '@/scaling/pubsub/publisher.service'; import type { AgentBackgroundJobService } from '../background/agent-background-job.service'; @@ -359,3 +359,60 @@ it('rejects preview and foreign executions on the production cancel route', asyn service.requestCancel({ ...production, userId: 'other-user' }), ).rejects.toBeInstanceOf(NotFoundError); }); + +it('foreground Stop leaves detached jobs running locally and on other mains', async () => { + const { service, backgroundJobs, publisher } = makeService(); + const controller = new AbortController(); + service.register(context, controller); + await service.requestCancel({ ...context, scope: 'foreground' }); + expect(controller.signal.aborted).toBe(true); + expect(backgroundJobs.cancelForParent).not.toHaveBeenCalled(); + await service.handleForegroundCancel(context); + expect(backgroundJobs.cancelForParent).not.toHaveBeenCalled(); + expect(publisher.publishCommand).not.toHaveBeenCalled(); +}); + +it('relays foreground scope before a remote execution registers', async () => { + const mainA = makeService(); + const mainB = makeService(); + await mainB.service.requestCancel({ ...context, scope: 'foreground' }); + const { surface: _surface, ...wireContext } = context; + const payload = { ...wireContext, productionN8nChat: false }; + expect(mainB.publisher.publishCommand).toHaveBeenCalledWith({ + command: 'cancel-agent-chat-foreground', + payload, + }); + await mainA.service.handleForegroundCancel(payload); + const controller = new AbortController(); + mainA.service.register(context, controller); + expect(controller.signal.aborted).toBe(true); + expect(controller.signal.reason).toMatchObject({ + name: 'AbortError', + message: 'Response stopped', + }); + expect(mainA.backgroundJobs.cancelForParent).not.toHaveBeenCalled(); + expect(mainB.backgroundJobs.cancelForParent).not.toHaveBeenCalled(); +}); + +it('foreground Stop removes a suspended checkpoint without canceling detached jobs', async () => { + const { service, repository, checkpointStorage, backgroundJobs } = makeService(); + const suspended = { ...running, status: 'success' as const, hitlStatus: 'suspended' as const }; + repository.findOneBy.mockResolvedValue(suspended); + repository.findLatestByThreadId.mockResolvedValue(suspended); + await service.requestCancel({ ...context, scope: 'foreground' }); + expect(checkpointStorage.delete).toHaveBeenCalledWith('run-1', 'agent-1'); + expect(backgroundJobs.cancelForParent).not.toHaveBeenCalled(); +}); + +it('cancels detached jobs on the owner before a relayed run registers', async () => { + const { service, backgroundJobs } = makeService(); + await service.handleCancel(context); + expect(backgroundJobs.cancelForParent).toHaveBeenCalledWith( + context.agentId, + context.threadId, + 'draft-chat:user-1', + ); + const controller = new AbortController(); + service.register(context, controller); + expect(controller.signal.aborted).toBe(true); +}); diff --git a/packages/cli/src/modules/agents/__tests__/agent-message-queue-consumer.service.test.ts b/packages/cli/src/modules/agents/__tests__/agent-message-queue-consumer.service.test.ts index 2bf38c8a226..1cc081cfc97 100644 --- a/packages/cli/src/modules/agents/__tests__/agent-message-queue-consumer.service.test.ts +++ b/packages/cli/src/modules/agents/__tests__/agent-message-queue-consumer.service.test.ts @@ -122,6 +122,26 @@ describe('AgentMessageQueueConsumer', () => { afterEach(() => consumer.stop()); + it('keeps ownership loss separate from a user Stop', async () => { + const item = claim('session'); + const lease = new AbortController(); + const failure = new DOMException('Execution ownership lost', 'AbortError'); + executions.getAbortSignal.mockReturnValue(lease.signal); + repository.findThreadIds.mockResolvedValue(['session']); + queue.claimNext.mockResolvedValueOnce(item).mockResolvedValue(null); + testRuns.executePreparedDraftRun.mockImplementation(async () => { + lease.abort(failure); + throw failure; + }); + consumer.start(); + await vi.waitFor(() => + expect(queue.settle).toHaveBeenCalledWith('session', item.admission.executionId), + ); + const userSignal = chatExecutions.register.mock.calls[0][1].signal; + expect(userSignal.aborted).toBe(false); + expect(queue.recordFailure).toHaveBeenCalledWith(item, failure, userSignal); + expect(sender.send).toHaveBeenCalledWith(expect.objectContaining({ type: 'error' })); + }); it('runs independent sessions without waiting and reuses each claimed execution', async () => { const first = claim('first'); first.payload.message = 'edited first message'; @@ -335,6 +355,32 @@ describe('AgentMessageQueueConsumer', () => { expect(sender.send).toHaveBeenCalledWith(expect.objectContaining({ type: 'error' })); }); + it('ends delivery without an error when Stop follows queue acceptance', async () => { + const item = claim('session'); + repository.findThreadIds.mockResolvedValue(['session']); + queue.claimNext.mockResolvedValueOnce(item).mockResolvedValue(null); + chatExecutions.register.mockImplementation((_context, controller) => { + controller.abort(new DOMException('Response stopped', 'AbortError')); + }); + consumer.start(); + await vi.waitFor(() => + expect(queue.settle).toHaveBeenCalledWith('session', item.admission.executionId), + ); + expect(queue.recordFailure).toHaveBeenCalledWith( + item, + expect.objectContaining({ name: 'AbortError' }), + expect.objectContaining({ aborted: true }), + ); + expect(testRuns.executePreparedDraftRun).not.toHaveBeenCalled(); + expect(sender.send).not.toHaveBeenCalledWith(expect.objectContaining({ type: 'error' })); + expect(sender.send).toHaveBeenCalledWith({ + type: 'done', + sessionId: 'session', + executionId: item.admission.executionId, + }); + expect(sender.close).toHaveBeenCalledOnce(); + }); + it('leaves integration work pending until its bridge is available and stops admission during shutdown', async () => { const item = claim('session', true); const checked = createDeferredPromise(); diff --git a/packages/cli/src/modules/agents/__tests__/agent-turn-execution.service.test.ts b/packages/cli/src/modules/agents/__tests__/agent-turn-execution.service.test.ts new file mode 100644 index 00000000000..a900549b6d5 --- /dev/null +++ b/packages/cli/src/modules/agents/__tests__/agent-turn-execution.service.test.ts @@ -0,0 +1,52 @@ +import { Logger } from '@n8n/backend-common'; +import { mock } from 'vitest-mock-extended'; + +import { AgentChatExecutionService } from '../agent-chat-execution.service'; +import { AgentExecutionService } from '../agent-execution.service'; +import { AgentMessageQueueService } from '../agent-message-queue.service'; +import { AgentMessageSteeringService } from '../agent-message-steering.service'; +import { AgentToolApprovalService } from '../agent-tool-approval.service'; +import { AgentTurnExecutionService, type AgentTurnRequest } from '../agent-turn-execution.service'; + +it('ends a preview turn stopped during preparation without recording an error', async () => { + const executions = mock(); + const service = new AgentTurnExecutionService( + mock(), + executions, + mock(), + mock(), + mock(), + mock(), + ); + const controller = new AbortController(); + const turn: AgentTurnRequest = { + type: 'start', + input: 'Continue', + options: { abortSignal: controller.signal }, + recording: { + projectId: 'project', + agentId: 'agent', + agentName: 'Agent', + threadId: 'thread', + resourceId: 'draft-chat:user', + userMessage: 'Continue', + source: 'chat', + access: { accessScope: 'user', ownerId: 'user' }, + }, + }; + const chunks = []; + for await (const chunk of service.execute({ + agentInstance: mock(), + toolRegistry: new Map(), + mcpServerAttributions: new Map(), + context: turn.recording, + chatSurface: 'preview', + prepare: async () => { + controller.abort(); + return turn; + }, + })) + chunks.push(chunk); + expect(chunks).toEqual([]); + expect(executions.startExecutionRecording).not.toHaveBeenCalled(); +}); diff --git a/packages/cli/src/modules/agents/__tests__/execution-recorder.test.ts b/packages/cli/src/modules/agents/__tests__/execution-recorder.test.ts index 00faf14b62f..beed4ec2d0a 100644 --- a/packages/cli/src/modules/agents/__tests__/execution-recorder.test.ts +++ b/packages/cli/src/modules/agents/__tests__/execution-recorder.test.ts @@ -2,6 +2,7 @@ import type { BuiltTool, StreamChunk } from '@n8n/agents'; import { ExecutionRecorder, type TimelineEvent } from '../execution-recorder'; import { buildToolRegistry } from '../tool-registry'; +import { executionToMessagesDto } from '../utils/execution-to-message-mapper'; function makeToolCallChunk(toolName: string, input: unknown, toolCallId = 'tc1'): StreamChunk { return { type: 'tool-call', toolCallId, toolName, input } satisfies StreamChunk; @@ -12,6 +13,41 @@ function makeToolResultChunk(toolName: string, output: unknown, toolCallId = 'tc } describe('ExecutionRecorder', () => { + it.each([true, false])( + 'keeps canceled tool state in saved chat messages (started: %s)', + (started) => { + const recorder = new ExecutionRecorder(); + if (started) recorder.record(makeToolCallChunk('update_plan', {}, 'tc-canceled')); + recorder.record({ + type: 'tool-result', + toolCallId: 'tc-canceled', + toolName: 'update_plan', + output: 'Tool call canceled', + canceled: true, + }); + const { timeline } = recorder.getMessageRecord(); + expect(timeline[0]).toMatchObject({ type: 'tool-call', canceled: true, success: true }); + const messages = executionToMessagesDto({ + id: 'execution-1', + userMessage: 'Update the comparison', + author: null, + timeline, + attachments: null, + status: 'cancelled', + error: null, + createdAt: new Date(), + }); + expect(messages.find((message) => message.role === 'assistant')?.content).toEqual([ + expect.objectContaining({ + type: 'tool-call', + toolCallId: 'tc-canceled', + state: 'resolved', + canceled: true, + }), + ]); + }, + ); + it('scrubs titles in the initial signal', () => { const recorder = new ExecutionRecorder(undefined, undefined, { tasks: [ diff --git a/packages/cli/src/modules/agents/agent-chat-execution.service.ts b/packages/cli/src/modules/agents/agent-chat-execution.service.ts index eb970155e3a..bf57a3bfe18 100644 --- a/packages/cli/src/modules/agents/agent-chat-execution.service.ts +++ b/packages/cli/src/modules/agents/agent-chat-execution.service.ts @@ -33,6 +33,7 @@ type CancelExecutionPayload = PubSubCommandMap['cancel-agent-chat-execution']; // compatibility with mains still running the previous version. type ExecutionContext = Omit & { surface: AgentChatSurface; + scope?: 'foreground' | 'all'; }; export interface CancelSuspendedRunParams { @@ -134,18 +135,22 @@ export class AgentChatExecutionService { if (execution.status !== 'running') return await this.cancelRecordedSuspension(context); this.cancelOrRemember(context); if (!this.instanceSettings.isMultiMain) return true; - const { surface, ...rest } = context; + const { surface, scope, ...rest } = context; await this.publisher.publishCommand({ - command: 'cancel-agent-chat-execution', + command: + scope === 'foreground' + ? 'cancel-agent-chat-foreground' + : 'cancel-agent-chat-execution', payload: { ...rest, productionN8nChat: surface === 'n8n-chat' }, }); return true; } finally { - await this.backgroundJobService.cancelForParent( - context.agentId, - context.threadId, - chatSurfaceMemoryResourceId(context.surface, context.userId), - ); + if (context.scope !== 'foreground') + await this.backgroundJobService.cancelForParent( + context.agentId, + context.threadId, + chatSurfaceMemoryResourceId(context.surface, context.userId), + ); } }, ); @@ -153,9 +158,22 @@ export class AgentChatExecutionService { @OnPubSubEvent('cancel-agent-chat-execution', { instanceType: 'main' }) async handleCancel(payload: CancelExecutionPayload): Promise { + await this.handleCancellation(payload); + } + + @OnPubSubEvent('cancel-agent-chat-foreground', { instanceType: 'main' }) + async handleForegroundCancel(payload: CancelExecutionPayload): Promise { + await this.handleCancellation(payload, 'foreground'); + } + + private async handleCancellation( + payload: CancelExecutionPayload, + scope?: ExecutionContext['scope'], + ): Promise { const { productionN8nChat, ...rest } = payload; const context: ExecutionContext = { ...rest, + scope, surface: productionN8nChat ? 'n8n-chat' : 'preview', }; if (await this.cancelLocalWithChildren(context)) return; @@ -166,7 +184,15 @@ export class AgentChatExecutionService { if (await this.cancelLocalWithChildren(context)) return; const execution = await this.getOwnedExecution(context); if (!execution) return; - if (execution.status === 'running') this.cancelOrRemember(context); + if (execution.status === 'running') { + this.cancelOrRemember(context); + if (context.scope !== 'foreground') + await this.backgroundJobService.cancelForParent( + context.agentId, + context.threadId, + chatSurfaceMemoryResourceId(context.surface, context.userId), + ); + } await this.cancelRecordedSuspension(context); }, ); @@ -174,6 +200,7 @@ export class AgentChatExecutionService { private async cancelLocalWithChildren(context: ExecutionContext): Promise { if (!this.cancelLocal(context)) return false; + if (context.scope === 'foreground') return true; await this.backgroundJobService.cancelForParent( context.agentId, context.threadId, @@ -203,7 +230,11 @@ export class AgentChatExecutionService { execution.context.userId !== context.userId ) return false; - execution.controller.abort(PARENT_TASK_CANCELLED_REASON); + execution.controller.abort( + context.scope === 'foreground' + ? new DOMException('Response stopped', 'AbortError') + : PARENT_TASK_CANCELLED_REASON, + ); return true; } @@ -240,6 +271,7 @@ export class AgentChatExecutionService { return await this.cancelSuspended({ agentId: context.agentId, runId: pending.runId, + cancelBackgroundJobs: context.scope !== 'foreground', resourceId: chatSurfaceMemoryResourceId(context.surface, context.userId), }); } diff --git a/packages/cli/src/modules/agents/agent-chat.controller.ts b/packages/cli/src/modules/agents/agent-chat.controller.ts index a7b032a5951..ac39d74a8d6 100644 --- a/packages/cli/src/modules/agents/agent-chat.controller.ts +++ b/packages/cli/src/modules/agents/agent-chat.controller.ts @@ -2,6 +2,7 @@ import { type AgentBackgroundJobsResponse, type AgentChatAttachmentPayload, AgentChatMessageDto, + AgentChatCancelDto, AgentChatQueueUpdateDto, AgentChatQueueSteerDto, AgentChatQueueReorderDto, @@ -24,6 +25,7 @@ import { Patch, Post, ProjectScope, + Query, RestController, } from '@n8n/decorators'; import { scrubSecretsInText } from '@n8n/utils/scrub-secrets'; @@ -457,6 +459,7 @@ export class AgentChatController { @Param('agentId') agentId: string, @Param('threadId') threadId: string, @Param('executionId') executionId: string, + @Query payload: AgentChatCancelDto = {}, ) { const cancelRequested = await this.chatExecutionService.requestCancel({ projectId: req.params.projectId, @@ -465,6 +468,7 @@ export class AgentChatController { executionId, userId: req.user.id, surface: 'preview', + ...(payload.scope ? { scope: payload.scope } : {}), }); return { cancelRequested }; } @@ -476,6 +480,7 @@ export class AgentChatController { _res: Response, @Param('agentId') agentId: string, @Param('runId') runId: string, + @Query payload: AgentChatCancelDto = {}, ) { const { projectId } = req.params; const agent = await this.agentsService.findById(agentId, projectId); @@ -485,6 +490,7 @@ export class AgentChatController { agentId, runId, resourceId: draftChatMemoryResourceId(req.user.id), + ...(payload.scope === 'foreground' ? { cancelBackgroundJobs: false } : {}), }); return { cancelled }; } diff --git a/packages/cli/src/modules/agents/agent-message-queue-consumer.service.ts b/packages/cli/src/modules/agents/agent-message-queue-consumer.service.ts index a2f4f51f853..9b1da326dfa 100644 --- a/packages/cli/src/modules/agents/agent-message-queue-consumer.service.ts +++ b/packages/cli/src/modules/agents/agent-message-queue-consumer.service.ts @@ -166,6 +166,15 @@ export class AgentMessageQueueConsumer { } } catch (error) { await this.queue.recordFailure(claim, error, controller.signal); + const { isAbortError } = await import('@n8n/agents'); + if ( + controller.signal.aborted && + isAbortError(controller.signal.reason) && + isAbortError(error) + ) { + sender?.send({ type: 'done', sessionId: thread.id, executionId: admission.executionId }); + return; + } sender?.send(toChatErrorEvent(error, 'Chat failed')); this.logger.warn('Queued agent message failed', { queueId: item.id, diff --git a/packages/cli/src/modules/agents/agent-turn-execution.service.ts b/packages/cli/src/modules/agents/agent-turn-execution.service.ts index 4a3b59966f7..6287d8a4c3a 100644 --- a/packages/cli/src/modules/agents/agent-turn-execution.service.ts +++ b/packages/cli/src/modules/agents/agent-turn-execution.service.ts @@ -138,6 +138,11 @@ export class AgentTurnExecutionService { const stream = await this.admitTurn(preparedTurn, config, recorder, state, chatControl); yield* this.streamTurn(stream, preparedTurn, config, recorder, state); } catch (error) { + if ( + (await this.isStoppedChatTurn(turn, config)) && + !(error instanceof AgentExecutionRecordingError) + ) + return; state.executionError = error; recorder.record({ type: 'error', error }); recorder.record({ type: 'finish', finishReason: 'error' }); @@ -155,6 +160,18 @@ export class AgentTurnExecutionService { } } + private async isStoppedChatTurn( + turn: AgentTurnRequest | undefined, + config: ExecuteTurnConfig, + ): Promise { + const signal = turn?.options.abortSignal; + if (!signal?.aborted || config.chatSurface === undefined) return false; + const { isAbortError } = await import('@n8n/agents'); + return Boolean( + config.chatSurface !== undefined && signal?.aborted && isAbortError(signal.reason), + ); + } + private createChatExecutionControl( turn: AgentTurnRequest, surface: AgentChatSurface, @@ -227,6 +244,8 @@ export class AgentTurnExecutionService { } continue; } + // The runtime also emits an error chunk for a requested stop. + if (chunk.type === 'error' && (await this.isStoppedChatTurn(turn, config))) continue; recorder.record(chunk); if (chunk.type === 'tool-call-suspended') state.suspendedRunId = chunk.runId; if (chunk.type === 'error') state.executionError = chunk.error; diff --git a/packages/cli/src/modules/agents/execution-recorder.ts b/packages/cli/src/modules/agents/execution-recorder.ts index cf111cb07e5..addc4235661 100644 --- a/packages/cli/src/modules/agents/execution-recorder.ts +++ b/packages/cli/src/modules/agents/execution-recorder.ts @@ -235,6 +235,7 @@ export type TimelineEvent = startTime: number; endTime: number; success: boolean; + canceled?: boolean; workflowId?: string; workflowName?: string; workflowExecutionId?: string; @@ -429,6 +430,7 @@ export class ExecutionRecorder { chunk.toolName, chunk.output, chunk.isError === true, + chunk.canceled === true, ); break; case 'finish': @@ -673,6 +675,7 @@ export class ExecutionRecorder { name: string, output: unknown, isError: boolean, + canceled: boolean, ): void { const recordedOutput = sanitizeExecutionLogValue( isError ? normaliseToolErrorOutput(output) : output, @@ -688,6 +691,7 @@ export class ExecutionRecorder { if (pendingTimeline) { pendingTimeline.output = recordedOutput; pendingTimeline.success = !isError; + if (canceled) pendingTimeline.canceled = true; // `tool-execution-end` (real per-tool finish) normally closed this entry // already; only fall back to the batched result time if it never fired. if (pendingTimeline.endTime === 0) { @@ -719,6 +723,7 @@ export class ExecutionRecorder { startTime: now, endTime: now, success: !isError, + ...(canceled ? { canceled: true } : {}), workflowId: entry?.workflowId, workflowName: entry?.workflowName, triggerType: entry?.triggerType, diff --git a/packages/cli/src/modules/agents/utils/execution-to-message-mapper.ts b/packages/cli/src/modules/agents/utils/execution-to-message-mapper.ts index 436c606fe35..ad401b329c3 100644 --- a/packages/cli/src/modules/agents/utils/execution-to-message-mapper.ts +++ b/packages/cli/src/modules/agents/utils/execution-to-message-mapper.ts @@ -90,6 +90,7 @@ function timelineToolCallToPart(event: ToolCallTimelineEvent): AgentPersistedMes toolName: event.name, toolCallId: event.toolCallId, input: event.input, + ...(event.canceled ? { canceled: true } : {}), ...(event.startTime > 0 ? { startTime: event.startTime } : {}), ...(event.endTime > 0 ? { endTime: event.endTime } : {}), ...(event.childTrace ? { childTrace: event.childTrace } : {}), diff --git a/packages/cli/src/scaling/constants.ts b/packages/cli/src/scaling/constants.ts index 4bbbde84df5..54fb43dd6da 100644 --- a/packages/cli/src/scaling/constants.ts +++ b/packages/cli/src/scaling/constants.ts @@ -47,6 +47,7 @@ export const IMMEDIATE_COMMANDS = new Set([ 'relay-agent-execution-update', 'relay-agent-queued-chat', 'cancel-agent-chat-execution', + 'cancel-agent-chat-foreground', 'relay-agent-message-queue-update', 'relay-agent-background-tasks-update', 'relay-agent-update', diff --git a/packages/cli/src/scaling/pubsub/pubsub.event-map.ts b/packages/cli/src/scaling/pubsub/pubsub.event-map.ts index 586eb262c27..59b5f650cb0 100644 --- a/packages/cli/src/scaling/pubsub/pubsub.event-map.ts +++ b/packages/cli/src/scaling/pubsub/pubsub.event-map.ts @@ -153,6 +153,15 @@ export type PubSubCommandMap = { productionN8nChat?: boolean; }; + 'cancel-agent-chat-foreground': { + projectId: string; + agentId: string; + threadId: string; + executionId: string; + userId: string; + productionN8nChat?: boolean; + }; + 'relay-agent-message-queue-update': { data: PushPayload<'agentMessageQueueUpdated'>; userIds: string[]; diff --git a/packages/cli/src/scaling/pubsub/pubsub.types.ts b/packages/cli/src/scaling/pubsub/pubsub.types.ts index 47bd1cf3714..4d1961d7913 100644 --- a/packages/cli/src/scaling/pubsub/pubsub.types.ts +++ b/packages/cli/src/scaling/pubsub/pubsub.types.ts @@ -63,6 +63,7 @@ export namespace PubSub { export type RelayAgentExecutionUpdate = ToCommand<'relay-agent-execution-update'>; export type RelayAgentQueuedChat = ToCommand<'relay-agent-queued-chat'>; export type CancelAgentChatExecution = ToCommand<'cancel-agent-chat-execution'>; + export type CancelAgentChatForeground = ToCommand<'cancel-agent-chat-foreground'>; export type RelayAgentMessageQueueUpdate = ToCommand<'relay-agent-message-queue-update'>; export type RelayAgentBackgroundJobsUpdate = ToCommand<'relay-agent-background-tasks-update'>; export type RelayAgentUpdate = ToCommand<'relay-agent-update'>; @@ -115,6 +116,7 @@ export namespace PubSub { | Commands.RelayAgentExecutionUpdate | Commands.RelayAgentQueuedChat | Commands.CancelAgentChatExecution + | Commands.CancelAgentChatForeground | Commands.RelayAgentMessageQueueUpdate | Commands.RelayAgentBackgroundJobsUpdate | Commands.RelayAgentUpdate From f474fcee42454eeb37bda29f88c4f7da495667a8 Mon Sep 17 00:00:00 2001 From: Eugene Molodkin Date: Fri, 16 Oct 2026 18:10:29 +0200 Subject: [PATCH 2/5] fix(core): Preserve Agent cancellation during task registration (no-changelog) --- .../agent-chat-execution.service.test.ts | 56 +++++++++++++++ ...ent-execution-orchestrator.service.test.ts | 2 +- .../agents/agent-chat-execution.service.ts | 48 ++++++++----- .../agent-background-job.service.test.ts | 71 +++++++++++++++++++ .../agent-background-job.service.ts | 46 +++++++++--- 5 files changed, 192 insertions(+), 31 deletions(-) diff --git a/packages/cli/src/modules/agents/__tests__/agent-chat-execution.service.test.ts b/packages/cli/src/modules/agents/__tests__/agent-chat-execution.service.test.ts index 14167db7f5d..2f3faaee937 100644 --- a/packages/cli/src/modules/agents/__tests__/agent-chat-execution.service.test.ts +++ b/packages/cli/src/modules/agents/__tests__/agent-chat-execution.service.test.ts @@ -10,6 +10,7 @@ import type { AgentMessageSteeringService } from '../agent-message-steering.serv import type { Publisher } from '@/scaling/pubsub/publisher.service'; import type { AgentBackgroundJobService } from '../background/agent-background-job.service'; +import { PARENT_TASK_CANCELLED_REASON } from '../background/sub-agent-background-state'; import { AgentChatExecutionService } from '../agent-chat-execution.service'; import type { AgentExecutionService } from '../agent-execution.service'; import type { AgentExecutionUpdateBroadcaster } from '../agent-execution-update-broadcaster'; @@ -416,3 +417,58 @@ it('cancels detached jobs on the owner before a relayed run registers', async () service.register(context, controller); expect(controller.signal.aborted).toBe(true); }); + +it('relays cancel-all after the foreground reply settles', async () => { + const mainA = makeService(); + const mainB = makeService(); + const controller = new AbortController(); + mainA.service.register(context, controller); + await mainA.service.requestCancel({ ...context, scope: 'foreground' }); + await mainA.service.settle(context.executionId, async () => {}); + const cancelled = { ...running, status: 'cancelled' as const }; + for (const main of [mainA, mainB]) { + main.repository.findOneBy.mockResolvedValue(cancelled); + main.repository.findLatestByThreadId.mockResolvedValue(cancelled); + } + + await mainB.service.requestCancel({ ...context, scope: 'all' }); + const { surface: _surface, ...wireContext } = context; + const payload = { ...wireContext, productionN8nChat: false }; + expect(mainB.publisher.publishCommand).toHaveBeenCalledWith({ + command: 'cancel-agent-chat-execution', + payload, + }); + await mainA.service.handleCancel(payload); + expect(mainA.backgroundJobs.cancelForParent).toHaveBeenCalledWith( + context.agentId, + context.threadId, + 'draft-chat:user-1', + ); +}); + +it('ignores a delayed cancel-all relay after a later reply settles', async () => { + const { service, repository, backgroundJobs } = makeService(); + repository.findOneBy.mockResolvedValue({ ...running, status: 'cancelled' }); + repository.findLatestByThreadId.mockResolvedValue({ + ...running, + id: 'execution-2', + status: 'success', + }); + await service.handleCancel(context); + expect(backgroundJobs.cancelForParent).not.toHaveBeenCalled(); +}); + +it.each([ + ['foreground', 'all'], + ['all', 'foreground'], +] as const)( + 'keeps the all scope before registration when requests arrive as %s then %s', + async (first, second) => { + const { service } = makeService(); + await service.requestCancel({ ...context, scope: first }); + await service.requestCancel({ ...context, scope: second }); + const controller = new AbortController(); + service.register(context, controller); + expect(controller.signal.reason).toBe(PARENT_TASK_CANCELLED_REASON); + }, +); diff --git a/packages/cli/src/modules/agents/__tests__/agent-execution-orchestrator.service.test.ts b/packages/cli/src/modules/agents/__tests__/agent-execution-orchestrator.service.test.ts index d986ae8b7f8..a05ab7cdae6 100644 --- a/packages/cli/src/modules/agents/__tests__/agent-execution-orchestrator.service.test.ts +++ b/packages/cli/src/modules/agents/__tests__/agent-execution-orchestrator.service.test.ts @@ -1135,7 +1135,7 @@ describe('AgentExecutionOrchestratorService', () => { userId, }); }); - await expect(collect(stream)).rejects.toMatchObject({ name: 'AbortError' }); + await expect(collect(stream)).resolves.toEqual([]); expect(sdkStart).not.toHaveBeenCalled(); expect(executionService.finalizeExecution).toHaveBeenCalledWith( 'execution-1', diff --git a/packages/cli/src/modules/agents/agent-chat-execution.service.ts b/packages/cli/src/modules/agents/agent-chat-execution.service.ts index bf57a3bfe18..01de9aabac2 100644 --- a/packages/cli/src/modules/agents/agent-chat-execution.service.ts +++ b/packages/cli/src/modules/agents/agent-chat-execution.service.ts @@ -132,18 +132,20 @@ export class AgentChatExecutionService { try { await this.steering.close(context.threadId, context.executionId); if (this.cancelLocal(context)) return true; - if (execution.status !== 'running') return await this.cancelRecordedSuspension(context); - this.cancelOrRemember(context); - if (!this.instanceSettings.isMultiMain) return true; - const { surface, scope, ...rest } = context; - await this.publisher.publishCommand({ - command: - scope === 'foreground' - ? 'cancel-agent-chat-foreground' - : 'cancel-agent-chat-execution', - payload: { ...rest, productionN8nChat: surface === 'n8n-chat' }, - }); - return true; + const running = execution.status === 'running'; + if (running) this.cancelOrRemember(context); + // A task insert can outlive the reply on another main. + if (this.instanceSettings.isMultiMain && (running || context.scope !== 'foreground')) { + const { surface, scope, ...rest } = context; + await this.publisher.publishCommand({ + command: + scope === 'foreground' + ? 'cancel-agent-chat-foreground' + : 'cancel-agent-chat-execution', + payload: { ...rest, productionN8nChat: surface === 'n8n-chat' }, + }); + } + return running || (await this.cancelRecordedSuspension(context)); } finally { if (context.scope !== 'foreground') await this.backgroundJobService.cancelForParent( @@ -186,13 +188,16 @@ export class AgentChatExecutionService { if (!execution) return; if (execution.status === 'running') { this.cancelOrRemember(context); - if (context.scope !== 'foreground') - await this.backgroundJobService.cancelForParent( - context.agentId, - context.threadId, - chatSurfaceMemoryResourceId(context.surface, context.userId), - ); + } else { + const latest = await this.executionRepository.findLatestByThreadId(context.threadId); + if (latest?.id !== context.executionId) return; } + if (context.scope !== 'foreground') + await this.backgroundJobService.cancelForParent( + context.agentId, + context.threadId, + chatSurfaceMemoryResourceId(context.surface, context.userId), + ); await this.cancelRecordedSuspension(context); }, ); @@ -211,7 +216,12 @@ export class AgentChatExecutionService { private cancelOrRemember(context: ExecutionContext): void { // Registration can finish while ownership validation awaits the database. - if (this.cancelLocal(context) || this.pendingCancellations.has(context.executionId)) return; + if (this.cancelLocal(context)) return; + const pending = this.pendingCancellations.get(context.executionId); + if (pending) { + if (context.scope !== 'foreground') pending.context = context; + return; + } const timer = setTimeout( () => this.pendingCancellations.delete(context.executionId), AgentChatExecutionService.PENDING_CANCEL_TTL_MS, diff --git a/packages/cli/src/modules/agents/background/__tests__/agent-background-job.service.test.ts b/packages/cli/src/modules/agents/background/__tests__/agent-background-job.service.test.ts index 463ad0ebca9..5fa3cb3e890 100644 --- a/packages/cli/src/modules/agents/background/__tests__/agent-background-job.service.test.ts +++ b/packages/cli/src/modules/agents/background/__tests__/agent-background-job.service.test.ts @@ -1,6 +1,7 @@ import type { Logger } from '@n8n/backend-common'; import type { AgentsConfig } from '@n8n/config'; import { Container } from '@n8n/di'; +import { createDeferredPromise } from '@n8n/utils/promise/deferred-promise'; import type { Mock } from 'vitest'; import { WorkflowOperationError } from 'n8n-workflow'; import { mock } from 'vitest-mock-extended'; @@ -12,6 +13,7 @@ import type { Publisher } from '@/scaling/pubsub/publisher.service'; import type { N8NCheckpointStorage } from '../../integrations/n8n-checkpoint-storage'; import type { AgentBackgroundJob } from '../../entities/agent-background-job.entity'; import type { AgentExecutionUpdateBroadcaster } from '../../agent-execution-update-broadcaster'; +import { AgentChatExecutionService } from '../../agent-chat-execution.service'; import type { AgentBackgroundJobRepository } from '../../repositories/agent-background-job.repository'; import type { AgentExecutionRepository } from '../../repositories/agent-execution.repository'; import type { AgentMessageRepository } from '../../repositories/agent-message.repository'; @@ -648,6 +650,75 @@ describe('background task notifications', () => { }); describe('registerSubAgentJob', () => { + it.each([false, true])( + 'handles a delayed registration after foreground Stop (cancel all: %s)', + async (cancelAll) => { + const { service, jobRepository } = setup(); + const chatExecutions = new AgentChatExecutionService( + mock(), + mock(), + mock(), + mock(), + mock(), + mock(), + mock(), + mock(), + service, + ); + const context = { + projectId: 'project-1', + agentId: 'agent-1', + threadId: 'thread-1', + executionId: 'execution-1', + userId: 'user-1', + }; + const controller = new AbortController(); + chatExecutions.register({ ...context, surface: 'preview' }, controller); + const inserted = createDeferredPromise(); + jobRepository.insertSubAgentJobIfCapacity.mockReturnValue(inserted.promise); + const registration = service.registerSubAgentJob(registerParams); + + await chatExecutions.handleForegroundCancel(context); + const foregroundReason: unknown = controller.signal.reason; + expect(foregroundReason).toMatchObject({ name: 'AbortError', message: 'Response stopped' }); + if (cancelAll) await chatExecutions.handleCancel(context); + expect(controller.signal.reason).toBe(foregroundReason); + expect(jobRepository.settleIfActive).not.toHaveBeenCalled(); + + jobRepository.findByParentThread.mockResolvedValue([makeJob()]); + inserted.resolve(true); + await registration; + + if (cancelAll) { + expect(jobRepository.settleIfActive).toHaveBeenCalledWith( + 'job-1', + { status: 'cancelled' }, + undefined, + ); + } else expect(jobRepository.settleIfActive).not.toHaveBeenCalled(); + + jobRepository.settleIfActive.mockClear(); + await service.registerSubAgentJob({ ...registerParams, id: 'job-2' }); + expect(jobRepository.settleIfActive).not.toHaveBeenCalled(); + }, + ); + + it.each([ + { parentAgentId: 'other-agent' }, + { parentThreadId: 'other-thread' }, + { parentResourceId: 'draft-chat:other-user' }, + ])('keeps an unrelated pending registration: %s', async (other) => { + const { service, jobRepository } = setup(); + const inserted = createDeferredPromise(); + jobRepository.insertSubAgentJobIfCapacity.mockReturnValue(inserted.promise); + const registration = service.registerSubAgentJob({ ...registerParams, ...other }); + await service.cancelForParent('agent-1', 'thread-1', 'draft-chat:user-1'); + jobRepository.findByParentThread.mockResolvedValue([makeJob(other)]); + inserted.resolve(true); + await registration; + expect(jobRepository.settleIfActive).not.toHaveBeenCalled(); + }); + it('returns started with the job id and a ~30min timeout', async () => { const { service, jobRepository } = setup(); diff --git a/packages/cli/src/modules/agents/background/agent-background-job.service.ts b/packages/cli/src/modules/agents/background/agent-background-job.service.ts index 64823f90388..0dc26bc9043 100644 --- a/packages/cli/src/modules/agents/background/agent-background-job.service.ts +++ b/packages/cli/src/modules/agents/background/agent-background-job.service.ts @@ -146,6 +146,10 @@ function latestStopGroup(jobs: AgentBackgroundJob[]): AgentBackgroundJob[] { @Service() export class AgentBackgroundJobService { private readonly abortControllers = new Map(); + private readonly pendingSubAgentJobs = new Set<{ + params: Pick; + cancelRequested: boolean; + }>(); constructor( private readonly jobRepository: AgentBackgroundJobRepository, @@ -169,18 +173,28 @@ export class AgentBackgroundJobService { async registerSubAgentJob( params: Omit, ): Promise { - const inserted = await this.jobRepository.insertSubAgentJobIfCapacity( - { - ...params, - kind: 'subagent', - timeoutAt: new Date(Date.now() + SUB_AGENT_BACKGROUND_TIMEOUT_MS), - }, - MAX_RUNNING_JOBS_PER_THREAD, - ); - if (!inserted) return { status: 'limit-reached' }; - this.updateBroadcaster.notifyBackgroundJobsUpdated(params.parentAgentId, params.parentThreadId); + const registration = { params, cancelRequested: false }; + this.pendingSubAgentJobs.add(registration); + try { + const inserted = await this.jobRepository.insertSubAgentJobIfCapacity( + { + ...params, + kind: 'subagent', + timeoutAt: new Date(Date.now() + SUB_AGENT_BACKGROUND_TIMEOUT_MS), + }, + MAX_RUNNING_JOBS_PER_THREAD, + ); + if (!inserted) return { status: 'limit-reached' }; + if (registration.cancelRequested) await this.cancel(params.parentThreadId, params.id); + this.updateBroadcaster.notifyBackgroundJobsUpdated( + params.parentAgentId, + params.parentThreadId, + ); - return { status: 'started', jobId: params.id }; + return { status: 'started', jobId: params.id }; + } finally { + this.pendingSubAgentJobs.delete(registration); + } } /** @@ -765,6 +779,16 @@ export class AgentBackgroundJobService { parentThreadId: string, parentResourceId: string, ): Promise { + // A pending insert can finish after the query. Keep its cancellation intent until registration ends. + for (const registration of this.pendingSubAgentJobs) { + const { params } = registration; + if ( + params.parentAgentId === parentAgentId && + params.parentThreadId === parentThreadId && + params.parentResourceId === parentResourceId + ) + registration.cancelRequested = true; + } const jobs = await this.jobRepository.findByParentThread(parentThreadId); for (const job of jobs) { if ( From f1deb15d2bb10f2ac8a10a04224c9385da528a22 Mon Sep 17 00:00:00 2001 From: Eugene Molodkin Date: Fri, 16 Oct 2026 18:13:42 +0200 Subject: [PATCH 3/5] test(core): Cover Agent foreground cancellation routes (no-changelog) --- .../__tests__/agent-chat.controller.test.ts | 76 ++++++++++++++----- 1 file changed, 55 insertions(+), 21 deletions(-) diff --git a/packages/cli/src/modules/agents/__tests__/agent-chat.controller.test.ts b/packages/cli/src/modules/agents/__tests__/agent-chat.controller.test.ts index 5a7d480ba0a..fb01453e225 100644 --- a/packages/cli/src/modules/agents/__tests__/agent-chat.controller.test.ts +++ b/packages/cli/src/modules/agents/__tests__/agent-chat.controller.test.ts @@ -825,30 +825,64 @@ describe('AgentChatController budget notice', () => { }); }); +describe('AgentChatController execution cancellation', () => { + it.each([undefined, 'all', 'foreground'] as const)( + 'forwards scope %s for the current preview execution', + async (scope) => { + const { controller, chatExecutionService } = makeController(); + chatExecutionService.requestCancel.mockResolvedValue(true); + await expect( + controller.cancelChatExecution( + { params: { projectId: 'project-1' }, user: { id: 'user-1' } } as never, + {} as never, + 'agent-1', + 'thread-1', + 'execution-1', + scope ? { scope } : undefined, + ), + ).resolves.toEqual({ cancelRequested: true }); + expect(chatExecutionService.requestCancel).toHaveBeenCalledWith({ + projectId: 'project-1', + agentId: 'agent-1', + threadId: 'thread-1', + executionId: 'execution-1', + userId: 'user-1', + surface: 'preview', + ...(scope ? { scope } : {}), + }); + }, + ); +}); + describe('AgentChatController HITL cancellation', () => { - it('cancels a suspended run for the current preview user', async () => { - const { controller, agentExecutionOrchestratorService, agentsService } = makeController(); - agentsService.findById.mockResolvedValue({ id: 'agent-1' } as never); - agentExecutionOrchestratorService.cancelChatRun.mockResolvedValue(true); + it.each([undefined, 'all', 'foreground'] as const)( + 'cancels a suspended run for the current preview user with scope %s', + async (scope) => { + const { controller, agentExecutionOrchestratorService, agentsService } = makeController(); + agentsService.findById.mockResolvedValue({ id: 'agent-1' } as never); + agentExecutionOrchestratorService.cancelChatRun.mockResolvedValue(true); - await expect( - controller.cancelChatRun( - { - params: { projectId: 'project-1' }, - user: { id: 'user-1' }, - } as never, - {} as never, - 'agent-1', - 'run-1', - ), - ).resolves.toEqual({ cancelled: true }); + await expect( + controller.cancelChatRun( + { + params: { projectId: 'project-1' }, + user: { id: 'user-1' }, + } as never, + {} as never, + 'agent-1', + 'run-1', + scope ? { scope } : undefined, + ), + ).resolves.toEqual({ cancelled: true }); - expect(agentExecutionOrchestratorService.cancelChatRun).toHaveBeenCalledWith({ - agentId: 'agent-1', - runId: 'run-1', - resourceId: 'draft-chat:user-1', - }); - }); + expect(agentExecutionOrchestratorService.cancelChatRun).toHaveBeenCalledWith({ + agentId: 'agent-1', + runId: 'run-1', + resourceId: 'draft-chat:user-1', + ...(scope === 'foreground' ? { cancelBackgroundJobs: false } : {}), + }); + }, + ); }); describe('AgentChatController attachment cleanup on failed turns', () => { From e8fd60c10d8485ffd6f5eb058b140795b1e877de Mon Sep 17 00:00:00 2001 From: Eugene Molodkin Date: Fri, 16 Oct 2026 18:42:59 +0200 Subject: [PATCH 4/5] fix(core): Report stopped Agent chat streams as cancelled (no-changelog) --- packages/@n8n/agents/src/types/sdk/agent.ts | 1 + .../discovery/__tests__/stream-status.test.ts | 9 ++- .../evaluations/discovery/stream-status.ts | 2 + ...ent-execution-orchestrator.service.test.ts | 68 ++++++++++++++++++- .../agents/__tests__/agent-sse-stream.test.ts | 6 +- .../agents/agent-turn-execution.service.ts | 12 +++- 6 files changed, 88 insertions(+), 10 deletions(-) diff --git a/packages/@n8n/agents/src/types/sdk/agent.ts b/packages/@n8n/agents/src/types/sdk/agent.ts index 3b0670baf7b..59c31b9200f 100644 --- a/packages/@n8n/agents/src/types/sdk/agent.ts +++ b/packages/@n8n/agents/src/types/sdk/agent.ts @@ -33,6 +33,7 @@ export const FINISH_REASONS = [ 'error', 'other', 'paused', + 'cancelled', ] as const; export type FinishReason = (typeof FINISH_REASONS)[number]; diff --git a/packages/@n8n/instance-ai/evaluations/discovery/__tests__/stream-status.test.ts b/packages/@n8n/instance-ai/evaluations/discovery/__tests__/stream-status.test.ts index 7fcf994fac0..8bd69915f7a 100644 --- a/packages/@n8n/instance-ai/evaluations/discovery/__tests__/stream-status.test.ts +++ b/packages/@n8n/instance-ai/evaluations/discovery/__tests__/stream-status.test.ts @@ -65,7 +65,10 @@ describe('resolveStreamStatus', () => { }, ); - it('never reads a cancelled run as a clean finish', () => { - expect(resolveStreamStatus(streamResult({ status: 'cancelled' }), false)).toBe('timed-out'); - }); + it.each([streamResult({ status: 'cancelled' }), streamResult({ finishReason: 'cancelled' })])( + 'never reads a cancelled run as a clean finish', + (result) => { + expect(resolveStreamStatus(result, false)).toBe('timed-out'); + }, + ); }); diff --git a/packages/@n8n/instance-ai/evaluations/discovery/stream-status.ts b/packages/@n8n/instance-ai/evaluations/discovery/stream-status.ts index 04a975d509f..0b454c51293 100644 --- a/packages/@n8n/instance-ai/evaluations/discovery/stream-status.ts +++ b/packages/@n8n/instance-ai/evaluations/discovery/stream-status.ts @@ -23,6 +23,8 @@ function statusForFinishReason(finishReason: FinishReason): DiscoveryStreamStatu return 'step-exhausted'; case 'paused': return 'suspended'; + case 'cancelled': + return 'timed-out'; case 'tool-calls': case 'length': case 'content-filter': diff --git a/packages/cli/src/modules/agents/__tests__/agent-execution-orchestrator.service.test.ts b/packages/cli/src/modules/agents/__tests__/agent-execution-orchestrator.service.test.ts index a05ab7cdae6..ba19535f79a 100644 --- a/packages/cli/src/modules/agents/__tests__/agent-execution-orchestrator.service.test.ts +++ b/packages/cli/src/modules/agents/__tests__/agent-execution-orchestrator.service.test.ts @@ -883,6 +883,7 @@ describe('AgentExecutionOrchestratorService', () => { announceExecution = true, onBudgetNotice, budget, + runtimeChunks, }: { abortSignal?: AbortSignal; previewChat?: boolean; @@ -890,9 +891,10 @@ describe('AgentExecutionOrchestratorService', () => { announceExecution?: boolean; onBudgetNotice?: () => void; budget?: BudgetGuardrailConfig; + runtimeChunks?: StreamChunk[]; } = {}) { const fixtures = makeService(); - const runtime = makeRuntime(undefined, undefined, budget); + const runtime = makeRuntime(runtimeChunks, undefined, budget); fixtures.runtimeCacheService.getRuntime.mockResolvedValue(runtime); fixtures.checkpointStorage.getStatus.mockResolvedValue({ status: 'active', @@ -1281,6 +1283,70 @@ describe('AgentExecutionOrchestratorService', () => { }, ); + it.each([ + { name: 'requested stop', previewChat: true, abort: true, expectedReason: 'cancelled' }, + { name: 'model failure', previewChat: true, abort: false, expectedReason: 'error' }, + { + name: 'ownership loss', + previewChat: true, + abort: true, + abortReason: new OperationalError('Execution ownership lost'), + expectedReason: 'error', + }, + { name: 'non-chat abort', previewChat: false, abort: true, expectedReason: 'error' }, + ])('emits the correct finish reason after $name', async (scenario) => { + const controller = new AbortController(); + const error = new Error('Runtime ended'); + const usage = { promptTokens: 2, completionTokens: 3, totalTokens: 5, cost: 0.01 }; + const { stream, executionService, chatExecutionService } = makeTurn({ + previewChat: scenario.previewChat, + abortSignal: controller.signal, + runtimeChunks: [ + { type: 'text-start', id: 'text-1' }, + { type: 'text-delta', id: 'text-1', delta: 'partial answer' }, + { type: 'error', error }, + { type: 'finish', finishReason: 'error', model: 'mock-model', usage }, + ], + }); + if (scenario.name === 'ownership loss') + executionService.getAbortSignal.mockReturnValue(controller.signal); + const chunks: AgentExecutionStreamChunk[] = []; + for await (const chunk of stream) { + chunks.push(chunk); + if (chunk.type !== 'text-delta' || !scenario.abort) continue; + if (scenario.name === 'requested stop') { + await chatExecutionService.handleCancel({ + projectId, + agentId, + threadId: 'thread-1', + executionId: 'execution-1', + userId, + scope: 'foreground', + }); + } else controller.abort(scenario.abortReason); + } + + expect(chunks.filter((chunk) => chunk.type === 'finish')).toEqual([ + { type: 'finish', finishReason: scenario.expectedReason, model: 'mock-model', usage }, + ]); + expect(chunks.filter((chunk) => chunk.type === 'error')).toEqual( + scenario.expectedReason === 'cancelled' ? [] : [{ type: 'error', error }], + ); + if (scenario.expectedReason === 'cancelled') { + expect(executionService.finalizeExecution).toHaveBeenCalledWith( + 'execution-1', + expect.objectContaining({ + record: expect.objectContaining({ + assistantResponse: 'partial answer', + finishReason: 'cancelled', + error: null, + usage: { promptTokens: 2, completionTokens: 3, totalTokens: 5 }, + }), + }), + ); + } + }); + it('releases the runtime when preparation fails after acquisition', async () => { const fixtures = makeTurn(); const error = new Error('preparation failed'); diff --git a/packages/cli/src/modules/agents/__tests__/agent-sse-stream.test.ts b/packages/cli/src/modules/agents/__tests__/agent-sse-stream.test.ts index 9d59de63f25..b00c6316741 100644 --- a/packages/cli/src/modules/agents/__tests__/agent-sse-stream.test.ts +++ b/packages/cli/src/modules/agents/__tests__/agent-sse-stream.test.ts @@ -270,13 +270,13 @@ describe('agent-sse-stream — stream completion', () => { ]); }); - it('forwards the finish reason and leaves completion delivery to the caller', async () => { + it.each(['stop', 'cancelled'] as const)('forwards the %s finish reason', async (finishReason) => { const events = await collectEvents([ { type: 'text-delta', id: 't-1', delta: 'hello' }, { type: 'text-end', id: 't-1' }, { type: 'finish', - finishReason: 'stop', + finishReason, usage: { promptTokens: 1, completionTokens: 1, totalTokens: 2 }, }, ]); @@ -284,7 +284,7 @@ describe('agent-sse-stream — stream completion', () => { expect(events).toEqual([ { type: 'text-delta', id: 't-1', delta: 'hello' }, { type: 'text-end', id: 't-1' }, - { type: 'finish', finishReason: 'stop' }, + { type: 'finish', finishReason }, ]); }); diff --git a/packages/cli/src/modules/agents/agent-turn-execution.service.ts b/packages/cli/src/modules/agents/agent-turn-execution.service.ts index 6287d8a4c3a..53e9cc8c801 100644 --- a/packages/cli/src/modules/agents/agent-turn-execution.service.ts +++ b/packages/cli/src/modules/agents/agent-turn-execution.service.ts @@ -231,7 +231,7 @@ export class AgentTurnExecutionService { ): AsyncGenerator { const attributionTracker = createAttributionTracker(config.mcpServerAttributions); - for await (const chunk of streamAgentChunks(stream)) { + for await (let chunk of streamAgentChunks(stream)) { if (chunk.type === 'input-boundary') { chunk.acknowledge(); continue; @@ -244,8 +244,14 @@ export class AgentTurnExecutionService { } continue; } - // The runtime also emits an error chunk for a requested stop. - if (chunk.type === 'error' && (await this.isStoppedChatTurn(turn, config))) continue; + // The runtime reports requested stops as errors. Keep chat outcomes consistent. + if ( + (chunk.type === 'error' || (chunk.type === 'finish' && chunk.finishReason === 'error')) && + (await this.isStoppedChatTurn(turn, config)) + ) { + if (chunk.type === 'error') continue; + chunk = { ...chunk, finishReason: 'cancelled' }; + } recorder.record(chunk); if (chunk.type === 'tool-call-suspended') state.suspendedRunId = chunk.runId; if (chunk.type === 'error') state.executionError = chunk.error; From 7e8ecff54d215621d8748114a957fec639175ab2 Mon Sep 17 00:00:00 2001 From: Eugene Molodkin Date: Fri, 16 Oct 2026 18:51:04 +0200 Subject: [PATCH 5/5] refactor(core): Keep foreground cancellation independent of guardrails (no-changelog) --- .../runtime/__tests__/agent-runtime.test.ts | 56 ------------------- .../runtime/guardrails/guardrail-runner.ts | 8 +-- .../src/runtime/tools/tool-call-runner.ts | 6 -- .../@n8n/agents/src/types/sdk/guardrail.ts | 8 +-- 4 files changed, 4 insertions(+), 74 deletions(-) diff --git a/packages/@n8n/agents/src/runtime/__tests__/agent-runtime.test.ts b/packages/@n8n/agents/src/runtime/__tests__/agent-runtime.test.ts index a2c7556728a..5aa7cbc30ed 100644 --- a/packages/@n8n/agents/src/runtime/__tests__/agent-runtime.test.ts +++ b/packages/@n8n/agents/src/runtime/__tests__/agent-runtime.test.ts @@ -1815,62 +1815,6 @@ describe('AgentRuntime — guardrails', () => { expect(call.state === 'rejected' && call.error).toContain('tool.stop'); }); - it.each(['generate', 'stream'] as const)( - 'records canceled calls without errors and stops the next %s model turn', - async (mode) => { - const canceled: GuardrailDecision = { action: 'stop', code: 'user.cancel', canceled: true }; - const calls = ['tc-1', 'tc-2'].map((toolCallId) => ({ - toolCallId, - toolName: 'echo', - args: { v: 'x' }, - })); - generateText.mockResolvedValueOnce(makeGenerateWithToolCalls(calls)); - streamText.mockReturnValueOnce(makeStreamWithToolCalls(calls)); - const handler = vi.fn(async () => 'Must not run'); - const hook = makeGuardrail({ - before: vi.fn().mockResolvedValueOnce({ action: 'allow' }).mockResolvedValue(canceled), - beforeTool: vi.fn().mockResolvedValue(canceled), - }); - const runtime = createRuntimeWithEchoTool(handler); - const options = { guardrails: guardrailsOption(hook) }; - if (mode === 'generate') { - const result = await runtime.generate('go', options); - const blocks = result.messages.flatMap((message) => - isLlmMessage(message) - ? message.content.filter((block) => block.type === 'tool-call') - : [], - ); - expect(blocks).toHaveLength(2); - expect(blocks).toEqual([ - expect.objectContaining({ state: 'resolved', canceled: true }), - expect.objectContaining({ state: 'resolved', canceled: true }), - ]); - expect(result.error).toBeUndefined(); - expect(result.guardrail).toEqual({ code: 'user.cancel', canceled: true }); - expect(generateText).toHaveBeenCalledTimes(1); - } else { - const { stream } = await runtime.stream('go', options); - const chunks = await collectChunks(stream); - const results = chunks.filter((chunk) => chunk.type === 'tool-result'); - expect(results).toHaveLength(2); - for (const result of results) { - expect(result.canceled).toBe(true); - expect(result.output).toContain('user.cancel'); - expect(result.output).not.toContain('User said'); - expect(result.isError).not.toBe(true); - } - expect(chunks.some((chunk) => chunk.type === 'error')).toBe(false); - expect(chunks.at(-1)).toMatchObject({ - type: 'finish', - guardrail: { code: 'user.cancel', canceled: true }, - }); - expect(streamText).toHaveBeenCalledTimes(1); - } - expect(handler).not.toHaveBeenCalled(); - expect(hook.afterTool).not.toHaveBeenCalled(); - }, - ); - it('passes the handler result to afterTool() when the tool call is allowed', async () => { generateText .mockResolvedValueOnce(makeGenerateWithToolCall('tc-1', 'echo', { v: 'x' })) diff --git a/packages/@n8n/agents/src/runtime/guardrails/guardrail-runner.ts b/packages/@n8n/agents/src/runtime/guardrails/guardrail-runner.ts index fe290702bfb..c67b96204f6 100644 --- a/packages/@n8n/agents/src/runtime/guardrails/guardrail-runner.ts +++ b/packages/@n8n/agents/src/runtime/guardrails/guardrail-runner.ts @@ -46,9 +46,7 @@ export class GuardrailRunner { async before(ctx: GuardrailModelCallContext): Promise { for (const hook of this.hooks) { const decision = await hook.before?.(ctx); - if (decision?.action === 'stop') { - return { code: decision.code, ...(decision.canceled ? { canceled: true } : {}) }; - } + if (decision?.action === 'stop') return { code: decision.code }; } return undefined; } @@ -62,9 +60,7 @@ export class GuardrailRunner { async beforeTool(ctx: GuardrailToolCallContext): Promise { for (const hook of this.hooks) { const decision = await hook.beforeTool?.(ctx); - if (decision?.action === 'stop') { - return { code: decision.code, ...(decision.canceled ? { canceled: true } : {}) }; - } + if (decision?.action === 'stop') return { code: decision.code }; } return undefined; } diff --git a/packages/@n8n/agents/src/runtime/tools/tool-call-runner.ts b/packages/@n8n/agents/src/runtime/tools/tool-call-runner.ts index 84bc8948364..2b4b458f2d3 100644 --- a/packages/@n8n/agents/src/runtime/tools/tool-call-runner.ts +++ b/packages/@n8n/agents/src/runtime/tools/tool-call-runner.ts @@ -137,12 +137,6 @@ export class ToolCallRunner { if (guardrails && guardCtx && !params.previouslySuspended) { const stop = await guardrails.beforeTool(guardCtx); if (stop) { - if (stop.canceled) - return this.buildCancelledOutcome( - params, - 'Tool call canceled', - `[Tool call stopped by guardrail: ${stop.code}]`, - ); return await this.toolError( params, new Error(`Tool call stopped by guardrail: ${stop.code}`), diff --git a/packages/@n8n/agents/src/types/sdk/guardrail.ts b/packages/@n8n/agents/src/types/sdk/guardrail.ts index cc0c93e51ef..8512501e46e 100644 --- a/packages/@n8n/agents/src/types/sdk/guardrail.ts +++ b/packages/@n8n/agents/src/types/sdk/guardrail.ts @@ -12,13 +12,9 @@ export interface BuiltGuardrail { /** @internal */ readonly _config: Record; } -export type GuardrailDecision = { action: 'allow' } | ({ action: 'stop' } & GuardrailStop); +export type GuardrailDecision = { action: 'allow' } | { action: 'stop'; code: string }; -export type GuardrailStop = { - code: string; - /** Record a stopped model or tool call as canceled. */ - canceled?: true; -}; +export type GuardrailStop = { code: string }; export type GuardrailModelCallSource = 'turn' | 'title' | 'memory'; export interface GuardrailModelCallContext {