diff --git a/crates/libsy-llm-client/src/observability.rs b/crates/libsy-llm-client/src/observability.rs index c55d46ad8..88b5dcf1e 100644 --- a/crates/libsy-llm-client/src/observability.rs +++ b/crates/libsy-llm-client/src/observability.rs @@ -195,6 +195,7 @@ fn client_call_error_type(error: &LibsyError) -> Cow<'static, str> { fn llm_client_error_type(error: &LlmClientError) -> Cow<'static, str> { match error { + LlmClientError::RoutedCall { failure } => Cow::Borrowed(failure.class().stable_tag()), LlmClientError::InvalidRequest { .. } => Cow::Borrowed("invalid_request"), LlmClientError::RequestTranslation(_) => Cow::Borrowed("request_translation"), LlmClientError::RequestEncoding(_) => Cow::Borrowed("request_encoding"), diff --git a/crates/libsy-llm-client/src/run.rs b/crates/libsy-llm-client/src/run.rs index 8103a3bb6..779546e74 100644 --- a/crates/libsy-llm-client/src/run.rs +++ b/crates/libsy-llm-client/src/run.rs @@ -21,7 +21,8 @@ use http::StatusCode; use parking_lot::Mutex; use switchyard_libsy::{Algorithm, CallModel, LibsyError, Result, RoutingOutcome, drive}; use switchyard_protocol::{ - LlmClientError, ModelId, Request, Response, RoutedLlmClient, RoutingFallbackReason, + LlmClientError, ModelId, ProviderTargetsExhaustedSummary, Request, Response, RoutedCallFailure, + RoutedCallFailureClass, RoutedLlmClient, RoutingDisposition, RoutingFallbackReason, }; use switchyard_translation::prepare_request_for_target; @@ -305,11 +306,17 @@ async fn call_one( } /// Whether a failed candidate is worth routing around. +/// +/// A [`LlmClientError::RoutedCall`] is authoritative: the routing host already classified +/// the failure, so its [`RoutingDisposition`] decides whether another candidate may serve +/// the request and no status is re-interpreted here. The remaining variants keep the +/// legacy status-shaped inference for clients that do not classify their own failures. fn fallback_reason(error: &LibsyError) -> Option { let LibsyError::ClientCall { source, .. } = error else { return None; }; match source { + LlmClientError::RoutedCall { failure } => routed_call_fallback(failure), LlmClientError::ContextWindowExceeded { .. } => Some(RoutingFallbackReason::ContextWindow), LlmClientError::Transport { .. } | LlmClientError::Timeout { .. } => { Some(RoutingFallbackReason::Unavailable) @@ -326,6 +333,31 @@ fn fallback_reason(error: &LibsyError) -> Option { } } +/// Maps a host-classified failure to fallback policy, or `None` to surface it terminally. +/// +/// The disposition alone decides whether routing advances. The class only names why, for +/// the reasoning published on the hop. +fn routed_call_fallback(failure: &RoutedCallFailure) -> Option { + match failure.disposition() { + RoutingDisposition::Stop => None, + RoutingDisposition::NextTarget => Some(match failure.class() { + RoutedCallFailureClass::ContextWindow => RoutingFallbackReason::ContextWindow, + // An exhaustion every one of whose real failures was an overflow is an overflow + // for the whole logical model, not target unavailability. Reporting it as + // `Unavailable` is the conflation this contract exists to remove, and it is + // what a caller would have to undo to know the request itself needs reshaping. + RoutedCallFailureClass::ProviderTargetsExhausted + if failure + .exhausted() + .is_some_and(ProviderTargetsExhaustedSummary::is_context_window_exhaustion) => + { + RoutingFallbackReason::ContextWindow + } + _ => RoutingFallbackReason::Unavailable, + }), + } +} + /// Resolves a routed call's selected model to the client that serves it. /// /// An algorithm routes among named targets; which provider each target lives on is the @@ -447,8 +479,9 @@ mod tests { use http::StatusCode; use switchyard_libsy::{Driver, RoutingOutcome}; use switchyard_protocol::{ - ContentBlock, LlmResponse, LlmResponseChunk, LlmResponseStreamEvent, completion_text, - text_request, text_response, + ContentBlock, LlmResponse, LlmResponseChunk, LlmResponseStreamEvent, + ProviderTargetsExhaustedSummary, RoutedFailureCount, completion_text, text_request, + text_response, }; use wiremock::matchers::method; use wiremock::{Mock, MockServer, ResponseTemplate}; @@ -755,6 +788,156 @@ mod tests { Ok(()) } + /// A host-classified failure follows its explicit disposition, so a status that the + /// legacy path would have treated as a fallback stops the route when the host says so. + #[test] + fn routed_call_fallback_follows_disposition_not_status() { + let error = |source| LibsyError::client_call("target", source); + let stop = RoutedCallFailure::new( + RoutedCallFailureClass::RateLimit, + RoutingDisposition::Stop, + None, + Some(429), + ) + .expect("valid direct failure"); + assert_eq!( + fallback_reason(&error(LlmClientError::RoutedCall { failure: stop })), + None + ); + + let advance = RoutedCallFailure::new( + RoutedCallFailureClass::ProviderRejected, + RoutingDisposition::NextTarget, + None, + Some(400), + ) + .expect("valid direct failure"); + assert_eq!( + fallback_reason(&error(LlmClientError::RoutedCall { failure: advance })), + Some(RoutingFallbackReason::Unavailable) + ); + } + + /// The class names why the hop happened: a context overflow is a request-shape + /// failure, every other advanceable class is target unavailability. + #[test] + fn routed_call_fallback_maps_class_to_reason() { + let error = |source| LibsyError::client_call("target", source); + let advanceable = |class| { + fallback_reason(&error(LlmClientError::RoutedCall { + failure: RoutedCallFailure::new(class, RoutingDisposition::NextTarget, None, None) + .expect("valid direct failure"), + })) + }; + + assert_eq!( + advanceable(RoutedCallFailureClass::ContextWindow), + Some(RoutingFallbackReason::ContextWindow) + ); + for class in [ + RoutedCallFailureClass::CircuitOpen, + RoutedCallFailureClass::TargetIncompatible, + RoutedCallFailureClass::Overloaded, + RoutedCallFailureClass::Transport, + RoutedCallFailureClass::AttemptTimeout, + ] { + assert_eq!( + advanceable(class), + Some(RoutingFallbackReason::Unavailable), + "{} should report target unavailability", + class.stable_tag() + ); + } + } + + /// Provider exhaustion may still move to another candidate, and its bounded summary + /// survives the hop without routing inspecting a status. + #[test] + fn routed_call_fallback_advances_on_provider_exhaustion() { + let summary = ProviderTargetsExhaustedSummary::new( + 1, + 1, + vec![ + RoutedFailureCount::new(RoutedCallFailureClass::CircuitOpen, 1), + RoutedFailureCount::new(RoutedCallFailureClass::RateLimit, 1), + ], + None, + ) + .expect("valid partition"); + assert_eq!( + fallback_reason(&LibsyError::client_call( + "target", + LlmClientError::RoutedCall { + failure: RoutedCallFailure::targets_exhausted(summary), + }, + )), + Some(RoutingFallbackReason::Unavailable) + ); + } + + /// An exhaustion whose every real failure was an overflow is an overflow for the whole + /// logical model, so the hop says so rather than reporting target unavailability. A + /// bypassed circuit is not a real failure and does not break that conclusion; a real + /// failure of any other class does, and an all-bypassed exhaustion proves nothing. + #[test] + fn routed_call_fallback_reports_an_all_overflow_exhaustion_as_context_window() { + let exhaustion = |attempted, bypassed, failures| { + let summary = ProviderTargetsExhaustedSummary::new(attempted, bypassed, failures, None) + .expect("valid partition"); + fallback_reason(&LibsyError::client_call( + "target", + LlmClientError::RoutedCall { + failure: RoutedCallFailure::targets_exhausted(summary), + }, + )) + }; + + assert_eq!( + exhaustion( + 2, + 0, + vec![RoutedFailureCount::new( + RoutedCallFailureClass::ContextWindow, + 2 + )] + ), + Some(RoutingFallbackReason::ContextWindow) + ); + assert_eq!( + exhaustion( + 1, + 1, + vec![ + RoutedFailureCount::new(RoutedCallFailureClass::CircuitOpen, 1), + RoutedFailureCount::new(RoutedCallFailureClass::ContextWindow, 1), + ] + ), + Some(RoutingFallbackReason::ContextWindow) + ); + assert_eq!( + exhaustion( + 2, + 0, + vec![ + RoutedFailureCount::new(RoutedCallFailureClass::ContextWindow, 1), + RoutedFailureCount::new(RoutedCallFailureClass::RateLimit, 1), + ] + ), + Some(RoutingFallbackReason::Unavailable) + ); + assert_eq!( + exhaustion( + 0, + 2, + vec![RoutedFailureCount::new( + RoutedCallFailureClass::CircuitOpen, + 2 + )] + ), + Some(RoutingFallbackReason::Unavailable) + ); + } + #[test] fn fallback_only_accepts_context_and_unavailable_failures() { let error = |source| LibsyError::client_call("target", source); diff --git a/crates/libsy/src/algorithms/escalation.rs b/crates/libsy/src/algorithms/escalation.rs index 593c22925..e63a23e10 100644 --- a/crates/libsy/src/algorithms/escalation.rs +++ b/crates/libsy/src/algorithms/escalation.rs @@ -8,6 +8,7 @@ use std::sync::Arc; use async_trait::async_trait; use switchyard_protocol::{ AggLlmResponse, LlmClientError, LlmResponse, Message, ModelId, Request, Response, Role, + RoutedCallFailureClass, RoutingDisposition, }; use super::util::classifier_contract::ClassifierContractConfig; @@ -112,6 +113,21 @@ impl Classifier for EscalationClassifier { source: LlmClientError::ContextWindowExceeded { .. }, .. }) => return Ok((decisive(&self.capable), None)), + // A routing host classifies its own failures, so the same condition arrives as + // a typed class instead: an advanceable overflow, or a target that cannot serve + // this request's shape, is the escalation signal the legacy variant carried. + Err(LibsyError::ClientCall { + source: LlmClientError::RoutedCall { failure }, + .. + }) if failure.disposition() == RoutingDisposition::NextTarget + && matches!( + failure.class(), + RoutedCallFailureClass::ContextWindow + | RoutedCallFailureClass::TargetIncompatible + ) => + { + return Ok((decisive(&self.capable), None)); + } Err(e) => return Err(e), }; // The call resolves when its stream handle arrives; transport can still fail while @@ -173,7 +189,7 @@ mod tests { use parking_lot::Mutex; use switchyard_protocol::{ ContentBlock, LlmClientError, LlmResponse, LlmResponseChunk, Metadata, Request, Response, - completion_text, text_request, text_response, + RoutedCallFailure, completion_text, text_request, text_response, }; use super::*; @@ -389,6 +405,88 @@ mod tests { Ok(()) } + #[tokio::test] + async fn falls_back_to_capable_on_an_advanceable_routed_failure() -> Result<()> { + // A routing host returns its own classification instead of the legacy variant. Both + // advanceable request-shape classes must escalate to capable exactly as an overflow + // does; otherwise a staged route terminates instead of reaching the strong tier. + for class in [ + RoutedCallFailureClass::ContextWindow, + RoutedCallFailureClass::TargetIncompatible, + ] { + let serve = move |target: ModelId, _request: Request| async move { + match target.as_str() { + "efficient" => Err(LlmClientError::RoutedCall { + failure: RoutedCallFailure::new( + class, + RoutingDisposition::NextTarget, + None, + None, + ) + .expect("valid direct failure"), + }), + "judge" => panic!("the judge must not be consulted when efficient escalates"), + _ => Ok(reply("capable answer")), + } + }; + + let (selected_model, response) = + test_drive(escalation_router()?, classify_request(), serve).await?; + + assert_eq!( + selected_model, + "capable", + "{} should escalate to capable", + class.stable_tag() + ); + assert_eq!( + response.llm_response.as_agg().map(completion_text), + Some("capable answer".to_string()) + ); + } + Ok(()) + } + + /// The disposition is authoritative: a host that says `Stop` must not be second-guessed + /// into an escalation, even for a class the `NextTarget` arm accepts. + #[tokio::test] + async fn surfaces_a_terminal_routed_failure() { + let serve = |target: ModelId, _request: Request| async move { + match target.as_str() { + "efficient" => Err(LlmClientError::RoutedCall { + failure: RoutedCallFailure::new( + RoutedCallFailureClass::ContextWindow, + RoutingDisposition::Stop, + None, + None, + ) + .expect("valid direct failure"), + }), + other => panic!("{other} must not be called after a terminal failure"), + } + }; + + let Err(error) = test_drive( + escalation_router().expect("escalation router"), + classify_request(), + serve, + ) + .await + else { + panic!("a Stop disposition is terminal for the route"); + }; + assert!( + matches!( + error, + LibsyError::ClientCall { + source: LlmClientError::RoutedCall { .. }, + .. + } + ), + "expected the host's routed failure to surface, got {error:?}" + ); + } + /// A transport failure while buffering efficient must bypass the judge and serve capable. #[tokio::test] async fn falls_back_when_efficient_stream_transport_fails() -> Result<()> { diff --git a/crates/libsy/src/algorithms/util/llm_judge.rs b/crates/libsy/src/algorithms/util/llm_judge.rs index 0210bda27..ad1083caa 100644 --- a/crates/libsy/src/algorithms/util/llm_judge.rs +++ b/crates/libsy/src/algorithms/util/llm_judge.rs @@ -300,6 +300,7 @@ pub(crate) fn libsy_error_reason(error: &LibsyError) -> &'static str { /// Returns a bounded reason from the error kind and HTTP status only. fn client_error_reason(error: &LlmClientError) -> &'static str { match error { + LlmClientError::RoutedCall { failure } => failure.class().stable_tag(), LlmClientError::Timeout { .. } => "timeout", LlmClientError::Transport { .. } => "transport", LlmClientError::UpstreamHttp { status, .. } if status.is_server_error() => "upstream_5xx", diff --git a/crates/protocol/src/client.rs b/crates/protocol/src/client.rs index 6b24afcc6..9c5d9d787 100644 --- a/crates/protocol/src/client.rs +++ b/crates/protocol/src/client.rs @@ -101,11 +101,423 @@ pub enum LlmClientError { source: BoxError, }, + /// A routing host classified the failure itself and instructed routing what to do. + /// + /// Prefer this over the transport- and status-shaped variants above: it carries an + /// explicit [`RoutingDisposition`], so routing never re-derives fallback policy from + /// an HTTP status. + #[error("routed call failed: {failure}")] + RoutedCall { + /// Provider-neutral classification and routing instruction. + failure: RoutedCallFailure, + }, + /// A string message. Useful in testing, but prefer adding variants over using this. #[error("{0}")] General(String), } +/// Largest retry advice, in milliseconds, a routed-call failure may carry. +/// +/// Advice above this bound is not accepted: a routing host must bound its own +/// provider hint before it becomes public contract. +pub const MAX_ROUTED_RETRY_AFTER_MS: u64 = 300_000; + +/// Largest number of provider candidates one exhaustion summary may account for. +pub const MAX_PROVIDER_TARGETS: u8 = 16; + +/// Whether routing may replace the failed target with another candidate. +/// +/// This is the routing host's explicit instruction. Routing consumes it directly and +/// never re-derives fallback policy from a transport or HTTP status. +#[derive(Clone, Copy, Debug, Eq, Hash, Ord, PartialEq, PartialOrd)] +pub enum RoutingDisposition { + /// The failure is terminal for this route; do not try another target. + Stop, + /// Another eligible target may serve the request. + NextTarget, +} + +impl RoutingDisposition { + /// Stable value embedded in routing reasoning and telemetry. + pub const fn as_str(self) -> &'static str { + match self { + Self::Stop => "stop", + Self::NextTarget => "next_target", + } + } +} + +impl std::fmt::Display for RoutingDisposition { + fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { + f.write_str(self.as_str()) + } +} + +/// Provider-neutral class of a routed-call failure. +/// +/// The class names *what kind* of failure occurred, never which provider produced it. +/// It carries no provider body, message, target name, or health evidence: a routing +/// host keeps that private. New classes may be added, so match non-exhaustively. +#[non_exhaustive] +#[derive(Clone, Copy, Debug, Eq, Hash, Ord, PartialEq, PartialOrd)] +pub enum RoutedCallFailureClass { + /// The target was skipped because its availability circuit was open. + CircuitOpen, + /// Every provider candidate for the selected model was attempted or bypassed. + ProviderTargetsExhausted, + /// The target cannot serve this request's capabilities or profile. + TargetIncompatible, + /// The request exceeds the target's context window. + ContextWindow, + /// A rate, spend, or quota limit was enforced. + RateLimit, + /// The provider or model timed out processing the request. + ProviderTimeout, + /// The routing host's own per-attempt timeout elapsed. + AttemptTimeout, + /// The provider identified itself as overloaded or out of capacity. + Overloaded, + /// The call failed before a provider response head arrived. + Transport, + /// The provider returned a failure that identifies no more specific class. + Upstream, + /// The provider response could not be decoded or validated. + InvalidResponse, + /// The provider rejected the request on its own terms. + ProviderRejected, + /// Host policy denied the call. + PolicyDenied, + /// The credential for the target could not be obtained or used. + CredentialUnavailable, + /// The target's configuration is invalid. + Configuration, + /// The call's work or attempt budget is spent. + WorkBudget, + /// The caller cancelled, or the route deadline elapsed. + Cancelled, + /// The failure could not be classified. + Unknown, +} + +impl RoutedCallFailureClass { + /// Stable tag for reasoning, telemetry, and aggregate ordering. + /// + /// These values are contract: they are immutable once published, so a host may + /// order and label by them across releases. + pub const fn stable_tag(self) -> &'static str { + match self { + Self::CircuitOpen => "circuit_open", + Self::ProviderTargetsExhausted => "provider_targets_exhausted", + Self::TargetIncompatible => "target_incompatible", + Self::ContextWindow => "context_window", + Self::RateLimit => "rate_limit", + Self::ProviderTimeout => "provider_timeout", + Self::AttemptTimeout => "attempt_timeout", + Self::Overloaded => "overloaded", + Self::Transport => "transport", + Self::Upstream => "upstream", + Self::InvalidResponse => "invalid_response", + Self::ProviderRejected => "provider_rejected", + Self::PolicyDenied => "policy_denied", + Self::CredentialUnavailable => "credential_unavailable", + Self::Configuration => "configuration", + Self::WorkBudget => "work_budget", + Self::Cancelled => "cancelled", + Self::Unknown => "unknown", + } + } +} + +impl std::fmt::Display for RoutedCallFailureClass { + fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { + f.write_str(self.stable_tag()) + } +} + +/// A routed-call contract invariant a caller tried to violate. +/// +/// A routing host maps this to its own terminal routing-invariant error rather than +/// guessing a status: an unenforceable summary is a host bug, not a provider outcome. +#[derive(Clone, Copy, Debug, Eq, Error, Hash, PartialEq)] +#[error("routed-call contract violated: {reason}")] +pub struct RoutedFailureInvariant { + reason: &'static str, +} + +impl RoutedFailureInvariant { + /// Which invariant was violated. + pub const fn reason(self) -> &'static str { + self.reason + } +} + +/// How many candidates failed with one class. +#[derive(Clone, Copy, Debug, Eq, Hash, PartialEq)] +pub struct RoutedFailureCount { + /// The failure class these candidates shared. + pub class: RoutedCallFailureClass, + /// How many candidates failed this way. Never zero in a valid summary. + pub count: u8, +} + +impl RoutedFailureCount { + /// Builds one class entry. + pub const fn new(class: RoutedCallFailureClass, count: u8) -> Self { + Self { class, count } + } +} + +/// Bounded evidence that every provider candidate for one model was spent. +/// +/// The summary is a partition: each attempted candidate and each candidate bypassed by +/// an open circuit is counted exactly once, under exactly one class. It names no +/// target, carries no provider text, and never nests another exhaustion. +/// +/// [`Self::new`] is the only constructor, so a value of this type always satisfies +/// every invariant in the contract. +#[derive(Clone, Debug, Eq, PartialEq)] +pub struct ProviderTargetsExhaustedSummary { + attempted: u8, + bypassed: u8, + failures: Vec, + retry_after_ms: Option, +} + +impl ProviderTargetsExhaustedSummary { + /// Builds a validated summary, ordering entries by immutable + /// [`stable_tag`](RoutedCallFailureClass::stable_tag) rather than declaration order. + /// + /// # Errors + /// + /// Returns [`RoutedFailureInvariant`] unless every invariant holds: + /// `attempted + bypassed` is in `1..=`[`MAX_PROVIDER_TARGETS`]; `failures` holds at + /// most [`MAX_PROVIDER_TARGETS`] entries; each class appears + /// once with a nonzero count; no entry is + /// [`ProviderTargetsExhausted`](RoutedCallFailureClass::ProviderTargetsExhausted), + /// so exhaustion never recurses; the counts sum to `attempted + bypassed`; + /// the [`CircuitOpen`](RoutedCallFailureClass::CircuitOpen) count equals `bypassed` + /// exactly, leaving every other count to sum to `attempted`; and any retry advice + /// is within [`MAX_ROUTED_RETRY_AFTER_MS`]. + pub fn new( + attempted: u8, + bypassed: u8, + failures: Vec, + retry_after_ms: Option, + ) -> Result { + let total = attempted + .checked_add(bypassed) + .ok_or(invariant("candidate total overflows"))?; + if total == 0 || total > MAX_PROVIDER_TARGETS { + return Err(invariant("candidate total is outside 1..=16")); + } + // Reject an over-long entry list before reserving for it: a valid partition has at + // most one entry per counted candidate, so a caller-supplied length above the + // candidate bound can never become a valid summary and must not size an allocation. + if failures.len() > MAX_PROVIDER_TARGETS as usize { + return Err(invariant("class entry count exceeds the candidate bound")); + } + + // One pass proves the partition: unique nonzero classes, no nested exhaustion, + // and the two sums that pin bypasses to CircuitOpen and the rest to attempts. + let mut counted: u32 = 0; + let mut circuit_open: u32 = 0; + let mut seen: Vec = Vec::with_capacity(failures.len()); + for entry in &failures { + if entry.count == 0 { + return Err(invariant("class entry has a zero count")); + } + if entry.class == RoutedCallFailureClass::ProviderTargetsExhausted { + return Err(invariant("exhaustion cannot nest inside a summary")); + } + if seen.contains(&entry.class) { + return Err(invariant("class entry is repeated")); + } + seen.push(entry.class); + counted += u32::from(entry.count); + if entry.class == RoutedCallFailureClass::CircuitOpen { + circuit_open = u32::from(entry.count); + } + } + if counted != u32::from(total) { + return Err(invariant("class counts do not sum to the candidate total")); + } + if circuit_open != u32::from(bypassed) { + return Err(invariant("CircuitOpen count does not equal bypassed")); + } + if retry_after_ms.is_some_and(|ms| ms > MAX_ROUTED_RETRY_AFTER_MS) { + return Err(invariant("retry advice exceeds the 300000ms cap")); + } + + let mut failures = failures; + failures.sort_by_key(|entry| entry.class.stable_tag()); + Ok(Self { + attempted, + bypassed, + failures, + retry_after_ms, + }) + } + + /// Candidates that reached a real provider attempt. + pub const fn attempted(&self) -> u8 { + self.attempted + } + + /// Candidates skipped before contact because their circuit was open. + pub const fn bypassed(&self) -> u8 { + self.bypassed + } + + /// Per-class counts, ordered by stable tag. + pub fn failures(&self) -> &[RoutedFailureCount] { + &self.failures + } + + /// Entries for candidates that reached a real provider attempt. + /// + /// A candidate bypassed by an open circuit is not a real failure: it produced no + /// provider evidence, so the aggregate rules are stated over these entries only. + pub fn real_failures(&self) -> impl Iterator { + self.failures + .iter() + .filter(|entry| entry.class != RoutedCallFailureClass::CircuitOpen) + } + + /// Whether every candidate that reached a real attempt overflowed its context window. + /// + /// This is a property of the whole logical model rather than one target: the request + /// as shaped could not be served by any candidate, which is a request-shape condition + /// and not target unavailability. An exhaustion with no real failure at all — every + /// candidate bypassed by an open circuit — is not an overflow, because nothing proved + /// the model could not serve the turn. + pub fn is_context_window_exhaustion(&self) -> bool { + let mut real = self.real_failures(); + real.next() + .is_some_and(|entry| entry.class == RoutedCallFailureClass::ContextWindow) + && real.all(|entry| entry.class == RoutedCallFailureClass::ContextWindow) + } + + /// Aggregate retry advice, when every exhausted candidate supplied a truthful hint. + pub const fn retry_after_ms(&self) -> Option { + self.retry_after_ms + } +} + +/// A routing host's typed, provider-neutral call failure. +/// +/// It carries the host's explicit [`RoutingDisposition`] so routing never re-derives +/// fallback policy from an HTTP status, plus a bounded class, an optional bounded +/// provider status, and optional bounded retry advice. It deliberately holds no +/// provider body, message, target name, or availability evidence. +#[derive(Clone, Debug, Eq, PartialEq)] +pub struct RoutedCallFailure { + class: RoutedCallFailureClass, + disposition: RoutingDisposition, + retry_after_ms: Option, + provider_status: Option, + exhausted: Option, +} + +impl RoutedCallFailure { + /// Builds a direct failure for any class other than + /// [`ProviderTargetsExhausted`](RoutedCallFailureClass::ProviderTargetsExhausted), + /// which has a fixed shape and is built by [`Self::targets_exhausted`]. + /// + /// # Errors + /// + /// Returns [`RoutedFailureInvariant`] for the exhaustion class, or for retry advice + /// above [`MAX_ROUTED_RETRY_AFTER_MS`]. + pub fn new( + class: RoutedCallFailureClass, + disposition: RoutingDisposition, + retry_after_ms: Option, + provider_status: Option, + ) -> Result { + if class == RoutedCallFailureClass::ProviderTargetsExhausted { + return Err(invariant("exhaustion requires a bounded summary")); + } + if retry_after_ms.is_some_and(|ms| ms > MAX_ROUTED_RETRY_AFTER_MS) { + return Err(invariant("retry advice exceeds the 300000ms cap")); + } + Ok(Self { + class, + disposition, + retry_after_ms, + provider_status, + exhausted: None, + }) + } + + /// Builds the provider-exhaustion failure from a validated summary. + /// + /// The shape is fixed by the contract, so this cannot fail: the class is + /// [`ProviderTargetsExhausted`](RoutedCallFailureClass::ProviderTargetsExhausted), + /// the disposition is [`NextTarget`](RoutingDisposition::NextTarget) because another + /// logical model may still serve the request, there is no provider status, and the + /// retry advice is the summary's. + pub fn targets_exhausted(summary: ProviderTargetsExhaustedSummary) -> Self { + Self { + class: RoutedCallFailureClass::ProviderTargetsExhausted, + disposition: RoutingDisposition::NextTarget, + retry_after_ms: summary.retry_after_ms(), + provider_status: None, + exhausted: Some(summary), + } + } + + /// What kind of failure this is. + pub const fn class(&self) -> RoutedCallFailureClass { + self.class + } + + /// Whether routing may try another target. + pub const fn disposition(&self) -> RoutingDisposition { + self.disposition + } + + /// Bounded retry advice, when the host had a truthful one. + pub const fn retry_after_ms(&self) -> Option { + self.retry_after_ms + } + + /// The provider's status code, when one identified the failure. + pub const fn provider_status(&self) -> Option { + self.provider_status + } + + /// The bounded exhaustion summary, present only for the exhaustion class. + pub const fn exhausted(&self) -> Option<&ProviderTargetsExhaustedSummary> { + self.exhausted.as_ref() + } +} + +impl std::fmt::Display for RoutedCallFailure { + fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { + write!(f, "{} ({})", self.class.stable_tag(), self.disposition)?; + if let Some(status) = self.provider_status { + write!(f, ", provider status {status}")?; + } + if let Some(summary) = &self.exhausted { + write!( + f, + ", {} attempted, {} bypassed", + summary.attempted(), + summary.bypassed() + )?; + } + if let Some(retry_after_ms) = self.retry_after_ms { + write!(f, ", retry after {retry_after_ms}ms")?; + } + Ok(()) + } +} + +/// Builds a contract violation. Reasons are fixed strings, never caller or provider text. +const fn invariant(reason: &'static str) -> RoutedFailureInvariant { + RoutedFailureInvariant { reason } +} + /// Why routing replaced a selected target with another eligible target. #[derive(Clone, Copy, Debug, Eq, PartialEq)] pub enum RoutingFallbackReason { @@ -140,3 +552,347 @@ pub trait RoutedLlmClient: Send + Sync { /// Make a request async fn call(&self, request: Request) -> Result; } + +#[cfg(test)] +mod tests { + use super::*; + + /// Every class, so the tag table below cannot silently miss one. + const ALL_CLASSES: [RoutedCallFailureClass; 18] = [ + RoutedCallFailureClass::CircuitOpen, + RoutedCallFailureClass::ProviderTargetsExhausted, + RoutedCallFailureClass::TargetIncompatible, + RoutedCallFailureClass::ContextWindow, + RoutedCallFailureClass::RateLimit, + RoutedCallFailureClass::ProviderTimeout, + RoutedCallFailureClass::AttemptTimeout, + RoutedCallFailureClass::Overloaded, + RoutedCallFailureClass::Transport, + RoutedCallFailureClass::Upstream, + RoutedCallFailureClass::InvalidResponse, + RoutedCallFailureClass::ProviderRejected, + RoutedCallFailureClass::PolicyDenied, + RoutedCallFailureClass::CredentialUnavailable, + RoutedCallFailureClass::Configuration, + RoutedCallFailureClass::WorkBudget, + RoutedCallFailureClass::Cancelled, + RoutedCallFailureClass::Unknown, + ]; + + /// Stable tags are published contract: a host orders and labels by them, so changing + /// one silently reorders aggregates and breaks telemetry continuity. + #[test] + fn stable_tags_are_immutable_and_unique() { + let expected = [ + "circuit_open", + "provider_targets_exhausted", + "target_incompatible", + "context_window", + "rate_limit", + "provider_timeout", + "attempt_timeout", + "overloaded", + "transport", + "upstream", + "invalid_response", + "provider_rejected", + "policy_denied", + "credential_unavailable", + "configuration", + "work_budget", + "cancelled", + "unknown", + ]; + let tags: Vec<&str> = ALL_CLASSES.iter().map(|class| class.stable_tag()).collect(); + assert_eq!(tags, expected); + + let mut unique = tags.clone(); + unique.sort_unstable(); + unique.dedup(); + assert_eq!(unique.len(), tags.len(), "stable tags must be unique"); + } + + #[test] + fn dispositions_carry_stable_tags() { + assert_eq!(RoutingDisposition::Stop.as_str(), "stop"); + assert_eq!(RoutingDisposition::NextTarget.as_str(), "next_target"); + } + + #[test] + fn summary_accepts_a_complete_partition() { + let summary = ProviderTargetsExhaustedSummary::new( + 3, + 1, + vec![ + RoutedFailureCount::new(RoutedCallFailureClass::RateLimit, 2), + RoutedFailureCount::new(RoutedCallFailureClass::CircuitOpen, 1), + RoutedFailureCount::new(RoutedCallFailureClass::Overloaded, 1), + ], + Some(1_500), + ) + .expect("valid partition"); + + assert_eq!(summary.attempted(), 3); + assert_eq!(summary.bypassed(), 1); + assert_eq!(summary.retry_after_ms(), Some(1_500)); + } + + /// Entries order by immutable tag, never by declaration order: `CircuitOpen` is + /// declared first but sorts after `attempt_timeout`. + #[test] + fn summary_orders_entries_by_stable_tag() { + let summary = ProviderTargetsExhaustedSummary::new( + 2, + 1, + vec![ + RoutedFailureCount::new(RoutedCallFailureClass::CircuitOpen, 1), + RoutedFailureCount::new(RoutedCallFailureClass::Upstream, 1), + RoutedFailureCount::new(RoutedCallFailureClass::AttemptTimeout, 1), + ], + None, + ) + .expect("valid partition"); + + let tags: Vec<&str> = summary + .failures() + .iter() + .map(|entry| entry.class.stable_tag()) + .collect(); + assert_eq!(tags, ["attempt_timeout", "circuit_open", "upstream"]); + } + + #[test] + fn summary_rejects_every_broken_invariant() { + // Empty candidate set: exhaustion needs at least one candidate. + assert!(ProviderTargetsExhaustedSummary::new(0, 0, Vec::new(), None).is_err()); + + // More class entries than there can be candidates: rejected before any entry is + // read, so a caller-supplied length never sizes an allocation. + assert!( + ProviderTargetsExhaustedSummary::new( + 1, + 0, + vec![ + RoutedFailureCount::new(RoutedCallFailureClass::RateLimit, 1); + MAX_PROVIDER_TARGETS as usize + 1 + ], + None, + ) + .is_err() + ); + + // Above the 16-candidate bound. + assert!( + ProviderTargetsExhaustedSummary::new( + 17, + 0, + vec![RoutedFailureCount::new( + RoutedCallFailureClass::RateLimit, + 17 + )], + None, + ) + .is_err() + ); + + // Zero count. + assert!( + ProviderTargetsExhaustedSummary::new( + 1, + 0, + vec![ + RoutedFailureCount::new(RoutedCallFailureClass::RateLimit, 1), + RoutedFailureCount::new(RoutedCallFailureClass::Upstream, 0), + ], + None, + ) + .is_err() + ); + + // Repeated class. + assert!( + ProviderTargetsExhaustedSummary::new( + 2, + 0, + vec![ + RoutedFailureCount::new(RoutedCallFailureClass::RateLimit, 1), + RoutedFailureCount::new(RoutedCallFailureClass::RateLimit, 1), + ], + None, + ) + .is_err() + ); + + // Exhaustion never nests inside its own summary. + assert!( + ProviderTargetsExhaustedSummary::new( + 1, + 0, + vec![RoutedFailureCount::new( + RoutedCallFailureClass::ProviderTargetsExhausted, + 1, + )], + None, + ) + .is_err() + ); + + // Counts do not sum to attempted + bypassed. + assert!( + ProviderTargetsExhaustedSummary::new( + 3, + 0, + vec![RoutedFailureCount::new( + RoutedCallFailureClass::RateLimit, + 2 + )], + None, + ) + .is_err() + ); + + // CircuitOpen count must equal bypassed exactly. + assert!( + ProviderTargetsExhaustedSummary::new( + 1, + 1, + vec![ + RoutedFailureCount::new(RoutedCallFailureClass::CircuitOpen, 2), + RoutedFailureCount::new(RoutedCallFailureClass::RateLimit, 1), + ], + None, + ) + .is_err() + ); + + // A bypass counted as an attempt leaves the non-CircuitOpen sum wrong. + assert!( + ProviderTargetsExhaustedSummary::new( + 2, + 1, + vec![ + RoutedFailureCount::new(RoutedCallFailureClass::CircuitOpen, 1), + RoutedFailureCount::new(RoutedCallFailureClass::RateLimit, 1), + ], + None, + ) + .is_err() + ); + + // Retry advice above the cap is not truthful public contract. + assert!( + ProviderTargetsExhaustedSummary::new( + 1, + 0, + vec![RoutedFailureCount::new( + RoutedCallFailureClass::RateLimit, + 1 + )], + Some(MAX_ROUTED_RETRY_AFTER_MS + 1), + ) + .is_err() + ); + } + + #[test] + fn summary_accepts_advice_at_the_cap() { + let summary = ProviderTargetsExhaustedSummary::new( + 1, + 0, + vec![RoutedFailureCount::new( + RoutedCallFailureClass::RateLimit, + 1, + )], + Some(MAX_ROUTED_RETRY_AFTER_MS), + ) + .expect("advice at the cap is in bounds"); + assert_eq!(summary.retry_after_ms(), Some(MAX_ROUTED_RETRY_AFTER_MS)); + } + + #[test] + fn direct_failure_carries_disposition_and_no_summary() { + let failure = RoutedCallFailure::new( + RoutedCallFailureClass::RateLimit, + RoutingDisposition::NextTarget, + Some(2_000), + Some(429), + ) + .expect("valid direct failure"); + + assert_eq!(failure.class(), RoutedCallFailureClass::RateLimit); + assert_eq!(failure.disposition(), RoutingDisposition::NextTarget); + assert_eq!(failure.retry_after_ms(), Some(2_000)); + assert_eq!(failure.provider_status(), Some(429)); + assert!(failure.exhausted().is_none()); + } + + #[test] + fn direct_failure_rejects_exhaustion_class_and_over_cap_advice() { + assert!( + RoutedCallFailure::new( + RoutedCallFailureClass::ProviderTargetsExhausted, + RoutingDisposition::NextTarget, + None, + None, + ) + .is_err() + ); + assert!( + RoutedCallFailure::new( + RoutedCallFailureClass::RateLimit, + RoutingDisposition::NextTarget, + Some(MAX_ROUTED_RETRY_AFTER_MS + 1), + None, + ) + .is_err() + ); + } + + /// The exhaustion failure's shape is fixed: `NextTarget`, no provider status, and + /// retry advice that always equals the summary's. + #[test] + fn exhaustion_failure_has_the_contract_shape() { + let summary = ProviderTargetsExhaustedSummary::new( + 2, + 0, + vec![ + RoutedFailureCount::new(RoutedCallFailureClass::RateLimit, 1), + RoutedFailureCount::new(RoutedCallFailureClass::Overloaded, 1), + ], + Some(4_000), + ) + .expect("valid partition"); + let failure = RoutedCallFailure::targets_exhausted(summary.clone()); + + assert_eq!( + failure.class(), + RoutedCallFailureClass::ProviderTargetsExhausted + ); + assert_eq!(failure.disposition(), RoutingDisposition::NextTarget); + assert_eq!(failure.provider_status(), None); + assert_eq!(failure.retry_after_ms(), summary.retry_after_ms()); + assert_eq!(failure.exhausted(), Some(&summary)); + } + + /// The rendered failure is bounded and provider-neutral: tags and numbers only. + #[test] + fn display_is_bounded_and_provider_neutral() { + let failure = RoutedCallFailure::new( + RoutedCallFailureClass::Overloaded, + RoutingDisposition::NextTarget, + Some(1_000), + Some(529), + ) + .expect("valid direct failure"); + assert_eq!( + failure.to_string(), + "overloaded (next_target), provider status 529, retry after 1000ms" + ); + + let error = LlmClientError::RoutedCall { failure }; + assert_eq!( + error.to_string(), + "routed call failed: overloaded (next_target), provider status 529, retry after 1000ms" + ); + } +} diff --git a/crates/protocol/src/format.rs b/crates/protocol/src/format.rs index 745f4120d..35c1b17ee 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. + #[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/switchyard-runner/src/failure.rs b/crates/switchyard-runner/src/failure.rs index 8113ef25b..3c1889a07 100644 --- a/crates/switchyard-runner/src/failure.rs +++ b/crates/switchyard-runner/src/failure.rs @@ -5,7 +5,10 @@ use libsy::LibsyError; use strum_macros::IntoStaticStr; -use switchyard_protocol::{LlmClientError, ModelId}; +use switchyard_protocol::{ + LlmClientError, ModelId, ProviderTargetsExhaustedSummary, RoutedCallFailure, + RoutedCallFailureClass, +}; use crate::RunnerError; @@ -138,6 +141,12 @@ fn client_error_summary( _ => None, }); let (kind, upstream_status) = match error { + // A routing host classified this itself. Its class is provider-neutral and + // already bounded, so it is projected onto the nearest telemetry kind rather + // than widening this enum, and its bounded provider status is carried through. + LlmClientError::RoutedCall { failure } => { + (routed_call_kind(failure), failure.provider_status()) + } LlmClientError::UpstreamHttp { status, .. } => { (RouteErrorKind::UpstreamHttp, Some(status.as_u16())) } @@ -158,6 +167,37 @@ fn client_error_summary( summary(kind, phase, upstream_status, target) } +/// Projects a provider-neutral routed-call class onto the nearest telemetry kind. +/// +/// The routed-call contract classifies more finely than this enum does. Collapsing here +/// keeps `RouteErrorKind` stable while still recording something truthful; the full class +/// remains available on the failure itself. +fn routed_call_kind(failure: &RoutedCallFailure) -> RouteErrorKind { + // An exhaustion whose every real failure was an overflow is a context-window outcome + // for the whole logical model; recording it as `Other` would lose the one fact that + // tells an operator the request needs reshaping rather than the targets need fixing. + if failure + .exhausted() + .is_some_and(ProviderTargetsExhaustedSummary::is_context_window_exhaustion) + { + return RouteErrorKind::ContextWindowExceeded; + } + match failure.class() { + RoutedCallFailureClass::ContextWindow => RouteErrorKind::ContextWindowExceeded, + RoutedCallFailureClass::AttemptTimeout | RoutedCallFailureClass::ProviderTimeout => { + RouteErrorKind::Timeout + } + RoutedCallFailureClass::Transport => RouteErrorKind::Transport, + RoutedCallFailureClass::InvalidResponse => RouteErrorKind::InvalidResponse, + RoutedCallFailureClass::Configuration => RouteErrorKind::Configuration, + RoutedCallFailureClass::TargetIncompatible => RouteErrorKind::InvalidRequest, + RoutedCallFailureClass::Upstream | RoutedCallFailureClass::ProviderRejected => { + RouteErrorKind::UpstreamHttp + } + _ => RouteErrorKind::Other, + } +} + fn summary( kind: RouteErrorKind, phase: RouteErrorPhase, @@ -175,6 +215,83 @@ fn summary( #[cfg(test)] mod tests { use super::*; + use switchyard_protocol::{RoutedFailureCount, RoutingDisposition}; + + /// A host-classified failure keeps its meaning in telemetry: the class projects onto + /// the nearest kind, and an exhaustion whose every real failure was an overflow is + /// recorded as a context-window outcome rather than an unclassified one. + #[test] + fn routed_call_summaries_keep_their_classification() { + let direct = |class, status| { + routed_call_kind( + &RoutedCallFailure::new(class, RoutingDisposition::NextTarget, None, status) + .expect("valid direct failure"), + ) + }; + assert!(matches!( + direct(RoutedCallFailureClass::ContextWindow, None), + RouteErrorKind::ContextWindowExceeded + )); + assert!(matches!( + direct(RoutedCallFailureClass::AttemptTimeout, None), + RouteErrorKind::Timeout + )); + assert!(matches!( + direct(RoutedCallFailureClass::Upstream, Some(500)), + RouteErrorKind::UpstreamHttp + )); + assert!(matches!( + direct(RoutedCallFailureClass::PolicyDenied, None), + RouteErrorKind::Other + )); + + let exhaustion = |attempted, bypassed, failures| { + let summary = ProviderTargetsExhaustedSummary::new(attempted, bypassed, failures, None) + .expect("valid partition"); + routed_call_kind(&RoutedCallFailure::targets_exhausted(summary)) + }; + assert!(matches!( + exhaustion( + 2, + 0, + vec![RoutedFailureCount::new( + RoutedCallFailureClass::ContextWindow, + 2 + )] + ), + RouteErrorKind::ContextWindowExceeded + )); + assert!(matches!( + exhaustion( + 2, + 0, + vec![ + RoutedFailureCount::new(RoutedCallFailureClass::ContextWindow, 1), + RoutedFailureCount::new(RoutedCallFailureClass::RateLimit, 1), + ] + ), + RouteErrorKind::Other + )); + } + + /// The bounded provider status survives onto the summary, and no provider text does. + #[test] + fn routed_call_summary_carries_only_bounded_evidence() { + let failure = RoutedCallFailure::new( + RoutedCallFailureClass::Overloaded, + RoutingDisposition::NextTarget, + None, + Some(529), + ) + .expect("valid direct failure"); + let summary = client_error_summary( + &LlmClientError::RoutedCall { failure }, + RouteErrorPhase::BeforeResponse, + None, + ); + assert_eq!(summary.upstream_status, Some(529)); + assert!(summary.target.is_none()); + } const SECRET: &str = "patient name is Jane Doe"; diff --git a/crates/switchyard-server/src/lib.rs b/crates/switchyard-server/src/lib.rs index 8b6f06de8..ead0f4259 100644 --- a/crates/switchyard-server/src/lib.rs +++ b/crates/switchyard-server/src/lib.rs @@ -26,7 +26,7 @@ use std::time::{Duration, Instant}; use axum::body::Body; use axum::extract::rejection::{JsonRejection, QueryRejection}; use axum::extract::{DefaultBodyLimit, Query, Request as HttpRequest, State}; -use axum::http::header::CONTENT_TYPE; +use axum::http::header::{CONTENT_TYPE, RETRY_AFTER}; use axum::http::{HeaderMap, HeaderName, HeaderValue, StatusCode}; use axum::middleware::Next; use axum::response::{IntoResponse, Response}; @@ -38,7 +38,10 @@ use parking_lot::Mutex; use serde::{Deserialize, Serialize}; use serde_json::{Value, json}; use switchyard_llm_client::{AuxiliaryOperation, ClientRouter, RunObservation, RunObserver}; -use switchyard_protocol::{LlmClientError, Metadata, ModelId, Request, Usage}; +use switchyard_protocol::{ + LlmClientError, Metadata, ModelId, ProviderTargetsExhaustedSummary, Request, RoutedCallFailure, + RoutedCallFailureClass, Usage, +}; use switchyard_runner::{ CallerAuthKind, DecisionTarget, ModelCapabilities, Route, RunOutput, Runner, RunnerError, }; @@ -1223,6 +1226,7 @@ fn runner_error(error: RunnerError) -> Response { fn client_error(error: &LlmClientError) -> Response { match error { + LlmClientError::RoutedCall { failure } => routed_call_error(failure), LlmClientError::InvalidRequest { message } | LlmClientError::RequestTranslation(message) => error_response( StatusCode::BAD_REQUEST, @@ -1273,6 +1277,79 @@ fn client_error(error: &LlmClientError) -> Response { } } +// A host-classified routing failure already states what went wrong in bounded, +// provider-neutral terms, so the status follows the class and the message carries no +// provider body. `Retry-After` is emitted only when the host supplied truthful advice. +fn routed_call_error(failure: &RoutedCallFailure) -> Response { + let (status, code) = match failure.class() { + // Exhaustion is the one class carrying an aggregate, so its status comes from the + // bounded partition rather than the class alone. A summary is structurally + // guaranteed here; the fallback keeps the function total without a panic. + RoutedCallFailureClass::ProviderTargetsExhausted => failure.exhausted().map_or( + (StatusCode::SERVICE_UNAVAILABLE, "model_targets_unavailable"), + exhaustion_status, + ), + RoutedCallFailureClass::CircuitOpen | RoutedCallFailureClass::Overloaded => { + (StatusCode::SERVICE_UNAVAILABLE, "model_targets_unavailable") + } + RoutedCallFailureClass::AttemptTimeout | RoutedCallFailureClass::ProviderTimeout => { + (StatusCode::GATEWAY_TIMEOUT, "model_attempts_timed_out") + } + RoutedCallFailureClass::RateLimit => (StatusCode::TOO_MANY_REQUESTS, "model_rate_limited"), + RoutedCallFailureClass::ContextWindow | RoutedCallFailureClass::TargetIncompatible => { + (StatusCode::UNPROCESSABLE_ENTITY, "model_route_incompatible") + } + _ => (StatusCode::BAD_GATEWAY, "routing_failure"), + }; + let mut response = error_response(status, failure.to_string(), "upstream_error", code); + if let Some(retry_after) = retry_after_header(failure.retry_after_ms()) { + response.headers_mut().insert(RETRY_AFTER, retry_after); + } + response +} + +// Resolves an exhaustion to a status in the precedence the aggregate contract defines, +// reading only the bounded partition: exact availability evidence or a bypassed circuit +// first, then host or provider timeouts, then the homogeneous rate-limit and +// request-shape cases, and a plain routing failure for anything mixed. No provider +// status is consulted, and a bypassed candidate is never a real failure. +fn exhaustion_status(summary: &ProviderTargetsExhaustedSummary) -> (StatusCode, &'static str) { + let any = |class| summary.failures().iter().any(|entry| entry.class == class); + let every_real_failure_is = |classes: &[RoutedCallFailureClass]| { + summary + .failures() + .iter() + .filter(|entry| entry.class != RoutedCallFailureClass::CircuitOpen) + .all(|entry| classes.contains(&entry.class)) + }; + + if summary.bypassed() > 0 + || any(RoutedCallFailureClass::Overloaded) + || any(RoutedCallFailureClass::Transport) + { + (StatusCode::SERVICE_UNAVAILABLE, "model_targets_unavailable") + } else if any(RoutedCallFailureClass::AttemptTimeout) + || any(RoutedCallFailureClass::ProviderTimeout) + { + (StatusCode::GATEWAY_TIMEOUT, "model_attempts_timed_out") + } else if every_real_failure_is(&[RoutedCallFailureClass::RateLimit]) { + (StatusCode::TOO_MANY_REQUESTS, "model_rate_limited") + } else if every_real_failure_is(&[ + RoutedCallFailureClass::ContextWindow, + RoutedCallFailureClass::TargetIncompatible, + ]) { + (StatusCode::UNPROCESSABLE_ENTITY, "model_route_incompatible") + } else { + (StatusCode::BAD_GATEWAY, "routing_failure") + } +} + +// Milliseconds round up to whole HTTP seconds so advice is never shorter than the host's. +fn retry_after_header(retry_after_ms: Option) -> Option { + let seconds = retry_after_ms?.div_ceil(1_000); + HeaderValue::from_str(&seconds.to_string()).ok() +} + // Provider errors are often JSON documents; expose their message without // embedding the entire document as an escaped string in our error envelope. fn upstream_error_message(body: &str) -> String { @@ -1319,6 +1396,9 @@ impl ApiError { "message": self.message.clone(), } }), + WireFormat::BedrockConverse => json!({ + "message": self.message.clone(), + }), WireFormat::OpenAiChat | WireFormat::OpenAiResponses => json!({ "error": { "message": self.message.clone(), @@ -1690,11 +1770,180 @@ fn endpoint_listing(has_routing_log: bool) -> String { #[cfg(test)] mod tests { use switchyard_llm_client::LlmCallObservation; + use switchyard_protocol::{ + ProviderTargetsExhaustedSummary, RoutedFailureCount, RoutingDisposition, + }; use tokio::io::{AsyncReadExt, AsyncWriteExt}; use tokio::sync::{Notify, oneshot}; use super::*; + // A host-classified routing failure maps to a status from its class alone; no + // provider status or body decides the terminal response. + #[test] + fn routed_call_status_follows_failure_class() { + for (class, expected) in [ + ( + RoutedCallFailureClass::CircuitOpen, + StatusCode::SERVICE_UNAVAILABLE, + ), + ( + RoutedCallFailureClass::Overloaded, + StatusCode::SERVICE_UNAVAILABLE, + ), + ( + RoutedCallFailureClass::AttemptTimeout, + StatusCode::GATEWAY_TIMEOUT, + ), + ( + RoutedCallFailureClass::ProviderTimeout, + StatusCode::GATEWAY_TIMEOUT, + ), + ( + RoutedCallFailureClass::RateLimit, + StatusCode::TOO_MANY_REQUESTS, + ), + ( + RoutedCallFailureClass::ContextWindow, + StatusCode::UNPROCESSABLE_ENTITY, + ), + ( + RoutedCallFailureClass::TargetIncompatible, + StatusCode::UNPROCESSABLE_ENTITY, + ), + ( + RoutedCallFailureClass::PolicyDenied, + StatusCode::BAD_GATEWAY, + ), + (RoutedCallFailureClass::Unknown, StatusCode::BAD_GATEWAY), + ] { + let failure = RoutedCallFailure::new(class, RoutingDisposition::Stop, None, Some(200)) + .expect("valid direct failure"); + let response = client_error(&LlmClientError::RoutedCall { failure }); + assert_eq!( + response.status(), + expected, + "{} should map to {expected}", + class.stable_tag() + ); + assert!(response.headers().get(RETRY_AFTER).is_none()); + } + } + + // An exhaustion whose every real failure is a rate limit is quota back-off, not + // unavailability, and its bounded advice rounds up to whole HTTP seconds so the client + // never retries earlier than the host advised. + #[test] + fn routed_call_exhaustion_maps_to_rate_limited_with_rounded_retry_after() { + let summary = ProviderTargetsExhaustedSummary::new( + 2, + 0, + vec![RoutedFailureCount::new( + RoutedCallFailureClass::RateLimit, + 2, + )], + Some(1_200), + ) + .expect("valid partition"); + let response = client_error(&LlmClientError::RoutedCall { + failure: RoutedCallFailure::targets_exhausted(summary), + }); + + assert_eq!(response.status(), StatusCode::TOO_MANY_REQUESTS); + assert_eq!( + response + .headers() + .get(RETRY_AFTER) + .and_then(|value| value.to_str().ok()), + Some("2") + ); + } + + // The terminal status of an exhaustion follows the bounded aggregate in the contract's + // precedence order, so a mixed route reports the most consequential thing that happened + // rather than collapsing every combination back into one unavailability code. + #[test] + fn routed_call_exhaustion_status_follows_aggregate_precedence() { + for (name, attempted, bypassed, failures, expected) in [ + ( + "availability + rate-limit", + 2, + 0, + vec![ + RoutedFailureCount::new(RoutedCallFailureClass::Overloaded, 1), + RoutedFailureCount::new(RoutedCallFailureClass::RateLimit, 1), + ], + (StatusCode::SERVICE_UNAVAILABLE, "model_targets_unavailable"), + ), + ( + "circuit-open + timeout", + 1, + 1, + vec![ + RoutedFailureCount::new(RoutedCallFailureClass::CircuitOpen, 1), + RoutedFailureCount::new(RoutedCallFailureClass::AttemptTimeout, 1), + ], + (StatusCode::SERVICE_UNAVAILABLE, "model_targets_unavailable"), + ), + ( + "timeout + rate-limit", + 2, + 0, + vec![ + RoutedFailureCount::new(RoutedCallFailureClass::ProviderTimeout, 1), + RoutedFailureCount::new(RoutedCallFailureClass::RateLimit, 1), + ], + (StatusCode::GATEWAY_TIMEOUT, "model_attempts_timed_out"), + ), + ( + "rate-limit + context", + 2, + 0, + vec![ + RoutedFailureCount::new(RoutedCallFailureClass::RateLimit, 1), + RoutedFailureCount::new(RoutedCallFailureClass::ContextWindow, 1), + ], + (StatusCode::BAD_GATEWAY, "routing_failure"), + ), + ( + "context + incompatible", + 2, + 0, + vec![ + RoutedFailureCount::new(RoutedCallFailureClass::ContextWindow, 1), + RoutedFailureCount::new(RoutedCallFailureClass::TargetIncompatible, 1), + ], + (StatusCode::UNPROCESSABLE_ENTITY, "model_route_incompatible"), + ), + ] { + let summary = ProviderTargetsExhaustedSummary::new(attempted, bypassed, failures, None) + .expect("valid partition"); + assert_eq!( + exhaustion_status(&summary), + expected, + "{name} should resolve to {expected:?}" + ); + + // The resolved status reaches the wire, and no provider text rides with it. + let failure = RoutedCallFailure::targets_exhausted(summary); + let message = failure.to_string(); + let response = client_error(&LlmClientError::RoutedCall { failure }); + assert_eq!(response.status(), expected.0, "{name} status on the wire"); + assert_eq!( + response + .extensions() + .get::() + .map(|error| error.0.as_str()), + Some(message.as_str()), + "{name} keeps the bounded provider-neutral message" + ); + assert!( + response.headers().get(RETRY_AFTER).is_none(), + "{name} carries no advice, so no Retry-After is emitted" + ); + } + } + /// A successful judge call lands in the per-session routing snapshot under its /// model id with the classifier tier, while routed calls stay off the observer's /// log path — they are logged with terminal usage when the served response is diff --git a/crates/switchyard-server/src/sse.rs b/crates/switchyard-server/src/sse.rs index d94cb220e..c53cca3ab 100644 --- a/crates/switchyard-server/src/sse.rs +++ b/crates/switchyard-server/src/sse.rs @@ -65,6 +65,7 @@ pub(crate) fn frame_stream( fn frame_event(target_format: WireFormat, value: Value) -> Result { match target_format { WireFormat::OpenAiChat => Event::default().json_data(value), + WireFormat::BedrockConverse => Event::default().json_data(value), WireFormat::AnthropicMessages | WireFormat::OpenAiResponses => { let event_type = value .get("type") @@ -87,6 +88,14 @@ fn error_event(target_format: WireFormat, message: String) -> Event { }) .to_string(), ), + WireFormat::BedrockConverse => Event::default().data( + json!({ + "modelStreamErrorException": { + "message": message, + } + }) + .to_string(), + ), WireFormat::AnthropicMessages | WireFormat::OpenAiResponses => { Event::default().event("error").data( json!({ 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..edcf1db0b --- /dev/null +++ b/crates/switchyard-translation/src/codecs/bedrock/buffered.rs @@ -0,0 +1,665 @@ +// SPDX-FileCopyrightText: Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved. +// SPDX-License-Identifier: Apache-2.0 + +//! Buffered codec for Amazon Bedrock Converse request and response JSON. + +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, OutputParams, + ProviderExtensions, ResponseOutput, Role, SamplingParams, 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, push_lossy, validate_request_capabilities, +}; + +/// Format codec for Amazon Bedrock's provider-neutral Converse JSON contract. +pub struct BedrockConverseCodec; + +impl FormatCodec for BedrockConverseCodec { + fn format(&self) -> FormatId { + WireFormat::BedrockConverse.into() + } + + fn decode_request(&self, body: &Value, policy: &TranslationPolicy) -> Result { + let body = crate::util::object(body, "$")?; + let mut diagnostics = Vec::new(); + let inference = body.get("inferenceConfig").and_then(Value::as_object); + let mut request = LlmRequest { + sampling: SamplingParams { + temperature: inference + .and_then(|value| value.get("temperature")) + .and_then(Value::as_f64), + top_p: inference + .and_then(|value| value.get("topP")) + .and_then(Value::as_f64), + top_k: None, + }, + output: OutputParams { + max_output_tokens: inference + .and_then(|value| value.get("maxTokens")) + .and_then(Value::as_u64), + response_format: None, + }, + preservation: capture_request_preservation( + WireFormat::BedrockConverse, + &Value::Object(body.clone()), + policy, + ), + ..LlmRequest::default() + }; + + if let Some(system) = body.get("system").and_then(Value::as_array) { + let content = decode_content(system, Role::System, &mut diagnostics, policy)?; + if !content.is_empty() { + request.instructions.push(InstructionBlock { + role: Role::System, + content, + }); + } + } + if let Some(messages) = body.get("messages").and_then(Value::as_array) { + for (index, message) in messages.iter().enumerate() { + let message = + message + .as_object() + .ok_or_else(|| TranslationError::InvalidValue { + path: format!("$.messages[{index}]"), + message: "expected an object".to_string(), + })?; + let role = match message.get("role").and_then(Value::as_str) { + Some("user") => Role::User, + Some("assistant") => Role::Assistant, + Some(other) => { + return Err(TranslationError::unsupported_role( + format!("$.messages[{index}].role"), + other, + )); + } + None => Role::User, + }; + let content = message + .get("content") + .and_then(Value::as_array) + .map(|blocks| decode_content(blocks, role, &mut diagnostics, policy)) + .transpose()? + .unwrap_or_default(); + request.messages.push(Message { role, content }); + } + } + if let Some(tool_config) = body.get("toolConfig").and_then(Value::as_object) { + request.tools = decode_tools(tool_config.get("tools")); + request.tool_choice = tool_config.get("toolChoice").map(decode_tool_choice); + } + if let Some(stop_sequences) = inference.and_then(|value| value.get("stopSequences")) { + request + .extensions + .fields + .insert("stop".to_string(), stop_sequences.clone()); + } + if let Some(fields) = body.get("additionalModelRequestFields") { + request + .extensions + .fields + .insert("additionalModelRequestFields".to_string(), fields.clone()); + } + request.extensions.fields.extend(provider_extensions( + body, + &[ + "messages", + "system", + "inferenceConfig", + "toolConfig", + "additionalModelRequestFields", + ], + )); + + Ok(DecodedRequest { + request, + diagnostics, + }) + } + + fn encode_request( + &self, + request: &LlmRequest, + policy: &TranslationPolicy, + ) -> Result { + if let Some(body) = + exact_preserved_request(&request.preservation, WireFormat::BedrockConverse, policy) + { + return Ok(EncodedRequest { + body, + diagnostics: Vec::new(), + }); + } + + let mut diagnostics = Vec::new(); + validate_request_capabilities(request, &mut diagnostics, policy)?; + let mut body = Map::new(); + let system = request + .instructions + .iter() + .flat_map(|instruction| instruction.content.iter()) + .map(|block| encode_content_block(block, &mut diagnostics, policy)) + .collect::>>()?; + if !system.is_empty() { + body.insert("system".to_string(), Value::Array(system)); + } + body.insert( + "messages".to_string(), + Value::Array( + request + .messages + .iter() + .map(|message| encode_message(message, &mut diagnostics, policy)) + .collect::>>()?, + ), + ); + + let mut inference = Map::new(); + if let Some(value) = request.output.max_output_tokens { + inference.insert("maxTokens".to_string(), Value::from(value)); + } + if let Some(value) = request.sampling.temperature { + inference.insert("temperature".to_string(), Value::from(value)); + } + if let Some(value) = request.sampling.top_p { + inference.insert("topP".to_string(), Value::from(value)); + } + if let Some(value) = request.extensions.fields.get("stop") { + inference.insert("stopSequences".to_string(), value.clone()); + } + if !inference.is_empty() { + body.insert("inferenceConfig".to_string(), Value::Object(inference)); + } + if !request.tools.is_empty() || request.tool_choice.is_some() { + let mut config = Map::new(); + if !request.tools.is_empty() { + config.insert( + "tools".to_string(), + Value::Array(request.tools.iter().map(encode_tool).collect()), + ); + } + if let Some(choice) = &request.tool_choice { + config.insert("toolChoice".to_string(), encode_tool_choice(choice)); + } + body.insert("toolConfig".to_string(), Value::Object(config)); + } + if let Some(value) = request + .extensions + .fields + .get("additionalModelRequestFields") + { + body.insert("additionalModelRequestFields".to_string(), value.clone()); + } + + Ok(EncodedRequest { + body: embed_preservation(Value::Object(body), &request.preservation, policy), + diagnostics, + }) + } + + fn decode_response(&self, body: &Value, policy: &TranslationPolicy) -> Result { + let body = crate::util::object(body, "$")?; + let mut diagnostics = Vec::new(); + let output = body + .get("output") + .and_then(Value::as_object) + .and_then(|output| output.get("message")) + .and_then(Value::as_object); + let content = output + .and_then(|message| message.get("content")) + .and_then(Value::as_array) + .map(|blocks| decode_content(blocks, Role::Assistant, &mut diagnostics, policy)) + .transpose()? + .unwrap_or_default(); + let stop_reason = body + .get("stopReason") + .and_then(Value::as_str) + .map(decode_stop_reason); + let usage = decode_usage(body.get("usage")); + let outputs = output + .map(|message| ResponseOutput { + role: match message.get("role").and_then(Value::as_str) { + Some("user") => Role::User, + _ => Role::Assistant, + }, + content, + stop_reason, + }) + .into_iter() + .collect(); + + Ok(DecodedResponse { + response: AggLlmResponse { + outputs, + usage, + extensions: ProviderExtensions { + fields: provider_extensions(body, &["output", "stopReason", "usage"]), + }, + preservation: capture_response_preservation( + WireFormat::BedrockConverse, + &Value::Object(body.clone()), + policy, + ), + ..AggLlmResponse::default() + }, + diagnostics, + }) + } + + fn encode_response( + &self, + response: &AggLlmResponse, + policy: &TranslationPolicy, + ) -> Result { + if let Some(body) = + exact_preserved_response(&response.preservation, WireFormat::BedrockConverse, policy) + { + return Ok(EncodedResponse { + body, + diagnostics: Vec::new(), + }); + } + let mut diagnostics = Vec::new(); + let output = response.outputs.first(); + let message = output + .map(|output| { + let content = output + .content + .iter() + .map(|block| encode_content_block(block, &mut diagnostics, policy)) + .collect::>>()?; + Ok::(json!({ + "role": encode_role(output.role), + "content": content + })) + }) + .transpose()?; + let mut body = Map::new(); + if let Some(message) = message { + body.insert("output".to_string(), json!({"message": message})); + } + if let Some(reason) = output.and_then(|output| output.stop_reason) { + body.insert( + "stopReason".to_string(), + Value::String(encode_stop_reason(reason).to_string()), + ); + } + let usage = encode_usage(&response.usage); + if !usage.is_empty() { + body.insert("usage".to_string(), Value::Object(usage)); + } + Ok(EncodedResponse { + body: embed_preservation(Value::Object(body), &response.preservation, policy), + diagnostics, + }) + } +} + +fn decode_content( + blocks: &[Value], + role: Role, + diagnostics: &mut Vec, + policy: &TranslationPolicy, +) -> Result> { + blocks + .iter() + .enumerate() + .map(|(index, block)| decode_content_block(block, role, index, diagnostics, policy)) + .collect() +} + +fn decode_content_block( + block: &Value, + role: Role, + index: usize, + diagnostics: &mut Vec, + policy: &TranslationPolicy, +) -> Result { + let object = block + .as_object() + .ok_or_else(|| TranslationError::InvalidValue { + path: format!("$.content[{index}]"), + message: "expected an object".to_string(), + })?; + if let Some(text) = object.get("text").and_then(Value::as_str) { + return Ok(ContentBlock::Text { + text: text.to_string(), + }); + } + if let Some(tool) = object.get("toolUse").and_then(Value::as_object) { + return Ok(ContentBlock::ToolCall(ToolCall { + id: tool + .get("toolUseId") + .and_then(Value::as_str) + .unwrap_or_default() + .to_string(), + name: tool + .get("name") + .and_then(Value::as_str) + .unwrap_or_default() + .to_string(), + arguments: tool.get("input").cloned().unwrap_or_else(|| json!({})), + })); + } + if let Some(tool) = object.get("toolResult").and_then(Value::as_object) { + let content = tool + .get("content") + .and_then(Value::as_array) + .map(|blocks| decode_tool_result_content(blocks)) + .unwrap_or_default(); + return Ok(ContentBlock::ToolResult(ToolResult { + tool_call_id: tool + .get("toolUseId") + .and_then(Value::as_str) + .unwrap_or_default() + .to_string(), + content, + is_error: tool + .get("status") + .and_then(Value::as_str) + .map(|status| status == "error"), + })); + } + if let Some(reasoning) = object.get("reasoningContent").and_then(Value::as_object) { + let reasoning = reasoning + .get("reasoningText") + .and_then(Value::as_object) + .unwrap_or(reasoning); + return Ok(ContentBlock::Reasoning { + text: reasoning + .get("text") + .and_then(Value::as_str) + .unwrap_or_default() + .to_string(), + signature: reasoning + .get("signature") + .and_then(Value::as_str) + .map(ToOwned::to_owned), + // Bedrock Converse models reasoning as text plus a signature; it carries no + // opaque detail objects to replay. + details: Vec::new(), + }); + } + if let Some(image) = object.get("image").and_then(Value::as_object) { + let format = image.get("format").and_then(Value::as_str); + let bytes = image + .get("source") + .and_then(Value::as_object) + .and_then(|source| source.get("bytes")) + .and_then(Value::as_str); + if let Some(data) = bytes { + return Ok(ContentBlock::Image { + source: ImageSource::Base64 { + media_type: format.map(|format| format!("image/{format}")), + data: data.to_string(), + }, + }); + } + } + push_lossy( + diagnostics, + policy, + format!("unsupported Bedrock Converse content block for {role:?}"), + )?; + Ok(ContentBlock::Unknown { + provider: WireFormat::BedrockConverse.into(), + raw: block.clone(), + }) +} + +fn decode_tool_result_content(blocks: &[Value]) -> Vec { + blocks + .iter() + .map(|block| { + if let Some(text) = block.get("text").and_then(Value::as_str) { + ContentBlock::Text { + text: text.to_string(), + } + } else if let Some(value) = block.get("json") { + ContentBlock::Text { + text: value.to_string(), + } + } else { + ContentBlock::Unknown { + provider: WireFormat::BedrockConverse.into(), + raw: block.clone(), + } + } + }) + .collect() +} + +fn encode_message( + message: &Message, + diagnostics: &mut Vec, + policy: &TranslationPolicy, +) -> Result { + Ok(json!({ + "role": encode_role(message.role), + "content": message + .content + .iter() + .map(|block| encode_content_block(block, diagnostics, policy)) + .collect::>>()?, + })) +} + +fn encode_role(role: Role) -> &'static str { + match role { + Role::Assistant => "assistant", + Role::User | Role::Tool | Role::System | Role::Developer => "user", + } +} + +fn encode_content_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, .. + } => json!({ + "reasoningContent": {"reasoningText": {"text": text, "signature": signature}} + }), + ContentBlock::ToolCall(call) => json!({ + "toolUse": {"toolUseId": call.id, "name": call.name, "input": call.arguments} + }), + ContentBlock::ToolResult(result) => json!({ + "toolResult": { + "toolUseId": result.tool_call_id, + "content": result.content.iter().map(encode_tool_result_block).collect::>(), + "status": if result.is_error == Some(true) { "error" } else { "success" } + } + }), + ContentBlock::Image { + source: ImageSource::Base64 { media_type, data }, + } => { + let format = media_type + .as_deref() + .and_then(|value| value.strip_prefix("image/")) + .unwrap_or("png"); + json!({"image": {"format": format, "source": {"bytes": data}}}) + } + ContentBlock::Unknown { provider, raw } + if provider.as_str() == WireFormat::BedrockConverse.as_str() => + { + raw.clone() + } + unsupported => { + push_lossy( + diagnostics, + policy, + format!("unsupported content block encoded for Bedrock Converse: {unsupported:?}"), + )?; + json!({"text": unsupported_text(unsupported)}) + } + }) +} + +fn unsupported_text(block: &ContentBlock) -> String { + match block { + ContentBlock::Audio { .. } => "[audio]", + ContentBlock::Video { .. } => "[video]", + ContentBlock::File { .. } => "[file]", + ContentBlock::Image { .. } => "[image]", + _ => "[unsupported content]", + } + .to_string() +} + +fn encode_tool_result_block(block: &ContentBlock) -> Value { + match block { + ContentBlock::Text { text } | ContentBlock::Refusal { text } => json!({"text": text}), + ContentBlock::Unknown { raw, .. } => json!({"json": raw}), + other => json!({"text": unsupported_text(other)}), + } +} + +fn decode_tools(value: Option<&Value>) -> Vec { + value + .and_then(Value::as_array) + .into_iter() + .flatten() + .filter_map(|tool| tool.get("toolSpec").and_then(Value::as_object)) + .filter_map(|tool| { + let name = tool.get("name")?.as_str()?.to_string(); + Some(ToolDefinition { + name, + description: tool + .get("description") + .and_then(Value::as_str) + .map(ToOwned::to_owned), + parameters: tool + .get("inputSchema") + .and_then(|schema| schema.get("json")) + .cloned() + .unwrap_or_else(|| json!({})), + strict: None, + }) + }) + .collect() +} + +fn encode_tool(tool: &ToolDefinition) -> Value { + json!({ + "toolSpec": { + "name": tool.name, + "description": tool.description, + "inputSchema": {"json": tool.parameters}, + } + }) +} + +fn decode_tool_choice(value: &Value) -> ToolChoice { + if value.get("auto").is_some() { + ToolChoice::Auto + } else if value.get("any").is_some() { + ToolChoice::Required + } else if let Some(name) = value + .get("tool") + .and_then(Value::as_object) + .and_then(|tool| tool.get("name")) + .and_then(Value::as_str) + { + ToolChoice::Tool { + name: name.to_string(), + } + } else { + ToolChoice::Raw(value.clone()) + } +} + +fn encode_tool_choice(choice: &ToolChoice) -> Value { + match choice { + ToolChoice::Auto => json!({"auto": {}}), + ToolChoice::Required => json!({"any": {}}), + ToolChoice::Tool { name } => json!({"tool": {"name": name}}), + ToolChoice::None => json!({"none": {}}), + ToolChoice::Raw(value) => value.clone(), + } +} + +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, + _ => StopReason::Unknown, + } +} + +fn encode_stop_reason(reason: StopReason) -> &'static str { + match reason { + StopReason::EndTurn => "end_turn", + StopReason::MaxTokens => "max_tokens", + StopReason::ToolUse => "tool_use", + StopReason::ContentFilter => "content_filtered", + StopReason::Error => "error", + StopReason::Unknown => "unknown", + } +} + +fn decode_usage(value: Option<&Value>) -> Usage { + let value = value.and_then(Value::as_object); + let mut usage = Usage { + input_tokens: value + .and_then(|value| value.get("inputTokens")) + .and_then(Value::as_u64), + output_tokens: value + .and_then(|value| value.get("outputTokens")) + .and_then(Value::as_u64), + total_tokens: value + .and_then(|value| value.get("totalTokens")) + .and_then(Value::as_u64), + ..Usage::default() + }; + if let Some(value) = value + .and_then(|value| value.get("cacheReadInputTokens")) + .and_then(Value::as_u64) + { + usage.set_cached_input_tokens(value); + } + if let Some(value) = value + .and_then(|value| value.get("cacheWriteInputTokens")) + .and_then(Value::as_u64) + { + usage.set_cache_creation_input_tokens(value); + } + usage +} + +fn encode_usage(usage: &Usage) -> Map { + let mut value = Map::new(); + if let Some(tokens) = usage.input_tokens { + value.insert("inputTokens".to_string(), Value::from(tokens)); + } + if let Some(tokens) = usage.output_tokens { + value.insert("outputTokens".to_string(), Value::from(tokens)); + } + if let Some(tokens) = usage.total_tokens { + value.insert("totalTokens".to_string(), Value::from(tokens)); + } + if let Some(tokens) = usage.cached_input_tokens() { + value.insert("cacheReadInputTokens".to_string(), Value::from(tokens)); + } + if let Some(tokens) = usage.cache_creation_input_tokens() { + value.insert("cacheWriteInputTokens".to_string(), Value::from(tokens)); + } + value +} diff --git a/crates/switchyard-translation/src/codecs/bedrock/mod.rs b/crates/switchyard-translation/src/codecs/bedrock/mod.rs new file mode 100644 index 000000000..1a73a1e28 --- /dev/null +++ b/crates/switchyard-translation/src/codecs/bedrock/mod.rs @@ -0,0 +1,9 @@ +// SPDX-FileCopyrightText: Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved. +// SPDX-License-Identifier: Apache-2.0 + +//! Amazon Bedrock Converse wire-format codecs. + +mod buffered; +pub mod stream; + +pub use buffered::BedrockConverseCodec; 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..bc01eb710 --- /dev/null +++ b/crates/switchyard-translation/src/codecs/bedrock/stream.rs @@ -0,0 +1,369 @@ +// SPDX-FileCopyrightText: Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved. +// SPDX-License-Identifier: Apache-2.0 + +//! Amazon Bedrock ConverseStream event payload codec. +//! +//! AWS EventStream carrier framing is intentionally host-owned. This codec consumes and +//! produces the JSON payload carried by each framed EventStream message. + +use serde_json::{Map, Value, json}; + +use crate::LlmResponseChunk; +use crate::codecs::stream::{StreamCodec, StreamTranslationState, record_source_identity}; +use crate::format::{FormatId, WireFormat}; + +/// Stream codec for Bedrock ConverseStream JSON event payloads. +pub struct BedrockConverseStreamCodec; + +impl StreamCodec for BedrockConverseStreamCodec { + fn format(&self) -> FormatId { + WireFormat::BedrockConverse.into() + } + + fn decode_event( + &self, + state: &mut StreamTranslationState, + event: &Value, + ) -> Vec { + decode_bedrock_event(state, event) + } + + fn encode_event( + &self, + state: &mut StreamTranslationState, + event: LlmResponseChunk, + ) -> Vec { + encode_bedrock_event(state, event) + } + + fn finish(&self, state: &mut StreamTranslationState) -> Vec { + finish_bedrock_stream(state) + } +} + +fn decode_bedrock_event( + state: &mut StreamTranslationState, + event: &Value, +) -> Vec { + let Some(object) = event.as_object() else { + return vec![LlmResponseChunk::DecodeError { + message: "Bedrock stream event is not an object".to_string(), + }]; + }; + if object.get("messageStart").is_some() { + state.saw_message_start = true; + return vec![LlmResponseChunk::MessageStart { + id: None, + model: None, + }]; + } + if let Some(start) = object.get("contentBlockStart").and_then(Value::as_object) { + return decode_content_start(start); + } + if let Some(delta) = object.get("contentBlockDelta").and_then(Value::as_object) { + return decode_content_delta(delta); + } + if let Some(stop) = object.get("messageStop").and_then(Value::as_object) { + let reason = stop + .get("stopReason") + .and_then(Value::as_str) + .map(ToOwned::to_owned); + state.stop_reason = reason.clone(); + return vec![LlmResponseChunk::MessageStop { reason }]; + } + if let Some(metadata) = object.get("metadata").and_then(Value::as_object) + && let Some(usage) = metadata.get("usage") + { + capture_usage(state, usage); + return vec![LlmResponseChunk::Usage(state.usage.clone())]; + } + if let Some((name, _)) = object.iter().find(|(name, _)| name.ends_with("Exception")) { + return vec![LlmResponseChunk::StreamError { + message: format!("Bedrock stream failed with {name}"), + }]; + } + Vec::new() +} + +fn decode_content_start(object: &Map) -> Vec { + let index = object + .get("contentBlockIndex") + .and_then(Value::as_u64) + .unwrap_or(0) as usize; + let start = object.get("start").and_then(Value::as_object); + let tool = start + .and_then(|start| start.get("toolUse")) + .and_then(Value::as_object); + tool.map(|tool| { + vec![LlmResponseChunk::ToolCallDelta { + index, + id: tool + .get("toolUseId") + .and_then(Value::as_str) + .map(ToOwned::to_owned), + name: tool + .get("name") + .and_then(Value::as_str) + .map(ToOwned::to_owned), + arguments_delta: None, + }] + }) + .unwrap_or_default() +} + +fn decode_content_delta(object: &Map) -> Vec { + let index = object + .get("contentBlockIndex") + .and_then(Value::as_u64) + .unwrap_or(0) as usize; + let Some(delta) = object.get("delta").and_then(Value::as_object) else { + return Vec::new(); + }; + if let Some(text) = delta.get("text").and_then(Value::as_str) { + return vec![LlmResponseChunk::TextDelta { + index, + text: text.to_string(), + }]; + } + if let Some(input) = delta + .get("toolUse") + .and_then(Value::as_object) + .and_then(|tool| tool.get("input")) + .and_then(Value::as_str) + { + return vec![LlmResponseChunk::ToolCallDelta { + index, + id: None, + name: None, + arguments_delta: Some(input.to_string()), + }]; + } + if let Some(reasoning) = delta.get("reasoningContent").and_then(Value::as_object) + && let Some(text) = reasoning.get("text").and_then(Value::as_str) + { + return vec![LlmResponseChunk::ReasoningDelta { + index, + text: text.to_string(), + }]; + } + Vec::new() +} + +fn encode_bedrock_event(state: &mut StreamTranslationState, event: LlmResponseChunk) -> Vec { + if state.errored { + return Vec::new(); + } + match event { + LlmResponseChunk::MessageStart { id, model } => { + record_source_identity(state, id, model); + if state.emitted_message_start { + Vec::new() + } else { + state.emitted_message_start = true; + vec![json!({"messageStart": {"role": "assistant"}})] + } + } + LlmResponseChunk::TextDelta { text, .. } => { + state.output_tokens_seen += 1; + let index = ensure_text_block(state); + vec![json!({ + "contentBlockDelta": { + "contentBlockIndex": index, + "delta": {"text": text} + } + })] + } + // Bedrock Converse has no wire slot for opaque reasoning details, so only the + // normalized text that accompanies them survives, on the reasoning-delta path. + LlmResponseChunk::ReasoningDelta { text, .. } + | LlmResponseChunk::ReasoningDetailsDelta { text, .. } => { + if text.is_empty() { + return Vec::new(); + } + let index = ensure_reasoning_block(state); + vec![json!({ + "contentBlockDelta": { + "contentBlockIndex": index, + "delta": {"reasoningContent": {"text": text}} + } + })] + } + LlmResponseChunk::ToolCallDelta { + index, + id, + name, + arguments_delta, + } => encode_tool_delta(state, index, id, name, arguments_delta), + LlmResponseChunk::Usage(usage) => { + state.usage = usage; + state.saw_backend_usage = true; + Vec::new() + } + LlmResponseChunk::MessageStop { reason } => { + state.stop_reason = reason.or_else(|| state.stop_reason.clone()); + Vec::new() + } + LlmResponseChunk::StreamError { message } | LlmResponseChunk::DecodeError { message } => { + state.finished = true; + state.errored = true; + vec![json!({"modelStreamErrorException": {"message": message}})] + } + } +} + +fn ensure_text_block(state: &mut StreamTranslationState) -> usize { + if let Some(index) = state.text_block_index { + return index; + } + let index = state.next_content_index; + state.next_content_index += 1; + state.text_block_index = Some(index); + state.text_block_started = true; + state.emitted_content_block = true; + index +} + +fn ensure_reasoning_block(state: &mut StreamTranslationState) -> usize { + if let Some(index) = state.reasoning_block_index { + return index; + } + let index = state.next_content_index; + state.next_content_index += 1; + state.reasoning_block_index = Some(index); + state.reasoning_block_started = true; + state.emitted_content_block = true; + index +} + +fn encode_tool_delta( + state: &mut StreamTranslationState, + index: usize, + id: Option, + name: Option, + arguments_delta: Option, +) -> Vec { + let tool = state.tool_states.entry(index).or_default(); + if id.is_some() { + tool.id = id; + } + if name.is_some() { + tool.name = name; + } + let mut out = Vec::new(); + if !tool.started { + let Some(name) = tool.name.clone() else { + if let Some(delta) = arguments_delta { + tool.pending_arguments.push_str(&delta); + } + return out; + }; + let content_index = state.next_content_index; + state.next_content_index += 1; + tool.content_index = Some(content_index); + tool.started = true; + state.emitted_content_block = true; + out.push(json!({ + "contentBlockStart": { + "contentBlockIndex": content_index, + "start": {"toolUse": { + "toolUseId": tool.id.clone().unwrap_or_default(), + "name": name + }} + } + })); + } + if let Some(delta) = arguments_delta { + tool.arguments.push_str(&delta); + out.push(json!({ + "contentBlockDelta": { + "contentBlockIndex": tool.content_index.unwrap_or(index), + "delta": {"toolUse": {"input": delta}} + } + })); + } + out +} + +fn finish_bedrock_stream(state: &mut StreamTranslationState) -> Vec { + if state.finished { + return Vec::new(); + } + let mut out = Vec::new(); + if !state.emitted_message_start { + out.push(json!({"messageStart": {"role": "assistant"}})); + state.emitted_message_start = true; + } + if let Some(index) = state.text_block_index.take() { + out.push(json!({"contentBlockStop": {"contentBlockIndex": index}})); + } + if let Some(index) = state.reasoning_block_index.take() { + out.push(json!({"contentBlockStop": {"contentBlockIndex": index}})); + } + for tool in state.tool_states.values_mut() { + if tool.started { + if let Some(index) = tool.content_index { + out.push(json!({"contentBlockStop": {"contentBlockIndex": index}})); + } + tool.started = false; + } + } + out.push(json!({ + "messageStop": {"stopReason": bedrock_stop_reason(state.stop_reason.as_deref())} + })); + out.push(json!({"metadata": {"usage": encode_usage(state)}})); + state.finished = true; + out +} + +fn capture_usage(state: &mut StreamTranslationState, value: &Value) { + let value = value.as_object(); + state.usage.input_tokens = value + .and_then(|value| value.get("inputTokens")) + .and_then(Value::as_u64); + state.usage.output_tokens = value + .and_then(|value| value.get("outputTokens")) + .and_then(Value::as_u64); + state.usage.total_tokens = value + .and_then(|value| value.get("totalTokens")) + .and_then(Value::as_u64); + if let Some(tokens) = value + .and_then(|value| value.get("cacheReadInputTokens")) + .and_then(Value::as_u64) + { + state.usage.set_cached_input_tokens(tokens); + } + if let Some(tokens) = value + .and_then(|value| value.get("cacheWriteInputTokens")) + .and_then(Value::as_u64) + { + state.usage.set_cache_creation_input_tokens(tokens); + } + state.saw_backend_usage = true; +} + +fn encode_usage(state: &StreamTranslationState) -> Value { + let output_tokens = state + .usage + .output_tokens + .unwrap_or(state.output_tokens_seen); + let input_tokens = state.usage.input_tokens.unwrap_or(0); + json!({ + "inputTokens": input_tokens, + "outputTokens": output_tokens, + "totalTokens": state + .usage + .total_tokens + .unwrap_or(input_tokens.saturating_add(output_tokens)), + "cacheReadInputTokens": state.usage.cached_input_tokens(), + "cacheWriteInputTokens": state.usage.cache_creation_input_tokens(), + }) +} + +fn bedrock_stop_reason(reason: Option<&str>) -> &'static str { + match reason { + Some("max_tokens" | "length") => "max_tokens", + Some("tool_use" | "tool_calls") => "tool_use", + Some("content_filter" | "content_filtered" | "guardrail_intervened") => "content_filtered", + _ => "end_turn", + } +} diff --git a/crates/switchyard-translation/src/codecs/mod.rs b/crates/switchyard-translation/src/codecs/mod.rs index 78d3a007c..59b02ef5d 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; pub mod responses; diff --git a/crates/switchyard-translation/src/codecs/stream.rs b/crates/switchyard-translation/src/codecs/stream.rs index 0ec9b7f88..cb8a0a3b1 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::stream::BedrockConverseStreamCodec; use crate::codecs::openai_chat::OpenAiChatStreamCodec; use crate::codecs::responses::OpenAiResponsesStreamCodec; use crate::engine::{FormatRegistry, TranslationEngine}; @@ -175,6 +176,7 @@ impl StreamCodecRegistry { pub fn with_builtins() -> Self { let mut registry = Self::new(); registry.register(OpenAiChatStreamCodec); + registry.register(BedrockConverseStreamCodec); registry.register(AnthropicMessagesStreamCodec); registry.register(OpenAiResponsesStreamCodec); registry diff --git a/crates/switchyard-translation/src/engine.rs b/crates/switchyard-translation/src/engine.rs index abd30955c..1c090ff81 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::{ @@ -59,6 +60,7 @@ impl FormatRegistry { pub fn with_builtins() -> Self { let mut registry = Self::new(); registry.register(OpenAiChatCodec); + registry.register(BedrockConverseCodec); registry.register(AnthropicMessagesCodec); registry.register(OpenAiResponsesCodec); registry diff --git a/crates/switchyard-translation/src/helpers.rs b/crates/switchyard-translation/src/helpers.rs index 061517e24..a196dde9e 100644 --- a/crates/switchyard-translation/src/helpers.rs +++ b/crates/switchyard-translation/src/helpers.rs @@ -207,9 +207,45 @@ fn stamp_streamed_response_model( message.insert("model".to_string(), Value::String(served_model.to_string())); } } + WireFormat::BedrockConverse => {} } } +/// Decodes a stream of already de-framed provider JSON events into neutral IR chunks. +/// +/// HTTP hosts use this when the carrier is not SSE, such as Amazon Bedrock's +/// AWS EventStream transport. The host remains responsible for validating and +/// removing carrier framing; Switchyard owns only the provider event semantics. +pub fn decode_event_stream( + events: S, + source: WireFormat, +) -> std::result::Result +where + S: Stream> + Send + 'static, +{ + let source_format: FormatId = source.into(); + let codec = StreamCodecRegistry::with_builtins() + .codec(source_format.clone()) + .map_err(|error| LlmClientError::ResponseTranslation(error.to_string()))?; + let mut state = StreamTranslationState { + source: Some(source_format.clone()), + ..StreamTranslationState::default() + }; + let stream = Box::pin(try_stream! { + futures::pin_mut!(events); + while let Some(event) = events.next().await { + let value = event?; + let normalized = codec.decode_event(&mut state, &value); + yield LlmResponseStreamEvent::preserved( + source_format.clone(), + value, + normalized, + ); + } + }); + Ok(stream) +} + /// Decodes provider SSE bytes into normalized stream events. /// /// Operates on raw bytes, not any HTTP client type: the caller adapts its @@ -346,8 +382,8 @@ mod tests { }; use super::{ - decode_aggregated_response, decode_request, decode_stream, encode_aggregated_response, - encode_request, encode_stream, stamp_streamed_response_model, + decode_aggregated_response, decode_event_stream, decode_request, decode_stream, + encode_aggregated_response, encode_request, encode_stream, stamp_streamed_response_model, }; use crate::{LlmResponseStream, LlmStreamError, WireFormat}; @@ -873,6 +909,37 @@ mod tests { Ok(()) } + #[test] + fn decode_event_stream_accepts_deframed_bedrock_events() -> Result<(), BoxError> { + let values = stream::iter(vec![ + Ok::(json!({"messageStart": {"role": "assistant"}})), + Ok(json!({"contentBlockDelta": { + "contentBlockIndex": 0, + "delta": {"text": "bedrock"} + }})), + Ok(json!({"messageStop": {"stopReason": "end_turn"}})), + ]); + let chunks = + block_on(decode_event_stream(values, WireFormat::BedrockConverse)?.collect::>()) + .into_iter() + .collect::, _>>()?; + + assert_eq!(text_of(&chunks), "bedrock"); + assert!( + chunks + .iter() + .flat_map(LlmResponseStreamEvent::normalized) + .any(|chunk| matches!( + chunk, + LlmResponseChunk::MessageStop { + reason: Some(reason), + .. + } if reason == "end_turn" + )) + ); + Ok(()) + } + #[test] fn decode_stream_propagates_source_errors() -> Result<(), BoxError> { // A transport error mid-stream surfaces as an error item, not a panic. diff --git a/crates/switchyard-translation/src/sse.rs b/crates/switchyard-translation/src/sse.rs index bf24aabbb..90504c221 100644 --- a/crates/switchyard-translation/src/sse.rs +++ b/crates/switchyard-translation/src/sse.rs @@ -63,6 +63,10 @@ pub(crate) fn is_terminal_event(format: WireFormat, event: &Value) -> bool { .and_then(Value::as_str), Some("response.completed" | "response.incomplete" | "response.failed") ), + // Bedrock Converse reaches Switchyard through `decode_event_stream`, not this SSE + // path, because its carrier is AWS EventStream. Its semantic terminal is still + // `messageStop`; a trailing `metadata` event may follow it. + WireFormat::BedrockConverse => event.get("messageStop").is_some(), } } diff --git a/crates/switchyard-translation/tests/lossless_roundtrip.rs b/crates/switchyard-translation/tests/lossless_roundtrip.rs index c9375b481..c4bc22a54 100644 --- a/crates/switchyard-translation/tests/lossless_roundtrip.rs +++ b/crates/switchyard-translation/tests/lossless_roundtrip.rs @@ -9,10 +9,11 @@ use switchyard_translation::{ PRESERVATION_METADATA_KEY, PreservationPolicy, TranslationEngine, TranslationPolicy, WireFormat, }; -const FORMATS: [WireFormat; 3] = [ +const FORMATS: [WireFormat; 4] = [ WireFormat::OpenAiChat, WireFormat::AnthropicMessages, WireFormat::OpenAiResponses, + WireFormat::BedrockConverse, ]; type TestResult = std::result::Result<(), Box>; @@ -248,14 +249,24 @@ fn embed_policy() -> TranslationPolicy { } } -// Returns both possible two-hop orders through the non-source formats. -fn distinct_hop_orders(source: WireFormat) -> [(WireFormat, WireFormat); 2] { +// Returns every ordered two-hop path through distinct non-source formats. +fn distinct_hop_orders(source: WireFormat) -> Vec<(WireFormat, WireFormat)> { let others = FORMATS .iter() .copied() .filter(|format| *format != source) .collect::>(); - [(others[0], others[1]), (others[1], others[0])] + others + .iter() + .copied() + .flat_map(|first| { + others + .iter() + .copied() + .filter(move |second| *second != first) + .map(move |second| (first, second)) + }) + .collect() } // Asserts a translated body carries the exact original body in preservation metadata. @@ -283,6 +294,7 @@ fn format_key(format: WireFormat) -> &'static str { WireFormat::OpenAiChat => "openai_chat", WireFormat::AnthropicMessages => "anthropic_messages", WireFormat::OpenAiResponses => "openai_responses", + WireFormat::BedrockConverse => "bedrock_converse", } } @@ -537,6 +549,44 @@ fn request_fixture(format: WireFormat) -> Value { "store": false, "truncation": "auto" }), + WireFormat::BedrockConverse => json!({ + "system": [{"text": "Follow exact instructions."}], + "messages": [ + {"role": "user", "content": [{"text": "Inspect this payload."}]}, + {"role": "assistant", "content": [{ + "toolUse": { + "toolUseId": "call_lookup", + "name": "lookup", + "input": {"query": "rust"} + } + }]}, + {"role": "user", "content": [{ + "toolResult": { + "toolUseId": "call_lookup", + "content": [{"text": "result"}], + "status": "success" + } + }]} + ], + "inferenceConfig": { + "maxTokens": 888, + "temperature": 0.2, + "topP": 0.91, + "stopSequences": ["STOP"] + }, + "toolConfig": { + "tools": [{ + "toolSpec": { + "name": "lookup", + "description": "Lookup data", + "inputSchema": {"json": {"type": "object"}} + } + }], + "toolChoice": {"auto": {}} + }, + "additionalModelRequestFields": {"top_k": 20}, + "requestMetadata": {"trace": "bedrock-request"} + }), } } @@ -676,5 +726,29 @@ fn response_fixture(format: WireFormat) -> Value { "metadata": {"trace": "responses-response", "kept": {"nested": true}}, "service_tier": "default" }), + WireFormat::BedrockConverse => json!({ + "output": { + "message": { + "role": "assistant", + "content": [ + {"text": "Here is the answer."}, + {"toolUse": { + "toolUseId": "call_lookup", + "name": "lookup", + "input": {"query": "rust"} + }} + ] + } + }, + "stopReason": "tool_use", + "usage": { + "inputTokens": 10, + "outputTokens": 5, + "totalTokens": 15, + "cacheReadInputTokens": 2 + }, + "metrics": {"latencyMs": 42}, + "additionalModelResponseFields": {"trace": "bedrock-response"} + }), } } diff --git a/crates/switchyard-translation/tests/request_translation.rs b/crates/switchyard-translation/tests/request_translation.rs index d27b21cd9..af896d7dc 100644 --- a/crates/switchyard-translation/tests/request_translation.rs +++ b/crates/switchyard-translation/tests/request_translation.rs @@ -2389,3 +2389,51 @@ fn responses_flat_file_data_survives_into_chat() -> TestResult { assert_eq!(file["file"]["filename"], "report.pdf"); Ok(()) } + +// Bedrock Converse owns model placement in the HTTP path, so the JSON body contains +// only the provider-neutral conversation contract. +#[test] +fn openai_chat_translates_to_bedrock_converse_request() -> TestResult { + let engine = TranslationEngine::default(); + let body = json!({ + "model": "route-name", + "messages": [ + {"role": "system", "content": "Be concise."}, + {"role": "user", "content": "Hello"} + ], + "tools": [{ + "type": "function", + "function": { + "name": "lookup", + "description": "Lookup data", + "parameters": {"type": "object"} + } + }], + "tool_choice": "auto", + "max_tokens": 64, + "temperature": 0.2 + }); + + let output = engine + .translate_request( + WireFormat::OpenAiChat, + WireFormat::BedrockConverse, + &body, + &TranslationPolicy::default(), + )? + .body; + + assert!(output.get("model").is_none()); + assert_eq!(output["system"], json!([{"text": "Be concise."}])); + assert_eq!( + output["messages"][0], + json!({"role": "user", "content": [{"text": "Hello"}]}) + ); + assert_eq!( + output["toolConfig"]["tools"][0]["toolSpec"]["name"], + "lookup" + ); + assert_eq!(output["toolConfig"]["toolChoice"], json!({"auto": {}})); + assert_eq!(output["inferenceConfig"]["maxTokens"], 64); + Ok(()) +} diff --git a/crates/switchyard-translation/tests/response_translation.rs b/crates/switchyard-translation/tests/response_translation.rs index f261b5050..8958b941c 100644 --- a/crates/switchyard-translation/tests/response_translation.rs +++ b/crates/switchyard-translation/tests/response_translation.rs @@ -724,3 +724,43 @@ fn content_filter_and_refusal_translate_across_formats() -> TestResult { assert_eq!(output["stop_details"]["category"], "cyber"); Ok(()) } + +#[test] +fn bedrock_converse_response_translates_to_openai_chat() -> TestResult { + let engine = TranslationEngine::default(); + let body = json!({ + "output": { + "message": { + "role": "assistant", + "content": [ + {"text": "Let me check."}, + {"toolUse": { + "toolUseId": "tool-1", + "name": "lookup", + "input": {"query": "weather"} + }} + ] + } + }, + "stopReason": "tool_use", + "usage": {"inputTokens": 8, "outputTokens": 4, "totalTokens": 12} + }); + + let output = engine + .translate_response( + WireFormat::BedrockConverse, + WireFormat::OpenAiChat, + &body, + &TranslationPolicy::default(), + )? + .body; + + assert_eq!(output["choices"][0]["message"]["content"], "Let me check."); + assert_eq!( + output["choices"][0]["message"]["tool_calls"][0]["function"]["name"], + "lookup" + ); + assert_eq!(output["choices"][0]["finish_reason"], "tool_calls"); + assert_eq!(output["usage"]["total_tokens"], 12); + Ok(()) +} diff --git a/crates/switchyard-translation/tests/stream_translation.rs b/crates/switchyard-translation/tests/stream_translation.rs index 3c043d59a..973cb0c8d 100644 --- a/crates/switchyard-translation/tests/stream_translation.rs +++ b/crates/switchyard-translation/tests/stream_translation.rs @@ -1546,3 +1546,46 @@ fn responses_decode_emits_tool_arguments_once() -> TestResult { assert_eq!(seen, arguments); Ok(()) } + +#[test] +fn bedrock_converse_stream_translates_to_openai_chat() -> TestResult { + let engine = TranslationEngine::default(); + let mut state = + StreamTranslationState::new(WireFormat::BedrockConverse, WireFormat::OpenAiChat); + let events = [ + json!({"messageStart": {"role": "assistant"}}), + json!({"contentBlockDelta": { + "contentBlockIndex": 0, + "delta": {"text": "Hello"} + }}), + json!({"messageStop": {"stopReason": "end_turn"}}), + json!({"metadata": { + "usage": {"inputTokens": 3, "outputTokens": 1, "totalTokens": 4} + }}), + ]; + let mut translated = Vec::new(); + for event in events { + translated.extend(engine.translate_event( + &mut state, + WireFormat::BedrockConverse, + WireFormat::OpenAiChat, + &event, + )?); + } + translated.extend(engine.finish_stream(&mut state, WireFormat::OpenAiChat)?); + + assert!(translated.iter().any(|event| { + event["choices"][0]["delta"]["content"] == Value::String("Hello".to_string()) + })); + assert!( + translated + .iter() + .any(|event| event["choices"][0]["finish_reason"] == "stop") + ); + let usage = translated + .iter() + .find(|event| event["usage"]["total_tokens"] == 4) + .ok_or("missing usage chat event")?; + assert_eq!(usage["usage"]["total_tokens"], 4); + Ok(()) +}