mirror of
https://github.com/n8n-io/n8n.git
synced 2026-10-17 05:29:44 +02:00
fix(core): Report stopped Agent chat streams as cancelled (no-changelog)
This commit is contained in:
parent
f1deb15d2b
commit
e8fd60c10d
|
|
@ -33,6 +33,7 @@ export const FINISH_REASONS = [
|
|||
'error',
|
||||
'other',
|
||||
'paused',
|
||||
'cancelled',
|
||||
] as const;
|
||||
|
||||
export type FinishReason = (typeof FINISH_REASONS)[number];
|
||||
|
|
|
|||
|
|
@ -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');
|
||||
},
|
||||
);
|
||||
});
|
||||
|
|
|
|||
|
|
@ -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':
|
||||
|
|
|
|||
|
|
@ -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');
|
||||
|
|
|
|||
|
|
@ -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 },
|
||||
]);
|
||||
});
|
||||
|
||||
|
|
|
|||
|
|
@ -231,7 +231,7 @@ export class AgentTurnExecutionService {
|
|||
): AsyncGenerator<AgentExecutionStreamChunk> {
|
||||
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;
|
||||
|
|
|
|||
Loading…
Reference in New Issue
Block a user