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..2b4b458f2d3 100644 --- a/packages/@n8n/agents/src/runtime/tools/tool-call-runner.ts +++ b/packages/@n8n/agents/src/runtime/tools/tool-call-runner.ts @@ -183,7 +183,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 +328,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/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/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/@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-chat-execution.service.test.ts b/packages/cli/src/modules/agents/__tests__/agent-chat-execution.service.test.ts index d355f97a5ff..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 @@ -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,9 +6,11 @@ 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'; +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'; @@ -359,3 +360,115 @@ 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); +}); + +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-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', () => { 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..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', @@ -1135,7 +1137,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', @@ -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-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-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/__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..01de9aabac2 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 { @@ -131,21 +132,27 @@ 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, ...rest } = context; - await this.publisher.publishCommand({ - command: '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 { - 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 +160,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 +186,18 @@ 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); + } 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); }, ); @@ -174,6 +205,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, @@ -184,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, @@ -203,7 +240,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 +281,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..53e9cc8c801 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, @@ -214,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; @@ -227,6 +244,14 @@ export class AgentTurnExecutionService { } 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; 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 ( 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