fix(ai): align Anthropic stream handling (#38733)

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<typeof AnthropicEvent>
 
 interface ParserState {
   readonly tools: ToolStream.State<number>
+  readonly reasoningSignatures: Readonly<Record<number, string>>
   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<StepResult>([state, NO_EVENTS])
 }
@@ -958,7 +997,11 @@ export const protocol = Protocol.make({
   },
   stream: {
     event: Protocol.jsonEvent(AnthropicEvent),
-    initial: () => ({ tools: ToolStream.empty<number>(), lifecycle: Lifecycle.initial() }),
+    initial: () => ({
+      tools: ToolStream.empty<number>(),
+      reasoningSignatures: {},
+      lifecycle: Lifecycle.initial(),
+    }),
     step,
   },
 })

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[],

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<AnthropicMessages.AnthropicMessagesBody>(
+        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, {

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" } },
           )