From 49bec25ae59d72406affc263de549809cd5ef56c Mon Sep 17 00:00:00 2001 From: Aiden Cline <63023139+rekram1-node@users.noreply.github.com> Date: Fri, 24 Jul 2026 14:23:35 -0500 Subject: [PATCH] fix(ai): align Anthropic stream handling (#38733) --- .../ai/src/protocols/anthropic-messages.ts | 89 ++++++++--- packages/ai/src/protocols/utils/lifecycle.ts | 15 +- .../test/provider/anthropic-messages.test.ts | 139 ++++++++++++++++++ packages/ai/test/tool-runtime.test.ts | 3 + 4 files changed, 217 insertions(+), 29 deletions(-) diff --git a/packages/ai/src/protocols/anthropic-messages.ts b/packages/ai/src/protocols/anthropic-messages.ts index b3a35f986e8..2812e0136f7 100644 --- a/packages/ai/src/protocols/anthropic-messages.ts +++ b/packages/ai/src/protocols/anthropic-messages.ts @@ -9,6 +9,7 @@ import { LLMEvent, Usage, type CacheHint, + type FinishReasonDetails, type FinishReason, type JsonSchema, type LLMRequest, @@ -288,7 +289,12 @@ type AnthropicEvent = Schema.Schema.Type interface ParserState { readonly tools: ToolStream.State + readonly reasoningSignatures: Readonly> readonly usage?: Usage + readonly pendingFinish?: { + readonly reason: FinishReasonDetails + readonly providerMetadata?: ProviderMetadata + } readonly lifecycle: Lifecycle.State } @@ -763,6 +769,10 @@ const onContentBlockStart = (state: ParserState, event: AnthropicEvent): StepRes tools: ToolStream.start(state.tools, event.index, { id: block.id ?? String(event.index), name: block.name ?? "", + input: + block.input !== undefined && (!ProviderShared.isRecord(block.input) || Object.keys(block.input).length > 0) + ? ProviderShared.encodeJson(block.input) + : undefined, providerExecuted: block.type === "server_tool_use", }), }, @@ -777,20 +787,31 @@ const onContentBlockStart = (state: ParserState, event: AnthropicEvent): StepRes ] } - if (block.type === "text" && block.text) { + if (block.type === "text" && block.text !== undefined) { const events: LLMEvent[] = [] + const id = `text-${event.index ?? 0}` + const lifecycle = Lifecycle.textStart(state.lifecycle, events, id) return [ - { ...state, lifecycle: Lifecycle.textDelta(state.lifecycle, events, `text-${event.index ?? 0}`, block.text) }, + { ...state, lifecycle: block.text ? Lifecycle.textDelta(lifecycle, events, id, block.text) : lifecycle }, events, ] } - if (block.type === "thinking" && block.thinking) { + if (block.type === "thinking" && block.thinking !== undefined) { const events: LLMEvent[] = [] + const id = `reasoning-${event.index ?? 0}` + const providerMetadata = block.signature === undefined ? undefined : anthropicMetadata({ signature: block.signature }) + const lifecycle = Lifecycle.reasoningStart(state.lifecycle, events, id, providerMetadata) return [ { ...state, - lifecycle: Lifecycle.reasoningDelta(state.lifecycle, events, `reasoning-${event.index ?? 0}`, block.thinking), + lifecycle: block.thinking + ? Lifecycle.reasoningDelta(lifecycle, events, id, block.thinking, providerMetadata) + : lifecycle, + reasoningSignatures: + event.index === undefined || block.signature === undefined + ? state.reasoningSignatures + : { ...state.reasoningSignatures, [event.index]: block.signature }, }, events, ] @@ -799,7 +820,7 @@ const onContentBlockStart = (state: ParserState, event: AnthropicEvent): StepRes // Redacted thinking surfaces as an empty reasoning part carrying the opaque // payload as `redactedData` metadata (same model as Vercel's // @ai-sdk/anthropic). The existing content_block_stop closes the part. - if (block.type === "redacted_thinking" && block.data) { + if (block.type === "redacted_thinking" && block.data !== undefined) { const events: LLMEvent[] = [] return [ { @@ -847,18 +868,13 @@ const onContentBlockDelta = Effect.fn("AnthropicMessages.onContentBlockDelta")(f } if (delta?.type === "signature_delta" && delta.signature) { - const events: LLMEvent[] = [] + const index = event.index ?? 0 return [ { ...state, - lifecycle: Lifecycle.reasoningEnd( - state.lifecycle, - events, - `reasoning-${event.index ?? 0}`, - anthropicMetadata({ signature: delta.signature }), - ), + reasoningSignatures: { ...state.reasoningSignatures, [index]: delta.signature }, }, - events, + NO_EVENTS, ] satisfies StepResult } @@ -889,31 +905,53 @@ const onContentBlockStop = Effect.fn("AnthropicMessages.onContentBlockStop")(fun const result = yield* ToolStream.finish(ADAPTER, state.tools, event.index) const events: LLMEvent[] = [] const resultEvents = result.events ?? [] + const signature = state.reasoningSignatures[event.index] const lifecycle = resultEvents.length ? Lifecycle.stepStart(state.lifecycle, events) : Lifecycle.reasoningEnd( Lifecycle.textEnd(state.lifecycle, events, `text-${event.index}`), events, `reasoning-${event.index}`, + signature === undefined ? undefined : anthropicMetadata({ signature }), ) events.push(...resultEvents) - return [{ ...state, lifecycle, tools: result.tools }, events] satisfies StepResult + const reasoningSignatures = { ...state.reasoningSignatures } + delete reasoningSignatures[event.index] + return [{ ...state, lifecycle, tools: result.tools, reasoningSignatures }, events] satisfies StepResult }) const onMessageDelta = (state: ParserState, event: AnthropicEvent): StepResult => { const usage = mergeUsage(state.usage, mapUsage(event.usage)) + return [ + { + ...state, + usage, + pendingFinish: { + reason: { + normalized: mapFinishReason(event.delta?.stop_reason), + raw: event.delta?.stop_reason ?? undefined, + }, + providerMetadata: + event.delta?.stop_sequence === null || event.delta?.stop_sequence === undefined + ? undefined + : anthropicMetadata({ stopSequence: event.delta.stop_sequence }), + }, + }, + NO_EVENTS, + ] +} + +const onMessageStop = (state: ParserState): StepResult => { const events: LLMEvent[] = [] const lifecycle = Lifecycle.finish(state.lifecycle, events, { - reason: { - normalized: mapFinishReason(event.delta?.stop_reason), - raw: event.delta?.stop_reason ?? undefined, + reason: state.pendingFinish?.reason ?? { + normalized: "unknown", + raw: undefined, }, - usage, - providerMetadata: event.delta?.stop_sequence - ? anthropicMetadata({ stopSequence: event.delta.stop_sequence }) - : undefined, + usage: state.usage, + providerMetadata: state.pendingFinish?.providerMetadata, }) - return [{ ...state, lifecycle, usage }, events] + return [{ ...state, lifecycle }, events] } // Prefix `error.type` so overloads, rate limits, and quota errors are visible @@ -938,6 +976,7 @@ const step = (state: ParserState, event: AnthropicEvent) => { if (event.type === "content_block_delta") return onContentBlockDelta(state, event) if (event.type === "content_block_stop") return onContentBlockStop(state, event) if (event.type === "message_delta") return Effect.succeed(onMessageDelta(state, event)) + if (event.type === "message_stop") return Effect.succeed(onMessageStop(state)) if (event.type === "error") return onError(event) return Effect.succeed([state, NO_EVENTS]) } @@ -958,7 +997,11 @@ export const protocol = Protocol.make({ }, stream: { event: Protocol.jsonEvent(AnthropicEvent), - initial: () => ({ tools: ToolStream.empty(), lifecycle: Lifecycle.initial() }), + initial: () => ({ + tools: ToolStream.empty(), + reasoningSignatures: {}, + lifecycle: Lifecycle.initial(), + }), step, }, }) diff --git a/packages/ai/src/protocols/utils/lifecycle.ts b/packages/ai/src/protocols/utils/lifecycle.ts index 761cff3690f..1ee8b608f67 100644 --- a/packages/ai/src/protocols/utils/lifecycle.ts +++ b/packages/ai/src/protocols/utils/lifecycle.ts @@ -14,16 +14,19 @@ export const stepStart = (state: State, events: LLMEvent[]): State => { return { ...state, stepStarted: true } } -export const textDelta = (state: State, events: LLMEvent[], id: string, text: string): State => { +export const textStart = (state: State, events: LLMEvent[], id: string, providerMetadata?: ProviderMetadata): State => { + if (state.text.has(id)) return state const stepped = stepStart(state, events) - if (stepped.text.has(id)) { - events.push(LLMEvent.textDelta({ id, text })) - return stepped - } - events.push(LLMEvent.textStart({ id }), LLMEvent.textDelta({ id, text })) + events.push(LLMEvent.textStart({ id, providerMetadata })) return { ...stepped, text: new Set([...stepped.text, id]) } } +export const textDelta = (state: State, events: LLMEvent[], id: string, text: string): State => { + const started = textStart(state, events, id) + events.push(LLMEvent.textDelta({ id, text })) + return started +} + export const reasoningStart = ( state: State, events: LLMEvent[], diff --git a/packages/ai/test/provider/anthropic-messages.test.ts b/packages/ai/test/provider/anthropic-messages.test.ts index acdab6a7b0c..004a378bcff 100644 --- a/packages/ai/test/provider/anthropic-messages.test.ts +++ b/packages/ai/test/provider/anthropic-messages.test.ts @@ -506,6 +506,7 @@ describe("Anthropic Messages route", () => { expect(response.events.find((event) => event.type === "reasoning-end")).toMatchObject({ providerMetadata: { anthropic: { signature: "sig_1" } }, }) + expect(response.events.find((event) => event.type === "reasoning-delta" && event.text === "")).toBeUndefined() expect(response.message.content).toEqual([ { type: "text", text: "Hello!" }, { type: "reasoning", text: "thinking", providerMetadata: { anthropic: { signature: "sig_1" } } }, @@ -518,6 +519,139 @@ describe("Anthropic Messages route", () => { }), ) + it.effect("requires message_stop before completing a streamed message", () => + Effect.gen(function* () { + const error = yield* LLMClient.generate(request).pipe( + Effect.provide( + fixedResponse( + sseEvents( + { type: "message_start", message: { usage: { input_tokens: 5 } } }, + { type: "content_block_start", index: 0, content_block: { type: "text", text: "" } }, + { type: "content_block_delta", index: 0, delta: { type: "text_delta", text: "Hello" } }, + { type: "content_block_stop", index: 0 }, + { type: "message_delta", delta: { stop_reason: "end_turn" }, usage: { output_tokens: 1 } }, + ), + ), + ), + Effect.flip, + ) + + expect(error.reason).toMatchObject({ + _tag: "InvalidProviderOutput", + message: "Provider stream ended without a terminal finish event", + }) + }), + ) + + it.effect("round-trips omitted thinking carried only by a signature delta", () => + Effect.gen(function* () { + const response = yield* LLMClient.generate(request).pipe( + Effect.provide( + fixedResponse( + sseEvents( + { type: "message_start", message: { usage: { input_tokens: 5 } } }, + { + type: "content_block_start", + index: 0, + content_block: { type: "thinking", thinking: "", signature: "" }, + }, + { type: "content_block_delta", index: 0, delta: { type: "signature_delta", signature: "sig_1" } }, + { type: "content_block_stop", index: 0 }, + { type: "message_delta", delta: { stop_reason: "end_turn" }, usage: { output_tokens: 1 } }, + { type: "message_stop" }, + ), + ), + ), + ) + + expect(response.message.content).toEqual([ + { type: "reasoning", text: "", providerMetadata: { anthropic: { signature: "sig_1" } } }, + ]) + + const prepared = yield* LLMClient.prepare( + LLM.request({ model, messages: [response.message], cache: "none" }), + ) + expect(prepared.body.messages).toEqual([ + { role: "assistant", content: [{ type: "thinking", thinking: "", signature: "sig_1" }] }, + ]) + }), + ) + + it.effect("retains a thinking signature supplied in content_block_start", () => + Effect.gen(function* () { + const response = yield* LLMClient.generate(request).pipe( + Effect.provide( + fixedResponse( + sseEvents( + { type: "message_start", message: { usage: { input_tokens: 5 } } }, + { + type: "content_block_start", + index: 0, + content_block: { type: "thinking", thinking: "", signature: "sig_1" }, + }, + { type: "content_block_stop", index: 0 }, + { type: "message_delta", delta: { stop_reason: "end_turn" }, usage: { output_tokens: 1 } }, + { type: "message_stop" }, + ), + ), + ), + ) + + expect(response.message.content).toEqual([ + { type: "reasoning", text: "", providerMetadata: { anthropic: { signature: "sig_1" } } }, + ]) + expect(response.events.find((event) => event.type === "reasoning-end")).toMatchObject({ + providerMetadata: { anthropic: { signature: "sig_1" } }, + }) + }), + ) + + it.effect("retains complete tool input from content_block_start", () => + Effect.gen(function* () { + const response = yield* LLMClient.generate(request).pipe( + Effect.provide( + fixedResponse( + sseEvents( + { type: "message_start", message: { usage: { input_tokens: 5 } } }, + { + type: "content_block_start", + index: 0, + content_block: { type: "tool_use", id: "call_1", name: "lookup", input: { query: "weather" } }, + }, + { type: "content_block_stop", index: 0 }, + { type: "message_delta", delta: { stop_reason: "tool_use" }, usage: { output_tokens: 1 } }, + { type: "message_stop" }, + ), + ), + ), + ) + + expect(response.toolCalls).toMatchObject([ + { id: "call_1", name: "lookup", input: { query: "weather" } }, + ]) + }), + ) + + it.effect("retains empty text blocks", () => + Effect.gen(function* () { + const response = yield* LLMClient.generate(request).pipe( + Effect.provide( + fixedResponse( + sseEvents( + { type: "message_start", message: { usage: { input_tokens: 5 } } }, + { type: "content_block_start", index: 0, content_block: { type: "text", text: "" } }, + { type: "content_block_stop", index: 0 }, + { type: "message_delta", delta: { stop_reason: "end_turn" }, usage: { output_tokens: 1 } }, + { type: "message_stop" }, + ), + ), + ), + ) + + expect(response.message.content).toEqual([{ type: "text", text: "" }]) + }), + ) + it.effect("parses redacted thinking into empty reasoning with redactedData metadata", () => Effect.gen(function* () { const body = sseEvents( @@ -629,6 +763,7 @@ describe("Anthropic Messages route", () => { delta: { stop_reason: "model_context_window_exceeded" }, usage: { output_tokens: 1 }, }, + { type: "message_stop" }, ), ), ), @@ -646,6 +781,7 @@ describe("Anthropic Messages route", () => { sseEvents( { type: "message_start", message: { usage: { input_tokens: 5 } } }, { type: "message_delta", delta: { stop_reason: "pause_turn" }, usage: { output_tokens: 1 } }, + { type: "message_stop" }, ), ), ), @@ -664,6 +800,7 @@ describe("Anthropic Messages route", () => { { type: "content_block_delta", index: 0, delta: { type: "input_json_delta", partial_json: ':"weather"}' } }, { type: "content_block_stop", index: 0 }, { type: "message_delta", delta: { stop_reason: "tool_use" }, usage: { output_tokens: 1 } }, + { type: "message_stop" }, ) const response = yield* LLMClient.generate( LLMRequest.update(request, { @@ -849,6 +986,7 @@ describe("Anthropic Messages route", () => { { type: "content_block_delta", index: 2, delta: { type: "text_delta", text: "Found it." } }, { type: "content_block_stop", index: 2 }, { type: "message_delta", delta: { stop_reason: "end_turn" }, usage: { output_tokens: 8 } }, + { type: "message_stop" }, ) const response = yield* LLMClient.generate( LLMRequest.update(request, { @@ -912,6 +1050,7 @@ describe("Anthropic Messages route", () => { }, { type: "content_block_stop", index: 1 }, { type: "message_delta", delta: { stop_reason: "end_turn" }, usage: { output_tokens: 1 } }, + { type: "message_stop" }, ) const response = yield* LLMClient.generate( LLMRequest.update(request, { diff --git a/packages/ai/test/tool-runtime.test.ts b/packages/ai/test/tool-runtime.test.ts index aa4c85f6efe..356ee779c8b 100644 --- a/packages/ai/test/tool-runtime.test.ts +++ b/packages/ai/test/tool-runtime.test.ts @@ -539,6 +539,7 @@ describe("LLMClient tools", () => { }, { type: "content_block_stop", index: 1 }, { type: "message_delta", delta: { stop_reason: "tool_use" }, usage: { output_tokens: 5 } }, + { type: "message_stop" }, ) : sseEvents( { type: "message_start", message: { usage: { input_tokens: 5 } } }, @@ -546,6 +547,7 @@ describe("LLMClient tools", () => { { type: "content_block_delta", index: 0, delta: { type: "text_delta", text: "Done." } }, { type: "content_block_stop", index: 0 }, { type: "message_delta", delta: { stop_reason: "end_turn" }, usage: { output_tokens: 1 } }, + { type: "message_stop" }, ), { headers: { "content-type": "text/event-stream" } }, ) @@ -801,6 +803,7 @@ describe("LLMClient tools", () => { { type: "content_block_delta", index: 2, delta: { type: "text_delta", text: "Done." } }, { type: "content_block_stop", index: 2 }, { type: "message_delta", delta: { stop_reason: "end_turn" }, usage: { output_tokens: 8 } }, + { type: "message_stop" }, ), { headers: { "content-type": "text/event-stream" } }, )