From 312cb6ba133cc0cb63f0606e33616928b3cfbc68 Mon Sep 17 00:00:00 2001 From: Ting-Hong Shieh Date: Sat, 3 Oct 2026 13:06:54 +0800 Subject: [PATCH 1/2] feat(translation): support Bedrock Converse as a native wire format Signed-off-by: Ting-Hong Shieh --- crates/protocol/src/format.rs | 4 + crates/protocol/src/stream.rs | 5 +- crates/switchyard-server/src/lib.rs | 16 +- crates/switchyard-server/src/sse.rs | 6 +- crates/switchyard-translation/README.md | 31 +- .../src/codecs/anthropic/stream.rs | 16 +- .../src/codecs/bedrock.rs | 12 + .../src/codecs/bedrock/buffered.rs | 958 ++++++++++++++++++ .../src/codecs/bedrock/stream.rs | 557 ++++++++++ .../switchyard-translation/src/codecs/mod.rs | 1 + .../src/codecs/stream.rs | 4 + crates/switchyard-translation/src/engine.rs | 15 +- crates/switchyard-translation/src/helpers.rs | 59 +- crates/switchyard-translation/src/sse.rs | 1 + .../tests/bedrock_translation.rs | 562 ++++++++++ .../tests/lossless_roundtrip.rs | 8 +- .../tests/request_translation.rs | 5 +- 17 files changed, 2235 insertions(+), 25 deletions(-) create mode 100644 crates/switchyard-translation/src/codecs/bedrock.rs create mode 100644 crates/switchyard-translation/src/codecs/bedrock/buffered.rs create mode 100644 crates/switchyard-translation/src/codecs/bedrock/stream.rs create mode 100644 crates/switchyard-translation/tests/bedrock_translation.rs diff --git a/crates/protocol/src/format.rs b/crates/protocol/src/format.rs index 745f4120d..eac1e29b2 100644 --- a/crates/protocol/src/format.rs +++ b/crates/protocol/src/format.rs @@ -20,6 +20,9 @@ pub enum WireFormat { /// OpenAI Responses API. #[serde(rename = "openai_responses")] OpenAiResponses, + /// Amazon Bedrock Converse API; the host owns AWS transport framing. + #[serde(rename = "bedrock_converse")] + BedrockConverse, } impl WireFormat { @@ -29,6 +32,7 @@ impl WireFormat { Self::OpenAiChat => "openai_chat", Self::AnthropicMessages => "anthropic_messages", Self::OpenAiResponses => "openai_responses", + Self::BedrockConverse => "bedrock_converse", } } } diff --git a/crates/protocol/src/stream.rs b/crates/protocol/src/stream.rs index c59a66f77..32e582f46 100644 --- a/crates/protocol/src/stream.rs +++ b/crates/protocol/src/stream.rs @@ -506,7 +506,10 @@ fn stop_reason_from_str(reason: Option<&str>) -> StopReason { match reason { Some("length" | "max_tokens") => StopReason::MaxTokens, Some("tool_calls" | "function_call" | "tool_use") => StopReason::ToolUse, - Some("content_filter") => StopReason::ContentFilter, + Some("content_filter" | "content_filtered" | "guardrail_intervened") => { + StopReason::ContentFilter + } + Some("malformed_model_output" | "malformed_tool_use") => StopReason::Error, Some("stop" | "end_turn" | "stop_sequence") | None => StopReason::EndTurn, Some(_) => StopReason::Unknown, } diff --git a/crates/switchyard-server/src/lib.rs b/crates/switchyard-server/src/lib.rs index 0f1738936..13f468fb2 100644 --- a/crates/switchyard-server/src/lib.rs +++ b/crates/switchyard-server/src/lib.rs @@ -1389,13 +1389,15 @@ impl ApiError { "message": self.message.clone(), } }), - WireFormat::OpenAiChat | WireFormat::OpenAiResponses => json!({ - "error": { - "message": self.message.clone(), - "type": self.error_type, - "code": self.code, - } - }), + WireFormat::OpenAiChat | WireFormat::OpenAiResponses | WireFormat::BedrockConverse => { + json!({ + "error": { + "message": self.message.clone(), + "type": self.error_type, + "code": self.code, + } + }) + } }; let mut response = (self.status, Json(body)).into_response(); response diff --git a/crates/switchyard-server/src/sse.rs b/crates/switchyard-server/src/sse.rs index 942b0ee07..67132f811 100644 --- a/crates/switchyard-server/src/sse.rs +++ b/crates/switchyard-server/src/sse.rs @@ -82,6 +82,10 @@ fn frame_event( let data = redactor.json(serde_json::to_string(&value)?); match target_format { WireFormat::OpenAiChat => Ok(Event::default().data(data)), + WireFormat::BedrockConverse => Err(serde_json::Error::io(std::io::Error::new( + std::io::ErrorKind::Unsupported, + "Bedrock requires AWS EventStream framing, unavailable on SSE endpoints", + ))), WireFormat::AnthropicMessages | WireFormat::OpenAiResponses => { let event_type = value .get("type") @@ -95,7 +99,7 @@ fn frame_event( fn error_event(target_format: WireFormat, message: String, redactor: &Redactor) -> Event { match target_format { - WireFormat::OpenAiChat => Event::default().data( + WireFormat::OpenAiChat | WireFormat::BedrockConverse => Event::default().data( redactor.json( json!({ "error": { diff --git a/crates/switchyard-translation/README.md b/crates/switchyard-translation/README.md index 10ed01f11..ce3faf253 100644 --- a/crates/switchyard-translation/README.md +++ b/crates/switchyard-translation/README.md @@ -1,11 +1,38 @@ # switchyard-translation -Pure Rust translation between OpenAI Chat Completions, OpenAI Responses, and Anthropic Messages -request, response, and streaming formats. +Pure Rust translation between OpenAI Chat Completions, OpenAI Responses, Anthropic Messages, +and Amazon Bedrock Converse request, response, and streaming formats. The crate translates through provider-neutral LLM types from `switchyard-protocol` and does not depend on provider SDKs, HTTP servers, Python, or FFI bindings. +## Bedrock Converse + +Use `WireFormat::BedrockConverse` (serialized as `bedrock_converse`) with `TranslationEngine` +for buffered bodies. Text, system instructions, function tools and tool results, common inference +settings, inline images, stop reasons, and cache-token usage have neutral mappings. JSON tool results +become serialized JSON text when normalized; same-format preservation retains their original JSON. +Native controls such as guardrails and additional model fields survive same-format request encoding. +Cross-format projection diagnoses their loss and rejects it under strict loss policy. + +ConverseStream codecs operate on the JSON union events after the host removes and validates AWS +EventStream framing. Feed those events to `decode_event_stream`, then use `encode_stream` for target +JSON events. `decode_stream` is for SSE bytes and rejects Bedrock. A Bedrock stream must include +`messageStart`, closed content blocks, `messageStop`, and terminal `metadata` with reported usage; +truncation, malformed events, or provider exceptions fail the stream. Encoding requires reported +input and output tokens; an absent total is derived from those counts and cache details. + +Same-format preserved events replay unchanged, including native reasoning signatures. Aggregated +Bedrock reasoning preserves signature and redacted-content fragments. Foreign opaque reasoning +stream details, including cross-format Anthropic/Bedrock signatures, are rejected. Tool arguments +may precede their ID and name; parallel calls are serialized into Bedrock blocks. A tool block +cannot resume after other content closes it. Encoding Bedrock rejects unsupported foreign media and provider-specific built-in tool +history. + +The host owns the model ID in the request URL, AWS credentials, SigV4 signing, regions, retries, +and binary EventStream framing. This crate adds no Bedrock HTTP client or server endpoint. Adding +the public `WireFormat` member requires downstream exhaustive matches to handle the new variant. + ## License Licensed under the Apache License, Version 2.0. See the diff --git a/crates/switchyard-translation/src/codecs/anthropic/stream.rs b/crates/switchyard-translation/src/codecs/anthropic/stream.rs index 908941a89..578c499b1 100644 --- a/crates/switchyard-translation/src/codecs/anthropic/stream.rs +++ b/crates/switchyard-translation/src/codecs/anthropic/stream.rs @@ -211,7 +211,21 @@ fn encode_anthropic_stream( })); out } - LlmResponseChunk::ReasoningDetailsDelta { text, .. } => { + LlmResponseChunk::ReasoningDetailsDelta { text, details, .. } => { + if details.iter().any(|detail| { + detail + .get("type") + .and_then(Value::as_str) + .is_some_and(|kind| kind.starts_with("bedrock.")) + }) { + return encode_anthropic_stream( + state, + LlmResponseChunk::DecodeError { + message: "Bedrock reasoning signatures have no Anthropic stream mapping" + .into(), + }, + ); + } if text.is_empty() { return Vec::new(); } diff --git a/crates/switchyard-translation/src/codecs/bedrock.rs b/crates/switchyard-translation/src/codecs/bedrock.rs new file mode 100644 index 000000000..ede642098 --- /dev/null +++ b/crates/switchyard-translation/src/codecs/bedrock.rs @@ -0,0 +1,12 @@ +// SPDX-FileCopyrightText: Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved. +// SPDX-License-Identifier: Apache-2.0 + +//! Bedrock Converse buffered bodies and ConverseStream event payloads. + +mod buffered; +pub(crate) mod stream; + +pub use buffered::BedrockConverseCodec; +pub use stream::BedrockConverseStreamCodec; + +pub(crate) use buffered::request_projection_diagnostics; diff --git a/crates/switchyard-translation/src/codecs/bedrock/buffered.rs b/crates/switchyard-translation/src/codecs/bedrock/buffered.rs new file mode 100644 index 000000000..22162ab4f --- /dev/null +++ b/crates/switchyard-translation/src/codecs/bedrock/buffered.rs @@ -0,0 +1,958 @@ +// SPDX-FileCopyrightText: Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved. +// SPDX-License-Identifier: Apache-2.0 + +//! Buffered Bedrock Converse JSON conversion. The host supplies the URL-owned model ID. + +use base64::{Engine as _, engine::general_purpose::STANDARD}; +use serde_json::{Map, Value, json}; + +use crate::codecs::common::provider_extensions; +use crate::codecs::{ + DecodedRequest, DecodedResponse, EncodedRequest, EncodedResponse, FormatCodec, +}; +use crate::diagnostic::TranslationDiagnostic; +use crate::error::{Result, TranslationError}; +use crate::format::{FormatId, WireFormat}; +use crate::llm::{ + AggLlmResponse, ContentBlock, ImageSource, InstructionBlock, LlmRequest, Message, + ProviderExtensions, ResponseOutput, Role, StopReason, ToolCall, ToolChoice, ToolDefinition, + ToolResult, Usage, +}; +use crate::policy::TranslationPolicy; +use crate::util::{ + capture_request_preservation, capture_response_preservation, embed_preservation, + exact_preserved_request, exact_preserved_response, object, push_lossy, push_unknown_field, + reject_responses_builtin_tool_item, validate_request_capabilities, +}; + +const NATIVE_REQUEST: &str = "_bedrock_converse_request"; +const NATIVE_RESPONSE: &str = "_bedrock_converse_response"; + +/// Buffered codec for Bedrock Converse bodies. +pub struct BedrockConverseCodec; + +impl FormatCodec for BedrockConverseCodec { + fn format(&self) -> FormatId { + WireFormat::BedrockConverse.into() + } + + fn decode_request(&self, body: &Value, policy: &TranslationPolicy) -> Result { + let obj = object(body, "$")?; + let mut diagnostics = Vec::new(); + check_unknown( + obj, + &[ + "messages", + "system", + "inferenceConfig", + "toolConfig", + "additionalModelRequestFields", + "additionalModelResponseFieldPaths", + "guardrailConfig", + "promptVariables", + "requestMetadata", + "performanceConfig", + "serviceTier", + "outputConfig", + "metadata", + ], + "$", + &mut diagnostics, + policy, + )?; + let mut request = LlmRequest { + preservation: capture_request_preservation(WireFormat::BedrockConverse, body, policy), + ..LlmRequest::default() + }; + if let Some(system) = obj.get("system") { + let content = decode_content( + array(system, "$.system")?, + "$.system", + &mut diagnostics, + policy, + )?; + request.instructions.push(InstructionBlock { + role: Role::System, + content, + }); + } + let messages = array(required(obj, "messages", "$")?, "$.messages")?; + for (index, value) in messages.iter().enumerate() { + let path = format!("$.messages[{index}]"); + let message = object(value, &path)?; + let role = decode_role(required_string(message, "role", &path)?, &path)?; + let content = decode_content( + array( + required(message, "content", &path)?, + &format!("{path}.content"), + )?, + &format!("{path}.content"), + &mut diagnostics, + policy, + )?; + request.messages.push(Message { role, content }); + } + if let Some(config) = obj.get("inferenceConfig") { + let config = object(config, "$.inferenceConfig")?; + check_unknown( + config, + &["maxTokens", "temperature", "topP", "stopSequences"], + "$.inferenceConfig", + &mut diagnostics, + policy, + )?; + request.output.max_output_tokens = + optional_u64(config, "maxTokens", "$.inferenceConfig")?; + request.sampling.temperature = + optional_f64(config, "temperature", "$.inferenceConfig")?; + request.sampling.top_p = optional_f64(config, "topP", "$.inferenceConfig")?; + if let Some(stop) = config.get("stopSequences") { + if array(stop, "$.inferenceConfig.stopSequences")? + .iter() + .any(|v| !v.is_string()) + { + return Err(invalid( + "$.inferenceConfig.stopSequences", + "expected strings", + )); + } + request + .extensions + .fields + .insert("stop".into(), stop.clone()); + request + .extensions + .fields + .insert("stop_sequences".into(), stop.clone()); + } + } + if let Some(config) = obj.get("toolConfig") { + let config = object(config, "$.toolConfig")?; + for (index, value) in array( + required(config, "tools", "$.toolConfig")?, + "$.toolConfig.tools", + )? + .iter() + .enumerate() + { + let path = format!("$.toolConfig.tools[{index}].toolSpec"); + let tool = object(value, &format!("$.toolConfig.tools[{index}]"))?; + if tool.len() != 1 { + return Err(invalid( + "$.toolConfig.tools", + "tool must contain one union member", + )); + } + if let Some(spec) = value.get("toolSpec") { + let spec = object(spec, &path)?; + let schema = object( + required(spec, "inputSchema", &path)?, + &format!("{path}.inputSchema"), + )?; + let strict = spec + .get("strict") + .map(|v| { + v.as_bool().ok_or_else(|| { + invalid(&format!("{path}.strict"), "expected boolean") + }) + }) + .transpose()?; + request.tools.push(ToolDefinition { + name: required_string(spec, "name", &path)?.into(), + description: spec + .get("description") + .map(|v| string(v, &format!("{path}.description")).map(str::to_owned)) + .transpose()?, + parameters: required(schema, "json", &format!("{path}.inputSchema"))? + .clone(), + strict, + }); + } else { + push_lossy( + &mut diagnostics, + policy, + "Bedrock non-function tool has no neutral tool definition", + )?; + } + } + request.tool_choice = config + .get("toolChoice") + .map(decode_tool_choice) + .transpose()?; + } + request.extensions.fields.insert( + NATIVE_REQUEST.into(), + Value::Object(provider_extensions( + obj, + &[ + "messages", + "system", + "inferenceConfig", + "toolConfig", + "metadata", + ], + )), + ); + Ok(DecodedRequest { + request, + diagnostics, + }) + } + + fn encode_request( + &self, + request: &LlmRequest, + policy: &TranslationPolicy, + ) -> Result { + let mut diagnostics = Vec::new(); + validate_request_capabilities(request, &mut diagnostics, policy)?; + if let Some(body) = + exact_preserved_request(&request.preservation, WireFormat::BedrockConverse, policy) + { + return Ok(EncodedRequest { body, diagnostics }); + } + for (unsupported, label) in [ + (request.sampling.top_k.is_some(), "top_k"), + ( + request.reasoning.effort.is_some() || request.reasoning.raw.is_some(), + "reasoning controls", + ), + ( + request.output.response_format.is_some() + || request.output.is_schema_enforced == Some(true), + "response format", + ), + ( + request + .extensions + .fields + .contains_key("parallel_tool_calls"), + "parallel_tool_calls", + ), + ] { + if unsupported { + push_lossy( + &mut diagnostics, + policy, + format!("Bedrock Converse has no portable mapping for {label}"), + )?; + } + } + let mut body = Map::new(); + let mut system = Vec::new(); + for instruction in &request.instructions { + for block in &instruction.content { + system.push(encode_system_block(block, &mut diagnostics, policy)?); + } + } + let mut messages = Vec::new(); + for message in &request.messages { + if matches!(message.role, Role::System | Role::Developer) { + for block in &message.content { + system.push(encode_system_block(block, &mut diagnostics, policy)?); + } + } else { + messages.push(json!({"role": encode_role(message.role), + "content": encode_content(&message.content, &mut diagnostics, policy)?})); + } + } + if !system.is_empty() { + body.insert("system".into(), Value::Array(system)); + } + body.insert("messages".into(), Value::Array(messages)); + let mut inference = Map::new(); + if let Some(v) = request.output.max_output_tokens { + inference.insert("maxTokens".into(), v.into()); + } + if let Some(v) = request.sampling.temperature { + inference.insert("temperature".into(), v.into()); + } + if let Some(v) = request.sampling.top_p { + inference.insert("topP".into(), v.into()); + } + if let Some(v) = request + .extensions + .fields + .get("stop_sequences") + .or_else(|| request.extensions.fields.get("stop")) + { + let v = if v.is_string() { json!([v]) } else { v.clone() }; + inference.insert("stopSequences".into(), v); + } + if !inference.is_empty() { + body.insert("inferenceConfig".into(), Value::Object(inference)); + } + // Converse has no `none` tool-choice variant: disabling tools means omitting the config. + if request.tool_choice != Some(ToolChoice::None) && !request.tools.is_empty() { + let tools = request + .tools + .iter() + .map(|tool| { + let mut spec = + json!({"name": tool.name, "inputSchema": {"json": tool.parameters}}); + if let Some(v) = &tool.description { + spec["description"] = v.clone().into(); + } + if let Some(v) = tool.strict { + spec["strict"] = v.into(); + } + json!({"toolSpec": spec}) + }) + .collect::>(); + let mut config = json!({"tools": tools}); + if let Some(choice) = &request.tool_choice { + config["toolChoice"] = encode_tool_choice(choice)?; + } + body.insert("toolConfig".into(), config); + } else if request.tools.is_empty() + && matches!( + request.tool_choice, + Some(ToolChoice::Required | ToolChoice::Tool { .. }) + ) + { + return Err(invalid("$.toolConfig", "required tool choice needs tools")); + } + if let Some(native) = request + .extensions + .fields + .get(NATIVE_REQUEST) + .and_then(Value::as_object) + { + for key in [ + "additionalModelRequestFields", + "additionalModelResponseFieldPaths", + "guardrailConfig", + "promptVariables", + "requestMetadata", + "performanceConfig", + "serviceTier", + "outputConfig", + ] { + if let Some(value) = native.get(key) { + body.insert(key.into(), value.clone()); + } + } + } + Ok(EncodedRequest { + body: embed_preservation(Value::Object(body), &request.preservation, policy), + diagnostics, + }) + } + + fn decode_response(&self, body: &Value, policy: &TranslationPolicy) -> Result { + let obj = object(body, "$")?; + if let Some(error) = obj.iter().find(|(key, _)| key.ends_with("Exception")) { + return Err(TranslationError::UpstreamFailure { + error: json!({error.0: error.1}), + }); + } + let mut diagnostics = Vec::new(); + check_unknown( + obj, + &[ + "output", + "stopReason", + "usage", + "metrics", + "additionalModelResponseFields", + "trace", + "performanceConfig", + "serviceTier", + "metadata", + ], + "$", + &mut diagnostics, + policy, + )?; + let output = object(required(obj, "output", "$")?, "$.output")?; + let message = object(required(output, "message", "$.output")?, "$.output.message")?; + let role = decode_role( + required_string(message, "role", "$.output.message")?, + "$.output.message", + )?; + if role != Role::Assistant { + return Err(invalid("$.output.message.role", "expected assistant")); + } + let content = decode_content( + array( + required(message, "content", "$.output.message")?, + "$.output.message.content", + )?, + "$.output.message.content", + &mut diagnostics, + policy, + )?; + let reason = required_string(obj, "stopReason", "$")?; + let response = AggLlmResponse { + outputs: vec![ResponseOutput { + role, + content, + stop_reason: Some(decode_stop_reason(reason)), + url_citations: Vec::new(), + }], + usage: decode_usage(required(obj, "usage", "$")?)?, + extensions: ProviderExtensions { + fields: Map::from_iter([( + NATIVE_RESPONSE.into(), + Value::Object(provider_extensions(obj, &["output", "usage", "metadata"])), + )]), + }, + preservation: capture_response_preservation(WireFormat::BedrockConverse, body, policy), + ..AggLlmResponse::default() + }; + Ok(DecodedResponse { + response, + diagnostics, + }) + } + + fn encode_response( + &self, + response: &AggLlmResponse, + policy: &TranslationPolicy, + ) -> Result { + super::super::responses::validate_response_output(response, WireFormat::BedrockConverse)?; + if let Some(body) = + exact_preserved_response(&response.preservation, WireFormat::BedrockConverse, policy) + { + return Ok(EncodedResponse { + body, + diagnostics: Vec::new(), + }); + } + let mut diagnostics = Vec::new(); + if response.outputs.len() > 1 { + push_lossy( + &mut diagnostics, + policy, + "Bedrock Converse returns one output message", + )?; + } + let output = response + .outputs + .first() + .ok_or_else(|| invalid("$.output", "expected an assistant output"))?; + if output.role != Role::Assistant { + return Err(invalid("$.output.message.role", "expected assistant")); + } + if !output.url_citations.is_empty() { + push_lossy( + &mut diagnostics, + policy, + "Bedrock output does not encode neutral URL citations", + )?; + } + let mut body = json!({"output": {"message": {"role": "assistant", + "content": encode_content(&output.content, &mut diagnostics, policy)?}}, + "stopReason": encode_stop_reason(output.stop_reason.unwrap_or(StopReason::EndTurn))?, + "usage": encode_usage(&response.usage)?}); + if let Some(native) = response + .extensions + .fields + .get(NATIVE_RESPONSE) + .and_then(Value::as_object) + { + for key in [ + "metrics", + "additionalModelResponseFields", + "trace", + "performanceConfig", + "serviceTier", + ] { + if let Some(v) = native.get(key) { + body[key] = v.clone(); + } + } + if let Some(reason) = native.get("stopReason").and_then(Value::as_str) + && Some(decode_stop_reason(reason)) == output.stop_reason + { + body["stopReason"] = reason.into(); + } + } + Ok(EncodedResponse { + body: embed_preservation(body, &response.preservation, policy), + diagnostics, + }) + } +} + +fn decode_content( + blocks: &[Value], + path: &str, + diagnostics: &mut Vec, + policy: &TranslationPolicy, +) -> Result> { + blocks + .iter() + .enumerate() + .map(|(i, block)| decode_block(block, &format!("{path}[{i}]"), diagnostics, policy)) + .collect() +} + +fn decode_block( + block: &Value, + path: &str, + diagnostics: &mut Vec, + policy: &TranslationPolicy, +) -> Result { + let obj = object(block, path)?; + if obj.len() != 1 { + return Err(invalid(path, "content block must contain one union member")); + } + if let Some(text) = obj.get("text") { + return Ok(ContentBlock::Text { + text: string(text, path)?.into(), + }); + } + if let Some(tool) = obj.get("toolUse") { + let tool = object(tool, path)?; + return Ok(ContentBlock::ToolCall(ToolCall { + id: required_string(tool, "toolUseId", path)?.into(), + name: required_string(tool, "name", path)?.into(), + arguments: required(tool, "input", path)?.clone(), + })); + } + if let Some(tool) = obj.get("toolResult") { + let tool = object(tool, path)?; + let mut content = Vec::new(); + for (i, value) in array(required(tool, "content", path)?, path)? + .iter() + .enumerate() + { + let p = format!("{path}.toolResult.content[{i}]"); + if object(value, &p)?.len() != 1 { + return Err(invalid(&p, "tool result must contain one union member")); + } + if value.get("json").is_some() { + content.push(ContentBlock::Text { + text: value["json"].to_string(), + }); + } else { + content.push(decode_block(value, &p, diagnostics, policy)?); + } + } + let is_error = match tool.get("status").map(|v| string(v, path)).transpose()? { + Some("error") => Some(true), + Some("success") => Some(false), + None => None, + _ => return Err(invalid(path, "invalid tool result status")), + }; + return Ok(ContentBlock::ToolResult(ToolResult { + tool_call_id: required_string(tool, "toolUseId", path)?.into(), + content, + is_error, + })); + } + if let Some(reasoning) = obj.get("reasoningContent") { + let reasoning = object(reasoning, path)?; + if reasoning.len() != 1 { + return Err(invalid(path, "reasoning must contain one union member")); + } + if let Some(text) = reasoning.get("reasoningText") { + let text = object(text, path)?; + return Ok(ContentBlock::Reasoning { + text: required_string(text, "text", path)?.into(), + signature: text + .get("signature") + .map(|v| string(v, path).map(str::to_owned)) + .transpose()?, + details: Vec::new(), + }); + } + if reasoning.contains_key("redactedContent") { + return Ok(ContentBlock::Unknown { + provider: WireFormat::BedrockConverse.into(), + raw: block.clone(), + }); + } + return Err(invalid(path, "invalid reasoning content")); + } + if let Some(image) = obj.get("image") { + let image = object(image, path)?; + let format = required_string(image, "format", path)?; + if let Some(data) = image.get("source").and_then(|s| s.get("bytes")) { + return Ok(ContentBlock::Image { + source: ImageSource::Base64 { + media_type: Some(format!("image/{format}")), + data: string(data, path)?.into(), + }, + }); + } + } + push_unknown_field(diagnostics, policy, path)?; + Ok(ContentBlock::Unknown { + provider: WireFormat::BedrockConverse.into(), + raw: block.clone(), + }) +} + +fn encode_content( + blocks: &[ContentBlock], + diagnostics: &mut Vec, + policy: &TranslationPolicy, +) -> Result> { + blocks + .iter() + .map(|b| encode_block(b, diagnostics, policy)) + .collect() +} + +fn encode_system_block( + block: &ContentBlock, + diagnostics: &mut Vec, + policy: &TranslationPolicy, +) -> Result { + match block { + ContentBlock::Text { .. } => encode_block(block, diagnostics, policy), + ContentBlock::Unknown { provider, raw } + if provider.as_str() == WireFormat::BedrockConverse.as_str() => + { + Ok(raw.clone()) + } + _ => Err(invalid("$.system", "unsupported system content")), + } +} + +fn encode_block( + block: &ContentBlock, + diagnostics: &mut Vec, + policy: &TranslationPolicy, +) -> Result { + Ok(match block { + ContentBlock::Text { text } | ContentBlock::Refusal { text } => json!({"text": text}), + ContentBlock::Reasoning { + text, + signature, + details, + } => { + let mut native_signature = String::new(); + let mut redacted = Vec::new(); + for detail in details { + match detail.get("type").and_then(Value::as_str) { + Some("bedrock.signature_delta") => { + if let Some(s) = detail.get("signature").and_then(Value::as_str) { + native_signature.push_str(s); + } + } + Some("bedrock.redacted_content") => { + if let Some(v) = detail.get("data") { + redacted.push(v.clone()); + } + } + _ => push_lossy( + diagnostics, + policy, + "opaque reasoning details have no Bedrock mapping", + )?, + } + } + if !redacted.is_empty() { + if !text.is_empty() || signature.is_some() || !native_signature.is_empty() { + return Err(invalid( + "$.reasoningContent", + "redacted reasoning cannot be combined with reasoning text or signatures", + )); + } + let mut bytes = Vec::new(); + for fragment in redacted { + let encoded = string(&fragment, "$.reasoningContent.redactedContent")?; + bytes.extend(STANDARD.decode(encoded).map_err(|_| { + invalid("$.reasoningContent.redactedContent", "invalid base64") + })?); + } + json!({"reasoningContent": {"redactedContent": STANDARD.encode(bytes)}}) + } else { + let mut reasoning = json!({"text": text}); + if let Some(signature) = signature { + reasoning["signature"] = signature.clone().into(); + } else if !native_signature.is_empty() { + reasoning["signature"] = native_signature.into(); + } + json!({"reasoningContent": {"reasoningText": reasoning}}) + } + } + ContentBlock::ToolCall(call) => { + let input = if let Value::String(raw) = &call.arguments { + serde_json::from_str::(raw) + .map_err(|_| invalid("$.toolUse.input", "invalid tool arguments JSON"))? + } else { + call.arguments.clone() + }; + json!({"toolUse": {"toolUseId": call.id, "name": call.name, "input": input}}) + } + ContentBlock::ToolResult(result) => { + let mut tool = json!({"toolUseId": result.tool_call_id, "content": encode_content(&result.content, diagnostics, policy)?}); + if let Some(is_error) = result.is_error { + tool["status"] = if is_error { "error" } else { "success" }.into(); + } + json!({"toolResult": tool}) + } + ContentBlock::Image { + source: ImageSource::Base64 { media_type, data }, + } => { + let format = media_type + .as_deref() + .and_then(|m| m.strip_prefix("image/")) + .ok_or_else(|| invalid("$.content.image", "inline image requires its MIME type"))?; + let format = if format == "jpg" { "jpeg" } else { format }; + if !matches!(format, "png" | "jpeg" | "gif" | "webp") { + return Err(invalid("$.content.image", "unsupported image format")); + } + json!({"image": {"format": format, "source": {"bytes": data}}}) + } + ContentBlock::Image { + source: ImageSource::Url { url, .. }, + } if url.starts_with("data:") => { + let (media, data) = url[5..] + .split_once(";base64,") + .ok_or_else(|| invalid("$.content.image", "expected a base64 data URL"))?; + return encode_block( + &ContentBlock::Image { + source: ImageSource::Base64 { + media_type: Some(media.into()), + data: data.into(), + }, + }, + diagnostics, + policy, + ); + } + ContentBlock::Unknown { provider, raw } + if provider.as_str() == WireFormat::BedrockConverse.as_str() => + { + raw.clone() + } + ContentBlock::Unknown { provider, raw } => { + reject_responses_builtin_tool_item(provider, raw, WireFormat::BedrockConverse)?; + return Err(TranslationError::UnsupportedTranslation { + from: provider.clone(), + to: WireFormat::BedrockConverse.into(), + }); + } + _ => { + return Err(invalid( + "$.content", + "content has no Bedrock Converse mapping", + )); + } + }) +} + +fn decode_tool_choice(value: &Value) -> Result { + let obj = object(value, "$.toolConfig.toolChoice")?; + if obj.len() != 1 { + return Err(invalid( + "$.toolConfig.toolChoice", + "expected one union member", + )); + } + if obj.get("auto").is_some_and(Value::is_object) { + Ok(ToolChoice::Auto) + } else if obj.get("any").is_some_and(Value::is_object) { + Ok(ToolChoice::Required) + } else if let Some(tool) = obj.get("tool") { + Ok(ToolChoice::Tool { + name: required_string( + object(tool, "$.toolConfig.toolChoice.tool")?, + "name", + "$.toolConfig.toolChoice.tool", + )? + .into(), + }) + } else { + Err(invalid("$.toolConfig.toolChoice", "unknown tool choice")) + } +} + +fn encode_tool_choice(choice: &ToolChoice) -> Result { + match choice { + ToolChoice::Auto => Ok(json!({"auto": {}})), + ToolChoice::Required => Ok(json!({"any": {}})), + ToolChoice::Tool { name } => Ok(json!({"tool": {"name": name}})), + _ => Err(invalid( + "$.toolConfig.toolChoice", + "unsupported tool choice", + )), + } +} + +fn decode_role(role: &str, path: &str) -> Result { + match role { + "user" => Ok(Role::User), + "assistant" => Ok(Role::Assistant), + _ => Err(TranslationError::unsupported_role( + format!("{path}.role"), + role, + )), + } +} +fn encode_role(role: Role) -> &'static str { + if role == Role::Assistant { + "assistant" + } else { + "user" + } +} + +pub(super) fn decode_stop_reason(reason: &str) -> StopReason { + match reason { + "end_turn" | "stop_sequence" => StopReason::EndTurn, + "max_tokens" => StopReason::MaxTokens, + "tool_use" => StopReason::ToolUse, + "content_filtered" | "guardrail_intervened" => StopReason::ContentFilter, + "malformed_model_output" | "malformed_tool_use" => StopReason::Error, + _ => StopReason::Unknown, + } +} +fn encode_stop_reason(reason: StopReason) -> Result<&'static str> { + match reason { + StopReason::EndTurn => Ok("end_turn"), + StopReason::MaxTokens => Ok("max_tokens"), + StopReason::ToolUse => Ok("tool_use"), + StopReason::ContentFilter => Ok("content_filtered"), + StopReason::Error => Ok("malformed_model_output"), + StopReason::Unknown => Err(invalid( + "$.stopReason", + "unknown stop reason has no Bedrock mapping", + )), + } +} + +pub(super) fn decode_usage(value: &Value) -> Result { + let obj = object(value, "$.usage")?; + Ok(Usage { + input_tokens: Some(unsigned( + required(obj, "inputTokens", "$.usage")?, + "$.usage.inputTokens", + )?), + output_tokens: Some(unsigned( + required(obj, "outputTokens", "$.usage")?, + "$.usage.outputTokens", + )?), + total_tokens: Some(unsigned( + required(obj, "totalTokens", "$.usage")?, + "$.usage.totalTokens", + )?), + cache: Usage::cache_details( + optional_u64(obj, "cacheReadInputTokens", "$.usage")?, + optional_u64(obj, "cacheWriteInputTokens", "$.usage")?, + ), + ..Usage::default() + }) +} +pub(super) fn encode_usage(usage: &Usage) -> Result { + let input = usage.input_tokens.ok_or_else(|| { + invalid( + "$.usage.inputTokens", + "Bedrock requires reported input tokens", + ) + })?; + let output = usage.output_tokens.ok_or_else(|| { + invalid( + "$.usage.outputTokens", + "Bedrock requires reported output tokens", + ) + })?; + let total = usage.total_tokens.unwrap_or_else(|| { + input + .saturating_add(output) + .saturating_add(usage.cached_input_tokens().unwrap_or(0)) + .saturating_add(usage.cache_creation_input_tokens().unwrap_or(0)) + }); + let mut value = json!({"inputTokens": input, "outputTokens": output, "totalTokens": total}); + if let Some(n) = usage.cached_input_tokens() { + value["cacheReadInputTokens"] = n.into(); + } + if let Some(n) = usage.cache_creation_input_tokens() { + value["cacheWriteInputTokens"] = n.into(); + } + Ok(value) +} + +pub(crate) fn request_projection_diagnostics( + request: &LlmRequest, + target: &FormatId, + policy: &TranslationPolicy, +) -> Result> { + let mut diagnostics = Vec::new(); + if target.as_str() == WireFormat::BedrockConverse.as_str() { + return Ok(diagnostics); + } + if let Some(native) = request + .extensions + .fields + .get(NATIVE_REQUEST) + .and_then(Value::as_object) + { + for key in native.keys().filter(|key| { + !matches!( + key.as_str(), + "messages" | "system" | "inferenceConfig" | "toolConfig" | "metadata" + ) + }) { + push_lossy( + &mut diagnostics, + policy, + format!("Bedrock {key} has no mapping to {target}"), + )?; + } + } + Ok(diagnostics) +} + +fn check_unknown( + obj: &Map, + known: &[&str], + path: &str, + diagnostics: &mut Vec, + policy: &TranslationPolicy, +) -> Result<()> { + for key in obj.keys() { + if !known.contains(&key.as_str()) { + push_unknown_field(diagnostics, policy, format!("{path}.{key}"))?; + } + } + Ok(()) +} +fn required<'a>(obj: &'a Map, key: &str, path: &str) -> Result<&'a Value> { + obj.get(key) + .ok_or_else(|| invalid(&format!("{path}.{key}"), "missing required field")) +} +fn required_string<'a>(obj: &'a Map, key: &str, path: &str) -> Result<&'a str> { + string(required(obj, key, path)?, &format!("{path}.{key}")) +} +fn string<'a>(value: &'a Value, path: &str) -> Result<&'a str> { + value + .as_str() + .ok_or_else(|| invalid(path, "expected string")) +} +fn array<'a>(value: &'a Value, path: &str) -> Result<&'a Vec> { + value + .as_array() + .ok_or_else(|| invalid(path, "expected array")) +} +fn unsigned(value: &Value, path: &str) -> Result { + value + .as_u64() + .ok_or_else(|| invalid(path, "expected non-negative integer")) +} + +fn optional_u64(obj: &Map, key: &str, path: &str) -> Result> { + obj.get(key) + .map(|v| unsigned(v, &format!("{path}.{key}"))) + .transpose() +} +fn optional_f64(obj: &Map, key: &str, path: &str) -> Result> { + obj.get(key) + .map(|v| { + v.as_f64() + .ok_or_else(|| invalid(&format!("{path}.{key}"), "expected number")) + }) + .transpose() +} +fn invalid(path: &str, message: &str) -> TranslationError { + TranslationError::InvalidValue { + path: path.into(), + message: message.into(), + } +} diff --git a/crates/switchyard-translation/src/codecs/bedrock/stream.rs b/crates/switchyard-translation/src/codecs/bedrock/stream.rs new file mode 100644 index 000000000..e9055d9cc --- /dev/null +++ b/crates/switchyard-translation/src/codecs/bedrock/stream.rs @@ -0,0 +1,557 @@ +// SPDX-FileCopyrightText: Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved. +// SPDX-License-Identifier: Apache-2.0 + +//! ConverseStream JSON events. AWS EventStream binary framing belongs to the host. + +use std::collections::{BTreeMap, BTreeSet}; + +use serde::{Deserialize, Serialize}; +use serde_json::{Value, json}; + +use crate::LlmResponseChunk; +use crate::codecs::stream::{StreamCodec, StreamTranslationState, record_source_identity}; +use crate::{FormatId, WireFormat}; + +use super::buffered::{decode_usage, encode_usage}; + +#[derive(Clone, Copy, Debug, PartialEq, Eq, Serialize, Deserialize)] +enum BlockKind { + Text, + Reasoning, + Tool, +} + +#[derive(Clone, Debug, Default, PartialEq, Serialize, Deserialize)] +pub(crate) struct BedrockStreamState { + started: bool, + stopped: bool, + metadata: bool, + decoded_blocks: BTreeMap, + closed_blocks: BTreeSet, + active: Option<(usize, BlockKind)>, + active_source_index: Option, + active_tool: Option, + replayed_stop: bool, + closed_tools: BTreeSet, +} + +/// Stream codec for ConverseStream's de-framed JSON union events. +pub struct BedrockConverseStreamCodec; + +impl StreamCodec for BedrockConverseStreamCodec { + fn format(&self) -> FormatId { + WireFormat::BedrockConverse.into() + } + + fn decode_event( + &self, + state: &mut StreamTranslationState, + event: &Value, + ) -> Vec { + match decode(state, event) { + Ok(chunks) => chunks, + Err(message) => vec![LlmResponseChunk::DecodeError { message }], + } + } + + fn encode_event( + &self, + state: &mut StreamTranslationState, + event: LlmResponseChunk, + ) -> Vec { + encode(state, event) + } + + fn observe_replayed_event( + &self, + state: &mut StreamTranslationState, + raw: &Value, + normalized: Vec, + ) { + for chunk in normalized { + drop(encode(state, chunk)); + } + let index = raw + .get("contentBlockStart") + .or_else(|| raw.get("contentBlockDelta")) + .and_then(|block| block.get("contentBlockIndex")) + .and_then(Value::as_u64) + .and_then(|index| usize::try_from(index).ok()); + if let Some(index) = index { + if let Some((active, _)) = state.bedrock.active.as_mut() { + *active = index; + } + if let Some(tool_index) = state.bedrock.active_tool + && let Some(tool) = state.tool_states.get_mut(&tool_index) + { + tool.content_index = Some(index); + } + state.next_content_index = state.next_content_index.max(index.saturating_add(1)); + } + if raw.get("contentBlockStop").is_some() { + state.bedrock.active = None; + state.bedrock.active_tool = None; + } + if raw.get("messageStop").is_some() { + state.bedrock.replayed_stop = true; + } + if raw.get("metadata").is_some() { + state.finished = true; + } + } + + fn finish(&self, state: &mut StreamTranslationState) -> Vec { + finish(state) + } +} + +fn decode( + state: &mut StreamTranslationState, + event: &Value, +) -> Result, String> { + let obj = event.as_object().ok_or("Bedrock event must be an object")?; + if obj.len() != 1 { + return Err("Bedrock event must contain one union member".into()); + } + let (key, payload) = obj.iter().next().ok_or("empty Bedrock event")?; + let payload = payload + .as_object() + .ok_or("Bedrock event payload must be an object")?; + if state.bedrock.metadata { + return Err("Bedrock event follows terminal metadata".into()); + } + if key.ends_with("Exception") { + return Ok(vec![LlmResponseChunk::StreamError { + message: format!( + "Bedrock {key}: {}", + payload + .get("message") + .and_then(Value::as_str) + .unwrap_or("upstream stream failed") + ), + }]); + } + if key == "messageStart" { + if state.bedrock.started { + return Err("duplicate Bedrock messageStart".into()); + } + if payload.get("role").and_then(Value::as_str) != Some("assistant") { + return Err("Bedrock response role must be assistant".into()); + } + state.bedrock.started = true; + state.saw_message_start = true; + return Ok(vec![LlmResponseChunk::MessageStart { + id: None, + model: None, + }]); + } + if !state.bedrock.started { + return Err("Bedrock event precedes messageStart".into()); + } + if key == "metadata" { + if !state.bedrock.stopped { + return Err("Bedrock metadata precedes messageStop".into()); + } + let usage = payload + .get("usage") + .ok_or("Bedrock metadata is missing usage")?; + state.usage = decode_usage(usage).map_err(|e| e.to_string())?; + state.bedrock.metadata = true; + state.saw_backend_usage = true; + return Ok(vec![LlmResponseChunk::Usage(state.usage.clone())]); + } + if state.bedrock.stopped { + return Err("Bedrock content follows messageStop".into()); + } + if key == "messageStop" { + if state + .bedrock + .decoded_blocks + .keys() + .any(|i| !state.bedrock.closed_blocks.contains(i)) + { + return Err("Bedrock messageStop precedes contentBlockStop".into()); + } + let reason = payload + .get("stopReason") + .and_then(Value::as_str) + .ok_or("Bedrock messageStop is missing stopReason")?; + state.bedrock.stopped = true; + state.stop_reason = Some(reason.into()); + let reason = match reason { + "content_filtered" | "guardrail_intervened" => "content_filter", + other => other, + }; + return Ok(vec![LlmResponseChunk::MessageStop { + reason: Some(reason.into()), + }]); + } + let index = payload + .get("contentBlockIndex") + .and_then(Value::as_u64) + .and_then(|n| usize::try_from(n).ok()) + .ok_or("Bedrock event is missing a valid contentBlockIndex")?; + if state.bedrock.closed_blocks.contains(&index) { + return Err("Bedrock event uses a closed content block".into()); + } + match key.as_str() { + "contentBlockStart" => { + if state.bedrock.decoded_blocks.contains_key(&index) { + return Err("duplicate Bedrock contentBlockStart".into()); + } + let start = payload + .get("start") + .and_then(Value::as_object) + .ok_or("Bedrock contentBlockStart is missing start")?; + if start.len() != 1 { + return Err("Bedrock start must contain one union member".into()); + } + let tool = start + .get("toolUse") + .and_then(Value::as_object) + .ok_or("unsupported Bedrock contentBlockStart")?; + let id = tool + .get("toolUseId") + .and_then(Value::as_str) + .filter(|s| !s.is_empty()) + .ok_or("Bedrock tool start is missing toolUseId")?; + let name = tool + .get("name") + .and_then(Value::as_str) + .filter(|s| !s.is_empty()) + .ok_or("Bedrock tool start is missing name")?; + state.bedrock.decoded_blocks.insert(index, BlockKind::Tool); + state.decoded_tool_call = true; + Ok(vec![LlmResponseChunk::ToolCallDelta { + index, + id: Some(id.into()), + name: Some(name.into()), + arguments_delta: None, + }]) + } + "contentBlockDelta" => { + let delta = payload + .get("delta") + .and_then(Value::as_object) + .ok_or("Bedrock content delta is missing delta")?; + if delta.len() != 1 { + return Err("Bedrock delta must contain one union member".into()); + } + let (kind, chunk) = if let Some(text) = delta.get("text") { + ( + BlockKind::Text, + LlmResponseChunk::TextDelta { + index, + text: text + .as_str() + .ok_or("Bedrock text delta must be a string")? + .into(), + }, + ) + } else if let Some(tool) = delta.get("toolUse") { + if state.bedrock.decoded_blocks.get(&index) != Some(&BlockKind::Tool) { + return Err("Bedrock tool delta precedes tool start".into()); + } + ( + BlockKind::Tool, + LlmResponseChunk::ToolCallDelta { + index, + id: None, + name: None, + arguments_delta: Some( + tool.get("input") + .and_then(Value::as_str) + .ok_or("Bedrock tool input delta must be a string")? + .into(), + ), + }, + ) + } else if let Some(reasoning) = delta.get("reasoningContent") { + let reasoning = reasoning + .as_object() + .ok_or("Bedrock reasoning delta must be an object")?; + if reasoning.len() != 1 { + return Err("Bedrock reasoning delta must contain one union member".into()); + } + let chunk = if let Some(text) = reasoning.get("text") { + LlmResponseChunk::ReasoningDelta { + index, + text: text + .as_str() + .ok_or("Bedrock reasoning text must be a string")? + .into(), + } + } else if let Some(signature) = reasoning.get("signature") { + LlmResponseChunk::ReasoningDetailsDelta { + index, + text: String::new(), + details: vec![ + json!({"type": "bedrock.signature_delta", "signature": signature.as_str().ok_or("Bedrock signature must be a string")?}), + ], + } + } else if let Some(data) = reasoning.get("redactedContent") { + LlmResponseChunk::ReasoningDetailsDelta { + index, + text: String::new(), + details: vec![ + json!({"type": "bedrock.redacted_content", "data": data.as_str().ok_or("Bedrock redactedContent must be base64 text")?}), + ], + } + } else { + return Err("unsupported Bedrock reasoning delta".into()); + }; + (BlockKind::Reasoning, chunk) + } else { + return Err("unsupported Bedrock content delta".into()); + }; + if state + .bedrock + .decoded_blocks + .get(&index) + .is_some_and(|existing| existing != &kind) + { + return Err("Bedrock delta changes content block kind".into()); + } + state.bedrock.decoded_blocks.insert(index, kind); + Ok(vec![chunk]) + } + "contentBlockStop" => { + if !state.bedrock.decoded_blocks.contains_key(&index) { + return Err("Bedrock stop names an unknown content block".into()); + } + state.bedrock.closed_blocks.insert(index); + Ok(Vec::new()) + } + _ => Err(format!("unsupported Bedrock stream event: {key}")), + } +} + +fn message_start(state: &mut StreamTranslationState, out: &mut Vec) { + if !state.emitted_message_start { + state.emitted_message_start = true; + out.push(json!({"messageStart": {"role": "assistant"}})); + } +} + +fn close_block(state: &mut StreamTranslationState, out: &mut Vec) { + if let Some((index, _)) = state.bedrock.active.take() { + out.push(json!({"contentBlockStop": {"contentBlockIndex": index}})); + } + if let Some(index) = state.bedrock.active_tool.take() { + state.bedrock.closed_tools.insert(index); + } +} + +fn content_index( + state: &mut StreamTranslationState, + kind: BlockKind, + source_index: usize, + out: &mut Vec, +) -> usize { + message_start(state, out); + if let Some((index, active)) = state.bedrock.active + && active == kind + && state.bedrock.active_source_index == Some(source_index) + { + return index; + } + close_block(state, out); + let index = state.next_content_index; + state.next_content_index += 1; + state.bedrock.active = Some((index, kind)); + state.bedrock.active_source_index = Some(source_index); + index +} + +fn encode(state: &mut StreamTranslationState, event: LlmResponseChunk) -> Vec { + if state.errored + || (state.finished + && !matches!( + event, + LlmResponseChunk::StreamError { .. } | LlmResponseChunk::DecodeError { .. } + )) + { + return Vec::new(); + } + let mut out = Vec::new(); + match event { + LlmResponseChunk::MessageStart { id, model } => { + record_source_identity(state, id, model); + message_start(state, &mut out); + } + LlmResponseChunk::TextDelta { text, index } => { + let index = content_index(state, BlockKind::Text, index, &mut out); + out.push( + json!({"contentBlockDelta": {"contentBlockIndex": index, "delta": {"text": text}}}), + ); + } + LlmResponseChunk::ReasoningDelta { text, index } => { + let index = content_index(state, BlockKind::Reasoning, index, &mut out); + out.push(json!({"contentBlockDelta": {"contentBlockIndex": index, "delta": {"reasoningContent": {"text": text}}}})); + } + LlmResponseChunk::ReasoningDetailsDelta { + text, + details, + index, + } => { + if !text.is_empty() { + out.extend(encode( + state, + LlmResponseChunk::ReasoningDelta { index, text }, + )); + } + for detail in details { + let value = match detail.get("type").and_then(Value::as_str) { + Some("bedrock.signature_delta") => { + detail.get("signature").map(|v| json!({"signature": v})) + } + Some("bedrock.redacted_content") => { + detail.get("data").map(|v| json!({"redactedContent": v})) + } + _ => { + return encode( + state, + LlmResponseChunk::DecodeError { + message: "opaque reasoning details have no Bedrock stream mapping" + .into(), + }, + ); + } + }; + if let Some(value) = value { + let index = content_index(state, BlockKind::Reasoning, index, &mut out); + out.push(json!({"contentBlockDelta": {"contentBlockIndex": index, "delta": {"reasoningContent": value}}})); + } + } + } + LlmResponseChunk::ToolCallDelta { + index, + id, + name, + arguments_delta, + } => { + if state.bedrock.closed_tools.contains(&index) { + return encode( + state, + LlmResponseChunk::DecodeError { + message: "Bedrock cannot resume a tool block after other content".into(), + }, + ); + } + let tool = state.tool_states.entry(index).or_default(); + if id.is_some() { + tool.id = id; + } + if name.is_some() { + tool.name = name; + } + if let Some(delta) = arguments_delta { + tool.pending_arguments.push_str(&delta); + } + if state.bedrock.active_tool.is_none() || state.bedrock.active_tool == Some(index) { + emit_tool(state, index, &mut out); + } + } + LlmResponseChunk::MessageStop { reason } => { + state.stop_reason = reason.or_else(|| state.stop_reason.clone()); + } + LlmResponseChunk::Usage(usage) => { + state.usage = usage; + state.saw_backend_usage = true; + } + LlmResponseChunk::StreamError { message } | LlmResponseChunk::DecodeError { message } => { + state.errored = true; + state.finished = true; + out.push(json!({"modelStreamErrorException": {"message": message}})); + } + } + out +} + +fn emit_tool(state: &mut StreamTranslationState, index: usize, out: &mut Vec) { + let Some(tool) = state.tool_states.get(&index) else { + return; + }; + let (Some(id), Some(name)) = (tool.id.as_ref(), tool.name.as_ref()) else { + return; + }; + let id = id.clone(); + let name = name.clone(); + if !tool.started { + message_start(state, out); + close_block(state, out); + let content_index = state.next_content_index; + state.next_content_index += 1; + state.bedrock.active = Some((content_index, BlockKind::Tool)); + state.bedrock.active_tool = Some(index); + let tool = state.tool_states.entry(index).or_default(); + tool.started = true; + tool.content_index = Some(content_index); + out.push(json!({"contentBlockStart": {"contentBlockIndex": content_index, "start": {"toolUse": {"toolUseId": id, "name": name}}}})); + } + let tool = state.tool_states.entry(index).or_default(); + if !tool.pending_arguments.is_empty() { + let delta = std::mem::take(&mut tool.pending_arguments); + tool.arguments.push_str(&delta); + out.push(json!({"contentBlockDelta": {"contentBlockIndex": tool.content_index, "delta": {"toolUse": {"input": delta}}}})); + } +} + +fn finish(state: &mut StreamTranslationState) -> Vec { + if state.finished || state.errored { + return Vec::new(); + } + let mut out = Vec::new(); + message_start(state, &mut out); + // Bedrock blocks are sequential. Parallel tools after the first wait until it closes. + let pending = state.tool_states.keys().copied().collect::>(); + for index in pending { + if state.bedrock.active_tool != Some(index) + && state + .tool_states + .get(&index) + .is_some_and(|tool| tool.started) + { + continue; + } + emit_tool(state, index, &mut out); + if state + .tool_states + .get(&index) + .is_some_and(|tool| !tool.pending_arguments.is_empty() || !tool.started) + { + state.errored = true; + state.finished = true; + return vec![ + json!({"modelStreamErrorException": {"message": "tool stream ended without a tool ID and name"}}), + ]; + } + } + close_block(state, &mut out); + if !state.bedrock.replayed_stop { + let reason = match state.stop_reason.as_deref() { + Some("length" | "max_tokens") => "max_tokens", + Some("tool_calls" | "function_call" | "tool_use") => "tool_use", + Some("content_filter" | "content_filtered") => "content_filtered", + Some("guardrail_intervened") => "guardrail_intervened", + Some("stop_sequence") => "stop_sequence", + Some("malformed_model_output" | "malformed_tool_use") => "malformed_model_output", + _ => "end_turn", + }; + out.push(json!({"messageStop": {"stopReason": reason}})); + } + match encode_usage(&state.usage) { + Ok(usage) => out.push(json!({"metadata": {"usage": usage}})), + Err(error) => { + state.errored = true; + state.finished = true; + return vec![json!({"modelStreamErrorException": {"message": error.to_string()}})]; + } + } + state.finished = true; + out +} + +pub(crate) fn saw_terminal(state: &StreamTranslationState) -> bool { + state.bedrock.metadata +} diff --git a/crates/switchyard-translation/src/codecs/mod.rs b/crates/switchyard-translation/src/codecs/mod.rs index 3b36a0bb8..73fd3684b 100644 --- a/crates/switchyard-translation/src/codecs/mod.rs +++ b/crates/switchyard-translation/src/codecs/mod.rs @@ -4,6 +4,7 @@ //! Buffered wire-format codecs that translate between provider JSON and IR. pub mod anthropic; +pub mod bedrock; pub(crate) mod common; pub mod openai_chat; mod openai_media; diff --git a/crates/switchyard-translation/src/codecs/stream.rs b/crates/switchyard-translation/src/codecs/stream.rs index db83015a7..cc891d512 100644 --- a/crates/switchyard-translation/src/codecs/stream.rs +++ b/crates/switchyard-translation/src/codecs/stream.rs @@ -11,6 +11,7 @@ use serde_json::{Map, Value, json}; use crate::LlmResponseChunk; use crate::codecs::anthropic::AnthropicMessagesStreamCodec; +use crate::codecs::bedrock::BedrockConverseStreamCodec; use crate::codecs::openai_chat::OpenAiChatStreamCodec; use crate::codecs::responses::OpenAiResponsesStreamCodec; use crate::engine::{FormatRegistry, TranslationEngine}; @@ -89,6 +90,8 @@ pub struct StreamTranslationState { pub(crate) reasoning_block_index: Option, pub(crate) reasoning_block_started: bool, + #[serde(default)] + pub(crate) bedrock: crate::codecs::bedrock::stream::BedrockStreamState, } // One Responses reasoning output item under construction by the encoder. @@ -231,6 +234,7 @@ impl StreamCodecRegistry { registry.register(OpenAiChatStreamCodec); registry.register(AnthropicMessagesStreamCodec); registry.register(OpenAiResponsesStreamCodec); + registry.register(BedrockConverseStreamCodec); registry } diff --git a/crates/switchyard-translation/src/engine.rs b/crates/switchyard-translation/src/engine.rs index 3399c769e..4f181a5c0 100644 --- a/crates/switchyard-translation/src/engine.rs +++ b/crates/switchyard-translation/src/engine.rs @@ -11,6 +11,7 @@ use serde_json::Value; use crate::LlmResponseStreamEvent; use crate::codecs::FormatCodec; use crate::codecs::anthropic::AnthropicMessagesCodec; +use crate::codecs::bedrock::BedrockConverseCodec; use crate::codecs::openai_chat::OpenAiChatCodec; use crate::codecs::responses::OpenAiResponsesCodec; use crate::codecs::stream::{ @@ -61,6 +62,7 @@ impl FormatRegistry { registry.register(OpenAiChatCodec); registry.register(AnthropicMessagesCodec); registry.register(OpenAiResponsesCodec); + registry.register(BedrockConverseCodec); registry } @@ -138,13 +140,16 @@ impl TranslationEngine { policy: &TranslationPolicy, ) -> Result { let target = target.into(); + let mut diagnostics = + crate::codecs::bedrock::request_projection_diagnostics(request, &target, policy)?; let encoded = self .registry .codec(target)? .encode_request(request, policy)?; + diagnostics.extend(encoded.diagnostics); Ok(TranslationOutput { body: encoded.body, - diagnostics: encoded.diagnostics, + diagnostics, }) } @@ -162,6 +167,14 @@ impl TranslationEngine { .registry .codec(source.clone())? .decode_request(body, policy)?; + let mut decoded = decoded; + decoded + .diagnostics + .extend(crate::codecs::bedrock::request_projection_diagnostics( + &decoded.request, + &target, + policy, + )?); let encoded = self .registry .codec(target.clone())? diff --git a/crates/switchyard-translation/src/helpers.rs b/crates/switchyard-translation/src/helpers.rs index 95ebb75e7..17f5a9e48 100644 --- a/crates/switchyard-translation/src/helpers.rs +++ b/crates/switchyard-translation/src/helpers.rs @@ -83,15 +83,15 @@ pub fn encode_aggregated_response_with_extensions( &DEFAULT_TRANSLATION_POLICY, )? .body; - if let (Some(model), Value::Object(object)) = (served_model, &mut body) { + if wire_format != WireFormat::BedrockConverse + && let (Some(model), Value::Object(object)) = (served_model, &mut body) + { object.insert("model".to_string(), Value::String(model.to_string())); } Ok(body) } -/// A stream of wire-format event objects in one format — the unframed body of an -/// SSE response. The serving layer frames each `Value` (e.g. as an SSE -/// `data:`/`event:` block). +/// A stream of unframed provider JSON events. The host supplies SSE or AWS EventStream framing. pub type RawEventStream = Pin> + Send>>; @@ -192,7 +192,7 @@ pub fn encode_stream_with_extensions( &custom_tools, &mut custom_state, ) { - yield value; + if state.errored { Err(LlmStreamError::Upstream(value))?; } else { yield value; } } } }; @@ -223,6 +223,7 @@ fn stamp_streamed_response_model( response.insert("model".to_string(), Value::String(served_model.to_string())); } } + WireFormat::BedrockConverse => {} WireFormat::AnthropicMessages => { if let Some(message) = event.get_mut("message").and_then(Value::as_object_mut) { message.insert("model".to_string(), Value::String(served_model.to_string())); @@ -231,6 +232,48 @@ fn stamp_streamed_response_model( } } +/// Decodes already de-framed provider JSON events into neutral stream events. +/// +/// The host validates and removes the carrier framing. Bedrock requires both +/// `messageStop` and the final usage `metadata`; an incomplete stream fails at EOF. +pub fn decode_event_stream( + events: S, + source: WireFormat, +) -> std::result::Result +where + S: Stream> + Send + 'static, +{ + let format: FormatId = source.into(); + let codec = StreamCodecRegistry::with_builtins() + .codec(format.clone()) + .map_err(|e| LlmClientError::ResponseTranslation(e.to_string()))?; + let mut state = StreamTranslationState { + source: Some(format.clone()), + ..Default::default() + }; + let stream = try_stream! { + futures::pin_mut!(events); + let mut terminal = false; + while let Some(event) = events.next().await { + let value = event?; + let normalized = codec.decode_event(&mut state, &value); + terminal |= if source == WireFormat::BedrockConverse { + crate::codecs::bedrock::stream::saw_terminal(&state) + } else { sse::is_terminal_event(source, &value) }; + let failed = normalized.iter().any(|chunk| matches!(chunk, + crate::LlmResponseChunk::DecodeError { .. } | crate::LlmResponseChunk::StreamError { .. })); + yield LlmResponseStreamEvent::preserved(format.clone(), value, normalized); + if failed { return; } + } + if !terminal { + Err(LlmClientError::ResponseTranslation( + "provider event stream ended before its terminal event".into(), + ))?; + } + }; + Ok(Box::pin(stream)) +} + /// Decodes provider SSE bytes into normalized stream events. /// /// Operates on raw bytes, not any HTTP client type: the caller adapts its @@ -248,6 +291,12 @@ pub fn decode_stream( where S: Stream, LlmClientError>> + Send + 'static, { + if source == WireFormat::BedrockConverse { + return Err(LlmClientError::Configuration { + message: "Bedrock requires de-framed AWS EventStream JSON; use decode_event_stream" + .into(), + }); + } let marker = sse::done_marker(source); let source_format: FormatId = source.into(); // The source is always a built-in wire format, so this lookup cannot fail; a diff --git a/crates/switchyard-translation/src/sse.rs b/crates/switchyard-translation/src/sse.rs index bf24aabbb..fc3a8fdc2 100644 --- a/crates/switchyard-translation/src/sse.rs +++ b/crates/switchyard-translation/src/sse.rs @@ -53,6 +53,7 @@ pub(crate) fn is_terminal_event(format: WireFormat, event: &Value) -> bool { .and_then(Value::as_str) .is_some() }), + WireFormat::BedrockConverse => false, WireFormat::AnthropicMessages => { event.get("type").and_then(Value::as_str) == Some("message_stop") } diff --git a/crates/switchyard-translation/tests/bedrock_translation.rs b/crates/switchyard-translation/tests/bedrock_translation.rs new file mode 100644 index 000000000..137309376 --- /dev/null +++ b/crates/switchyard-translation/tests/bedrock_translation.rs @@ -0,0 +1,562 @@ +// SPDX-FileCopyrightText: Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved. +// SPDX-License-Identifier: Apache-2.0 + +use futures::{StreamExt, executor::block_on, stream}; +use serde_json::{Value, json}; +use switchyard_protocol::{LlmClientError, LlmResponse, LlmResponseChunk, LlmResponseStreamEvent}; +use switchyard_translation::{ + ContentBlock, LossyConversionPolicy, PreservationPolicy, StreamTranslationState, ToolChoice, + TranslationEngine, TranslationPolicy, UnknownFieldPolicy, WireFormat, decode_event_stream, + decode_stream, encode_aggregated_response, encode_stream, +}; + +type TestResult = Result<(), Box>; +const BEDROCK: WireFormat = WireFormat::BedrockConverse; + +fn normalized() -> TranslationPolicy { + TranslationPolicy { + preservation: PreservationPolicy::Disabled, + ..TranslationPolicy::default() + } +} + +fn request() -> Value { + json!({ + "system": [{"text": "Be brief."}], + "messages": [ + {"role": "user", "content": [{"text": "weather?"}]}, + {"role": "assistant", "content": [{"toolUse": {"toolUseId": "call_1", "name": "weather", "input": {"city": "Taipei"}}}]}, + {"role": "user", "content": [{"toolResult": {"toolUseId": "call_1", "content": [{"text": "sunny"}], "status": "error"}}]} + ], + "inferenceConfig": {"maxTokens": 123, "temperature": 0.2, "topP": 0.8, "stopSequences": ["END"]}, + "toolConfig": {"tools": [{"toolSpec": {"name": "weather", "strict": true, "inputSchema": {"json": {"type": "object"}}}}], "toolChoice": {"tool": {"name": "weather"}}} + }) +} + +fn usage() -> Value { + json!({"inputTokens": 10, "outputTokens": 5, "totalTokens": 23, + "cacheReadInputTokens": 6, "cacheWriteInputTokens": 2}) +} + +fn events() -> Vec { + vec![ + json!({"messageStart": {"role": "assistant"}}), + json!({"contentBlockDelta": {"contentBlockIndex": 0, "delta": {"text": "Hello"}}}), + json!({"contentBlockStop": {"contentBlockIndex": 0}}), + json!({"messageStop": {"stopReason": "end_turn"}}), + json!({"metadata": {"usage": usage(), "metrics": {"latencyMs": 12}}}), + ] +} + +fn raw_events( + values: Vec, +) -> Result { + decode_event_stream(stream::iter(values.into_iter().map(Ok)), BEDROCK) +} + +fn translate_events( + values: &[Value], + target: WireFormat, +) -> Result, Box> { + let engine = TranslationEngine::default(); + let mut state = StreamTranslationState::new(BEDROCK, target); + let mut result = Vec::new(); + for value in values { + result.extend(engine.translate_event(&mut state, BEDROCK, target, value)?); + } + result.extend(engine.finish_stream(&mut state, target)?); + Ok(result) +} + +#[test] +fn bedrock_text_request_decodes() -> TestResult { + let output = TranslationEngine::default().decode_request( + "bedrock_converse", + &json!({"messages": [{"role": "user", "content": [{"text": "hello"}]}]}), + &TranslationPolicy::default(), + )?; + assert_eq!( + output.request.messages[0].text_content(""), + Some("hello".into()) + ); + assert_eq!(serde_json::to_value(BEDROCK)?, "bedrock_converse"); + assert!(output.request.model.is_none()); + Ok(()) +} + +#[test] +fn native_requests_and_cross_format_tool_histories() -> TestResult { + let engine = TranslationEngine::default(); + let body = request(); + let decoded = engine + .decode_request(BEDROCK, &body, &normalized())? + .request; + assert_eq!( + engine + .encode_request(BEDROCK, &decoded, &normalized())? + .body, + body + ); + assert_eq!(decoded.output.max_output_tokens, Some(123)); + assert_eq!(decoded.tools[0].strict, Some(true)); + assert!( + matches!(&decoded.messages[2].content[0], ContentBlock::ToolResult(result) if result.is_error == Some(true)) + ); + for target in [ + WireFormat::OpenAiChat, + WireFormat::OpenAiResponses, + WireFormat::AnthropicMessages, + ] { + let projected = engine.translate_request(BEDROCK, target, &body, &normalized())?; + let back = engine + .translate_request(target, BEDROCK, &projected.body, &normalized())? + .body; + assert_eq!( + back["messages"][1]["content"][0]["toolUse"]["toolUseId"], "call_1", + "{target}: {back}" + ); + assert_eq!( + back["messages"][1]["content"][0]["toolUse"]["input"], + json!({"city": "Taipei"}) + ); + assert_eq!(back["inferenceConfig"]["maxTokens"], 123); + // OpenAI Responses does not provide a stop-sequence setting. + if target != WireFormat::OpenAiResponses { + assert_eq!(back["inferenceConfig"]["stopSequences"], json!(["END"])); + } + } + let mut rich = body.clone(); + rich["guardrailConfig"] = json!({"guardrailIdentifier": "guardrail", "guardrailVersion": "1"}); + rich["futureControl"] = json!({"enabled": true}); + assert_eq!( + engine + .translate_request(BEDROCK, BEDROCK, &rich, &TranslationPolicy::default())? + .body, + rich + ); + let strict = TranslationPolicy { + lossy_conversion_policy: LossyConversionPolicy::Reject, + ..normalized() + }; + assert!( + engine + .translate_request(BEDROCK, WireFormat::OpenAiChat, &rich, &strict) + .is_err() + ); + let mut json_result = body.clone(); + json_result["messages"][2]["content"][0]["toolResult"]["content"] = + json!([{"json": {"answer": 42}}]); + let chat = engine + .translate_request(BEDROCK, WireFormat::OpenAiChat, &json_result, &normalized())? + .body; + assert_eq!(chat["messages"][3]["role"], "tool"); + assert_eq!(chat["messages"][3]["tool_call_id"], "call_1"); + assert_eq!(chat["messages"][3]["content"], "{\"answer\":42}"); + assert_eq!(chat["messages"].as_array().map(Vec::len), Some(4)); + let mut no_tools = decoded; + no_tools.tool_choice = Some(ToolChoice::None); + assert!( + engine + .encode_request(BEDROCK, &no_tools, &normalized())? + .body + .get("toolConfig") + .is_none() + ); + Ok(()) +} + +#[test] +fn rejects_malformed_requests_and_reports_unsupported_controls() -> TestResult { + let engine = TranslationEngine::default(); + for body in [ + json!({"messages": "bad"}), + json!({"messages": [{"role": "mystery", "content": [{"text": "hello"}]}]}), + json!({"messages": [{"role": "user", "content": [{"text": "x", "image": {}}]}]}), + json!({"messages": [], "inferenceConfig": {"maxTokens": -1}}), + json!({"messages": [], "toolConfig": {"tools": [{"toolSpec": {"name": "t", "inputSchema": {"json": {}}}, "cachePoint": {"type": "default"}}]}}), + json!({"messages": [{"role": "assistant", "content": [{"reasoningContent": {"reasoningText": {"text": "x"}, "redactedContent": "YQ=="}}]}]}), + ] { + assert!( + engine + .decode_request(BEDROCK, &body, &normalized()) + .is_err(), + "{body}" + ); + } + let mut policy = normalized(); + policy.unknown_field_policy = UnknownFieldPolicy::Reject; + assert!( + engine + .decode_request(BEDROCK, &json!({"messages": [], "typo": true}), &policy) + .is_err() + ); + let mut ir = engine + .decode_request(BEDROCK, &request(), &normalized())? + .request; + ir.output.response_format = Some(json!({"type": "json_object"})); + assert!( + !engine + .encode_request(BEDROCK, &ir, &normalized())? + .diagnostics + .is_empty() + ); + policy.lossy_conversion_policy = LossyConversionPolicy::Reject; + assert!(engine.encode_request(BEDROCK, &ir, &policy).is_err()); + Ok(()) +} + +#[test] +fn buffered_responses_preserve_reasoning_usage_and_native_fields() -> TestResult { + let engine = TranslationEngine::default(); + let body = json!({"output": {"message": {"role": "assistant", "content": [ + {"reasoningContent": {"reasoningText": {"text": "consider", "signature": "signed"}}}, + {"text": "answer"}]}}, "stopReason": "guardrail_intervened", "usage": usage(), + "metrics": {"latencyMs": 12}, "trace": {"guardrail": {"action": "BLOCKED"}}}); + let mut malformed = body.clone(); + malformed["output"]["message"]["role"] = json!("user"); + assert!( + engine + .decode_response(BEDROCK, &malformed, &normalized()) + .is_err() + ); + malformed = body.clone(); + malformed["usage"] + .as_object_mut() + .ok_or("usage object")? + .remove("totalTokens"); + assert!( + engine + .decode_response(BEDROCK, &malformed, &normalized()) + .is_err() + ); + let decoded = engine + .decode_response(BEDROCK, &body, &normalized())? + .response; + assert_eq!( + engine + .encode_response(BEDROCK, &decoded, &normalized())? + .body, + body + ); + let chat = engine + .encode_response(WireFormat::OpenAiChat, &decoded, &normalized())? + .body; + let mut chat_with_reasoning = chat.clone(); + chat_with_reasoning["usage"] = json!({"prompt_tokens": 5, "completion_tokens": 10, "total_tokens": 15, + "completion_tokens_details": {"reasoning_tokens": 4}}); + let bedrock = engine + .translate_response( + WireFormat::OpenAiChat, + BEDROCK, + &chat_with_reasoning, + &normalized(), + )? + .body; + assert_eq!( + bedrock["usage"], + json!({"inputTokens": 5, "outputTokens": 10, "totalTokens": 15}) + ); + assert_eq!(chat["usage"]["prompt_tokens"], 18); + assert_eq!(chat["usage"]["completion_tokens"], 5); + assert_eq!(chat["choices"][0]["finish_reason"], "content_filter"); + assert_eq!(chat["choices"][0]["message"]["content"], "answer"); + assert_eq!( + chat["choices"][0]["message"]["reasoning_content"], + "consider" + ); + assert!( + encode_aggregated_response(&decoded, BEDROCK, Some("url-owned/model"))? + .get("model") + .is_none() + ); + assert!( + engine + .decode_response( + BEDROCK, + &json!({"modelStreamErrorException": {"message": "failed"}}), + &normalized() + ) + .is_err() + ); + Ok(()) +} + +#[test] +fn native_stream_replays_without_duplicate_terminal_events() -> TestResult { + let values = events(); + let replayed = block_on( + encode_stream( + raw_events(values.clone())?, + BEDROCK, + Some("url-owned/model".into()), + )? + .collect::>(), + ) + .into_iter() + .collect::, _>>()?; + assert_eq!(replayed, values); + let engine = TranslationEngine::default(); + let mut decoder = StreamTranslationState::new(BEDROCK, BEDROCK); + let mut encoder = StreamTranslationState::new(BEDROCK, BEDROCK); + for raw in [ + values[0].clone(), + json!({"contentBlockDelta": {"contentBlockIndex": 7, "delta": {"text": "partial"}}}), + ] { + let event = engine.decode_stream_event(&mut decoder, BEDROCK, raw.clone())?; + assert_eq!( + engine.encode_stream_event(&mut encoder, BEDROCK, event)?, + vec![raw] + ); + } + engine.encode_stream_event( + &mut encoder, + BEDROCK, + LlmResponseChunk::Usage(switchyard_protocol::Usage { + input_tokens: Some(1), + output_tokens: Some(2), + ..Default::default() + }) + .into(), + )?; + let finished = engine.finish_stream(&mut encoder, BEDROCK)?; + assert_eq!(finished[0]["contentBlockStop"]["contentBlockIndex"], 7); + assert_eq!(finished[2]["metadata"]["usage"]["totalTokens"], 3); + Ok(()) +} + +#[test] +fn stream_text_and_usage_translate_to_existing_formats() -> TestResult { + for target in [ + WireFormat::OpenAiChat, + WireFormat::OpenAiResponses, + WireFormat::AnthropicMessages, + ] { + let output = translate_events(&events(), target)?; + let aggregate = block_on( + LlmResponse::Stream(decode_event_stream( + stream::iter(output.into_iter().map(Ok)), + target, + )?) + .into_agg(), + )?; + assert_eq!( + aggregate.outputs[0].content, + vec![ContentBlock::Text { + text: "Hello".into() + }], + "{target}" + ); + assert_eq!(aggregate.usage.output_tokens, Some(5)); + assert_eq!(aggregate.usage.cached_input_tokens(), Some(6)); + } + Ok(()) +} + +#[test] +fn streaming_tools_keep_arguments_that_arrive_before_identity() -> TestResult { + let chunks = vec![ + LlmResponseChunk::TextDelta { + index: 0, + text: "calling tools".into(), + }, + LlmResponseChunk::ToolCallDelta { + index: 0, + id: None, + name: None, + arguments_delta: Some("{\"city\":".into()), + }, + LlmResponseChunk::ToolCallDelta { + index: 1, + id: Some("second".into()), + name: Some("weather".into()), + arguments_delta: Some("{}".into()), + }, + LlmResponseChunk::ToolCallDelta { + index: 0, + id: Some("first".into()), + name: Some("weather".into()), + arguments_delta: Some("\"Taipei\"}".into()), + }, + LlmResponseChunk::MessageStop { + reason: Some("tool_calls".into()), + }, + LlmResponseChunk::Usage(switchyard_protocol::Usage { + input_tokens: Some(1), + output_tokens: Some(2), + total_tokens: Some(3), + ..Default::default() + }), + ]; + let encoded = block_on( + encode_stream( + Box::pin(stream::iter(chunks.into_iter().map(|c| Ok(c.into())))), + BEDROCK, + None, + )? + .collect::>(), + ) + .into_iter() + .collect::, _>>()?; + let aggregate = block_on(LlmResponse::Stream(raw_events(encoded)?).into_agg())?; + let calls = aggregate.outputs[0] + .content + .iter() + .filter_map(|b| { + if let ContentBlock::ToolCall(c) = b { + Some(c) + } else { + None + } + }) + .collect::>(); + assert_eq!(calls.len(), 2); + assert!( + calls + .iter() + .any(|c| c.id == "first" && c.arguments == json!({"city": "Taipei"})) + ); + assert!( + calls + .iter() + .any(|c| c.id == "second" && c.arguments == json!({})) + ); + Ok(()) +} + +#[test] +fn reasoning_signatures_survive_stream_aggregation() -> TestResult { + let values = vec![ + json!({"messageStart": {"role": "assistant"}}), + json!({"contentBlockDelta": {"contentBlockIndex": 0, "delta": {"reasoningContent": {"text": "think"}}}}), + json!({"contentBlockDelta": {"contentBlockIndex": 0, "delta": {"reasoningContent": {"signature": "sig1"}}}}), + json!({"contentBlockDelta": {"contentBlockIndex": 0, "delta": {"reasoningContent": {"signature": "sig2"}}}}), + json!({"contentBlockStop": {"contentBlockIndex": 0}}), + json!({"messageStop": {"stopReason": "end_turn"}}), + json!({"metadata": {"usage": usage()}}), + ]; + assert_eq!(translate_events(&values, BEDROCK)?, values); + let mut redacted = values.clone(); + redacted.splice(1..4, [ + json!({"contentBlockDelta": {"contentBlockIndex": 0, "delta": {"reasoningContent": {"redactedContent": "YQ=="}}}}), + json!({"contentBlockDelta": {"contentBlockIndex": 0, "delta": {"reasoningContent": {"redactedContent": "Yg=="}}}}), + ]); + let redacted_aggregate = block_on(LlmResponse::Stream(raw_events(redacted)?).into_agg())?; + let body = TranslationEngine::default() + .encode_response(BEDROCK, &redacted_aggregate, &normalized())? + .body; + assert_eq!( + body["output"]["message"]["content"][0]["reasoningContent"]["redactedContent"], + "YWI=" + ); + let aggregate = block_on(LlmResponse::Stream(raw_events(values)?).into_agg())?; + let body = TranslationEngine::default() + .encode_response(BEDROCK, &aggregate, &normalized())? + .body; + assert_eq!( + body["output"]["message"]["content"][0]["reasoningContent"]["reasoningText"], + json!({"text": "think", "signature": "sig1sig2"}) + ); + Ok(()) +} + +#[test] +fn event_stream_errors_and_truncation_fail_without_success_terminal() -> TestResult { + for extra in [ + json!({"contentBlockDelta": {"contentBlockIndex": 0, "delta": {"text": "late"}}}), + json!({"modelStreamErrorException": {"message": "late"}}), + ] { + let mut values = events(); + values.push(extra); + let output = + block_on(encode_stream(raw_events(values)?, BEDROCK, None)?.collect::>()); + assert!(output.last().is_some_and(Result::is_err)); + } + let mut incomplete = events(); + incomplete.pop(); + assert!(block_on(LlmResponse::Stream(raw_events(incomplete)?).into_agg()).is_err()); + for values in [ + vec![ + json!({"messageStart": {"role": "assistant"}}), + json!({"throttlingException": {"message": "slow down"}}), + ], + vec![ + json!({"messageStart": {"role": "assistant"}}), + json!({"contentBlockDelta": {"delta": {"text": "bad"}}}), + ], + ] { + let output = + block_on(encode_stream(raw_events(values)?, BEDROCK, None)?.collect::>()); + assert!(output.last().is_some_and(Result::is_err)); + assert!( + !output + .iter() + .filter_map(|v| v.as_ref().ok()) + .any(|v| v.get("messageStop").is_some()) + ); + } + let incomplete_tool: Vec> = + vec![Ok(LlmResponseChunk::ToolCallDelta { + index: 0, + id: None, + name: None, + arguments_delta: Some("{}".into()), + } + .into())]; + let output = block_on( + encode_stream(Box::pin(stream::iter(incomplete_tool)), BEDROCK, None)?.collect::>(), + ); + assert!(output.last().is_some_and(Result::is_err)); + let unsupported = vec![LlmResponseChunk::ReasoningDetailsDelta { + index: 0, + text: String::new(), + details: vec![json!({"type": "anthropic.signature_delta", "signature": "opaque"})], + }]; + let output = block_on( + encode_stream( + Box::pin(stream::iter(unsupported.into_iter().map(|c| Ok(c.into())))), + BEDROCK, + None, + )? + .collect::>(), + ); + assert!(output.last().is_some_and(Result::is_err)); + let mut resume = StreamTranslationState::new(WireFormat::OpenAiChat, BEDROCK); + let engine = TranslationEngine::default(); + for chunk in [ + LlmResponseChunk::ToolCallDelta { + index: 0, + id: Some("tool".into()), + name: Some("t".into()), + arguments_delta: Some("{".into()), + }, + LlmResponseChunk::TextDelta { + index: 1, + text: "interleaved".into(), + }, + LlmResponseChunk::ToolCallDelta { + index: 0, + id: None, + name: None, + arguments_delta: Some("}".into()), + }, + ] { + engine.encode_stream_event(&mut resume, BEDROCK, chunk.into())?; + } + assert!(resume.errored); + let mut signature = StreamTranslationState::new(BEDROCK, WireFormat::AnthropicMessages); + let output = engine.encode_stream_event( + &mut signature, + WireFormat::AnthropicMessages, + LlmResponseChunk::ReasoningDetailsDelta { + index: 0, + text: String::new(), + details: vec![json!({"type": "bedrock.signature_delta", "signature": "opaque"})], + } + .into(), + )?; + assert!(signature.errored); + assert_eq!(output[0]["type"], "error"); + let bytes = stream::iter([Ok::, LlmClientError>(b"data: {}\n\n".to_vec())]); + assert!(decode_stream(bytes, BEDROCK).is_err()); + Ok(()) +} diff --git a/crates/switchyard-translation/tests/lossless_roundtrip.rs b/crates/switchyard-translation/tests/lossless_roundtrip.rs index db99a71fd..0a6b8c397 100644 --- a/crates/switchyard-translation/tests/lossless_roundtrip.rs +++ b/crates/switchyard-translation/tests/lossless_roundtrip.rs @@ -279,16 +279,13 @@ fn assert_embeds_original(body: &Value, group: &str, source: WireFormat, origina // Returns the JSON metadata key for a built-in wire format. fn format_key(format: WireFormat) -> &'static str { - match format { - WireFormat::OpenAiChat => "openai_chat", - WireFormat::AnthropicMessages => "anthropic_messages", - WireFormat::OpenAiResponses => "openai_responses", - } + format.as_str() } // Builds an intentionally broad request fixture for a provider format. fn request_fixture(format: WireFormat) -> Value { match format { + WireFormat::BedrockConverse => unreachable!("Bedrock preservation has separate fixtures"), WireFormat::OpenAiChat => json!({ "model": "gpt-5.2", "messages": [ @@ -544,6 +541,7 @@ fn request_fixture(format: WireFormat) -> Value { // Builds an intentionally broad response fixture for a provider format. fn response_fixture(format: WireFormat) -> Value { match format { + WireFormat::BedrockConverse => unreachable!("Bedrock preservation has separate fixtures"), WireFormat::OpenAiChat => json!({ "id": "chatcmpl_adversarial", "object": "chat.completion", diff --git a/crates/switchyard-translation/tests/request_translation.rs b/crates/switchyard-translation/tests/request_translation.rs index 3de0f5a47..9015eb855 100644 --- a/crates/switchyard-translation/tests/request_translation.rs +++ b/crates/switchyard-translation/tests/request_translation.rs @@ -282,6 +282,7 @@ fn native_abuse_identity_is_preserved_but_not_mapped() -> TestResult { "messages": [{"role": "user", "content": "hi"}], "max_tokens": 8 }), + WireFormat::BedrockConverse => unreachable!(), }; if source == WireFormat::AnthropicMessages { if let Some(identity) = identity { @@ -393,7 +394,7 @@ fn openai_target_prompt_preserves_native_request_fields() -> TestResult { WireFormat::OpenAiResponses => { expected["instructions"] = json!("target prompt\n\ncaller prompt"); } - WireFormat::AnthropicMessages => unreachable!(), + WireFormat::AnthropicMessages | WireFormat::BedrockConverse => unreachable!(), } let mut request = engine.decode_request(format, &body, &policy)?.request; @@ -665,7 +666,7 @@ fn anthropic_target_prompt_preserves_native_request_fields() -> TestResult { "messages": [{"role": "user", "content": "hi"}] }), WireFormat::OpenAiResponses => json!({"model": "route", "input": "hi"}), - WireFormat::AnthropicMessages => unreachable!(), + WireFormat::AnthropicMessages | WireFormat::BedrockConverse => unreachable!(), }; if has_fields { for (key, value) in fields.as_object().ok_or("expected fields object")? { From 3d043411ea478411723b3174ba1d9002b681e661 Mon Sep 17 00:00:00 2001 From: Ting-Hong Shieh Date: Sat, 3 Oct 2026 16:11:05 +0800 Subject: [PATCH 2/2] fix(translation): preserve Bedrock tool history and incomplete streams Signed-off-by: Ting-Hong Shieh --- crates/protocol/src/stream.rs | 2 +- crates/switchyard-translation/README.md | 12 +- .../src/codecs/anthropic/stream.rs | 7 +- .../src/codecs/bedrock/buffered.rs | 44 ++- .../src/codecs/bedrock/stream.rs | 2 + .../tests/bedrock_translation.rs | 282 +++++++++++++++++- 6 files changed, 328 insertions(+), 21 deletions(-) diff --git a/crates/protocol/src/stream.rs b/crates/protocol/src/stream.rs index 32e582f46..4c2775963 100644 --- a/crates/protocol/src/stream.rs +++ b/crates/protocol/src/stream.rs @@ -504,7 +504,7 @@ fn parse_tool_arguments(arguments: &str) -> Value { /// to a normalized [`StopReason`], covering the common OpenAI and Anthropic spellings. fn stop_reason_from_str(reason: Option<&str>) -> StopReason { match reason { - Some("length" | "max_tokens") => StopReason::MaxTokens, + Some("length" | "max_tokens" | "model_context_window_exceeded") => StopReason::MaxTokens, Some("tool_calls" | "function_call" | "tool_use") => StopReason::ToolUse, Some("content_filter" | "content_filtered" | "guardrail_intervened") => { StopReason::ContentFilter diff --git a/crates/switchyard-translation/README.md b/crates/switchyard-translation/README.md index ce3faf253..0618137c6 100644 --- a/crates/switchyard-translation/README.md +++ b/crates/switchyard-translation/README.md @@ -14,6 +14,12 @@ settings, inline images, stop reasons, and cache-token usage have neutral mappin become serialized JSON text when normalized; same-format preservation retains their original JSON. Native controls such as guardrails and additional model fields survive same-format request encoding. Cross-format projection diagnoses their loss and rejects it under strict loss policy. +Normalized encoding merges consecutive messages with the same Bedrock role in content order. +Tool history requires tool definitions. Disabling tools with history retains the required config +and reports a lossy conversion; strict loss policy rejects it. Without history, disabling tools +omits the config. Same-format preservation keeps the original request body. +Context-window exhaustion maps to the neutral token-limit stop reason so other formats mark the +response incomplete. Same-format preservation retains the original Bedrock stop reason. ConverseStream codecs operate on the JSON union events after the host removes and validates AWS EventStream framing. Feed those events to `decode_event_stream`, then use `encode_stream` for target @@ -23,8 +29,10 @@ truncation, malformed events, or provider exceptions fail the stream. Encoding r input and output tokens; an absent total is derived from those counts and cache details. Same-format preserved events replay unchanged, including native reasoning signatures. Aggregated -Bedrock reasoning preserves signature and redacted-content fragments. Foreign opaque reasoning -stream details, including cross-format Anthropic/Bedrock signatures, are rejected. Tool arguments +Bedrock reasoning preserves signature and redacted-content fragments. Translation to Anthropic +streams omits Bedrock signature fragments while retaining visible reasoning and answer text; +Bedrock redacted reasoning cannot be mapped and fails the stream. Encoding Bedrock rejects +foreign opaque reasoning stream details, including Anthropic signatures. Tool arguments may precede their ID and name; parallel calls are serialized into Bedrock blocks. A tool block cannot resume after other content closes it. Encoding Bedrock rejects unsupported foreign media and provider-specific built-in tool history. diff --git a/crates/switchyard-translation/src/codecs/anthropic/stream.rs b/crates/switchyard-translation/src/codecs/anthropic/stream.rs index 578c499b1..e12da3283 100644 --- a/crates/switchyard-translation/src/codecs/anthropic/stream.rs +++ b/crates/switchyard-translation/src/codecs/anthropic/stream.rs @@ -213,15 +213,12 @@ fn encode_anthropic_stream( } LlmResponseChunk::ReasoningDetailsDelta { text, details, .. } => { if details.iter().any(|detail| { - detail - .get("type") - .and_then(Value::as_str) - .is_some_and(|kind| kind.starts_with("bedrock.")) + detail.get("type").and_then(Value::as_str) == Some("bedrock.redacted_content") }) { return encode_anthropic_stream( state, LlmResponseChunk::DecodeError { - message: "Bedrock reasoning signatures have no Anthropic stream mapping" + message: "Bedrock redacted reasoning has no Anthropic stream mapping" .into(), }, ); diff --git a/crates/switchyard-translation/src/codecs/bedrock/buffered.rs b/crates/switchyard-translation/src/codecs/bedrock/buffered.rs index 22162ab4f..8ffe80507 100644 --- a/crates/switchyard-translation/src/codecs/bedrock/buffered.rs +++ b/crates/switchyard-translation/src/codecs/bedrock/buffered.rs @@ -245,15 +245,23 @@ impl FormatCodec for BedrockConverseCodec { system.push(encode_system_block(block, &mut diagnostics, policy)?); } } - let mut messages = Vec::new(); + let mut messages: Vec = Vec::new(); for message in &request.messages { if matches!(message.role, Role::System | Role::Developer) { for block in &message.content { system.push(encode_system_block(block, &mut diagnostics, policy)?); } } else { - messages.push(json!({"role": encode_role(message.role), - "content": encode_content(&message.content, &mut diagnostics, policy)?})); + let role = encode_role(message.role); + let content = encode_content(&message.content, &mut diagnostics, policy)?; + if let Some(previous) = messages.last_mut() + && previous["role"] == role + && let Some(blocks) = previous["content"].as_array_mut() + { + blocks.extend(content); + } else { + messages.push(json!({"role": role, "content": content})); + } } } if !system.is_empty() { @@ -282,8 +290,30 @@ impl FormatCodec for BedrockConverseCodec { if !inference.is_empty() { body.insert("inferenceConfig".into(), Value::Object(inference)); } - // Converse has no `none` tool-choice variant: disabling tools means omitting the config. - if request.tool_choice != Some(ToolChoice::None) && !request.tools.is_empty() { + let has_tool_history = request.messages.iter().any(|message| { + message.content.iter().any(|block| { + matches!( + block, + ContentBlock::ToolCall(_) | ContentBlock::ToolResult(_) + ) + }) + }); + if has_tool_history && request.tools.is_empty() { + return Err(invalid( + "$.toolConfig", + "tool history requires tool definitions", + )); + } + let is_tool_disabled = request.tool_choice == Some(ToolChoice::None); + // History still requires toolConfig, but Converse has no `none` choice. + if is_tool_disabled && has_tool_history { + push_lossy( + &mut diagnostics, + policy, + "Bedrock Converse cannot disable tools while retaining tool history", + )?; + } + if (!is_tool_disabled || has_tool_history) && !request.tools.is_empty() { let tools = request .tools .iter() @@ -300,7 +330,7 @@ impl FormatCodec for BedrockConverseCodec { }) .collect::>(); let mut config = json!({"tools": tools}); - if let Some(choice) = &request.tool_choice { + if !is_tool_disabled && let Some(choice) = &request.tool_choice { config["toolChoice"] = encode_tool_choice(choice)?; } body.insert("toolConfig".into(), config); @@ -797,7 +827,7 @@ fn encode_role(role: Role) -> &'static str { pub(super) fn decode_stop_reason(reason: &str) -> StopReason { match reason { "end_turn" | "stop_sequence" => StopReason::EndTurn, - "max_tokens" => StopReason::MaxTokens, + "max_tokens" | "model_context_window_exceeded" => StopReason::MaxTokens, "tool_use" => StopReason::ToolUse, "content_filtered" | "guardrail_intervened" => StopReason::ContentFilter, "malformed_model_output" | "malformed_tool_use" => StopReason::Error, diff --git a/crates/switchyard-translation/src/codecs/bedrock/stream.rs b/crates/switchyard-translation/src/codecs/bedrock/stream.rs index e9055d9cc..b6317bfa3 100644 --- a/crates/switchyard-translation/src/codecs/bedrock/stream.rs +++ b/crates/switchyard-translation/src/codecs/bedrock/stream.rs @@ -180,6 +180,7 @@ fn decode( state.stop_reason = Some(reason.into()); let reason = match reason { "content_filtered" | "guardrail_intervened" => "content_filter", + "model_context_window_exceeded" => "max_tokens", other => other, }; return Ok(vec![LlmResponseChunk::MessageStop { @@ -531,6 +532,7 @@ fn finish(state: &mut StreamTranslationState) -> Vec { if !state.bedrock.replayed_stop { let reason = match state.stop_reason.as_deref() { Some("length" | "max_tokens") => "max_tokens", + Some("model_context_window_exceeded") => "model_context_window_exceeded", Some("tool_calls" | "function_call" | "tool_use") => "tool_use", Some("content_filter" | "content_filtered") => "content_filtered", Some("guardrail_intervened") => "guardrail_intervened", diff --git a/crates/switchyard-translation/tests/bedrock_translation.rs b/crates/switchyard-translation/tests/bedrock_translation.rs index 137309376..d83d12f26 100644 --- a/crates/switchyard-translation/tests/bedrock_translation.rs +++ b/crates/switchyard-translation/tests/bedrock_translation.rs @@ -155,12 +155,13 @@ fn native_requests_and_cross_format_tool_histories() -> TestResult { assert_eq!(chat["messages"].as_array().map(Vec::len), Some(4)); let mut no_tools = decoded; no_tools.tool_choice = Some(ToolChoice::None); + let output = engine.encode_request(BEDROCK, &no_tools, &normalized())?; + assert!(output.body["toolConfig"]["tools"].is_array()); assert!( - engine - .encode_request(BEDROCK, &no_tools, &normalized())? - .body - .get("toolConfig") - .is_none() + output + .diagnostics + .iter() + .any(|d| d.code == "lossy_conversion") ); Ok(()) } @@ -550,7 +551,7 @@ fn event_stream_errors_and_truncation_fail_without_success_terminal() -> TestRes LlmResponseChunk::ReasoningDetailsDelta { index: 0, text: String::new(), - details: vec![json!({"type": "bedrock.signature_delta", "signature": "opaque"})], + details: vec![json!({"type": "bedrock.redacted_content", "data": "YQ=="})], } .into(), )?; @@ -560,3 +561,272 @@ fn event_stream_errors_and_truncation_fail_without_success_terminal() -> TestRes assert!(decode_stream(bytes, BEDROCK).is_err()); Ok(()) } + +#[test] +fn context_window_exhaustion_stays_incomplete_across_formats() -> TestResult { + let engine = TranslationEngine::default(); + let body = json!({"output": {"message": {"role": "assistant", "content": [{"text": "partial"}]}}, + "stopReason": "model_context_window_exceeded", "usage": usage()}); + let response = engine + .decode_response(BEDROCK, &body, &normalized())? + .response; + assert_eq!( + response.outputs[0].stop_reason, + Some(switchyard_protocol::StopReason::MaxTokens) + ); + assert_eq!( + engine + .encode_response(BEDROCK, &response, &normalized())? + .body, + body + ); + let mut values = events(); + values[3] = json!({"messageStop": {"stopReason": "model_context_window_exceeded"}}); + let aggregate = block_on(LlmResponse::Stream(raw_events(values.clone())?).into_agg())?; + assert_eq!( + aggregate.outputs[0].stop_reason, + Some(switchyard_protocol::StopReason::MaxTokens) + ); + let replayed = + block_on(encode_stream(raw_events(values.clone())?, BEDROCK, None)?.collect::>()) + .into_iter() + .collect::, _>>()?; + assert_eq!(replayed, values); + for target in [ + WireFormat::OpenAiChat, + WireFormat::OpenAiResponses, + WireFormat::AnthropicMessages, + ] { + let encoded = engine + .encode_response(target, &response, &normalized())? + .body; + match target { + WireFormat::OpenAiChat => assert_eq!(encoded["choices"][0]["finish_reason"], "length"), + WireFormat::OpenAiResponses => assert_eq!(encoded["status"], "incomplete"), + WireFormat::AnthropicMessages => assert_eq!(encoded["stop_reason"], "max_tokens"), + WireFormat::BedrockConverse => unreachable!(), + } + let projected = translate_events(&values, target)?; + let aggregate = block_on( + LlmResponse::Stream(decode_event_stream( + stream::iter(projected.into_iter().map(Ok)), + target, + )?) + .into_agg(), + )?; + assert_eq!( + aggregate.outputs[0].stop_reason, + Some(switchyard_protocol::StopReason::MaxTokens), + "{target}" + ); + } + let mut direct = StreamTranslationState::new(BEDROCK, BEDROCK); + engine.encode_stream_event( + &mut direct, + BEDROCK, + LlmResponseChunk::MessageStop { + reason: Some("model_context_window_exceeded".into()), + } + .into(), + )?; + engine.encode_stream_event( + &mut direct, + BEDROCK, + LlmResponseChunk::Usage(response.usage).into(), + )?; + assert!( + engine + .finish_stream(&mut direct, BEDROCK)? + .iter() + .any(|e| e["messageStop"]["stopReason"] == "model_context_window_exceeded") + ); + Ok(()) +} + +#[test] +fn consecutive_tool_results_and_user_turns_merge_without_reordering() -> TestResult { + let body = json!({"messages": [ + {"role": "user", "content": "weather?"}, + {"role": "assistant", "tool_calls": [ + {"id": "a", "type": "function", "function": {"name": "weather", "arguments": "{}"}}, + {"id": "b", "type": "function", "function": {"name": "weather", "arguments": "{}"}} + ]}, + {"role": "tool", "tool_call_id": "a", "content": "sunny"}, + {"role": "tool", "tool_call_id": "b", "content": "rainy"}, + {"role": "user", "content": "also tomorrow?"}, + {"role": "assistant", "content": "forecast"}, + {"role": "assistant", "content": "continued"}, + {"role": "user", "content": "thanks"} + ], "tools": [{"type": "function", "function": {"name": "weather", "parameters": {"type": "object"}}}]}); + let engine = TranslationEngine::default(); + let ir = engine + .decode_request(WireFormat::OpenAiChat, &body, &normalized())? + .request; + let encoded = engine.encode_request(BEDROCK, &ir, &normalized())?.body; + assert_eq!(encoded["messages"].as_array().map(Vec::len), Some(5)); + assert_eq!(encoded["messages"][2]["role"], "user"); + let content = encoded["messages"][2]["content"] + .as_array() + .ok_or("content array")?; + assert_eq!(content.len(), 3); + assert_eq!(content[0]["toolResult"]["toolUseId"], "a"); + assert_eq!(content[1]["toolResult"]["toolUseId"], "b"); + assert_eq!(content[2], json!({"text": "also tomorrow?"})); + assert_eq!( + encoded["messages"][3]["content"], + json!([{"text": "forecast"}, {"text": "continued"}]) + ); + let responses = json!({"input": [ + {"role": "user", "content": "weather?"}, + {"type": "function_call", "call_id": "a", "name": "weather", "arguments": "{}"}, + {"type": "function_call", "call_id": "b", "name": "weather", "arguments": "{}"}, + {"type": "function_call_output", "call_id": "a", "output": "sunny"}, + {"type": "function_call_output", "call_id": "b", "output": "rainy"}, + {"role": "user", "content": "also tomorrow?"}, + {"role": "assistant", "content": "forecast"}, + {"role": "assistant", "content": "continued"}, + {"role": "user", "content": "thanks"} + ], "tools": [{"type": "function", "name": "weather", "parameters": {"type": "object"}}]}); + let translated = engine + .translate_request( + WireFormat::OpenAiResponses, + BEDROCK, + &responses, + &normalized(), + )? + .body; + assert_eq!(translated["messages"], encoded["messages"]); + Ok(()) +} + +#[test] +fn disabled_tools_keep_required_history_configuration_or_fail() -> TestResult { + let engine = TranslationEngine::default(); + let mut ir = engine + .decode_request(BEDROCK, &request(), &normalized())? + .request; + ir.tool_choice = Some(ToolChoice::None); + let output = engine.encode_request(BEDROCK, &ir, &normalized())?; + assert!(output.body["toolConfig"]["tools"].is_array()); + assert!(output.body["toolConfig"].get("toolChoice").is_none()); + assert!( + output + .diagnostics + .iter() + .any(|d| d.code == "lossy_conversion") + ); + let strict = TranslationPolicy { + lossy_conversion_policy: LossyConversionPolicy::Reject, + ..normalized() + }; + assert!(engine.encode_request(BEDROCK, &ir, &strict).is_err()); + let mut no_history = ir.clone(); + no_history.messages.truncate(1); + assert!( + engine + .encode_request(BEDROCK, &no_history, &strict)? + .body + .get("toolConfig") + .is_none() + ); + ir.tools.clear(); + for choice in [ToolChoice::None, ToolChoice::Auto] { + ir.tool_choice = Some(choice); + assert!(engine.encode_request(BEDROCK, &ir, &normalized()).is_err()); + } + ir.messages = vec![ir.messages[0].clone()]; + ir.tool_choice = Some(ToolChoice::None); + assert!( + engine + .encode_request(BEDROCK, &ir, &strict)? + .body + .get("toolConfig") + .is_none() + ); + let mut prompt_body = request(); + for field in ["toolConfig", "system", "inferenceConfig"] { + prompt_body + .as_object_mut() + .ok_or("request object")? + .remove(field); + } + prompt_body["promptVariables"] = json!({"city": {"text": "Taipei"}}); + let preserved_policy = TranslationPolicy { + preservation: PreservationPolicy::InMemory, + ..strict + }; + let preserved = engine + .decode_request(BEDROCK, &prompt_body, &preserved_policy)? + .request; + assert!(preserved.tools.is_empty()); + assert_eq!( + engine + .encode_request(BEDROCK, &preserved, &preserved_policy)? + .body, + prompt_body + ); + Ok(()) +} + +#[test] +fn signed_bedrock_reasoning_keeps_anthropic_answer_but_redacted_fails() -> TestResult { + let values = vec![ + json!({"messageStart": {"role": "assistant"}}), + json!({"contentBlockDelta": {"contentBlockIndex": 0, "delta": {"reasoningContent": {"text": "think"}}}}), + json!({"contentBlockDelta": {"contentBlockIndex": 0, "delta": {"reasoningContent": {"signature": "opaque"}}}}), + json!({"contentBlockStop": {"contentBlockIndex": 0}}), + json!({"contentBlockDelta": {"contentBlockIndex": 1, "delta": {"text": "answer"}}}), + json!({"contentBlockStop": {"contentBlockIndex": 1}}), + json!({"messageStop": {"stopReason": "end_turn"}}), + json!({"metadata": {"usage": usage()}}), + ]; + let output = translate_events(&values, WireFormat::AnthropicMessages)?; + assert!(!output.iter().any(|event| event["type"] == "error")); + let aggregate = block_on( + LlmResponse::Stream(decode_event_stream( + stream::iter(output.into_iter().map(Ok)), + WireFormat::AnthropicMessages, + )?) + .into_agg(), + )?; + assert!( + aggregate.outputs[0] + .content + .iter() + .any(|b| matches!(b, ContentBlock::Reasoning { text, .. } if text == "think")) + ); + assert!( + aggregate.outputs[0] + .content + .iter() + .any(|b| matches!(b, ContentBlock::Text { text } if text == "answer")) + ); + let engine = TranslationEngine::default(); + let mut state = StreamTranslationState::new(BEDROCK, WireFormat::AnthropicMessages); + let output = engine.encode_stream_event( + &mut state, + WireFormat::AnthropicMessages, + LlmResponseChunk::ReasoningDetailsDelta { + index: 0, + text: "visible".into(), + details: vec![ + json!({"type": "bedrock.signature_delta", "signature": "opaque"}), + json!({"type": "bedrock.redacted_content", "data": "YQ=="}), + ], + } + .into(), + )?; + assert!(state.errored); + assert_eq!(output[0]["type"], "error"); + assert!( + output[0]["error"]["message"] + .as_str() + .is_some_and(|s| s.contains("redacted")) + ); + assert!( + engine + .finish_stream(&mut state, WireFormat::AnthropicMessages)? + .is_empty() + ); + Ok(()) +}