From 4f201f87a90b1f0fa99cfdc98cae2cea20b37178 Mon Sep 17 00:00:00 2001 From: Aiden Cline <63023139+rekram1-node@users.noreply.github.com> Date: Fri, 24 Jul 2026 11:15:23 -0500 Subject: [PATCH] fix(ai): align Bedrock stream handling (#38712) --- packages/ai/src/protocols/bedrock-converse.ts | 39 ++-- .../ai/src/protocols/bedrock-event-stream.ts | 18 +- .../ai/test/provider/bedrock-converse.test.ts | 170 +++++++++++++++++- 3 files changed, 208 insertions(+), 19 deletions(-) diff --git a/packages/ai/src/protocols/bedrock-converse.ts b/packages/ai/src/protocols/bedrock-converse.ts index 18de2642059..6dda74d4171 100644 --- a/packages/ai/src/protocols/bedrock-converse.ts +++ b/packages/ai/src/protocols/bedrock-converse.ts @@ -155,6 +155,12 @@ const BedrockUsageSchema = Schema.Struct({ }) type BedrockUsageSchema = Schema.Schema.Type +const BedrockStreamException = Schema.Struct({ + message: Schema.optional(Schema.String), + originalMessage: Schema.optional(Schema.String), + originalStatusCode: Schema.optional(Schema.Number), +}) + // Streaming event shape — the AWS event stream wraps each JSON payload by its // `:event-type` header (e.g. `messageStart`, `contentBlockDelta`). We // reconstruct that wrapping in `decodeFrames` below so the event schema can @@ -184,6 +190,9 @@ const BedrockEvent = Schema.Struct({ signature: Schema.optional(Schema.String), // Blob fields in Bedrock's JSON event stream are base64 strings. redactedContent: Schema.optional(Schema.String), + // Vercel's Bedrock provider exposes the same delta under + // Anthropic's shorter `data` spelling. + data: Schema.optional(Schema.String), }), ), }), @@ -203,11 +212,11 @@ const BedrockEvent = Schema.Struct({ metrics: Schema.optional(Schema.Unknown), }), ), - internalServerException: Schema.optional(Schema.Struct({ message: Schema.String })), - modelStreamErrorException: Schema.optional(Schema.Struct({ message: Schema.String })), - validationException: Schema.optional(Schema.Struct({ message: Schema.String })), - throttlingException: Schema.optional(Schema.Struct({ message: Schema.String })), - serviceUnavailableException: Schema.optional(Schema.Struct({ message: Schema.String })), + internalServerException: Schema.optional(BedrockStreamException), + modelStreamErrorException: Schema.optional(BedrockStreamException), + validationException: Schema.optional(BedrockStreamException), + throttlingException: Schema.optional(BedrockStreamException), + serviceUnavailableException: Schema.optional(BedrockStreamException), }) type BedrockEvent = Schema.Schema.Type @@ -531,14 +540,20 @@ const step = (state: ParserState, event: BedrockEvent) => const index = event.contentBlockDelta.contentBlockIndex const reasoning = event.contentBlockDelta.delta.reasoningContent const events: LLMEvent[] = [] - const lifecycle = reasoning.text - ? Lifecycle.reasoningDelta(state.lifecycle, events, `reasoning-${index}`, reasoning.text) - : reasoning.redactedContent !== undefined - ? Lifecycle.reasoningStart( + const redactedData = reasoning.redactedContent ?? reasoning.data + const providerMetadata = reasoning.signature + ? bedrockMetadata({ signature: reasoning.signature }) + : redactedData !== undefined + ? bedrockMetadata({ redactedData }) + : undefined + const lifecycle = + reasoning.text !== undefined || providerMetadata !== undefined + ? Lifecycle.reasoningDelta( state.lifecycle, events, `reasoning-${index}`, - bedrockMetadata({ redactedData: reasoning.redactedContent }), + reasoning.text ?? "", + providerMetadata, ) : state.lifecycle return [ @@ -618,7 +633,7 @@ const step = (state: ParserState, event: BedrockEvent) => } if (event.metadata) { - const usage = mapUsage(event.metadata.usage) + const usage = mapUsage(event.metadata.usage) ?? state.pendingFinish?.usage return [ { ...state, @@ -645,7 +660,7 @@ const step = (state: ParserState, event: BedrockEvent) => module: ADAPTER, method: "stream", reason: classifyProviderFailure({ - message: exception[1]?.message ?? "Bedrock Converse stream error", + message: exception[1]?.message ?? exception[1]?.originalMessage ?? "Bedrock Converse stream error", code: exception[0], }), }) diff --git a/packages/ai/src/protocols/bedrock-event-stream.ts b/packages/ai/src/protocols/bedrock-event-stream.ts index 0312ea7d57d..6d5b58ad8e5 100644 --- a/packages/ai/src/protocols/bedrock-event-stream.ts +++ b/packages/ai/src/protocols/bedrock-event-stream.ts @@ -53,8 +53,22 @@ const consumeFrames = (route: string) => (state: FrameBufferState, chunk: Uint8A }) cursor = { buffer: cursor.buffer, offset: cursor.offset + totalLength } - if (decoded.headers[":message-type"]?.value !== "event") continue - const eventType = decoded.headers[":event-type"]?.value + const messageType = decoded.headers[":message-type"]?.value + if (messageType === "error") { + const code = decoded.headers[":error-code"]?.value + const message = decoded.headers[":error-message"]?.value + return yield* ProviderShared.eventError( + route, + [code, message].filter((value): value is string => typeof value === "string").join(": ") || + "Bedrock Converse event-stream error", + ) + } + const eventType = + messageType === "event" + ? decoded.headers[":event-type"]?.value + : messageType === "exception" + ? decoded.headers[":exception-type"]?.value + : undefined if (typeof eventType !== "string") continue const payload = utf8.decode(decoded.body) if (!payload) continue diff --git a/packages/ai/test/provider/bedrock-converse.test.ts b/packages/ai/test/provider/bedrock-converse.test.ts index 05c0c669fc1..86f2240488c 100644 --- a/packages/ai/test/provider/bedrock-converse.test.ts +++ b/packages/ai/test/provider/bedrock-converse.test.ts @@ -43,6 +43,26 @@ const eventFrame = (type: string, payload: object) => body: utf8Encoder.encode(JSON.stringify(payload)), }) +const exceptionFrame = (type: string, payload: object) => + codec.encode({ + headers: { + ":message-type": { type: "string", value: "exception" }, + ":exception-type": { type: "string", value: type }, + ":content-type": { type: "string", value: "application/json" }, + }, + body: utf8Encoder.encode(JSON.stringify(payload)), + }) + +const errorFrame = (code: string, message: string) => + codec.encode({ + headers: { + ":message-type": { type: "string", value: "error" }, + ":error-code": { type: "string", value: code }, + ":error-message": { type: "string", value: message }, + }, + body: new Uint8Array(), + }) + const concat = (frames: ReadonlyArray) => { const total = frames.reduce((sum, frame) => sum + frame.length, 0) const out = new Uint8Array(total) @@ -363,6 +383,19 @@ describe("Bedrock Converse route", () => { }), ) + it.effect("preserves usage across later metadata events without usage", () => + Effect.gen(function* () { + const body = eventStreamBody( + ["messageStop", { stopReason: "end_turn" }], + ["metadata", { usage: { inputTokens: 5, outputTokens: 2, totalTokens: 7 } }], + ["metadata", { metrics: { latencyMs: 100 } }], + ) + const response = yield* LLMClient.generate(baseRequest).pipe(Effect.provide(fixedBytes(body))) + + expect(response.usage).toMatchObject({ inputTokens: 5, outputTokens: 2, totalTokens: 7 }) + }), + ) + it.effect("assembles streamed tool call input", () => Effect.gen(function* () { const body = eventStreamBody( @@ -478,6 +511,95 @@ describe("Bedrock Converse route", () => { }), ) + it.effect("preserves reasoning signatures when contentBlockStop is missing", () => + Effect.gen(function* () { + const response = yield* LLMClient.generate(baseRequest).pipe( + Effect.provide( + fixedBytes( + eventStreamBody( + ["messageStart", { role: "assistant" }], + [ + "contentBlockDelta", + { contentBlockIndex: 0, delta: { reasoningContent: { text: "Let me think." } } }, + ], + [ + "contentBlockDelta", + { contentBlockIndex: 0, delta: { reasoningContent: { signature: "sig_1" } } }, + ], + ["messageStop", { stopReason: "end_turn" }], + ), + ), + ), + ) + + expect(response.events.find((event) => event.type === "reasoning-delta" && event.text === "")).toEqual({ + type: "reasoning-delta", + id: "reasoning-0", + text: "", + providerMetadata: { bedrock: { signature: "sig_1" } }, + }) + expect(response.message.content).toEqual([ + { + type: "reasoning", + text: "Let me think.", + providerMetadata: { bedrock: { signature: "sig_1" } }, + }, + ]) + + const prepared = yield* LLMClient.prepare( + LLM.request({ model, messages: [response.message], cache: "none" }), + ) + expect(prepared.body.messages).toEqual([ + { + role: "assistant", + content: [{ reasoningContent: { reasoningText: { text: "Let me think.", signature: "sig_1" } } }], + }, + ]) + }), + ) + + it.effect("preserves signature-only reasoning blocks", () => + Effect.gen(function* () { + const body = eventStreamBody( + ["messageStart", { role: "assistant" }], + [ + "contentBlockDelta", + { contentBlockIndex: 0, delta: { reasoningContent: { signature: "sig_1" } } }, + ], + ["contentBlockStop", { contentBlockIndex: 0 }], + ["messageStop", { stopReason: "end_turn" }], + ) + const response = yield* LLMClient.generate(baseRequest).pipe(Effect.provide(fixedBytes(body))) + + expect(response.message.content).toEqual([ + { type: "reasoning", text: "", providerMetadata: { bedrock: { signature: "sig_1" } } }, + ]) + }), + ) + + it.effect("accepts Vercel-compatible redacted reasoning data deltas", () => + Effect.gen(function* () { + const redactedData = "cmVkYWN0ZWQtdGhpbmtpbmc=" + const body = eventStreamBody( + ["messageStart", { role: "assistant" }], + ["contentBlockDelta", { contentBlockIndex: 0, delta: { reasoningContent: { data: redactedData } } }], + ["contentBlockStop", { contentBlockIndex: 0 }], + ["messageStop", { stopReason: "end_turn" }], + ) + const response = yield* LLMClient.generate(baseRequest).pipe(Effect.provide(fixedBytes(body))) + + expect(response.events.find((event) => event.type === "reasoning-delta" && event.text === "")).toEqual({ + type: "reasoning-delta", + id: "reasoning-0", + text: "", + providerMetadata: { bedrock: { redactedData } }, + }) + expect(response.message.content).toEqual([ + { type: "reasoning", text: "", providerMetadata: { bedrock: { redactedData } } }, + ]) + }), + ) + it.effect("round-trips streamed redacted reasoning with tool use into a continuation request", () => Effect.gen(function* () { // Bedrock represents redactedContent blobs as base64 strings on its JSON @@ -511,6 +633,12 @@ describe("Bedrock Converse route", () => { ), ), ) + expect(response.events.find((event) => event.type === "reasoning-delta" && event.text === "")).toEqual({ + type: "reasoning-delta", + id: "reasoning-0", + text: "", + providerMetadata: { bedrock: { redactedData } }, + }) const prepared = yield* LLMClient.prepare( LLM.request({ model, @@ -543,10 +671,10 @@ describe("Bedrock Converse route", () => { it.effect("classifies throttlingException as a rate limit", () => Effect.gen(function* () { - const body = eventStreamBody( - ["messageStart", { role: "assistant" }], - ["throttlingException", { message: "Slow down" }], - ) + const body = concat([ + eventFrame("messageStart", { role: "assistant" }), + exceptionFrame("throttlingException", { message: "Slow down" }), + ]) const error = yield* LLMClient.generate(baseRequest).pipe(Effect.provide(fixedBytes(body)), Effect.flip) expect(error.reason).toMatchObject({ _tag: "RateLimit", message: "Slow down" }) @@ -557,7 +685,7 @@ describe("Bedrock Converse route", () => { Effect.gen(function* () { const error = yield* LLMClient.generate(baseRequest).pipe( Effect.provide( - fixedBytes(eventStreamBody(["validationException", { message: "Input is too long for requested model" }])), + fixedBytes(exceptionFrame("validationException", { message: "Input is too long for requested model" })), ), Effect.flip, ) @@ -570,6 +698,38 @@ describe("Bedrock Converse route", () => { }), ) + it.effect("uses originalMessage from model stream exception frames", () => + Effect.gen(function* () { + const error = yield* LLMClient.generate(baseRequest).pipe( + Effect.provide( + fixedBytes( + exceptionFrame("modelStreamErrorException", { + originalMessage: "Upstream model failed", + originalStatusCode: 500, + }), + ), + ), + Effect.flip, + ) + + expect(error.reason).toMatchObject({ _tag: "ProviderInternal", message: "Upstream model failed" }) + }), + ) + + it.effect("fails unmodeled AWS event-stream errors", () => + Effect.gen(function* () { + const error = yield* LLMClient.generate(baseRequest).pipe( + Effect.provide(fixedBytes(errorFrame("BadStream", "Stream failed"))), + Effect.flip, + ) + + expect(error.reason).toMatchObject({ + _tag: "InvalidProviderOutput", + message: "BadStream: Stream failed", + }) + }), + ) + it.effect("rejects requests with no auth path", () => Effect.gen(function* () { const unsignedModel = AmazonBedrock.configure({