fix(core): Prevent Anthropic agent threads from breaking permanently and keep run errors visible (backport to release-candidate/2.38.x) (#37722)

Co-authored-by: Robin Braumann <50590409+bjorger@users.noreply.github.com>
Co-authored-by: Cursor <cursoragent@cursor.com>
This commit is contained in:
n8n-assistant[bot] 2026-09-03 10:14:20 +00:00 committed by GitHub
parent 3a77307e04
commit 1007dff4be
No known key found for this signature in database
GPG Key ID: B5690EEEBB952194
12 changed files with 437 additions and 85 deletions

View File

@ -124,6 +124,58 @@ describe('toAiMessages + fromAiMessages — round-trip', () => {
});
});
it('keeps only the last Anthropic signature when assistant responses are adjacent', () => {
const input: Message[] = [
{
role: 'assistant',
content: [
{
type: 'reasoning',
text: 'First response reasoning',
providerMetadata: { anthropic: { signature: 'first-signature' } },
},
{ type: 'text', text: 'First response' },
],
},
{
role: 'assistant',
content: [
{
type: 'reasoning',
text: 'Second response reasoning',
providerMetadata: { anthropic: { signature: 'second-signature' } },
},
{ type: 'text', text: 'Second response' },
],
},
];
const aiMessages = toAiMessages(input);
const content = aiMessages.map(
(message) =>
(
message as {
content: Array<{
type: string;
text?: string;
providerOptions?: { anthropic?: { signature?: string } };
}>;
}
).content,
);
expect(content.flatMap((parts) => parts).filter((part) => part.type === 'text')).toEqual([
{ type: 'text', text: 'First response' },
{ type: 'text', text: 'Second response' },
]);
expect(
content
.flatMap((parts) => parts)
.map((part) => part.providerOptions?.anthropic?.signature)
.filter((signature) => signature !== undefined),
).toEqual(['second-signature']);
});
it('copies OpenAI reasoning replay metadata into providerOptions on replay', () => {
const providerMetadata = {
openai: { itemId: 'rs_123', reasoningEncryptedContent: 'encrypted' },
@ -656,6 +708,69 @@ describe('toAiMessages + fromAiMessages — round-trip', () => {
expect((block as { output: unknown }).output).toEqual({ result: 3 });
});
it('round-trips provider-executed tool results inside the assistant message', () => {
// Native web search: the AI SDK places the result in the assistant message,
// never in a role:tool message.
const results = [{ url: 'https://n8n.io', title: 'n8n', type: 'web_search_result' }];
const searchError = { type: 'web_search_tool_result_error', errorCode: 'max_uses_exceeded' };
const error = JSON.stringify(searchError);
const aiMessages: ModelMessage[] = [
{
role: 'assistant',
content: [
{
type: 'tool-call',
toolCallId: 'srvtoolu_1',
toolName: 'web_search',
input: { query: 'n8n' },
providerExecuted: true,
},
{
type: 'tool-result',
toolCallId: 'srvtoolu_1',
toolName: 'web_search',
output: { type: 'json', value: results },
},
{
type: 'tool-call',
toolCallId: 'srvtoolu_2',
toolName: 'web_search',
input: { query: 'more' },
providerExecuted: true,
},
{
type: 'tool-result',
toolCallId: 'srvtoolu_2',
toolName: 'web_search',
output: { type: 'error-json', value: searchError },
},
{ type: 'text', text: 'Here is what I found.' },
],
},
];
const persisted = fromAiMessages(aiMessages) as Message[];
expect(persisted[0].content).toMatchObject([
{ type: 'tool-call', state: 'resolved', output: results },
{ type: 'tool-call', state: 'rejected', error },
{ type: 'text' },
]);
const replayed = toAiMessages(persisted);
expect(replayed).toHaveLength(1);
expect(replayed[0].content).toMatchObject([
{ type: 'tool-call', toolCallId: 'srvtoolu_1', providerExecuted: true },
{ type: 'tool-result', toolCallId: 'srvtoolu_1', output: { type: 'json', value: results } },
{ type: 'tool-call', toolCallId: 'srvtoolu_2', providerExecuted: true },
{
type: 'tool-result',
toolCallId: 'srvtoolu_2',
output: { type: 'error-json', value: error },
},
{ type: 'text', text: 'Here is what I found.' },
]);
});
it('round-trip is structurally equivalent for a resolved tool-call', () => {
const original: Message[] = [
{

View File

@ -781,10 +781,10 @@ describe('AgentRuntime — reasoning-only turn', () => {
* reasoning-only message, `other`, and no usage. Nothing the user can see
* and nothing the loop can act on.
*/
function makeStreamReasoningOnly() {
function makeStreamReasoningOnly(finishReason = 'other') {
return {
stream: makeChunkStream([{ type: 'reasoning-delta', id: 'r-1', text: 'thinking...' }]),
finishReason: Promise.resolve('other'),
finishReason: Promise.resolve(finishReason),
usage: Promise.resolve({
inputTokens: undefined,
outputTokens: undefined,
@ -830,6 +830,42 @@ describe('AgentRuntime — reasoning-only turn', () => {
);
expect(String((error?.error as Error).message)).toContain('no output');
});
it.each([
{ finishReason: 'length', errorText: 'output token limit' },
{ finishReason: 'stop', errorText: 'without returning an answer' },
])(
'fails a reasoning-only $finishReason turn without retrying or persisting it',
async ({ finishReason, errorText }) => {
streamText.mockReturnValue(makeStreamReasoningOnly(finishReason));
const memory = new InMemoryMemory();
const runtime = new AgentRuntime({
name: 'test',
model: 'anthropic/claude-opus-5',
instructions: 'You are a test assistant.',
memory,
});
const result = await runtime.stream('make my workflow smarter', {
persistence: { threadId: 'thread-1', resourceId: 'user-1' },
});
const chunks = await collectChunks(result.stream);
expect(streamText).toHaveBeenCalledTimes(1);
expect(chunks).toEqual(
expect.arrayContaining([
expect.objectContaining({ type: 'error' }),
expect.objectContaining({ type: 'finish', finishReason: 'error' }),
]),
);
const error = chunks.find((chunk) => chunk.type === 'error');
expect(String(error?.error)).toContain(errorText);
const persisted = await memory.getMessages('thread-1', { resourceId: 'user-1' });
expect(
persisted.filter((message) => (message as { role?: string }).role === 'assistant'),
).toHaveLength(0);
},
);
});
// ---------------------------------------------------------------------------

View File

@ -43,12 +43,16 @@ If this affects the user's request, briefly let them know which server is unavai
}
/**
* Finish reasons that indicate the provider rejected or filtered the request
* when they arrive with zero output. `stop`/`length` with empty output are the
* model's own (odd but legal) choice; `tool-calls` always carries calls;
* `error` surfaces through the SDK's thrown error instead.
* Finish reasons that indicate the provider rejected, filtered, or truncated
* the request when they arrive with zero output. `tool-calls` always carries
* calls; `error` surfaces through the SDK's thrown error instead.
*/
const EMPTY_RESPONSE_ERROR_FINISH_REASONS = new Set(['other', 'unknown', 'content-filter']);
const EMPTY_RESPONSE_ERROR_FINISH_REASONS = new Set([
'length',
'other',
'unknown',
'content-filter',
]);
/**
* Whether a turn carries output the user can see or the loop can act on.
@ -69,6 +73,17 @@ function hasActionableContent(messages: AgentMessage[]): boolean {
);
}
function hasReasoningContent(messages: AgentMessage[]): boolean {
return messages.some(
(message) =>
'content' in message &&
Array.isArray(message.content) &&
message.content.some(
(content) => content.type === 'reasoning' || content.type === 'reasoning-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
@ -83,10 +98,23 @@ export function classifyModelTurnError(turn: {
providerError?: RawProviderError;
}): ModelTurnError | undefined {
if (hasActionableContent(turn.newMessages)) return undefined;
if (turn.aiFinishReason === 'stop' && hasReasoningContent(turn.newMessages)) {
return {
type: 'no_output',
message: 'The model finished without returning an answer. Try again or use another model.',
};
}
if (!EMPTY_RESPONSE_ERROR_FINISH_REASONS.has(turn.aiFinishReason)) return undefined;
const guidance =
'This can be a provider-side false positive — try rephrasing the message, clearing the chat history, or switching models.';
if (turn.aiFinishReason === 'length') {
return {
type: 'no_output',
message:
'The model reached its output token limit before it returned an answer. Reduce the request scope or use another model.',
};
}
if (turn.providerError) {
return {
type: turn.providerError.type,
@ -118,6 +146,11 @@ export function isEmptyModelTurn(turn: {
}): boolean {
if (turn.aiFinishReason === 'tool-calls') return false;
if (turn.structuredOutput !== undefined) return false;
// A reasoning-only stop or a truncated turn cannot recover under the same
// conditions. Retrying only consumes more tokens.
if (turn.errorReason && (turn.aiFinishReason === 'stop' || turn.aiFinishReason === 'length')) {
return false;
}
// 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.

View File

@ -121,6 +121,13 @@ function hasReplayableReasoningProviderOptions(
});
}
function isReasoningMergedByProvider(part: AiContentPart): boolean {
if (part.type !== 'reasoning') return false;
return Object.keys(part.providerOptions ?? {}).some(
(namespace) => getProviderQuirks(namespace).mergesAdjacentAssistantMessages === true,
);
}
export type ContentToolResultOutput = Extract<ToolResultPart['output'], { type: 'content' }>;
export function isContentToolResultOutput(value: unknown): value is ContentToolResultOutput {
@ -274,11 +281,15 @@ function toolCallToResultPart(
// rejected
const errorValue = block.error;
if (typeof errorValue === 'string') {
// Providers replay their own tool errors (e.g. a web search error code)
// from error-json; error-text is dropped with a warning.
return {
type: 'tool-result',
toolCallId: block.toolCallId,
toolName: block.toolName,
output: { type: 'error-text', value: errorValue },
output: block.providerExecuted
? { type: 'error-json', value: errorValue }
: { type: 'error-text', value: errorValue },
};
}
return {
@ -364,7 +375,8 @@ function fromAiContent(part: AiContentPart): MessageContent | undefined {
*
* For assistant messages with resolved/rejected tool-call blocks, this emits:
* 1. The assistant ModelMessage (tool-call parts only, no result fields)
* 2. One tool ModelMessage per settled tool-call block (resolved or rejected)
* 2. One tool ModelMessage per settled tool-call block (resolved or rejected),
* except provider-executed blocks, whose result stays in the assistant message
*
* Pending tool-call blocks are silently skipped (defense-in-depth; the strip
* step should already have removed them before forLlm() calls toAiMessages).
@ -418,9 +430,11 @@ function toAiMessageList(msg: Message): ModelMessage[] {
...(block.providerMetadata && { providerMetadata: block.providerMetadata }),
...(block.providerOptions && { providerOptions: block.providerOptions }),
} as ToolCallPart);
// Emit corresponding tool-result message immediately after
// Emit corresponding tool-result message immediately after. A
// provider-executed result belongs in the assistant message itself.
const resultPart = toolCallToResultPart(block);
resultMessages.push({ role: 'tool', content: [resultPart] });
if (block.providerExecuted) assistantParts.push(resultPart);
else resultMessages.push({ role: 'tool', content: [resultPart] });
} else {
const part = toAiContent(block);
if (part) assistantParts.push(part);
@ -458,7 +472,26 @@ function toAiMessageList(msg: Message): ModelMessage[] {
/** Convert n8n Messages to AI SDK ModelMessages for passing to stream/generateText. */
export function toAiMessages(messages: Message[]): ModelMessage[] {
return messages.flatMap(toAiMessageList);
const modelMessages = messages.flatMap(toAiMessageList);
const result: ModelMessage[] = [];
for (const [index, message] of modelMessages.entries()) {
if (
message.role !== 'assistant' ||
typeof message.content === 'string' ||
modelMessages[index + 1]?.role !== 'assistant'
) {
result.push(message);
continue;
}
// Keep replayable reasoning only on the last of the merged responses so
// signatures from separate turns never mix (see mergesAdjacentAssistantMessages).
const content = message.content.filter((part) => !isReasoningMergedByProvider(part));
if (content.length > 0) result.push({ ...message, content });
}
return result;
}
/**
@ -477,37 +510,41 @@ export function fromAiMessages(messages: ModelMessage[]): AgentMessage[] {
const toolCallIndex = new Map<string, ContentToolCall>();
const result: AgentMessage[] = [];
// Merge tool results back into the matching tool-call blocks
const settleToolCalls = (parts: readonly AiContentPart[]) => {
for (const part of parts) {
if (part.type !== 'tool-result') continue;
const block = toolCallIndex.get(part.toolCallId);
if (!block) continue; // orphan — drop
const { output } = part;
if (output.type === 'json' || output.type === 'text') {
const mutableBlock = block as Extract<ContentToolCall, { state: 'resolved' }>;
mutableBlock.state = 'resolved';
mutableBlock.output = output.value as JSONValue;
} else if (output.type === 'content') {
const mutableBlock = block as Extract<ContentToolCall, { state: 'resolved' }>;
mutableBlock.state = 'resolved';
mutableBlock.output = output as JSONValue;
} else if (output.type === 'error-json') {
const mutableBlock = block as Extract<ContentToolCall, { state: 'rejected' }>;
mutableBlock.state = 'rejected';
mutableBlock.error = JSON.stringify(output.value);
} else if (output.type === 'error-text') {
const mutableBlock = block as Extract<ContentToolCall, { state: 'rejected' }>;
mutableBlock.state = 'rejected';
mutableBlock.error = output.value;
} else {
const mutableBlock = block as Extract<ContentToolCall, { state: 'rejected' }>;
mutableBlock.state = 'rejected';
mutableBlock.error = JSON.stringify(output);
}
}
};
for (const msg of messages) {
if (msg.role === 'tool') {
// Merge tool results back into the matching tool-call blocks
const toolParts = msg.content as ToolResultPart[];
for (const part of toolParts) {
const block = toolCallIndex.get(part.toolCallId);
if (!block) continue; // orphan — drop
const { output } = part;
if (output.type === 'json' || output.type === 'text') {
const mutableBlock = block as Extract<ContentToolCall, { state: 'resolved' }>;
mutableBlock.state = 'resolved';
mutableBlock.output = output.value as JSONValue;
} else if (output.type === 'content') {
const mutableBlock = block as Extract<ContentToolCall, { state: 'resolved' }>;
mutableBlock.state = 'resolved';
mutableBlock.output = output as JSONValue;
} else if (output.type === 'error-json') {
const mutableBlock = block as Extract<ContentToolCall, { state: 'rejected' }>;
mutableBlock.state = 'rejected';
mutableBlock.error = JSON.stringify(output.value);
} else if (output.type === 'error-text') {
const mutableBlock = block as Extract<ContentToolCall, { state: 'rejected' }>;
mutableBlock.state = 'rejected';
mutableBlock.error = output.value;
} else {
const mutableBlock = block as Extract<ContentToolCall, { state: 'rejected' }>;
mutableBlock.state = 'rejected';
mutableBlock.error = JSON.stringify(output);
}
}
settleToolCalls(msg.content);
// Do not emit a separate n8n message for tool results
continue;
}
@ -531,6 +568,9 @@ export function fromAiMessages(messages: ModelMessage[]): AgentMessage[] {
toolCallIndex.set(block.toolCallId, block);
}
}
// Provider-executed tools (e.g. native web search) return their result
// inside the assistant message rather than in a role:tool message.
if (typeof rawContent !== 'string') settleToolCalls(rawContent);
}
}

View File

@ -18,6 +18,8 @@ export interface ProviderQuirks {
providerOptionsNamespace?: string;
/** providerMetadata keys on reasoning parts that must be copied to providerOptions and survive replay. */
reasoningReplayKeys?: string[];
/** Provider merges adjacent assistant messages into one; replayable reasoning may only survive on the last of them. */
mergesAdjacentAssistantMessages?: boolean;
/** Defaults merged under this provider's namespace into every tool's providerOptions (explicit tool values win). */
toolProviderOptionDefaults?: JSONObject;
/** Provider defaults to strict JSON Schema validation for structured output; relax for raw user schemas. */
@ -98,6 +100,11 @@ export const PROVIDER_QUIRKS: Partial<Record<ProviderId, ProviderQuirks>> = {
// exposes them in providerMetadata, not providerOptions. Shim until the
// provider copies them itself on replay.
reasoningReplayKeys: ['signature', 'redactedData'],
// QUIRK(anthropic): the provider merges adjacent assistant messages into
// one API message. Thinking blocks are only valid inside the response
// that produced them, so a merged message with signed reasoning from two
// responses is rejected with "thinking blocks ... cannot be modified".
mergesAdjacentAssistantMessages: true,
// QUIRK(anthropic): defaults every function tool to eager_input_streaming,
// which forwards the model's raw argument tokens without server-side JSON
// validation — malformed inputs (e.g. unquoted string values) then reach

View File

@ -16,6 +16,36 @@ async function drain(stream: ReadableStream<StreamChunk>): Promise<StreamChunk[]
}
describe('startStreamSession', () => {
it('emits only the provider error when a wrapper error follows', async () => {
const bus = new AgentEventBus();
const abortScope = bus.createAbortScope();
const providerError = new Error('Thinking blocks cannot be modified');
const readable = startStreamSession({
eventBus: bus,
abortScope,
runId: 'run-1',
options: undefined,
withRootSpan: async <T>(_op: unknown, _opts: unknown, _rid: unknown, fn: () => Promise<T>) =>
await fn(),
runLoop: async (guard) => {
await guard.write({ type: 'error', error: providerError });
throw new Error('No output generated. Check the stream for errors.');
},
flushTelemetry: vi.fn(),
cleanupRun: vi.fn(),
updateState: vi.fn(),
emitError: vi.fn(),
});
const chunks = await drain(readable);
expect(chunks.filter((chunk) => chunk.type === 'error')).toEqual([
{ type: 'error', error: providerError },
]);
expect(chunks.at(-1)).toMatchObject({ type: 'finish', finishReason: 'error' });
});
it('completes shutdown even when the abort persistence throws', async () => {
const bus = new AgentEventBus();
const abortScope = bus.createAbortScope();

View File

@ -12,6 +12,8 @@ import type { TokenUsage } from '../../types/sdk/agent';
export class StreamWriterGuard {
private closed = false;
private errorWritten = false;
constructor(private readonly writer: WritableStreamDefaultWriter<StreamChunk>) {
writer.closed
.then(() => {
@ -29,6 +31,7 @@ export class StreamWriterGuard {
if (this.closed) return;
try {
await this.writer.write(chunk);
if (chunk.type === 'error') this.errorWritten = true;
} catch {
// Downstream consumer is gone (cancelled/errored). Nothing useful to do.
}
@ -46,8 +49,8 @@ export class StreamWriterGuard {
}
/**
* Terminate the stream with an error: emit an `error` chunk and a terminal
* `finish` chunk, then close. Idempotent — a no-op if already closed.
* Terminate the stream with an error: ensure an `error` chunk exists, emit a
* terminal `finish` chunk, then close. Idempotent — a no-op if already closed.
*
* `finish` enriches the terminal chunk with usage/model so an aborted run
* still carries the tokens consumed before the stop.
@ -58,7 +61,7 @@ export class StreamWriterGuard {
finish?: { usage?: TokenUsage; model?: string },
): Promise<void> {
if (this.closed) return;
await this.write({ type: 'error', error });
if (!this.errorWritten) await this.write({ type: 'error', error });
await this.write({ type: 'finish', finishReason, ...finish });
await this.close();
}

View File

@ -185,6 +185,18 @@ describe('ExecutionRecorder', () => {
);
});
it('keeps the first stream error when a later wrapper error arrives', () => {
const recorder = new ExecutionRecorder();
recorder.record({ type: 'error', error: new Error('Thinking blocks cannot be modified') });
recorder.record({
type: 'error',
error: new Error('No output generated. Check the stream for errors.'),
});
expect(recorder.getMessageRecord().error).toBe('Thinking blocks cannot be modified');
});
it('captures text → tool call → text in order', () => {
const recorder = new ExecutionRecorder();

View File

@ -263,28 +263,19 @@ describe('execution-to-message-mapper', () => {
]);
});
it('includes the execution outcome on assistant messages', () => {
it('maps an execution error without model output into an assistant message', () => {
const result = executionToMessagesDto(
execution({
status: 'error',
timeline: [
{
type: 'tool-call',
kind: 'tool',
name: 'slow_tool',
toolCallId: 'call-1',
input: {},
output: undefined,
startTime: 100,
endTime: 0,
success: false,
},
],
error: 'Model request failed',
}),
);
expect(result[1]).toMatchObject({
expect(result[1]).toEqual({
id: 'execution-1:assistant',
role: 'assistant',
content: [{ type: 'text', text: 'Model request failed' }],
executionId: 'execution-1',
executionStatus: 'error',
});
});

View File

@ -478,7 +478,7 @@ export class ExecutionRecorder {
case 'error': {
this.flushReasoningBuffer();
this.flushTextBuffer();
this.error = normaliseStreamError(chunk.error);
this.error ??= normaliseStreamError(chunk.error);
break;
}
}

View File

@ -6,7 +6,7 @@ import type { TimelineEvent } from '../execution-recorder';
type ExecutionTranscript = Pick<
AgentExecution,
'id' | 'userMessage' | 'timeline' | 'attachments' | 'status'
'id' | 'userMessage' | 'timeline' | 'attachments' | 'status' | 'error'
>;
type ToolCallTimelineEvent = Extract<TimelineEvent, { type: 'tool-call' }>;
@ -167,6 +167,10 @@ export function executionToMessagesDto(execution: ExecutionTranscript): AgentPer
}
const assistantContent = assistantContentFromExecution(execution);
if (execution.status === 'error' && execution.error) {
const error = textPart(execution.error);
if (error) assistantContent.push(error);
}
if (assistantContent.length > 0) {
messages.push({
id: `${execution.id}:assistant`,