diff --git a/packages/ai/src/protocols/anthropic-messages.ts b/packages/ai/src/protocols/anthropic-messages.ts index 504835da5d5..24e3ea7eef0 100644 --- a/packages/ai/src/protocols/anthropic-messages.ts +++ b/packages/ai/src/protocols/anthropic-messages.ts @@ -1343,20 +1343,28 @@ const onMessageDelta = ( event: AnthropicEvent & { readonly delta?: AnthropicStreamDelta }, ): StepResult => { const usage = mergeUsage(state.usage, mapUsage(event.usage, state.providerMetadataKey), state.providerMetadataKey) + const pendingFinish = (() => { + const stopReason = event.delta?.stop_reason + if (stopReason === null || stopReason === undefined) return state.pendingFinish + + const stopSequence = event.delta?.stop_sequence + const finishMetadata = + stopSequence === null || stopSequence === undefined + ? state.pendingFinish?.providerMetadata + : providerMetadata(state.providerMetadataKey, { stopSequence }) + return { + reason: { + normalized: mapFinishReason(stopReason), + raw: stopReason, + }, + providerMetadata: finishMetadata, + } + })() 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 - : providerMetadata(state.providerMetadataKey, { stopSequence: event.delta.stop_sequence }), - }, + pendingFinish, }, NO_EVENTS, ] diff --git a/packages/ai/test/provider/anthropic-messages.test.ts b/packages/ai/test/provider/anthropic-messages.test.ts index 0703cb53550..1fbd15dc8c2 100644 --- a/packages/ai/test/provider/anthropic-messages.test.ts +++ b/packages/ai/test/provider/anthropic-messages.test.ts @@ -949,6 +949,41 @@ describe("Anthropic Messages route", () => { }), ) + it.effect("preserves terminal state across usage-only message deltas", () => + Effect.gen(function* () { + const response = yield* LLMClient.generate(request).pipe( + Effect.provide( + fixedResponse( + sseEvents( + { type: "message_start", message: { usage: { input_tokens: 5 } } }, + { + type: "message_delta", + delta: { stop_reason: "end_turn", stop_sequence: "X" }, + usage: { output_tokens: 8 }, + }, + { type: "message_delta", delta: {}, usage: { output_tokens: 10 } }, + { type: "message_stop" }, + ), + ), + ), + ) + + expect(response.usage).toMatchObject({ inputTokens: 5, outputTokens: 10, totalTokens: 15 }) + expect(response.finishReason).toEqual({ normalized: "stop", raw: "end_turn" }) + expect(response.events.find((event) => event.type === "step-finish")).toMatchObject({ + reason: { normalized: "stop", raw: "end_turn" }, + usage: { inputTokens: 5, outputTokens: 10, totalTokens: 15 }, + providerMetadata: { anthropic: { stopSequence: "X" } }, + }) + expect(response.events.at(-1)).toMatchObject({ + type: "finish", + reason: { normalized: "stop", raw: "end_turn" }, + usage: { inputTokens: 5, outputTokens: 10, totalTokens: 15 }, + providerMetadata: { anthropic: { stopSequence: "X" } }, + }) + }), + ) + it.effect("requires message_stop before completing a streamed message", () => Effect.gen(function* () { const error = yield* LLMClient.generate(request).pipe(