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;