mirror of
https://github.com/n8n-io/n8n.git
synced 2026-10-17 21:49:36 +02:00
fix(core): Recover model turns that end without usable output (#36423)
Co-authored-by: Claude Opus 5 (1M context) <noreply@anthropic.com>
This commit is contained in:
parent
7060434f1f
commit
bae82bdad5
|
|
@ -764,6 +764,73 @@ describe('AgentRuntime — empty stop turn retry', () => {
|
|||
});
|
||||
});
|
||||
|
||||
// ---------------------------------------------------------------------------
|
||||
// reasoning-only turn
|
||||
// ---------------------------------------------------------------------------
|
||||
|
||||
describe('AgentRuntime — reasoning-only turn', () => {
|
||||
beforeEach(() => {
|
||||
generateText.mockReset();
|
||||
streamText.mockReset();
|
||||
});
|
||||
|
||||
/**
|
||||
* The provider streamed thinking and then the stream ended without a
|
||||
* terminal chunk, so the SDK synthesizes the finish from its defaults: a
|
||||
* reasoning-only message, `other`, and no usage. Nothing the user can see
|
||||
* and nothing the loop can act on.
|
||||
*/
|
||||
function makeStreamReasoningOnly() {
|
||||
return {
|
||||
stream: makeChunkStream([{ type: 'reasoning-delta', id: 'r-1', text: 'thinking...' }]),
|
||||
finishReason: Promise.resolve('other'),
|
||||
usage: Promise.resolve({
|
||||
inputTokens: undefined,
|
||||
outputTokens: undefined,
|
||||
totalTokens: undefined,
|
||||
}),
|
||||
response: Promise.resolve({
|
||||
messages: [{ role: 'assistant', content: [{ type: 'reasoning', text: 'thinking...' }] }],
|
||||
}),
|
||||
toolCalls: Promise.resolve([]),
|
||||
};
|
||||
}
|
||||
|
||||
it('retries a reasoning-only turn instead of ending the run', async () => {
|
||||
streamText
|
||||
.mockReturnValueOnce(makeStreamReasoningOnly())
|
||||
.mockReturnValueOnce(makeStreamSuccess('Recovered'));
|
||||
|
||||
const { runtime } = createRuntime();
|
||||
const result = await runtime.stream('make my workflow smarter');
|
||||
const chunks = await collectChunks(result.stream);
|
||||
|
||||
expect(streamText).toHaveBeenCalledTimes(2);
|
||||
const text = chunks
|
||||
.filter((chunk): chunk is StreamChunk & { type: 'text-delta' } => chunk.type === 'text-delta')
|
||||
.map((chunk) => chunk.delta)
|
||||
.join('');
|
||||
expect(text).toBe('Recovered');
|
||||
});
|
||||
|
||||
it('reports an error rather than finishing silently when every attempt is reasoning-only', async () => {
|
||||
streamText.mockImplementation(() => makeStreamReasoningOnly());
|
||||
|
||||
const { runtime } = createRuntime();
|
||||
const result = await runtime.stream('make my workflow smarter');
|
||||
const chunks = await collectChunks(result.stream);
|
||||
|
||||
const finish = chunks.find(
|
||||
(chunk): chunk is StreamChunk & { type: 'finish' } => chunk.type === 'finish',
|
||||
);
|
||||
expect(finish?.finishReason).toBe('error');
|
||||
const error = chunks.find(
|
||||
(chunk): chunk is StreamChunk & { type: 'error' } => chunk.type === 'error',
|
||||
);
|
||||
expect(String((error?.error as Error).message)).toContain('no output');
|
||||
});
|
||||
});
|
||||
|
||||
// ---------------------------------------------------------------------------
|
||||
// generate() — graceful error contract
|
||||
// ---------------------------------------------------------------------------
|
||||
|
|
|
|||
|
|
@ -50,9 +50,31 @@ describe('isEmptyModelTurn', () => {
|
|||
).toBe(false);
|
||||
});
|
||||
|
||||
it('is false for non-stop finish reasons', () => {
|
||||
it('is false for a tool-calls turn', () => {
|
||||
expect(isEmptyModelTurn({ aiFinishReason: 'tool-calls', newMessages: [] })).toBe(false);
|
||||
expect(isEmptyModelTurn({ aiFinishReason: 'length', newMessages: [] })).toBe(false);
|
||||
});
|
||||
|
||||
it('is true for a truncated turn with no content', () => {
|
||||
expect(isEmptyModelTurn({ aiFinishReason: 'length', newMessages: [] })).toBe(true);
|
||||
});
|
||||
|
||||
it('is true for a reasoning-only turn whose stream died before its terminal chunk', () => {
|
||||
expect(
|
||||
isEmptyModelTurn({
|
||||
aiFinishReason: 'other',
|
||||
newMessages: [assistant([{ type: 'reasoning', text: 'thinking…' }])],
|
||||
}),
|
||||
).toBe(true);
|
||||
});
|
||||
|
||||
it('is false when the provider blocked the prompt', () => {
|
||||
expect(
|
||||
isEmptyModelTurn({
|
||||
aiFinishReason: 'other',
|
||||
newMessages: [],
|
||||
errorReason: { type: 'prompt_blocked', message: 'blocked: PROHIBITED_CONTENT' },
|
||||
}),
|
||||
).toBe(false);
|
||||
});
|
||||
|
||||
it('is false when structured output was produced', () => {
|
||||
|
|
|
|||
|
|
@ -50,6 +50,25 @@ If this affects the user's request, briefly let them know which server is unavai
|
|||
*/
|
||||
const EMPTY_RESPONSE_ERROR_FINISH_REASONS = new Set(['other', 'unknown', 'content-filter']);
|
||||
|
||||
/**
|
||||
* Whether a turn carries output the user can see or the loop can act on.
|
||||
* Reasoning is neither: a thinking block the model never turned into an answer
|
||||
* or a tool call leaves the run with nothing to show and nothing to do next.
|
||||
*/
|
||||
function hasActionableContent(messages: AgentMessage[]): boolean {
|
||||
return messages.some(
|
||||
(m) =>
|
||||
'content' in m &&
|
||||
Array.isArray(m.content) &&
|
||||
m.content.some(
|
||||
(c) =>
|
||||
(c.type === 'text' && c.text.trim().length > 0) ||
|
||||
c.type === 'tool-call' ||
|
||||
c.type === 'file',
|
||||
),
|
||||
);
|
||||
}
|
||||
|
||||
/**
|
||||
* Classify a turn that produced no output as a recognized failure, or return
|
||||
* `undefined` when it doesn't look like a provider rejection. Some providers
|
||||
|
|
@ -63,7 +82,7 @@ export function classifyModelTurnError(turn: {
|
|||
newMessages: AgentMessage[];
|
||||
providerError?: RawProviderError;
|
||||
}): ModelTurnError | undefined {
|
||||
if (turn.newMessages.length > 0) return undefined;
|
||||
if (hasActionableContent(turn.newMessages)) return undefined;
|
||||
if (!EMPTY_RESPONSE_ERROR_FINISH_REASONS.has(turn.aiFinishReason)) return undefined;
|
||||
|
||||
const guidance =
|
||||
|
|
@ -81,31 +100,29 @@ export function classifyModelTurnError(turn: {
|
|||
}
|
||||
|
||||
/**
|
||||
* True when a turn finished with `stop` but produced no usable output — no
|
||||
* non-whitespace text, no tool call, no file. Some providers (observed with
|
||||
* Kimi via Together) occasionally emit such a turn mid-task, which would
|
||||
* silently end the run with work half-done; callers retry a bounded number
|
||||
* of times before accepting it. Reasoning-only turns count as empty: they
|
||||
* carry no user-visible output and no action.
|
||||
* True when a turn produced no usable output — no non-whitespace text, no tool
|
||||
* call, no file. Providers emit such a turn mid-task in more than one shape: a
|
||||
* bare `stop` (observed with Kimi via Together), or a stream that dies before
|
||||
* its terminal chunk, leaving the SDK to synthesize a finish from its defaults
|
||||
* (`other`, no usage) around a reasoning-only message. Either way the run would
|
||||
* end silently with work half-done, so callers retry a bounded number of times
|
||||
* before accepting it. Reasoning-only turns count as empty: they carry no
|
||||
* user-visible output and no action. `tool-calls` is the one finish reason that
|
||||
* cannot be empty — the calls are the turn's output.
|
||||
*/
|
||||
export function isEmptyModelTurn(turn: {
|
||||
aiFinishReason: string;
|
||||
newMessages: AgentMessage[];
|
||||
structuredOutput?: unknown;
|
||||
errorReason?: ModelTurnError;
|
||||
}): boolean {
|
||||
if (turn.aiFinishReason !== 'stop') return false;
|
||||
if (turn.aiFinishReason === 'tool-calls') return false;
|
||||
if (turn.structuredOutput !== undefined) return false;
|
||||
return !turn.newMessages.some(
|
||||
(m) =>
|
||||
'content' in m &&
|
||||
Array.isArray(m.content) &&
|
||||
m.content.some(
|
||||
(c) =>
|
||||
(c.type === 'text' && c.text.trim().length > 0) ||
|
||||
c.type === 'tool-call' ||
|
||||
c.type === 'file',
|
||||
),
|
||||
);
|
||||
// A safety block is the provider's deterministic verdict on this prompt:
|
||||
// re-issuing it earns the same answer and discards the captured reason,
|
||||
// which is the only place the block is explained.
|
||||
if (turn.errorReason?.type === 'prompt_blocked') return false;
|
||||
return !hasActionableContent(turn.newMessages);
|
||||
}
|
||||
|
||||
/** Extract all settled (resolved or rejected) tool-call blocks from a flat list of agent messages. */
|
||||
|
|
|
|||
|
|
@ -144,6 +144,33 @@ describe('InstanceAiTerminalResponseGuard', () => {
|
|||
expect(decision.event?.type).toBe('text-delta');
|
||||
});
|
||||
|
||||
it('emits fallback when a completed run only produced a preamble before its tool calls', () => {
|
||||
const decision = guard().evaluateTerminal(
|
||||
[runStart(), rootText('Let me read the current source first.'), toolCall()],
|
||||
'completed',
|
||||
);
|
||||
|
||||
expect(decision.action).toBe('emit');
|
||||
expect(decision.reason).toBe('completed-silent');
|
||||
expect(decision.event?.type).toBe('text-delta');
|
||||
});
|
||||
|
||||
it('does not emit fallback when a completed run answers after its tool calls', () => {
|
||||
const decision = guard().evaluateTerminal(
|
||||
[
|
||||
runStart(),
|
||||
rootText('Let me read the current source first.'),
|
||||
toolCall(),
|
||||
rootText('Done.'),
|
||||
],
|
||||
'completed',
|
||||
);
|
||||
|
||||
expect(decision.action).toBe('none');
|
||||
expect(decision.visibilitySource).toBe('root-text');
|
||||
expect(decision.reason).toBe('already-visible');
|
||||
});
|
||||
|
||||
it('does not emit completed fallback when the message group already has root text', () => {
|
||||
const decision = new InstanceAiTerminalResponseGuard({
|
||||
runId,
|
||||
|
|
|
|||
|
|
@ -102,7 +102,9 @@ export class InstanceAiTerminalResponseGuard {
|
|||
reason: 'completed-after-error',
|
||||
};
|
||||
}
|
||||
if (visibility.hasRootText) {
|
||||
// Closing text only: a preamble the model never followed with an answer
|
||||
// leaves the user watching a turn that looks like it is still running.
|
||||
if (visibility.hasClosingRootText) {
|
||||
return {
|
||||
status,
|
||||
visibilitySource: 'root-text',
|
||||
|
|
@ -216,16 +218,28 @@ export class InstanceAiTerminalResponseGuard {
|
|||
|
||||
private getVisibility(events: InstanceAiEvent[]): {
|
||||
hasRootText: boolean;
|
||||
hasClosingRootText: boolean;
|
||||
hasRootError: boolean;
|
||||
hasMessageGroupRootText: boolean;
|
||||
hasCurrentRunFallback: boolean;
|
||||
hasAgentText: boolean;
|
||||
} {
|
||||
const currentRunEvents = events.filter((event) => event.runId === this.options.runId);
|
||||
// The loop runs another turn for every tool call, so the turn that ends the
|
||||
// run never carries one: root text before the last tool call is a preamble
|
||||
// ("Let me read the source…"), text after it is the answer. With no tool
|
||||
// calls the index is -1 and every text event counts.
|
||||
const lastRootToolCall = currentRunEvents.findLastIndex(
|
||||
(event) => event.agentId === this.options.rootAgentId && event.type === 'tool-call',
|
||||
);
|
||||
return {
|
||||
hasRootText: currentRunEvents.some(
|
||||
(event) => event.agentId === this.options.rootAgentId && hasText(event),
|
||||
),
|
||||
hasClosingRootText: currentRunEvents.some(
|
||||
(event, index) =>
|
||||
index > lastRootToolCall && event.agentId === this.options.rootAgentId && hasText(event),
|
||||
),
|
||||
// Any agent's text this run. Tool calls don't count — internal calls (e.g. complete-checkpoint) aren't a visible answer.
|
||||
hasAgentText: currentRunEvents.some((event) => hasText(event)),
|
||||
hasRootError: currentRunEvents.some(
|
||||
|
|
|
|||
|
|
@ -130,6 +130,7 @@ type Deps = {
|
|||
updateLast: Mock;
|
||||
};
|
||||
telemetry: { track: Mock };
|
||||
errorReporter: { report: Mock };
|
||||
logger: { warn: Mock; debug: Mock; error: Mock };
|
||||
runState: { getRunIdsForMessageGroup: Mock; cancelThread: Mock };
|
||||
suspendedThreads: { dropPendingConfirmationsForThread: Mock };
|
||||
|
|
@ -192,6 +193,7 @@ function createService(snapshotTree?: InstanceAiAgentNode): {
|
|||
updateLast: vi.fn(async () => {}),
|
||||
},
|
||||
telemetry: { track: vi.fn() },
|
||||
errorReporter: { report: vi.fn() },
|
||||
logger: { warn: vi.fn(), debug: vi.fn(), error: vi.fn() },
|
||||
runState: {
|
||||
getRunIdsForMessageGroup: vi.fn(() => ['run-1']),
|
||||
|
|
@ -221,6 +223,7 @@ function createService(snapshotTree?: InstanceAiAgentNode): {
|
|||
dbSnapshotStorage: deps.dbSnapshotStorage,
|
||||
agentMemory: {},
|
||||
telemetry: deps.telemetry,
|
||||
errorReporter: deps.errorReporter,
|
||||
logger: deps.logger,
|
||||
runState: deps.runState,
|
||||
suspendedThreads: deps.suspendedThreads,
|
||||
|
|
@ -464,6 +467,35 @@ describe('InstanceAiTerminalOutcomeService — terminal response guard wiring',
|
|||
expect(deps.eventBus.events.map((event) => event.type)).toEqual(['text-delta', 'run-finish']);
|
||||
});
|
||||
|
||||
it('reports a silent completed run so the stall is not lost', async () => {
|
||||
const { service, deps } = createService();
|
||||
|
||||
await service.evaluateTerminalResponse('thread-a', 'run-1', 'completed', {
|
||||
messageGroupId: 'group-1',
|
||||
});
|
||||
|
||||
expect(deps.errorReporter.report).toHaveBeenCalledWith(
|
||||
expect.any(Error),
|
||||
expect.objectContaining({
|
||||
component: 'instance-ai-terminal-guard',
|
||||
severity: 'warning',
|
||||
threadId: 'thread-a',
|
||||
runId: 'run-1',
|
||||
}),
|
||||
);
|
||||
});
|
||||
|
||||
it('does not report a completed run when silence is expected', async () => {
|
||||
const { service, deps } = createService();
|
||||
|
||||
await service.evaluateTerminalResponse('thread-a', 'run-1', 'completed', {
|
||||
messageGroupId: 'group-1',
|
||||
suppressCompletedFallback: true,
|
||||
});
|
||||
|
||||
expect(deps.errorReporter.report).not.toHaveBeenCalled();
|
||||
});
|
||||
|
||||
it('does not publish completed fallback output when silence is expected', async () => {
|
||||
const { service, deps } = createService();
|
||||
|
||||
|
|
|
|||
|
|
@ -842,6 +842,7 @@ function createTerminalGuardOrderService(): TerminalGuardOrderServiceInternals {
|
|||
dbSnapshotStorage: {},
|
||||
agentMemory: {},
|
||||
telemetry: service.telemetry,
|
||||
errorReporter: service.instanceAiErrorReporter,
|
||||
logger: service.logger,
|
||||
runState: service.runState,
|
||||
suspendedThreads: service.suspendedThreads,
|
||||
|
|
|
|||
|
|
@ -15,9 +15,12 @@ import {
|
|||
type WorkSummary,
|
||||
} from '@n8n/instance-ai';
|
||||
|
||||
import { OperationalError } from 'n8n-workflow';
|
||||
|
||||
import type { Telemetry } from '@/telemetry';
|
||||
|
||||
import type { InProcessEventBus } from './event-bus/in-process-event-bus';
|
||||
import type { InstanceAiErrorReporterService } from './instance-ai-error-reporter.service';
|
||||
import type { DbSnapshotStorage } from './storage/db-snapshot-storage';
|
||||
import type { SuspendedThreadPersistenceService } from './suspended-thread-persistence.service';
|
||||
import type {
|
||||
|
|
@ -104,6 +107,8 @@ export type InstanceAiTerminalOutcomeSnapshotStorage = Pick<
|
|||
|
||||
export type InstanceAiTerminalOutcomeTelemetry = Pick<Telemetry, 'track'>;
|
||||
|
||||
export type InstanceAiTerminalOutcomeErrorReporter = Pick<InstanceAiErrorReporterService, 'report'>;
|
||||
|
||||
export type InstanceAiTerminalOutcomeRunState = Pick<
|
||||
RunStateRegistry<User>,
|
||||
'getRunIdsForMessageGroup' | 'cancelThread'
|
||||
|
|
@ -131,6 +136,7 @@ export interface InstanceAiTerminalOutcomeServiceOptions {
|
|||
dbSnapshotStorage: InstanceAiTerminalOutcomeSnapshotStorage;
|
||||
agentMemory: PatchableThreadMemory;
|
||||
telemetry: InstanceAiTerminalOutcomeTelemetry;
|
||||
errorReporter: InstanceAiTerminalOutcomeErrorReporter;
|
||||
logger: Logger;
|
||||
runState: InstanceAiTerminalOutcomeRunState;
|
||||
suspendedThreads: InstanceAiTerminalOutcomeSuspendedThreads;
|
||||
|
|
@ -188,6 +194,8 @@ export class InstanceAiTerminalOutcomeService {
|
|||
|
||||
private readonly telemetry: InstanceAiTerminalOutcomeTelemetry;
|
||||
|
||||
private readonly errorReporter: InstanceAiTerminalOutcomeErrorReporter;
|
||||
|
||||
private readonly logger: Logger;
|
||||
|
||||
private readonly runState: InstanceAiTerminalOutcomeRunState;
|
||||
|
|
@ -206,6 +214,7 @@ export class InstanceAiTerminalOutcomeService {
|
|||
this.dbSnapshotStorage = options.dbSnapshotStorage;
|
||||
this.agentMemory = options.agentMemory;
|
||||
this.telemetry = options.telemetry;
|
||||
this.errorReporter = options.errorReporter;
|
||||
this.logger = options.logger;
|
||||
this.runState = options.runState;
|
||||
this.suspendedThreads = options.suspendedThreads;
|
||||
|
|
@ -305,6 +314,22 @@ export class InstanceAiTerminalOutcomeService {
|
|||
});
|
||||
}
|
||||
|
||||
// The run reported success while answering nothing, so no error path fires
|
||||
// and the fallback line is all the user gets. Alert on it: a stall that only
|
||||
// shows up as a generic placeholder is otherwise invisible to us.
|
||||
if (decision.reason === 'completed-silent') {
|
||||
this.errorReporter.report(
|
||||
new OperationalError('Instance AI run completed without a final response'),
|
||||
{
|
||||
component: 'instance-ai-terminal-guard',
|
||||
severity: 'warning',
|
||||
threadId,
|
||||
runId,
|
||||
messageGroupId,
|
||||
},
|
||||
);
|
||||
}
|
||||
|
||||
if (decision.reason === 'confirmation-invalid') {
|
||||
this.logger.warn('invalid_confirmation_payload', {
|
||||
threadId,
|
||||
|
|
|
|||
|
|
@ -904,6 +904,7 @@ export class InstanceAiService {
|
|||
dbSnapshotStorage: this.dbSnapshotStorage,
|
||||
agentMemory: this.agentMemory,
|
||||
telemetry: this.telemetry,
|
||||
errorReporter: this.instanceAiErrorReporter,
|
||||
logger: this.logger,
|
||||
runState: this.runState,
|
||||
suspendedThreads: this.suspendedThreads,
|
||||
|
|
|
|||
Loading…
Reference in New Issue
Block a user