diff --git a/packages/ai/src/protocols/bedrock-converse.ts b/packages/ai/src/protocols/bedrock-converse.ts index a839fbd1bd0..6e132baa1ff 100644 --- a/packages/ai/src/protocols/bedrock-converse.ts +++ b/packages/ai/src/protocols/bedrock-converse.ts @@ -1,4 +1,4 @@ -import { Effect, Schema } from "effect" +import { Effect, Encoding, Schema } from "effect" import { Route } from "../route/client.js" import { Endpoint } from "../route/endpoint.js" import { Protocol } from "../route/protocol.js" @@ -262,17 +262,13 @@ const providerMetadata = (key: string, metadata: Record): Provi const reasoningSignature = (part: ReasoningPart, providerMetadataKey: string) => { const metadata = part.providerMetadata?.[providerMetadataKey] - return ( - part.encrypted ?? - (ProviderShared.isRecord(metadata) && typeof metadata.signature === "string" ? metadata.signature : undefined) - ) + if (part.encrypted !== undefined) return part.encrypted + if (ProviderShared.isRecord(metadata) && typeof metadata.signature === "string") return metadata.signature } const reasoningRedactedData = (part: ReasoningPart, providerMetadataKey: string) => { const metadata = part.providerMetadata?.[providerMetadataKey] - return ProviderShared.isRecord(metadata) && typeof metadata.redactedData === "string" - ? metadata.redactedData - : undefined + if (ProviderShared.isRecord(metadata) && typeof metadata.redactedData === "string") return metadata.redactedData } const lowerToolCall = (part: ToolCallPart): BedrockToolUseBlock => ({ @@ -422,15 +418,15 @@ const fromRequest = Effect.fn("BedrockConverse.fromRequest")(function* (request: // Bedrock-Claude shares Anthropic's 4-breakpoint cap. Spend the budget in // tools → system → messages order to favour the highest-impact prefixes. const breakpoints = BedrockCache.breakpoints() - const toolConfig = - request.tools.length > 0 - ? { - tools: lowerTools(request.model.compatibility?.toolSchema, breakpoints, request.tools), - // Converse has no native "none". Keep definitions stable for prompt - // caching and omit only the unsupported choice. - toolChoice, - } - : undefined + const toolConfig = (() => { + if (request.tools.length === 0) return undefined + return { + tools: lowerTools(request.model.compatibility?.toolSchema, breakpoints, request.tools), + // Converse has no native "none". Keep definitions stable for prompt + // caching and omit only the unsupported choice. + toolChoice, + } + })() const system = request.system.length === 0 ? undefined : lowerSystem(breakpoints, request.system) const messages = yield* lowerMessages(request, breakpoints) if (breakpoints.dropped > 0) { @@ -438,22 +434,26 @@ const fromRequest = Effect.fn("BedrockConverse.fromRequest")(function* (request: `Bedrock Converse: dropped ${breakpoints.dropped} cache breakpoint(s); the API allows at most ${BedrockCache.BEDROCK_BREAKPOINT_CAP} per request.`, ) } - return { - modelId: request.model.id, - messages, - system, - inferenceConfig: + const inferenceConfig = (() => { + if ( generation?.maxTokens === undefined && generation?.temperature === undefined && generation?.topP === undefined && (generation?.stop === undefined || generation.stop.length === 0) - ? undefined - : { - maxTokens: generation?.maxTokens, - temperature: generation?.temperature, - topP: generation?.topP, - stopSequences: generation?.stop, - }, + ) + return undefined + return { + maxTokens: generation?.maxTokens, + temperature: generation?.temperature, + topP: generation?.topP, + stopSequences: generation?.stop, + } + })() + return { + modelId: request.model.id, + messages, + system, + inferenceConfig, toolConfig, // Converse's base inferenceConfig has no topK; Anthropic/Nova accept it // as a model-specific field, so it goes through additionalModelRequestFields. @@ -503,6 +503,16 @@ interface ParserState { readonly hasToolCalls: boolean readonly lifecycle: Lifecycle.State readonly reasoningSignatures: Readonly> + readonly reasoningRedactedContent: Readonly>> +} + +const encodeRedactedContent = (chunks: ReadonlyArray) => { + const bytes = new Uint8Array(chunks.reduce((total, chunk) => total + chunk.length, 0)) + chunks.reduce((offset, chunk) => { + bytes.set(chunk, offset) + return offset + chunk.length + }, 0) + return Encoding.encodeBase64(bytes) } const step = (state: ParserState, event: BedrockEvent) => @@ -550,23 +560,46 @@ const step = (state: ParserState, event: BedrockEvent) => const index = event.contentBlockDelta.contentBlockIndex const reasoning = event.contentBlockDelta.delta.reasoningContent const events: LLMEvent[] = [] - const redactedData = reasoning.redactedContent ?? reasoning.data - const metadata = reasoning.signature - ? providerMetadata(state.providerMetadataKey, { signature: reasoning.signature }) - : redactedData !== undefined - ? providerMetadata(state.providerMetadataKey, { redactedData }) - : undefined - const lifecycle = - reasoning.text !== undefined || metadata !== undefined - ? Lifecycle.reasoningDelta(state.lifecycle, events, `reasoning-${index}`, reasoning.text ?? "", metadata) - : state.lifecycle + const redactedChunks = yield* (() => { + if (reasoning.redactedContent === undefined) return Effect.succeed(undefined) + return Effect.fromResult(Encoding.decodeBase64(reasoning.redactedContent)).pipe( + Effect.map((chunk) => [...(state.reasoningRedactedContent[index] ?? []), chunk]), + Effect.mapError((cause) => + ProviderShared.eventError( + ADAPTER, + "Bedrock Converse reasoningContent.redactedContent contains invalid base64 data", + undefined, + cause, + ), + ), + ) + })() + const redactedData = redactedChunks === undefined ? reasoning.data : encodeRedactedContent(redactedChunks) + const metadata = (() => { + if (reasoning.signature) return providerMetadata(state.providerMetadataKey, { signature: reasoning.signature }) + if (redactedData !== undefined) return providerMetadata(state.providerMetadataKey, { redactedData }) + })() + const lifecycle = (() => { + if (reasoning.text === undefined && metadata === undefined) return state.lifecycle + return Lifecycle.reasoningDelta(state.lifecycle, events, `reasoning-${index}`, reasoning.text ?? "", metadata) + })() + const reasoningRedactedContent = (() => { + if (redactedChunks !== undefined) return { ...state.reasoningRedactedContent, [index]: redactedChunks } + if (reasoning.data === undefined) return state.reasoningRedactedContent + return Object.fromEntries( + Object.entries(state.reasoningRedactedContent).filter(([key]) => key !== String(index)), + ) + })() + const reasoningSignatures = (() => { + if (!reasoning.signature) return state.reasoningSignatures + return { ...state.reasoningSignatures, [index]: reasoning.signature } + })() return [ { ...state, lifecycle, - reasoningSignatures: reasoning.signature - ? { ...state.reasoningSignatures, [index]: reasoning.signature } - : state.reasoningSignatures, + reasoningSignatures, + reasoningRedactedContent, }, events, ] as const @@ -594,16 +627,24 @@ const step = (state: ParserState, event: BedrockEvent) => const result = yield* ToolStream.finish(ADAPTER, state.tools, index) const events: LLMEvent[] = [] const resultEvents = result.events ?? [] - const lifecycle = resultEvents.length - ? Lifecycle.stepStart(state.lifecycle, events) - : Lifecycle.reasoningEnd( - Lifecycle.textEnd(state.lifecycle, events, `text-${index}`), - events, - `reasoning-${index}`, - state.reasoningSignatures[index] - ? providerMetadata(state.providerMetadataKey, { signature: state.reasoningSignatures[index] }) - : undefined, - ) + const lifecycle = (() => { + if (resultEvents.length) return Lifecycle.stepStart(state.lifecycle, events) + const metadata = (() => { + const signature = state.reasoningSignatures[index] + if (signature) return providerMetadata(state.providerMetadataKey, { signature }) + const redactedContent = state.reasoningRedactedContent[index] + if (redactedContent) + return providerMetadata(state.providerMetadataKey, { + redactedData: encodeRedactedContent(redactedContent), + }) + })() + return Lifecycle.reasoningEnd( + Lifecycle.textEnd(state.lifecycle, events, `text-${index}`), + events, + `reasoning-${index}`, + metadata, + ) + })() events.push(...resultEvents) return [ { @@ -617,6 +658,9 @@ const step = (state: ParserState, event: BedrockEvent) => reasoningSignatures: Object.fromEntries( Object.entries(state.reasoningSignatures).filter(([key]) => key !== String(index)), ), + reasoningRedactedContent: Object.fromEntries( + Object.entries(state.reasoningRedactedContent).filter(([key]) => key !== String(index)), + ), }, events, ] as const @@ -678,23 +722,22 @@ const step = (state: ParserState, event: BedrockEvent) => const framing = BedrockEventStream.framing(ADAPTER) -const onHalt = (state: ParserState): ReadonlyArray => - state.pendingFinish - ? (() => { - const events: LLMEvent[] = [] - Lifecycle.finish(state.lifecycle, events, { - reason: { - ...state.pendingFinish.reason, - normalized: - state.pendingFinish.reason.normalized === "stop" && state.hasToolCalls - ? "tool-calls" - : state.pendingFinish.reason.normalized, - }, - usage: state.pendingFinish.usage, - }) - return events - })() - : [] +const onHalt = (state: ParserState): ReadonlyArray => { + if (!state.pendingFinish) return [] + const normalized = (() => { + if (state.pendingFinish.reason.normalized === "stop" && state.hasToolCalls) return "tool-calls" + return state.pendingFinish.reason.normalized + })() + const events: LLMEvent[] = [] + Lifecycle.finish(state.lifecycle, events, { + reason: { + ...state.pendingFinish.reason, + normalized, + }, + usage: state.pendingFinish.usage, + }) + return events +} // ============================================================================= // Protocol And Bedrock Route @@ -719,6 +762,7 @@ export const protocol = Protocol.make({ hasToolCalls: false, lifecycle: Lifecycle.initial(), reasoningSignatures: {}, + reasoningRedactedContent: {}, }), step, onHalt: (state) => Effect.succeed(onHalt(state)), diff --git a/packages/ai/test/provider/bedrock-converse.test.ts b/packages/ai/test/provider/bedrock-converse.test.ts index 828a650a986..95ce1140a05 100644 --- a/packages/ai/test/provider/bedrock-converse.test.ts +++ b/packages/ai/test/provider/bedrock-converse.test.ts @@ -923,7 +923,7 @@ describe("Bedrock Converse route", () => { Effect.gen(function* () { // Bedrock represents redactedContent blobs as base64 strings on its JSON // wire. The provider owns the payload and requires byte-exact replay. - const redactedData = "cmVkYWN0ZWQtdGhpbmtpbmc=" + const redactedData = "AQID" const response = yield* LLMClient.generate( LLMRequest.update(baseRequest, { tools: [ToolDefinition.make({ name: "lookup", description: "Lookup data", inputSchema: { type: "object" } })], @@ -933,10 +933,8 @@ describe("Bedrock Converse route", () => { fixedBytes( eventStreamBody( ["messageStart", { role: "assistant" }], - [ - "contentBlockDelta", - { contentBlockIndex: 0, delta: { reasoningContent: { redactedContent: redactedData } } }, - ], + ["contentBlockDelta", { contentBlockIndex: 0, delta: { reasoningContent: { redactedContent: "AQ==" } } }], + ["contentBlockDelta", { contentBlockIndex: 0, delta: { reasoningContent: { redactedContent: "AgM=" } } }], ["contentBlockStop", { contentBlockIndex: 0 }], [ "contentBlockStart", @@ -952,12 +950,17 @@ describe("Bedrock Converse route", () => { ), ), ) - expect(response.events.find((event) => event.type === "reasoning-delta" && event.text === "")).toEqual({ + expect(response.events.filter((event) => event.type === "reasoning-delta" && event.text === "").at(-1)).toEqual({ type: "reasoning-delta", id: "reasoning-0", text: "", providerMetadata: { bedrock: { redactedData } }, }) + expect(response.events.find((event) => event.type === "reasoning-end")).toEqual({ + type: "reasoning-end", + id: "reasoning-0", + providerMetadata: { bedrock: { redactedData } }, + }) const prepared = yield* compileRequest( LLM.request({ model, @@ -988,6 +991,73 @@ describe("Bedrock Converse route", () => { }), ) + it.effect("keeps redacted reasoning accumulation separate by content block index", () => + Effect.gen(function* () { + const response = yield* LLMClient.generate(baseRequest).pipe( + Effect.provide( + fixedBytes( + eventStreamBody( + ["messageStart", { role: "assistant" }], + ["contentBlockDelta", { contentBlockIndex: 2, delta: { reasoningContent: { redactedContent: "AQ==" } } }], + ["contentBlockDelta", { contentBlockIndex: 2, delta: { reasoningContent: { redactedContent: "Ag==" } } }], + ["contentBlockStop", { contentBlockIndex: 2 }], + ["contentBlockDelta", { contentBlockIndex: 7, delta: { reasoningContent: { redactedContent: "Aw==" } } }], + ["contentBlockDelta", { contentBlockIndex: 7, delta: { reasoningContent: { redactedContent: "BA==" } } }], + ["contentBlockStop", { contentBlockIndex: 7 }], + ["messageStop", { stopReason: "end_turn" }], + ), + ), + ), + ) + + expect(response.message.content).toEqual([ + { type: "reasoning", text: "", providerMetadata: { bedrock: { redactedData: "AQI=" } } }, + { type: "reasoning", text: "", providerMetadata: { bedrock: { redactedData: "AwQ=" } } }, + ]) + }), + ) + + it.effect("preserves split redacted reasoning 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: { redactedContent: "AQ==" } } }], + ["contentBlockDelta", { contentBlockIndex: 0, delta: { reasoningContent: { redactedContent: "AgM=" } } }], + ["messageStop", { stopReason: "end_turn" }], + ), + ), + ), + ) + + expect(response.message.content).toEqual([ + { type: "reasoning", text: "", providerMetadata: { bedrock: { redactedData: "AQID" } } }, + ]) + }), + ) + + it.effect("rejects invalid redacted reasoning base64 with the triggering event", () => + Effect.gen(function* () { + const payload = { contentBlockIndex: 0, delta: { reasoningContent: { redactedContent: "%%==" } } } + const error = yield* LLMClient.generate(baseRequest).pipe( + Effect.provide(fixedBytes(eventStreamBody(["contentBlockDelta", payload]))), + Effect.flip, + ) + + expect(error).toMatchObject({ + reason: { _tag: "InvalidProviderOutput" }, + message: "Bedrock Converse reasoningContent.redactedContent contains invalid base64 data", + }) + expect(JSON.parse(error.reason.body ?? "")).toMatchObject({ + headers: { ":event-type": { value: "contentBlockDelta" } }, + body: JSON.stringify(payload), + }) + expect(error.reason.cause).toBeInstanceOf(Error) + }), + ) + it.effect("ignores unknown normal stream events", () => Effect.gen(function* () { const body = concat([