From 3bcb90b2d08f397511752b4f6e5dfd0afb36c7fb Mon Sep 17 00:00:00 2001 From: Josh Avant <830519+joshavant@users.noreply.github.com> Date: Fri, 10 Jul 2026 20:54:37 -0500 Subject: [PATCH] fix(anthropic): stage thinking signatures until block stop (#104043) --- src/agents/anthropic-transport-stream.test.ts | 162 +++++++++++++++++- src/agents/anthropic-transport-stream.ts | 22 ++- 2 files changed, 179 insertions(+), 5 deletions(-) diff --git a/src/agents/anthropic-transport-stream.test.ts b/src/agents/anthropic-transport-stream.test.ts index eb85aafc4215..ebe3710ab98b 100644 --- a/src/agents/anthropic-transport-stream.test.ts +++ b/src/agents/anthropic-transport-stream.test.ts @@ -25,13 +25,55 @@ type AnthropicStreamOptions = Parameters[2]; type RequestTransportConfig = Parameters[1]; function createSseResponse(events: Record[] = []): Response { - const body = events.map((event) => `data: ${JSON.stringify(event)}\n\n`).join(""); + const body = serializeSseEvents(events); return new Response(body, { status: 200, headers: { "content-type": "text/event-stream" }, }); } +function serializeSseEvents(events: Record[]): string { + return events.map((event) => `data: ${JSON.stringify(event)}\n\n`).join(""); +} + +function createFailingSseResponse(events: Record[], error: Error): Response { + const encoder = new TextEncoder(); + let sentEvents = false; + const body = new ReadableStream({ + pull(controller) { + if (!sentEvents) { + sentEvents = true; + controller.enqueue(encoder.encode(serializeSseEvents(events))); + return; + } + controller.error(error); + }, + }); + return new Response(body, { + status: 200, + headers: { "content-type": "text/event-stream" }, + }); +} + +function createInterruptedThinkingEvents(): Record[] { + return [ + { + type: "message_start", + message: { id: "msg_1", usage: { input_tokens: 6, output_tokens: 0 } }, + }, + { + type: "content_block_start", + index: 0, + content_block: { type: "thinking", thinking: "step by step", signature: "" }, + }, + { + type: "content_block_delta", + index: 0, + delta: { type: "signature_delta", signature: "partial-signature" }, + }, + ]; +} + function createStalledSseResponse(params: { onCancel: (reason: unknown) => void }): Response { const encoder = new TextEncoder(); const body = new ReadableStream({ @@ -1941,6 +1983,124 @@ describe("anthropic transport stream", () => { }); }); + it.each([ + { + label: "the stream ends before content_block_stop", + response: () => createSseResponse(createInterruptedThinkingEvents()), + stopReason: "stop", + }, + { + label: "the provider errors before content_block_stop", + response: () => + createSseResponse([ + ...createInterruptedThinkingEvents(), + { type: "error", error: { message: "provider failed" } }, + ]), + stopReason: "error", + }, + { + label: "the response body fails", + response: () => + createFailingSseResponse( + createInterruptedThinkingEvents(), + new Error("response body failed"), + ), + stopReason: "error", + }, + ])("does not persist signature deltas when $label", async ({ response, stopReason }) => { + guardedFetchMock.mockResolvedValueOnce(response()); + + const result = await runTransportStream( + makeAnthropicTransportModel(), + { messages: [{ role: "user", content: "think" }] } as AnthropicStreamContext, + { apiKey: "sk-ant-api" } as AnthropicStreamOptions, + ); + + expect(result.stopReason).toBe(stopReason); + expect(result.content[0]).toMatchObject({ + type: "thinking", + thinking: "step by step", + thinkingSignature: "", + }); + }); + + it("does not persist signature deltas when the request aborts", async () => { + const controller = new AbortController(); + guardedFetchMock.mockResolvedValueOnce( + createOpenRawSseResponse({ + body: serializeSseEvents(createInterruptedThinkingEvents()), + onCancel: () => undefined, + }), + ); + setTimeout(() => controller.abort(new Error("request aborted")), 20); + + const result = await runTransportStream( + makeAnthropicTransportModel(), + { messages: [{ role: "user", content: "think" }] } as AnthropicStreamContext, + { + apiKey: "sk-ant-api", + signal: controller.signal, + } as AnthropicStreamOptions, + ); + + expect(result.stopReason).toBe("aborted"); + expect(result.content[0]).toMatchObject({ + type: "thinking", + thinkingSignature: "", + }); + }); + + it("commits only stopped signatures across interleaved thinking blocks", async () => { + guardedFetchMock.mockResolvedValueOnce( + createSseResponse([ + { + type: "message_start", + message: { id: "msg_1", usage: { input_tokens: 6, output_tokens: 0 } }, + }, + { + type: "content_block_start", + index: 0, + content_block: { type: "thinking", thinking: "first", signature: "" }, + }, + { + type: "content_block_start", + index: 1, + content_block: { type: "thinking", thinking: "second", signature: "" }, + }, + { + type: "content_block_delta", + index: 1, + delta: { type: "signature_delta", signature: "complete-second" }, + }, + { + type: "content_block_delta", + index: 0, + delta: { type: "signature_delta", signature: "partial-first" }, + }, + { type: "content_block_stop", index: 1 }, + ]), + ); + + const result = await runTransportStream( + makeAnthropicTransportModel(), + { messages: [{ role: "user", content: "think" }] } as AnthropicStreamContext, + { apiKey: "sk-ant-api" } as AnthropicStreamOptions, + ); + + expect(result.content).toEqual([ + expect.objectContaining({ + type: "thinking", + thinking: "first", + thinkingSignature: "", + }), + expect.objectContaining({ + type: "thinking", + thinking: "second", + thinkingSignature: "complete-second", + }), + ]); + }); + it("captures OpenAI-style reasoning_content deltas from Anthropic-compatible streams", async () => { guardedFetchMock.mockResolvedValueOnce( createSseResponse([ diff --git a/src/agents/anthropic-transport-stream.ts b/src/agents/anthropic-transport-stream.ts index be4a77f4c715..0c2be2c2446c 100644 --- a/src/agents/anthropic-transport-stream.ts +++ b/src/agents/anthropic-transport-stream.ts @@ -1256,7 +1256,9 @@ export function createAnthropicMessagesTransportStreamFn(): StreamFn { ); const blocks = output.content; const blockIndexes = new Map(); - const signatureDeltaIndexes = new Set(); + // Signature deltas are opaque and only complete at content_block_stop. + // Keep partial bytes out of output so interrupted streams cannot poison replay. + const pendingThinkingSignatures = new Map(); const allowReasoningContentReplay = supportsReasoningContentReplay(model); const reasoningContentThinkingBlocks = new Map(); const reasoningContentTextBlocks = new Map(); @@ -1453,6 +1455,7 @@ export function createAnthropicMessagesTransportStreamFn(): StreamFn { refusalBuffer?.discard(); pendingTextEnds.length = 0; blockIndexes.clear(); + pendingThinkingSignatures.clear(); applyAnthropicFallbackBoundary({ output, boundary: fallbackBoundary, @@ -1492,6 +1495,7 @@ export function createAnthropicMessagesTransportStreamFn(): StreamFn { } continue; } + pendingThinkingSignatures.delete(index); if (contentBlock?.type === "text") { const text = typeof contentBlock.text === "string" @@ -1695,16 +1699,23 @@ export function createAnthropicMessagesTransportStreamFn(): StreamFn { typeof delta.signature === "string" ) { const signatureIndex = eventIndexKey(event.index); - if (!signatureDeltaIndexes.has(signatureIndex)) { - signatureDeltaIndexes.add(signatureIndex); + const pendingSignature = pendingThinkingSignatures.get(signatureIndex); + if (pendingSignature === undefined) { block.thinkingSignature = ""; + pendingThinkingSignatures.set(signatureIndex, delta.signature); + } else { + pendingThinkingSignatures.set(signatureIndex, pendingSignature + delta.signature); } - block.thinkingSignature = (block.thinkingSignature || "") + delta.signature; } continue; } if (event.type === "content_block_stop") { const eventIndex = typeof event.index === "number" ? event.index : undefined; + const pendingSignature = + eventIndex === undefined ? undefined : pendingThinkingSignatures.get(eventIndex); + if (eventIndex !== undefined) { + pendingThinkingSignatures.delete(eventIndex); + } const index = eventIndex === undefined ? undefined : blockIndexes.get(eventIndex); const block = index === undefined ? undefined : blocks[index]; if (eventIndex === undefined || index === undefined || !block) { @@ -1724,6 +1735,9 @@ export function createAnthropicMessagesTransportStreamFn(): StreamFn { continue; } if (block.type === "thinking") { + if (pendingSignature !== undefined) { + block.thinkingSignature = pendingSignature; + } eventSink.push({ type: "thinking_end", contentIndex: index,