From 769d6123d55b9d96aee501f874599d2157474836 Mon Sep 17 00:00:00 2001 From: Kit Langton Date: Sun, 26 Apr 2026 19:41:59 -0400 Subject: [PATCH] feat(llm): add Bedrock Converse adapter MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Implements the AWS Bedrock Converse streaming protocol as the 5th first-class adapter in @opencode-ai/llm. Single `bedrock-converse` adapter covers all underlying models (Anthropic, Llama, Mistral, Cohere, Nova, Titan) since Converse is uniform. Wire format: messages with text / reasoning / toolUse / toolResult content blocks, system blocks, inferenceConfig, toolConfig with toolSpec + toolChoice. Image / document / cache-point content types are still TODO. Streaming: AWS event stream binary framing via @smithy/eventstream-codec. Each frame is decoded then dispatched by `:event-type` header into the chunk schema. Bedrock splits the finish across `messageStop` (reason) and `metadata` (usage) — the parser stashes the reason and emits a single consolidated `request-finish` event when metadata arrives, with an `onHalt` fallback for truncated streams. Auth: two paths. Bearer API key (newer) when the consumer sets `model.headers.authorization = 'Bearer '`. SigV4 signing via aws4fetch otherwise — credentials live on `model.native.aws_credentials` and are signed at `toHttp` time so STS-vended tokens are picked up when the consumer rebuilds the model. The adapter rejects requests with neither auth path with a clear InvalidRequestError. Routing: `@ai-sdk/amazon-bedrock` lowers to `bedrock-converse` via the new `AmazonBedrock` provider routing module; the OpenCode `llm-bridge.ts` registers it. Cassette format: response bodies under `application/vnd.amazon.eventstream` and `application/octet-stream` content types are now stored as base64 with `bodyEncoding: 'base64'` on the response snapshot — text round-tripping mangled the CRC32 fields in event-stream frames. Existing cassettes (SSE/JSON) omit the field and decode as text unchanged. Tests: 11 deterministic fixtures (prepare / lower messages / lower tool config / decode text+usage / decode tool calls / decode reasoning / decode throttling exception / auth path validation / SigV4 plumbing) + 2 recorded cassettes against live Bedrock (`us.amazon.nova-micro-v1:0` in us-east-1) for streaming text and streaming tool calls. AGENTS.md: documents the Bedrock auth model, binary cassette format, and updates the protocol coverage / cassette backlog. Deps: @smithy/eventstream-codec, @smithy/util-utf8, aws4fetch (~40KB combined; matches AI SDK's approach). --- bun.lock | 9 + packages/llm/AGENTS.md | 6 +- packages/llm/package.json | 3 + packages/llm/src/index.ts | 2 + packages/llm/src/provider/amazon-bedrock.ts | 5 + packages/llm/src/provider/bedrock-converse.ts | 677 ++++++++++++++++++ .../bedrock-converse/streams-a-tool-call.json | 23 + .../bedrock-converse/streams-text.json | 23 + .../test/provider/bedrock-converse.test.ts | 375 ++++++++++ packages/llm/test/record-replay.ts | 48 +- packages/opencode/src/provider/llm-bridge.ts | 2 + 11 files changed, 1163 insertions(+), 10 deletions(-) create mode 100644 packages/llm/src/provider/amazon-bedrock.ts create mode 100644 packages/llm/src/provider/bedrock-converse.ts create mode 100644 packages/llm/test/fixtures/recordings/bedrock-converse/streams-a-tool-call.json create mode 100644 packages/llm/test/fixtures/recordings/bedrock-converse/streams-text.json create mode 100644 packages/llm/test/provider/bedrock-converse.test.ts diff --git a/bun.lock b/bun.lock index 037d4c86ed..9b1c8cfa0d 100644 --- a/bun.lock +++ b/bun.lock @@ -356,6 +356,9 @@ "name": "@opencode-ai/llm", "version": "1.14.25", "dependencies": { + "@smithy/eventstream-codec": "4.2.14", + "@smithy/util-utf8": "4.2.2", + "aws4fetch": "1.0.20", "effect": "catalog:", }, "devDependencies": { @@ -5652,6 +5655,10 @@ "@opencode-ai/desktop-electron/typescript": ["typescript@5.6.3", "", { "bin": { "tsc": "bin/tsc", "tsserver": "bin/tsserver" } }, "sha512-hjcS1mhfuyi4WW8IWtjP7brDrG2cuDZukyrYrSauoXGNgx0S7zceP07adYkJycEr56BOUTNPzbInooiN3fn1qw=="], + "@opencode-ai/llm/@smithy/eventstream-codec": ["@smithy/eventstream-codec@4.2.14", "", { "dependencies": { "@aws-crypto/crc32": "5.2.0", "@smithy/types": "^4.14.1", "@smithy/util-hex-encoding": "^4.2.2", "tslib": "^2.6.2" } }, "sha512-erZq0nOIpzfeZdCyzZjdJb4nVSKLUmSkaQUVkRGQTXs30gyUGeKnrYEg+Xe1W5gE3aReS7IgsvANwVPxSzY6Pw=="], + + "@opencode-ai/llm/@smithy/util-utf8": ["@smithy/util-utf8@4.2.2", "", { "dependencies": { "@smithy/util-buffer-from": "^4.2.2", "tslib": "^2.6.2" } }, "sha512-75MeYpjdWRe8M5E3AW0O4Cx3UadweS+cwdXjwYGBW5h/gxxnbeZ877sLPX/ZJA9GVTlL/qG0dXP29JWFCD1Ayw=="], + "@opencode-ai/ui/@solid-primitives/resize-observer": ["@solid-primitives/resize-observer@2.1.3", "", { "dependencies": { "@solid-primitives/event-listener": "^2.4.3", "@solid-primitives/rootless": "^1.5.2", "@solid-primitives/static-store": "^0.1.2", "@solid-primitives/utils": "^6.3.2" }, "peerDependencies": { "solid-js": "^1.6.12" } }, "sha512-zBLje5E06TgOg93S7rGPldmhDnouNGhvfZVKOp+oG2XU8snA+GoCSSCz1M+jpNAg5Ek2EakU5UVQqL152WmdXQ=="], "@opencode-ai/web/@shikijs/transformers": ["@shikijs/transformers@3.20.0", "", { "dependencies": { "@shikijs/core": "3.20.0", "@shikijs/types": "3.20.0" } }, "sha512-PrHHMRr3Q5W1qB/42kJW6laqFyWdhrPF2hNR9qjOm1xcSiAO3hAHo7HaVyHE6pMyevmy3i51O8kuGGXC78uK3g=="], @@ -6636,6 +6643,8 @@ "@opencode-ai/desktop/@actions/artifact/@actions/http-client": ["@actions/http-client@2.2.3", "", { "dependencies": { "tunnel": "^0.0.6", "undici": "^5.25.4" } }, "sha512-mx8hyJi/hjFvbPokCg4uRd4ZX78t+YyRPtnKWwIl+RzNaVuFpQHfmlGVfsKEJN8LwTCvL+DfVgAM04XaHkm6bA=="], + "@opencode-ai/llm/@smithy/eventstream-codec/@smithy/types": ["@smithy/types@4.14.1", "", { "dependencies": { "tslib": "^2.6.2" } }, "sha512-59b5HtSVrVR/eYNei3BUj3DCPKD/G7EtDDe7OEJE7i7FtQFugYo6MxbotS8mVJkLNVf8gYaAlEBwwtJ9HzhWSg=="], + "@opencode-ai/web/@shikijs/transformers/@shikijs/core": ["@shikijs/core@3.20.0", "", { "dependencies": { "@shikijs/types": "3.20.0", "@shikijs/vscode-textmate": "^10.0.2", "@types/hast": "^3.0.4", "hast-util-to-html": "^9.0.5" } }, "sha512-f2ED7HYV4JEk827mtMDwe/yQ25pRiXZmtHjWF8uzZKuKiEsJR7Ce1nuQ+HhV9FzDcbIo4ObBCD9GPTzNuy9S1g=="], "@opencode-ai/web/@shikijs/transformers/@shikijs/types": ["@shikijs/types@3.20.0", "", { "dependencies": { "@shikijs/vscode-textmate": "^10.0.2", "@types/hast": "^3.0.4" } }, "sha512-lhYAATn10nkZcBQ0BlzSbJA3wcmL5MXUUF8d2Zzon6saZDlToKaiRX60n2+ZaHJCmXEcZRWNzn+k9vplr8Jhsw=="], diff --git a/packages/llm/AGENTS.md b/packages/llm/AGENTS.md index 67c87761cd..faf8f896cf 100644 --- a/packages/llm/AGENTS.md +++ b/packages/llm/AGENTS.md @@ -157,6 +157,8 @@ recorded.effect("streams text", () => Effect.gen(function* () { Replay is the default. `RECORD=true` records fresh cassettes and requires the listed env vars. Cassettes are written as pretty-printed JSON so multi-interaction diffs stay reviewable. +**Binary response bodies.** Most providers stream text (SSE, JSON). AWS Bedrock streams binary AWS event-stream frames whose CRC32 fields would be mangled by a UTF-8 round-trip — those bodies are stored as base64 with `bodyEncoding: "base64"` on the response snapshot. Detection is by `Content-Type` (currently `application/vnd.amazon.eventstream` and `application/octet-stream`); cassettes for SSE/JSON adapters omit the field and decode as text. To support a new binary content type, extend `BINARY_CONTENT_TYPES` in `test/record-replay.ts`. + **Matching strategies.** Replay defaults to `defaultMatcher`, which finds an interaction by structurally comparing method, URL, allow-listed headers, and the canonical JSON body. This is the right choice for tool loops because each round's request differs (the message history grows). For scenarios where successive requests are byte-identical and expect different responses (retries, polling), pass `match: sequentialMatcher` in `RecordReplayOptions` — replay then walks the cassette in record order via an internal cursor. `scriptedResponses` (in `test/lib/http.ts`) is the deterministic counterpart for tests that don't need a live provider; it scripts response bodies in order without reading from disk. Do not blanket re-record an entire test file when adding one cassette. `RECORD=true` rewrites every recorded case that runs, and provider streams contain volatile IDs, timestamps, fingerprints, and obfuscation fields. Prefer deleting the one cassette you intend to refresh, or run a focused test pattern that only registers the scenario you want to record. Keep stable existing cassettes unchanged unless their request shape or expected behavior changed. @@ -182,7 +184,7 @@ Do not blanket re-record an entire test file when adding one cassette. `RECORD=t - [ ] Keep OpenAI Responses as a separate first-class protocol for providers that actually implement `/responses`; do not treat generic OpenAI-compatible providers as Responses-capable by default. - [x] Cover OpenAI-compatible provider families that can share the generic adapter first: DeepSeek, TogetherAI, Cerebras, Baseten, Fireworks, DeepInfra, and similar providers. - [ ] Decide which providers need thin dedicated wrappers over OpenAI-compatible Chat because they have custom parsing/options: Mistral, Groq, xAI, Perplexity, and Cohere. -- [ ] Add Bedrock Converse support or a clear compatibility layer before moving Amazon Bedrock traffic onto `packages/llm`. +- [x] Add Bedrock Converse support: wire format (messages / system / inferenceConfig / toolConfig), AWS event stream binary framing via `@smithy/eventstream-codec`, SigV4 signing via `aws4fetch` (or Bearer API key path), text/reasoning/tool/usage/finish decoding, deterministic + recorded integration tests. Cache hints, image/document content, and additional model-specific fields are still TODO. - [ ] Decide Vertex shape after Bedrock/OpenAI-compatible are stable: Vertex Gemini as Gemini target/http patch vs adapter, and Vertex Anthropic as Anthropic target/http patch vs adapter. - [ ] Add Gateway/OpenRouter-style routing support only after the generic OpenAI-compatible adapter and provider option patch model are stable. @@ -226,6 +228,6 @@ Do not blanket re-record an entire test file when adding one cassette. `RECORD=t - [ ] DeepInfra OpenAI-compatible Chat basic streaming text and tool-call flow. - [ ] Provider-error cassettes for stable, non-secret error bodies where the provider returns deterministic 4xx/5xx payloads. - [ ] Mistral, Groq, xAI, Perplexity, and Cohere basic/tool cassettes after deciding whether each stays generic OpenAI-compatible or gets a thin wrapper. -- [ ] Bedrock Converse basic text, tool use/result, and cache-hint cassettes after Bedrock support lands. +- [x] Bedrock Converse basic text and tool-call cassettes (recorded against `us.amazon.nova-micro-v1:0` in us-east-1). Cache-hint cassettes still TODO. - [ ] Vertex Gemini and Vertex Anthropic basic/tool cassettes after the Vertex adapter/patch shape is decided. - [ ] Gateway/OpenRouter routing-header cassettes after routing support lands. diff --git a/packages/llm/package.json b/packages/llm/package.json index cc916ea7c4..2140ffb28a 100644 --- a/packages/llm/package.json +++ b/packages/llm/package.json @@ -20,6 +20,9 @@ "@typescript/native-preview": "catalog:" }, "dependencies": { + "@smithy/eventstream-codec": "4.2.14", + "@smithy/util-utf8": "4.2.2", + "aws4fetch": "1.0.20", "effect": "catalog:" } } diff --git a/packages/llm/src/index.ts b/packages/llm/src/index.ts index ea69b03702..c3e035ddce 100644 --- a/packages/llm/src/index.ts +++ b/packages/llm/src/index.ts @@ -9,8 +9,10 @@ export * as LLM from "./llm" export * as ProviderPatch from "./provider/patch" export * as Schema from "./schema" export { AnthropicMessages } from "./provider/anthropic-messages" +export { AmazonBedrock } from "./provider/amazon-bedrock" export { Anthropic } from "./provider/anthropic" export { Azure } from "./provider/azure" +export { BedrockConverse } from "./provider/bedrock-converse" export { Gemini } from "./provider/gemini" export { Google } from "./provider/google" export { GitHubCopilot } from "./provider/github-copilot" diff --git a/packages/llm/src/provider/amazon-bedrock.ts b/packages/llm/src/provider/amazon-bedrock.ts new file mode 100644 index 0000000000..ae0ac3fcfb --- /dev/null +++ b/packages/llm/src/provider/amazon-bedrock.ts @@ -0,0 +1,5 @@ +import { ProviderRoute } from "../provider-route" + +export const provider = ProviderRoute.fixed("amazon-bedrock", "bedrock-converse") + +export * as AmazonBedrock from "./amazon-bedrock" diff --git a/packages/llm/src/provider/bedrock-converse.ts b/packages/llm/src/provider/bedrock-converse.ts new file mode 100644 index 0000000000..ccf2931a75 --- /dev/null +++ b/packages/llm/src/provider/bedrock-converse.ts @@ -0,0 +1,677 @@ +import { EventStreamCodec } from "@smithy/eventstream-codec" +import { fromUtf8, toUtf8 } from "@smithy/util-utf8" +import { AwsV4Signer } from "aws4fetch" +import { Effect, Schema, Stream } from "effect" +import { HttpClientRequest, HttpClientResponse } from "effect/unstable/http" +import { Adapter } from "../adapter" +import { capabilities, model as llmModel, type ModelInput } from "../llm" +import { + InvalidRequestError, + Usage, + type FinishReason, + type LLMEvent, + type LLMRequest, + type TextPart, + type ToolCallPart, + type ToolDefinition, + type ToolResultPart, +} from "../schema" +import { ProviderShared } from "./shared" + +const ADAPTER = "bedrock-converse" + +/** + * AWS credentials for SigV4 signing. Bedrock also supports Bearer API key auth + * — pass the key as `model.headers.authorization = "Bearer "` to take that + * path instead. STS-vended credentials should be refreshed by the consumer + * (rebuild the model) before they expire; the adapter does not refresh. + */ +export interface BedrockCredentials { + readonly region: string + readonly accessKeyId: string + readonly secretAccessKey: string + readonly sessionToken?: string +} + +export type BedrockConverseModelInput = Omit & { + /** + * Bearer API key (Bedrock's newer API key auth). Sets the `Authorization` + * header and bypasses SigV4 signing. Mutually exclusive with `credentials`. + */ + readonly apiKey?: string + /** + * AWS credentials for SigV4 signing. The adapter signs each request at + * `toHttp` time using `aws4fetch`. Mutually exclusive with `apiKey`. + */ + readonly credentials?: BedrockCredentials + readonly headers?: Record +} + +const BedrockTextBlock = Schema.Struct({ + text: Schema.String, +}) + +const BedrockToolUseBlock = Schema.Struct({ + toolUse: Schema.Struct({ + toolUseId: Schema.String, + name: Schema.String, + input: Schema.Unknown, + }), +}) +type BedrockToolUseBlock = Schema.Schema.Type + +const BedrockToolResultContentItem = Schema.Union([ + Schema.Struct({ text: Schema.String }), + Schema.Struct({ json: Schema.Unknown }), +]) + +const BedrockToolResultBlock = Schema.Struct({ + toolResult: Schema.Struct({ + toolUseId: Schema.String, + content: Schema.Array(BedrockToolResultContentItem), + status: Schema.optional(Schema.Literals(["success", "error"])), + }), +}) +type BedrockToolResultBlock = Schema.Schema.Type + +const BedrockReasoningBlock = Schema.Struct({ + reasoningContent: Schema.Struct({ + reasoningText: Schema.optional( + Schema.Struct({ + text: Schema.String, + signature: Schema.optional(Schema.String), + }), + ), + }), +}) + +const BedrockUserBlock = Schema.Union([BedrockTextBlock, BedrockToolResultBlock]) +const BedrockAssistantBlock = Schema.Union([BedrockTextBlock, BedrockReasoningBlock, BedrockToolUseBlock]) +type BedrockAssistantBlock = Schema.Schema.Type + +const BedrockMessage = Schema.Union([ + Schema.Struct({ role: Schema.Literal("user"), content: Schema.Array(BedrockUserBlock) }), + Schema.Struct({ role: Schema.Literal("assistant"), content: Schema.Array(BedrockAssistantBlock) }), +]) +type BedrockMessage = Schema.Schema.Type + +const BedrockSystem = Schema.Struct({ text: Schema.String }) + +const BedrockTool = Schema.Struct({ + toolSpec: Schema.Struct({ + name: Schema.String, + description: Schema.String, + inputSchema: Schema.Struct({ + json: Schema.Record(Schema.String, Schema.Unknown), + }), + }), +}) +type BedrockTool = Schema.Schema.Type + +const BedrockToolChoice = Schema.Union([ + Schema.Struct({ auto: Schema.Struct({}) }), + Schema.Struct({ any: Schema.Struct({}) }), + Schema.Struct({ tool: Schema.Struct({ name: Schema.String }) }), +]) + +const BedrockTargetFields = { + modelId: Schema.String, + messages: Schema.Array(BedrockMessage), + system: Schema.optional(Schema.Array(BedrockSystem)), + inferenceConfig: Schema.optional( + Schema.Struct({ + maxTokens: Schema.optional(Schema.Number), + temperature: Schema.optional(Schema.Number), + topP: Schema.optional(Schema.Number), + stopSequences: Schema.optional(Schema.Array(Schema.String)), + }), + ), + toolConfig: Schema.optional( + Schema.Struct({ + tools: Schema.Array(BedrockTool), + toolChoice: Schema.optional(BedrockToolChoice), + }), + ), + additionalModelRequestFields: Schema.optional(Schema.Record(Schema.String, Schema.Unknown)), +} +const BedrockConverseDraft = Schema.Struct(BedrockTargetFields) +type BedrockConverseDraft = Schema.Schema.Type +const BedrockConverseTarget = Schema.Struct(BedrockTargetFields) +export type BedrockConverseTarget = Schema.Schema.Type + +const BedrockUsageSchema = Schema.Struct({ + inputTokens: Schema.optional(Schema.Number), + outputTokens: Schema.optional(Schema.Number), + totalTokens: Schema.optional(Schema.Number), + cacheReadInputTokens: Schema.optional(Schema.Number), + cacheWriteInputTokens: Schema.optional(Schema.Number), +}) +type BedrockUsageSchema = Schema.Schema.Type + +// Streaming chunk 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 chunk schema can +// stay a plain discriminated record. +const BedrockChunk = Schema.Struct({ + messageStart: Schema.optional(Schema.Struct({ role: Schema.String })), + contentBlockStart: Schema.optional( + Schema.Struct({ + contentBlockIndex: Schema.Number, + start: Schema.optional( + Schema.Struct({ + toolUse: Schema.optional( + Schema.Struct({ toolUseId: Schema.String, name: Schema.String }), + ), + }), + ), + }), + ), + contentBlockDelta: Schema.optional( + Schema.Struct({ + contentBlockIndex: Schema.Number, + delta: Schema.optional( + Schema.Struct({ + text: Schema.optional(Schema.String), + toolUse: Schema.optional(Schema.Struct({ input: Schema.String })), + reasoningContent: Schema.optional( + Schema.Struct({ + text: Schema.optional(Schema.String), + signature: Schema.optional(Schema.String), + }), + ), + }), + ), + }), + ), + contentBlockStop: Schema.optional(Schema.Struct({ contentBlockIndex: Schema.Number })), + messageStop: Schema.optional( + Schema.Struct({ + stopReason: Schema.String, + additionalModelResponseFields: Schema.optional(Schema.Unknown), + }), + ), + metadata: Schema.optional( + Schema.Struct({ + usage: Schema.optional(BedrockUsageSchema), + 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 })), +}) +type BedrockChunk = Schema.Schema.Type + +const BedrockChunkJson = Schema.fromJsonString(BedrockChunk) +const BedrockTargetJson = Schema.fromJsonString(BedrockConverseTarget) +const decodeChunkSync = Schema.decodeUnknownSync(BedrockChunkJson) + +const decodeChunk = (data: string) => + Effect.try({ + try: () => decodeChunkSync(data), + catch: () => ProviderShared.chunkError(ADAPTER, "Invalid Bedrock Converse stream chunk", data), + }) + +const encodeTarget = Schema.encodeSync(BedrockTargetJson) +const decodeTarget = Schema.decodeUnknownEffect(BedrockConverseDraft.pipe(Schema.decodeTo(BedrockConverseTarget))) + +const invalid = (message: string) => new InvalidRequestError({ message }) + +const region = (request: LLMRequest) => { + const fromNative = request.model.native?.aws_region + if (typeof fromNative === "string" && fromNative !== "") return fromNative + if (typeof request.model.native?.region === "string") return request.model.native.region as string + return "us-east-1" +} + +const baseUrl = (request: LLMRequest) => { + const configured = request.model.baseURL + if (configured) return configured.replace(/\/+$/, "") + return `https://bedrock-runtime.${region(request)}.amazonaws.com` +} + +const text = (values: ReadonlyArray<{ readonly text: string }>) => values.map((part) => part.text).join("\n") + +const lowerTool = (tool: ToolDefinition): BedrockTool => ({ + toolSpec: { + name: tool.name, + description: tool.description, + inputSchema: { json: tool.inputSchema }, + }, +}) + +const lowerToolChoice = Effect.fn("BedrockConverse.lowerToolChoice")(function* ( + toolChoice: NonNullable, +) { + if (toolChoice.type === "none") return undefined + if (toolChoice.type === "required") return { any: {} } as const + if (toolChoice.type !== "tool") return { auto: {} } as const + if (!toolChoice.name) return yield* invalid("Bedrock Converse tool choice requires a tool name") + return { tool: { name: toolChoice.name } } as const +}) + +const lowerToolCall = (part: ToolCallPart): BedrockToolUseBlock => ({ + toolUse: { + toolUseId: part.id, + name: part.name, + input: part.input, + }, +}) + +const lowerToolResult = (part: ToolResultPart): BedrockToolResultBlock => { + const status = part.result.type === "error" ? ("error" as const) : ("success" as const) + const content = + part.result.type === "text" || part.result.type === "error" + ? [{ text: String(part.result.value) }] + : [{ json: part.result.value }] + return { toolResult: { toolUseId: part.id, content, status } } +} + +const lowerMessages = Effect.fn("BedrockConverse.lowerMessages")(function* (request: LLMRequest) { + const messages: BedrockMessage[] = [] + + for (const message of request.messages) { + if (message.role === "user") { + const content: Array> = [] + for (const part of message.content) { + if (part.type === "text") { + content.push({ text: part.text }) + continue + } + return yield* invalid("Bedrock Converse user messages only support text content for now") + } + messages.push({ role: "user", content }) + continue + } + + if (message.role === "assistant") { + const content: BedrockAssistantBlock[] = [] + for (const part of message.content) { + if (part.type === "text") { + content.push({ text: part.text }) + continue + } + if (part.type === "reasoning") { + content.push({ + reasoningContent: { + reasoningText: { text: part.text, signature: part.encrypted }, + }, + }) + continue + } + if (part.type === "tool-call") { + content.push(lowerToolCall(part)) + continue + } + return yield* invalid("Bedrock Converse assistant messages only support text, reasoning, and tool-call content for now") + } + messages.push({ role: "assistant", content }) + continue + } + + const content: BedrockToolResultBlock[] = [] + for (const part of message.content) { + if (part.type !== "tool-result") + return yield* invalid("Bedrock Converse tool messages only support tool-result content") + content.push(lowerToolResult(part)) + } + messages.push({ role: "user", content }) + } + + return messages +}) + +const prepare = Effect.fn("BedrockConverse.prepare")(function* (request: LLMRequest) { + const toolChoice = request.toolChoice ? yield* lowerToolChoice(request.toolChoice) : undefined + const useTools = request.tools.length > 0 && request.toolChoice?.type !== "none" + return { + modelId: request.model.id, + messages: yield* lowerMessages(request), + system: request.system.length === 0 ? undefined : request.system.map((part) => ({ text: part.text })), + inferenceConfig: + request.generation.maxTokens === undefined && + request.generation.temperature === undefined && + request.generation.topP === undefined && + (request.generation.stop === undefined || request.generation.stop.length === 0) + ? undefined + : { + maxTokens: request.generation.maxTokens, + temperature: request.generation.temperature, + topP: request.generation.topP, + stopSequences: request.generation.stop, + }, + toolConfig: useTools + ? { tools: request.tools.map(lowerTool), toolChoice } + : undefined, + } +}) + +const credentialsFromInput = (request: LLMRequest): BedrockCredentials | undefined => { + const native = request.model.native + if (!native) return undefined + const creds = native.aws_credentials + if (!creds || typeof creds !== "object") return undefined + const obj = creds as Record + if (typeof obj.accessKeyId !== "string" || typeof obj.secretAccessKey !== "string") return undefined + return { + region: typeof obj.region === "string" ? obj.region : region(request), + accessKeyId: obj.accessKeyId, + secretAccessKey: obj.secretAccessKey, + sessionToken: typeof obj.sessionToken === "string" ? obj.sessionToken : undefined, + } +} + +const isBearerAuth = (headers: Record | undefined) => { + const auth = headers?.authorization ?? headers?.Authorization + return typeof auth === "string" && auth.toLowerCase().startsWith("bearer ") +} + +const signRequest = ( + url: string, + body: string, + headers: Record, + credentials: BedrockCredentials, +) => + Effect.tryPromise({ + try: async () => { + const signer = new AwsV4Signer({ + url, + method: "POST", + headers: Object.entries(headers), + body, + region: credentials.region, + accessKeyId: credentials.accessKeyId, + secretAccessKey: credentials.secretAccessKey, + sessionToken: credentials.sessionToken, + service: "bedrock", + }) + const signed = await signer.sign() + const out: Record = {} + signed.headers.forEach((value, key) => { + out[key] = value + }) + return out + }, + catch: (error) => + new InvalidRequestError({ + message: `Bedrock Converse SigV4 signing failed: ${error instanceof Error ? error.message : String(error)}`, + }), + }) + +const toHttp = Effect.fn("BedrockConverse.toHttp")(function* (target: BedrockConverseTarget, request: LLMRequest) { + const url = `${baseUrl(request)}/model/${encodeURIComponent(target.modelId)}/converse-stream` + const body = encodeTarget(target) + const baseHeaders: Record = { + ...request.model.headers, + "content-type": "application/json", + } + + if (isBearerAuth(request.model.headers)) { + return HttpClientRequest.post(url).pipe( + HttpClientRequest.setHeaders(baseHeaders), + HttpClientRequest.bodyText(body, "application/json"), + ) + } + + const credentials = credentialsFromInput(request) + if (!credentials) { + return yield* invalid( + "Bedrock Converse requires either a Bearer API key in headers or AWS credentials in model.native.aws_credentials", + ) + } + const signed = yield* signRequest(url, body, baseHeaders, credentials) + return HttpClientRequest.post(url).pipe( + HttpClientRequest.setHeaders({ ...baseHeaders, ...signed }), + HttpClientRequest.bodyText(body, "application/json"), + ) +}) + +const mapFinishReason = (reason: string | undefined): FinishReason => { + if (reason === "end_turn" || reason === "stop_sequence") return "stop" + if (reason === "max_tokens") return "length" + if (reason === "tool_use") return "tool-calls" + if (reason === "content_filtered" || reason === "guardrail_intervened") return "content-filter" + return "unknown" +} + +const mapUsage = (usage: BedrockUsageSchema | undefined): Usage | undefined => { + if (!usage) return undefined + return new Usage({ + inputTokens: usage.inputTokens, + outputTokens: usage.outputTokens, + totalTokens: + usage.totalTokens ?? + ((usage.inputTokens ?? 0) + (usage.outputTokens ?? 0) || undefined), + cacheReadInputTokens: usage.cacheReadInputTokens, + cacheWriteInputTokens: usage.cacheWriteInputTokens, + native: usage, + }) +} + +interface ToolAccumulator { + readonly id: string + readonly name: string + readonly input: string +} + +interface ParserState { + readonly tools: Record + // Bedrock splits the finish into `messageStop` (carries `stopReason`) and + // `metadata` (carries usage). We accumulate both before emitting a single + // `request-finish` event so consumers see one terminal event with both. + readonly finishReason: FinishReason | undefined +} + +const finishToolCall = (tool: ToolAccumulator | undefined) => + Effect.gen(function* () { + if (!tool) return [] as ReadonlyArray + const input = yield* ProviderShared.parseJson( + ADAPTER, + tool.input || "{}", + `Invalid JSON input for Bedrock Converse tool call ${tool.name}`, + ) + return [{ type: "tool-call" as const, id: tool.id, name: tool.name, input }] + }) + +const processChunk = (state: ParserState, chunk: BedrockChunk) => + Effect.gen(function* () { + if (chunk.contentBlockStart?.start?.toolUse) { + const index = chunk.contentBlockStart.contentBlockIndex + return [ + { + ...state, + tools: { + ...state.tools, + [index]: { + id: chunk.contentBlockStart.start.toolUse.toolUseId, + name: chunk.contentBlockStart.start.toolUse.name, + input: "", + }, + }, + }, + [], + ] as const + } + + if (chunk.contentBlockDelta?.delta?.text) { + return [state, [{ type: "text-delta" as const, text: chunk.contentBlockDelta.delta.text }]] as const + } + + if (chunk.contentBlockDelta?.delta?.reasoningContent?.text) { + return [ + state, + [{ type: "reasoning-delta" as const, text: chunk.contentBlockDelta.delta.reasoningContent.text }], + ] as const + } + + if (chunk.contentBlockDelta?.delta?.toolUse) { + const index = chunk.contentBlockDelta.contentBlockIndex + const current = state.tools[index] + if (!current) { + return yield* ProviderShared.chunkError(ADAPTER, "Bedrock Converse tool delta is missing its tool call") + } + const next = { ...current, input: `${current.input}${chunk.contentBlockDelta.delta.toolUse.input}` } + return [ + { ...state, tools: { ...state.tools, [index]: next } }, + [ + { + type: "tool-input-delta" as const, + id: next.id, + name: next.name, + text: chunk.contentBlockDelta.delta.toolUse.input, + }, + ], + ] as const + } + + if (chunk.contentBlockStop) { + const events = yield* finishToolCall(state.tools[chunk.contentBlockStop.contentBlockIndex]) + const { [chunk.contentBlockStop.contentBlockIndex]: _, ...tools } = state.tools + return [{ ...state, tools }, events] as const + } + + if (chunk.messageStop) { + // Stash the reason — emit `request-finish` once `metadata` arrives with + // usage, so consumers see one terminal event carrying both. If metadata + // never arrives the `onHalt` fallback emits a usage-less finish. + return [{ ...state, finishReason: mapFinishReason(chunk.messageStop.stopReason) }, []] as const + } + + if (chunk.metadata) { + const reason = state.finishReason ?? "stop" + const usage = mapUsage(chunk.metadata.usage) + return [ + { ...state, finishReason: undefined }, + [{ type: "request-finish" as const, reason, usage }], + ] as const + } + + if (chunk.internalServerException || chunk.modelStreamErrorException || chunk.serviceUnavailableException) { + const message = + chunk.internalServerException?.message ?? + chunk.modelStreamErrorException?.message ?? + chunk.serviceUnavailableException?.message ?? + "Bedrock Converse stream error" + return [state, [{ type: "provider-error" as const, message, retryable: true }]] as const + } + + if (chunk.validationException || chunk.throttlingException) { + const message = + chunk.validationException?.message ?? chunk.throttlingException?.message ?? "Bedrock Converse error" + return [ + state, + [{ type: "provider-error" as const, message, retryable: chunk.throttlingException !== undefined }], + ] as const + } + + return [state, []] as const + }) + +// Bedrock streams responses using the AWS event stream binary protocol — each +// frame is `[length:4][headers-length:4][prelude-crc:4][headers][payload][crc:4]`. +// We use `@smithy/eventstream-codec` to validate framing and CRCs, then +// reconstruct the JSON wrapping by `:event-type` so the chunk schema can match. +const eventCodec = new EventStreamCodec(toUtf8, fromUtf8) +const utf8 = new TextDecoder() + +const concat = (left: Uint8Array, right: Uint8Array) => { + const next = new Uint8Array(left.length + right.length) + next.set(left) + next.set(right, left.length) + return next +} + +const consumeFrames = (state: Uint8Array, chunk: Uint8Array) => + Effect.gen(function* () { + let buffer = concat(state, chunk) + const out: string[] = [] + while (buffer.length >= 4) { + const totalLength = new DataView(buffer.buffer, buffer.byteOffset, buffer.byteLength).getUint32(0, false) + if (buffer.length < totalLength) break + + const decoded = yield* Effect.try({ + try: () => eventCodec.decode(buffer.subarray(0, totalLength)), + catch: (error) => + ProviderShared.chunkError( + ADAPTER, + `Failed to decode Bedrock Converse event-stream frame: ${ + error instanceof Error ? error.message : String(error) + }`, + ), + }) + buffer = buffer.slice(totalLength) + + if (decoded.headers[":message-type"]?.value !== "event") continue + const eventType = decoded.headers[":event-type"]?.value + if (typeof eventType !== "string") continue + const payload = utf8.decode(decoded.body) + if (!payload) continue + // The AWS event stream pads short payloads with a `p` field. Drop it + // before re-validating against the chunk schema. + const parsed = JSON.parse(payload) as Record + delete parsed.p + out.push(JSON.stringify({ [eventType]: parsed })) + } + return [buffer, out] as const + }) + +const parseStream = (response: HttpClientResponse.HttpClientResponse) => + response.stream.pipe( + Stream.mapError((error) => + ProviderShared.chunkError(ADAPTER, "Failed to read Bedrock Converse stream", String(error)), + ), + // Frame buffer: accumulate bytes, emit decoded JSON event strings as they + // become available. `mapAccumEffect` flattens the per-step `ReadonlyArray` + // automatically so the downstream stream sees one JSON string per element. + Stream.mapAccumEffect(() => new Uint8Array(0), consumeFrames), + Stream.mapEffect(decodeChunk), + Stream.mapAccumEffect( + (): ParserState => ({ tools: {}, finishReason: undefined }), + processChunk, + { + // If a stream ends after `messageStop` but before `metadata` (rare but + // possible on truncated transports), still surface a terminal finish. + onHalt: (state): ReadonlyArray => + state.finishReason ? [{ type: "request-finish", reason: state.finishReason }] : [], + }, + ), + ) + +export const adapter = Adapter.define({ + id: ADAPTER, + protocol: "bedrock-converse", + redact: (target) => target, + prepare, + validate: (draft) => decodeTarget(draft).pipe(Effect.mapError((error) => invalid(error.message))), + toHttp: (target, context) => toHttp(target, context.request), + parse: parseStream, +}) + +export const model = (input: BedrockConverseModelInput) => { + const { apiKey, credentials, headers, ...rest } = input + const authHeaders = apiKey ? { ...headers, authorization: `Bearer ${apiKey}` } : headers + return llmModel({ + ...rest, + provider: "bedrock", + protocol: "bedrock-converse", + headers: authHeaders, + capabilities: + input.capabilities ?? + capabilities({ + output: { reasoning: true }, + tools: { calls: true, streamingInput: true }, + cache: { prompt: true, contentBlocks: true }, + }), + native: credentials + ? { + ...input.native, + aws_credentials: credentials, + aws_region: credentials.region, + } + : input.native, + }) +} + +export * as BedrockConverse from "./bedrock-converse" diff --git a/packages/llm/test/fixtures/recordings/bedrock-converse/streams-a-tool-call.json b/packages/llm/test/fixtures/recordings/bedrock-converse/streams-a-tool-call.json new file mode 100644 index 0000000000..20c8d10b56 --- /dev/null +++ b/packages/llm/test/fixtures/recordings/bedrock-converse/streams-a-tool-call.json @@ -0,0 +1,23 @@ +{ + "version": 1, + "interactions": [ + { + "request": { + "method": "POST", + "url": "https://bedrock-runtime.us-east-1.amazonaws.com/model/us.amazon.nova-micro-v1%3A0/converse-stream", + "headers": { + "content-type": "application/json" + }, + "body": "{\"modelId\":\"us.amazon.nova-micro-v1:0\",\"messages\":[{\"role\":\"user\",\"content\":[{\"text\":\"Call get_weather with city exactly Paris.\"}]}],\"system\":[{\"text\":\"Call tools exactly as requested.\"}],\"inferenceConfig\":{\"maxTokens\":80,\"temperature\":0},\"toolConfig\":{\"tools\":[{\"toolSpec\":{\"name\":\"get_weather\",\"description\":\"Get current weather for a city.\",\"inputSchema\":{\"json\":{\"type\":\"object\",\"properties\":{\"city\":{\"type\":\"string\"}},\"required\":[\"city\"],\"additionalProperties\":false}}}}],\"toolChoice\":{\"tool\":{\"name\":\"get_weather\"}}}}" + }, + "response": { + "status": 200, + "headers": { + "content-type": "application/vnd.amazon.eventstream" + }, + "body": "AAAAsgAAAFKKQLTFCzpldmVudC10eXBlBwAMbWVzc2FnZVN0YXJ0DTpjb250ZW50LXR5cGUHABBhcHBsaWNhdGlvbi9qc29uDTptZXNzYWdlLXR5cGUHAAVldmVudHsicCI6ImFiY2RlZmdoaWprbG1ub3BxcnN0dXZ3eHl6QUJDREVGR0hJSktMTU5PUFFSU1RVVldYWVowIiwicm9sZSI6ImFzc2lzdGFudCJ9QP+CaAAAARYAAABXtNGquQs6ZXZlbnQtdHlwZQcAEWNvbnRlbnRCbG9ja1N0YXJ0DTpjb250ZW50LXR5cGUHABBhcHBsaWNhdGlvbi9qc29uDTptZXNzYWdlLXR5cGUHAAVldmVudHsiY29udGVudEJsb2NrSW5kZXgiOjAsInAiOiJhYmNkZWZnaGlqa2xtbm9wcXJzdHV2d3h5ekFCQ0RFRkdISUpLTE1OT1BRUlNUVVZXWFlaMDEyMzQiLCJzdGFydCI6eyJ0b29sVXNlIjp7Im5hbWUiOiJnZXRfd2VhdGhlciIsInRvb2xVc2VJZCI6InRvb2x1c2Vfc1NjQ2J5bmVSSFphQk1Ya25kV2JsYyJ9fX3TyNbnAAAA3QAAAFfhSKDWCzpldmVudC10eXBlBwARY29udGVudEJsb2NrRGVsdGENOmNvbnRlbnQtdHlwZQcAEGFwcGxpY2F0aW9uL2pzb24NOm1lc3NhZ2UtdHlwZQcABWV2ZW50eyJjb250ZW50QmxvY2tJbmRleCI6MCwiZGVsdGEiOnsidG9vbFVzZSI6eyJpbnB1dCI6IntcImNpdHlcIjpcIlBhcmlzXCJ9In19LCJwIjoiYWJjZGVmZ2hpamtsbW5vcHFyc3R1dnd4eXpBQkNERUZHSEkifVXxCVQAAACaAAAAVnycFBkLOmV2ZW50LXR5cGUHABBjb250ZW50QmxvY2tTdG9wDTpjb250ZW50LXR5cGUHABBhcHBsaWNhdGlvbi9qc29uDTptZXNzYWdlLXR5cGUHAAVldmVudHsiY29udGVudEJsb2NrSW5kZXgiOjAsInAiOiJhYmNkZWZnaGlqa2xtbm9wcXJzdHV2In2Jz2GTAAAArQAAAFHx+eUsCzpldmVudC10eXBlBwALbWVzc2FnZVN0b3ANOmNvbnRlbnQtdHlwZQcAEGFwcGxpY2F0aW9uL2pzb24NOm1lc3NhZ2UtdHlwZQcABWV2ZW50eyJwIjoiYWJjZGVmZ2hpamtsbW5vcHFyc3R1dnd4eXpBQkNERUZHSElKS0xNTk9QUVIiLCJzdG9wUmVhc29uIjoidG9vbF91c2UifRnHLhIAAADnAAAATm6yqDELOmV2ZW50LXR5cGUHAAhtZXRhZGF0YQ06Y29udGVudC10eXBlBwAQYXBwbGljYXRpb24vanNvbg06bWVzc2FnZS10eXBlBwAFZXZlbnR7Im1ldHJpY3MiOnsibGF0ZW5jeU1zIjo0Njh9LCJwIjoiYWJjZGVmZ2hpamtsbW5vcCIsInVzYWdlIjp7ImlucHV0VG9rZW5zIjo0MTksIm91dHB1dFRva2VucyI6MTYsInNlcnZlclRvb2xVc2FnZSI6e30sInRvdGFsVG9rZW5zIjo0MzV9fXyEqtI=", + "bodyEncoding": "base64" + } + } + ] +} diff --git a/packages/llm/test/fixtures/recordings/bedrock-converse/streams-text.json b/packages/llm/test/fixtures/recordings/bedrock-converse/streams-text.json new file mode 100644 index 0000000000..2b859cc25f --- /dev/null +++ b/packages/llm/test/fixtures/recordings/bedrock-converse/streams-text.json @@ -0,0 +1,23 @@ +{ + "version": 1, + "interactions": [ + { + "request": { + "method": "POST", + "url": "https://bedrock-runtime.us-east-1.amazonaws.com/model/us.amazon.nova-micro-v1%3A0/converse-stream", + "headers": { + "content-type": "application/json" + }, + "body": "{\"modelId\":\"us.amazon.nova-micro-v1:0\",\"messages\":[{\"role\":\"user\",\"content\":[{\"text\":\"Say hello.\"}]}],\"system\":[{\"text\":\"Reply with the single word 'Hello'.\"}],\"inferenceConfig\":{\"maxTokens\":16,\"temperature\":0}}" + }, + "response": { + "status": 200, + "headers": { + "content-type": "application/vnd.amazon.eventstream" + }, + "body": "AAAAgwAAAFIWASXzCzpldmVudC10eXBlBwAMbWVzc2FnZVN0YXJ0DTpjb250ZW50LXR5cGUHABBhcHBsaWNhdGlvbi9qc29uDTptZXNzYWdlLXR5cGUHAAVldmVudHsicCI6ImFiY2RlZiIsInJvbGUiOiJhc3Npc3RhbnQifQzWKawAAACnAAAAV1IqWLgLOmV2ZW50LXR5cGUHABFjb250ZW50QmxvY2tEZWx0YQ06Y29udGVudC10eXBlBwAQYXBwbGljYXRpb24vanNvbg06bWVzc2FnZS10eXBlBwAFZXZlbnR7ImNvbnRlbnRCbG9ja0luZGV4IjowLCJkZWx0YSI6eyJ0ZXh0IjoiSGVsbG8ifSwicCI6ImFiY2RlZmdoaSJ9Jr8EZAAAANYAAABXlpiRxws6ZXZlbnQtdHlwZQcAEWNvbnRlbnRCbG9ja0RlbHRhDTpjb250ZW50LXR5cGUHABBhcHBsaWNhdGlvbi9qc29uDTptZXNzYWdlLXR5cGUHAAVldmVudHsiY29udGVudEJsb2NrSW5kZXgiOjAsImRlbHRhIjp7InRleHQiOiIifSwicCI6ImFiY2RlZmdoaWprbG1ub3BxcnN0dXZ3eHl6QUJDREVGR0hJSktMTU5PUFFSU1RVVldYWVowMTIzNDU2NzgifagVAUAAAACJAAAAVlvc+UsLOmV2ZW50LXR5cGUHABBjb250ZW50QmxvY2tTdG9wDTpjb250ZW50LXR5cGUHABBhcHBsaWNhdGlvbi9qc29uDTptZXNzYWdlLXR5cGUHAAVldmVudHsiY29udGVudEJsb2NrSW5kZXgiOjAsInAiOiJhYmNkZSJ95xzwrwAAAJ0AAABRUNhdqgs6ZXZlbnQtdHlwZQcAC21lc3NhZ2VTdG9wDTpjb250ZW50LXR5cGUHABBhcHBsaWNhdGlvbi9qc29uDTptZXNzYWdlLXR5cGUHAAVldmVudHsicCI6ImFiY2RlZmdoaWprbG1ub3BxcnN0dXZ3eHl6QUIiLCJzdG9wUmVhc29uIjoiZW5kX3R1cm4ifXYxAh4AAADjAAAATpsyDvELOmV2ZW50LXR5cGUHAAhtZXRhZGF0YQ06Y29udGVudC10eXBlBwAQYXBwbGljYXRpb24vanNvbg06bWVzc2FnZS10eXBlBwAFZXZlbnR7Im1ldHJpY3MiOnsibGF0ZW5jeU1zIjo3NjF9LCJwIjoiYWJjZGVmZ2hpamtsbW5vIiwidXNhZ2UiOnsiaW5wdXRUb2tlbnMiOjEyLCJvdXRwdXRUb2tlbnMiOjIsInNlcnZlclRvb2xVc2FnZSI6e30sInRvdGFsVG9rZW5zIjoxNH19Mk8JgQ==", + "bodyEncoding": "base64" + } + } + ] +} diff --git a/packages/llm/test/provider/bedrock-converse.test.ts b/packages/llm/test/provider/bedrock-converse.test.ts new file mode 100644 index 0000000000..8159aa6d92 --- /dev/null +++ b/packages/llm/test/provider/bedrock-converse.test.ts @@ -0,0 +1,375 @@ +import { EventStreamCodec } from "@smithy/eventstream-codec" +import { fromUtf8, toUtf8 } from "@smithy/util-utf8" +import { describe, expect } from "bun:test" +import { Effect, Layer } from "effect" +import { LLM } from "../../src" +import { client } from "../../src/adapter" +import { BedrockConverse } from "../../src/provider/bedrock-converse" +import { testEffect } from "../lib/effect" +import { dynamicResponse } from "../lib/http" +import { recordedTests } from "../recorded-test" + +const codec = new EventStreamCodec(toUtf8, fromUtf8) +const utf8Encoder = new TextEncoder() + +// Build a single AWS event-stream frame for a Converse stream event. Each +// frame carries `:message-type=event` + `:event-type=` headers and a +// JSON payload body. +const eventFrame = (type: string, payload: object) => + codec.encode({ + headers: { + ":message-type": { type: "string", value: "event" }, + ":event-type": { type: "string", value: type }, + ":content-type": { type: "string", value: "application/json" }, + }, + body: utf8Encoder.encode(JSON.stringify(payload)), + }) + +const concat = (frames: ReadonlyArray) => { + const total = frames.reduce((sum, frame) => sum + frame.length, 0) + const out = new Uint8Array(total) + let offset = 0 + for (const frame of frames) { + out.set(frame, offset) + offset += frame.length + } + return out +} + +const eventStreamBody = (...payloads: ReadonlyArray) => + concat(payloads.map(([type, payload]) => eventFrame(type, payload))) + +const fixedBytes = (bytes: Uint8Array) => + dynamicResponse((input) => + Effect.succeed(input.respond(bytes, { headers: { "content-type": "application/vnd.amazon.eventstream" } })), + ) + +const model = BedrockConverse.model({ + id: "anthropic.claude-3-5-sonnet-20240620-v1:0", + baseURL: "https://bedrock-runtime.test", + apiKey: "test-bearer", +}) + +const baseRequest = LLM.request({ + id: "req_1", + model, + system: "You are concise.", + prompt: "Say hello.", + generation: { maxTokens: 64, temperature: 0 }, +}) + +const it = testEffect(Layer.empty) + +describe("Bedrock Converse adapter", () => { + it.effect("prepares Converse target with system, inference config, and messages", () => + Effect.gen(function* () { + const prepared = yield* client({ adapters: [BedrockConverse.adapter] }).prepare(baseRequest) + + expect(prepared.target).toEqual({ + modelId: "anthropic.claude-3-5-sonnet-20240620-v1:0", + system: [{ text: "You are concise." }], + messages: [{ role: "user", content: [{ text: "Say hello." }] }], + inferenceConfig: { maxTokens: 64, temperature: 0 }, + }) + }), + ) + + it.effect("prepares tool config with toolSpec and toolChoice", () => + Effect.gen(function* () { + const prepared = yield* client({ adapters: [BedrockConverse.adapter] }).prepare( + LLM.request({ + ...baseRequest, + tools: [ + { + name: "lookup", + description: "Lookup data", + inputSchema: { type: "object", properties: { query: { type: "string" } }, required: ["query"] }, + }, + ], + toolChoice: LLM.toolChoice({ type: "required" }), + }), + ) + + expect(prepared.target).toMatchObject({ + toolConfig: { + tools: [ + { + toolSpec: { + name: "lookup", + description: "Lookup data", + inputSchema: { + json: { type: "object", properties: { query: { type: "string" } }, required: ["query"] }, + }, + }, + }, + ], + toolChoice: { any: {} }, + }, + }) + }), + ) + + it.effect("lowers assistant tool-call + tool-result message history", () => + Effect.gen(function* () { + const prepared = yield* client({ adapters: [BedrockConverse.adapter] }).prepare( + LLM.request({ + id: "req_history", + model, + messages: [ + LLM.user("What is the weather?"), + LLM.assistant([LLM.toolCall({ id: "tool_1", name: "lookup", input: { query: "weather" } })]), + LLM.toolMessage({ id: "tool_1", name: "lookup", result: { forecast: "sunny" } }), + ], + }), + ) + + expect(prepared.target).toMatchObject({ + messages: [ + { role: "user", content: [{ text: "What is the weather?" }] }, + { + role: "assistant", + content: [{ toolUse: { toolUseId: "tool_1", name: "lookup", input: { query: "weather" } } }], + }, + { + role: "user", + content: [ + { + toolResult: { + toolUseId: "tool_1", + content: [{ json: { forecast: "sunny" } }], + status: "success", + }, + }, + ], + }, + ], + }) + }), + ) + + it.effect("decodes text-delta + messageStop + metadata usage from binary event stream", () => + Effect.gen(function* () { + const body = eventStreamBody( + ["messageStart", { role: "assistant" }], + ["contentBlockDelta", { contentBlockIndex: 0, delta: { text: "Hello" } }], + ["contentBlockDelta", { contentBlockIndex: 0, delta: { text: "!" } }], + ["contentBlockStop", { contentBlockIndex: 0 }], + ["messageStop", { stopReason: "end_turn" }], + ["metadata", { usage: { inputTokens: 5, outputTokens: 2, totalTokens: 7 } }], + ) + const response = yield* client({ adapters: [BedrockConverse.adapter] }) + .generate(baseRequest) + .pipe(Effect.provide(fixedBytes(body))) + + expect(LLM.outputText(response)).toBe("Hello!") + const finishes = response.events.filter((event) => event.type === "request-finish") + // Bedrock splits the finish across `messageStop` (carries reason) and + // `metadata` (carries usage). We consolidate them into a single + // terminal `request-finish` event with both. + expect(finishes).toHaveLength(1) + expect(finishes[0]).toMatchObject({ type: "request-finish", reason: "stop" }) + expect(LLM.outputUsage(response)).toMatchObject({ + inputTokens: 5, + outputTokens: 2, + totalTokens: 7, + }) + }), + ) + + it.effect("assembles streamed tool call input", () => + Effect.gen(function* () { + const body = eventStreamBody( + ["messageStart", { role: "assistant" }], + [ + "contentBlockStart", + { + contentBlockIndex: 0, + start: { toolUse: { toolUseId: "tool_1", name: "lookup" } }, + }, + ], + ["contentBlockDelta", { contentBlockIndex: 0, delta: { toolUse: { input: '{"query"' } } }], + ["contentBlockDelta", { contentBlockIndex: 0, delta: { toolUse: { input: ':"weather"}' } } }], + ["contentBlockStop", { contentBlockIndex: 0 }], + ["messageStop", { stopReason: "tool_use" }], + ) + const response = yield* client({ adapters: [BedrockConverse.adapter] }) + .generate( + LLM.request({ + ...baseRequest, + tools: [{ name: "lookup", description: "Lookup", inputSchema: { type: "object" } }], + }), + ) + .pipe(Effect.provide(fixedBytes(body))) + + expect(LLM.outputToolCalls(response)).toEqual([ + { type: "tool-call", id: "tool_1", name: "lookup", input: { query: "weather" } }, + ]) + const events = response.events.filter((event) => event.type === "tool-input-delta") + expect(events).toEqual([ + { type: "tool-input-delta", id: "tool_1", name: "lookup", text: '{"query"' }, + { type: "tool-input-delta", id: "tool_1", name: "lookup", text: ':"weather"}' }, + ]) + expect(response.events.at(-1)).toMatchObject({ type: "request-finish", reason: "tool-calls" }) + }), + ) + + it.effect("decodes reasoning deltas", () => + Effect.gen(function* () { + const body = eventStreamBody( + ["messageStart", { role: "assistant" }], + [ + "contentBlockDelta", + { contentBlockIndex: 0, delta: { reasoningContent: { text: "Let me think." } } }, + ], + ["contentBlockStop", { contentBlockIndex: 0 }], + ["messageStop", { stopReason: "end_turn" }], + ) + const response = yield* client({ adapters: [BedrockConverse.adapter] }) + .generate(baseRequest) + .pipe(Effect.provide(fixedBytes(body))) + + expect(LLM.outputReasoning(response)).toBe("Let me think.") + }), + ) + + it.effect("emits provider-error for throttlingException", () => + Effect.gen(function* () { + const body = eventStreamBody( + ["messageStart", { role: "assistant" }], + ["throttlingException", { message: "Slow down" }], + ) + const response = yield* client({ adapters: [BedrockConverse.adapter] }) + .generate(baseRequest) + .pipe(Effect.provide(fixedBytes(body))) + + expect(response.events.find((event) => event.type === "provider-error")).toEqual({ + type: "provider-error", + message: "Slow down", + retryable: true, + }) + }), + ) + + it.effect("rejects requests with no auth path", () => + Effect.gen(function* () { + const unsignedModel = BedrockConverse.model({ + id: "anthropic.claude-3-5-sonnet-20240620-v1:0", + baseURL: "https://bedrock-runtime.test", + }) + const error = yield* client({ adapters: [BedrockConverse.adapter] }) + .generate(LLM.request({ ...baseRequest, model: unsignedModel })) + .pipe(Effect.provide(fixedBytes(eventStreamBody(["messageStop", { stopReason: "end_turn" }]))), Effect.flip) + + expect(error.message).toContain("Bedrock Converse requires either a Bearer API key") + }), + ) + + it.effect("signs requests with SigV4 when AWS credentials are provided (deterministic plumbing check)", () => + Effect.gen(function* () { + const signed = BedrockConverse.model({ + id: "anthropic.claude-3-5-sonnet-20240620-v1:0", + baseURL: "https://bedrock-runtime.test", + credentials: { + region: "us-east-1", + accessKeyId: "AKIAIOSFODNN7EXAMPLE", + secretAccessKey: "wJalrXUtnFEMI/K7MDENG/bPxRfiCYEXAMPLEKEY", + }, + }) + const prepared = yield* client({ adapters: [BedrockConverse.adapter] }).prepare( + LLM.request({ ...baseRequest, model: signed }), + ) + + expect(prepared.adapter).toBe("bedrock-converse") + // The prepare phase doesn't sign — toHttp does. We assert the credential + // is plumbed onto the model native field for the signer to find. + expect(prepared.model.native).toMatchObject({ + aws_credentials: { region: "us-east-1", accessKeyId: "AKIAIOSFODNN7EXAMPLE" }, + aws_region: "us-east-1", + }) + }), + ) +}) + +// Live recorded integration tests. Run with `RECORD=true AWS_ACCESS_KEY_ID=... +// AWS_SECRET_ACCESS_KEY=... [AWS_SESSION_TOKEN=...] bun run test ...` to refresh +// cassettes; replay is the default and works without credentials. +// +// Region is pinned to us-east-1 in tests so the request URL is stable across +// machines on replay. If you need to record from a different region (e.g. your +// account has access elsewhere), pass `BEDROCK_RECORDING_REGION=eu-west-1` — +// but then commit the resulting cassette and others should record from the +// same region too. +const RECORDING_REGION = process.env.BEDROCK_RECORDING_REGION ?? "us-east-1" + +const recordedModel = () => + BedrockConverse.model({ + // Most newer Anthropic models on Bedrock require a cross-region inference + // profile (`us.` prefix). Nova does not require an Anthropic use-case form + // and is on-demand-throughput accessible by default for most accounts. + id: process.env.BEDROCK_MODEL_ID ?? "us.amazon.nova-micro-v1:0", + credentials: { + region: RECORDING_REGION, + accessKeyId: process.env.AWS_ACCESS_KEY_ID ?? "fixture", + secretAccessKey: process.env.AWS_SECRET_ACCESS_KEY ?? "fixture", + sessionToken: process.env.AWS_SESSION_TOKEN, + }, + }) + +const recorded = recordedTests({ + prefix: "bedrock-converse", + requires: ["AWS_ACCESS_KEY_ID", "AWS_SECRET_ACCESS_KEY"], +}) + +describe("Bedrock Converse recorded", () => { + recorded.effect("streams text", () => + Effect.gen(function* () { + const llm = client({ adapters: [BedrockConverse.adapter] }) + const response = yield* llm.generate( + LLM.request({ + id: "recorded_bedrock_text", + model: recordedModel(), + system: "Reply with the single word 'Hello'.", + prompt: "Say hello.", + generation: { maxTokens: 16, temperature: 0 }, + }), + ) + + expect(LLM.outputText(response)).toMatch(/hello/i) + expect(response.events.at(-1)).toMatchObject({ type: "request-finish" }) + }), + ) + + recorded.effect("streams a tool call", () => + Effect.gen(function* () { + const llm = client({ adapters: [BedrockConverse.adapter] }) + const response = yield* llm.generate( + LLM.request({ + id: "recorded_bedrock_tool_call", + model: recordedModel(), + system: "Call tools exactly as requested.", + prompt: "Call get_weather with city exactly Paris.", + tools: [ + { + name: "get_weather", + description: "Get current weather for a city.", + inputSchema: { + type: "object", + properties: { city: { type: "string" } }, + required: ["city"], + additionalProperties: false, + }, + }, + ], + toolChoice: LLM.toolChoice({ type: "tool", name: "get_weather" }), + generation: { maxTokens: 80, temperature: 0 }, + }), + ) + + expect(response.events.some((event) => event.type === "tool-input-delta")).toBe(true) + expect(LLM.outputToolCalls(response)).toEqual([ + { type: "tool-call", id: expect.any(String), name: "get_weather", input: { city: "Paris" } }, + ]) + expect(response.events.at(-1)).toMatchObject({ type: "request-finish", reason: "tool-calls" }) + }), + ) +}) diff --git a/packages/llm/test/record-replay.ts b/packages/llm/test/record-replay.ts index ecc1346901..1c77e14ace 100644 --- a/packages/llm/test/record-replay.ts +++ b/packages/llm/test/record-replay.ts @@ -26,6 +26,11 @@ const ResponseSnapshot = Schema.Struct({ status: Schema.Number, headers: Schema.Record(Schema.String, Schema.String), body: Schema.String, + // Most provider responses are text (SSE, JSON). AWS Bedrock streams are + // binary AWS event-stream frames whose CRC32 fields would mangle through a + // UTF-8 round-trip — store those as base64. Older cassettes omit this field + // and decode as text by default. + bodyEncoding: Schema.optional(Schema.Literals(["text", "base64"])), }) const Interaction = Schema.Struct({ @@ -152,6 +157,36 @@ const responseHeaders = ( return merged } +// Content types whose payloads are binary frames or arbitrary bytes — they +// would not survive a UTF-8 text round-trip. The list intentionally matches +// the substrings that appear in `Content-Type` headers, not full values. +const BINARY_CONTENT_TYPES: ReadonlyArray = [ + "vnd.amazon.eventstream", + "octet-stream", +] + +const isBinaryContentType = (contentType: string | undefined) => { + if (!contentType) return false + const lower = contentType.toLowerCase() + return BINARY_CONTENT_TYPES.some((token) => lower.includes(token)) +} + +const captureResponseBody = ( + response: HttpClientResponse.HttpClientResponse, + contentType: string | undefined, +) => + Effect.gen(function* () { + if (!isBinaryContentType(contentType)) { + const text = yield* response.text + return { body: text, bodyEncoding: undefined as "text" | "base64" | undefined } + } + const bytes = yield* response.arrayBuffer + return { body: Buffer.from(bytes).toString("base64"), bodyEncoding: "base64" as const } + }) + +const decodeResponseBody = (snapshot: Schema.Schema.Type) => + snapshot.bodyEncoding === "base64" ? Buffer.from(snapshot.body, "base64") : snapshot.body + const fixtureMissing = (request: HttpClientRequest.HttpClientRequest, name: string) => new HttpClientError.HttpClientError({ reason: new HttpClientError.TransportError({ @@ -251,19 +286,16 @@ export const layer = ( return Effect.gen(function* () { const currentRequest = yield* snapshotRequest(request) const response = yield* upstream.execute(request) - const body = yield* response.text + const headers = responseHeaders(response, responseHeadersAllow) + const captured = yield* captureResponseBody(response, headers["content-type"]) const interaction: Interaction = { request: currentRequest, - response: { - status: response.status, - headers: responseHeaders(response, responseHeadersAllow), - body, - }, + response: { status: response.status, headers, body: captured.body, bodyEncoding: captured.bodyEncoding }, } const interactions = yield* Ref.updateAndGet(recorded, (prev) => [...prev, interaction]) yield* fileSystem.makeDirectory(dir, { recursive: true }).pipe(Effect.orDie) yield* fileSystem.writeFileString(file, formatCassette(interactions)).pipe(Effect.orDie) - return HttpClientResponse.fromWeb(request, new Response(body, interaction.response)) + return HttpClientResponse.fromWeb(request, new Response(decodeResponseBody(interaction.response), interaction.response)) }) } @@ -275,7 +307,7 @@ export const layer = ( const { interaction, detail } = yield* selectInteraction(cassette, incoming) if (!interaction) return yield* fixtureMismatch(request, name, detail) - return HttpClientResponse.fromWeb(request, new Response(interaction.response.body, interaction.response)) + return HttpClientResponse.fromWeb(request, new Response(decodeResponseBody(interaction.response), interaction.response)) }) }) }), diff --git a/packages/opencode/src/provider/llm-bridge.ts b/packages/opencode/src/provider/llm-bridge.ts index b756494143..105def8a2c 100644 --- a/packages/opencode/src/provider/llm-bridge.ts +++ b/packages/opencode/src/provider/llm-bridge.ts @@ -1,4 +1,5 @@ import * as LLM from "@opencode-ai/llm/llm" +import { AmazonBedrock } from "@opencode-ai/llm/provider/amazon-bedrock" import { Anthropic } from "@opencode-ai/llm/provider/anthropic" import { Azure } from "@opencode-ai/llm/provider/azure" import { GitHubCopilot } from "@opencode-ai/llm/provider/github-copilot" @@ -18,6 +19,7 @@ type Input = { } const PROVIDERS: Record = { + "@ai-sdk/amazon-bedrock": AmazonBedrock.provider, "@ai-sdk/anthropic": Anthropic.provider, "@ai-sdk/azure": Azure.provider, "@ai-sdk/baseten": OpenAICompatibleFamily.provider,