From b7cf353725153b92b6e9b2e17d479e5524d4b2c2 Mon Sep 17 00:00:00 2001 From: Clement Pakkam Isaac Date: Thu, 3 Sep 2026 17:00:36 -0700 Subject: [PATCH 1/3] feat(observability): expose translation diagnostics Signed-off-by: Clement Pakkam Isaac --- crates/libsy-llm-client/src/client.rs | 39 +++++-- crates/libsy-llm-client/src/metrics.rs | 65 ++++++++++- .../libsy-llm-client/tests/observability.rs | 62 ++++++++++ crates/switchyard-server/src/lib.rs | 24 +++- crates/switchyard-server/src/response.rs | 13 ++- crates/switchyard-server/tests/server.rs | 110 +++++++++++++++--- crates/switchyard-translation/src/helpers.rs | 86 ++++++++++---- docs/internal/metrics_reference.md | 16 +++ 8 files changed, 364 insertions(+), 51 deletions(-) diff --git a/crates/libsy-llm-client/src/client.rs b/crates/libsy-llm-client/src/client.rs index de7f2ca6c..06a6eaf7f 100644 --- a/crates/libsy-llm-client/src/client.rs +++ b/crates/libsy-llm-client/src/client.rs @@ -19,8 +19,9 @@ use switchyard_protocol::{ Response, RoutedLlmClient, }; use switchyard_translation::{ - WireFormat, decode_aggregated_response, decode_request, decode_stream, - encode_aggregated_response_with_extensions, encode_request, encode_stream_with_extensions, + WireFormat, decode_aggregated_response_with_diagnostics, decode_request_with_diagnostics, + decode_stream, encode_aggregated_response_with_extensions_and_diagnostics, + encode_request_with_diagnostics, encode_stream_with_extensions, }; use tracing::Instrument; @@ -240,8 +241,14 @@ impl TranslatingLlmClient { model: &ModelId, endpoint: UpstreamEndpoint, ) -> Result { - let mut body = encode_request(&llm_request, wire_format) + let encoded = encode_request_with_diagnostics(&llm_request, wire_format) .map_err(|error| LlmClientError::RequestEncoding(error.to_string()))?; + metrics::record_translation_diagnostics( + &encoded.diagnostics, + metrics::TranslationOperation::RequestEncode, + wire_format, + ); + let mut body = encoded.body; // `encode_request` round-trips a preserved same-format body verbatim, // which keeps the caller's original `model`; force the resolved model so // the upstream always sees the target id. @@ -501,9 +508,14 @@ impl TranslatingLlmClient { let body = serde_json::from_slice::(&body).map_err(|error| { LlmClientError::ResponseTranslation(format!("invalid upstream JSON: {error}")) })?; - let agg = decode_aggregated_response(&body, wire_format) + let decoded = decode_aggregated_response_with_diagnostics(&body, wire_format) .map_err(|error| LlmClientError::ResponseTranslation(error.to_string()))?; - LlmResponse::Agg(agg) + metrics::record_translation_diagnostics( + &decoded.diagnostics, + metrics::TranslationOperation::ResponseDecode, + wire_format, + ); + LlmResponse::Agg(decoded.response) } }; @@ -534,8 +546,14 @@ impl TranslatingLlmClient { model: Option<&ModelId>, wire_format: WireFormat, ) -> Result { - let llm_request = decode_request(wire_format, &raw_http_request) + let decoded = decode_request_with_diagnostics(wire_format, &raw_http_request) .map_err(|error| LlmClientError::RequestTranslation(error.to_string()))?; + metrics::record_translation_diagnostics( + &decoded.diagnostics, + metrics::TranslationOperation::RequestDecode, + wire_format, + ); + let llm_request = decoded.request; let request_extensions = llm_request.extensions.clone(); // The model that serves the call — the rewrite target when the caller pinned // one, else the request's own model. Mirrors `call_rewrite_model`'s own @@ -562,14 +580,19 @@ impl TranslatingLlmClient { match response.llm_response { LlmResponse::Agg(agg) => { - let body = encode_aggregated_response_with_extensions( + let encoded = encode_aggregated_response_with_extensions_and_diagnostics( &agg, wire_format, served_model.as_deref(), &request_extensions, ) .map_err(|error| LlmClientError::ResponseTranslation(error.to_string()))?; - Ok(RawResponse::Buffered(body)) + metrics::record_translation_diagnostics( + &encoded.diagnostics, + metrics::TranslationOperation::ResponseEncode, + wire_format, + ); + Ok(RawResponse::Buffered(encoded.body)) } LlmResponse::Stream(chunks) => { let events = encode_stream_with_extensions( diff --git a/crates/libsy-llm-client/src/metrics.rs b/crates/libsy-llm-client/src/metrics.rs index 76730ebfe..c2cc20445 100644 --- a/crates/libsy-llm-client/src/metrics.rs +++ b/crates/libsy-llm-client/src/metrics.rs @@ -12,12 +12,34 @@ use std::{ use opentelemetry::metrics::ObservableGauge; use opentelemetry::{KeyValue, global}; use switchyard_libsy::Result; -use switchyard_protocol::{ModelId, Response}; +use switchyard_protocol::{ModelId, Response, WireFormat}; +use switchyard_translation::{DiagnosticSeverity, TranslationDiagnostic}; static TOTAL_REQUESTS: AtomicU64 = AtomicU64::new(0); static TOTAL_ERRORS: AtomicU64 = AtomicU64::new(0); static TOTAL_GAUGES: OnceLock<(ObservableGauge, ObservableGauge)> = OnceLock::new(); +/// Runtime boundary at which a buffered translation diagnostic was emitted. +#[derive(Clone, Copy, Debug, Eq, PartialEq)] +pub enum TranslationOperation { + RequestDecode, + RequestEncode, + ResponseDecode, + ResponseEncode, +} + +impl TranslationOperation { + /// Returns the bounded metric-label value for this operation. + pub const fn as_str(self) -> &'static str { + match self { + Self::RequestDecode => "request_decode", + Self::RequestEncode => "request_encode", + Self::ResponseDecode => "response_decode", + Self::ResponseEncode => "response_encode", + } + } +} + /// Registers process-wide compatibility gauges with the installed global meter provider. pub fn initialize() { TOTAL_GAUGES.get_or_init(|| { @@ -104,6 +126,47 @@ pub(crate) fn record_retry_recovered() { .add(1, &[]); } +/// Records translation diagnostics without putting request-derived values in metric labels. +pub fn record_translation_diagnostics( + diagnostics: &[TranslationDiagnostic], + operation: TranslationOperation, + format: WireFormat, +) { + for diagnostic in diagnostics { + let severity = diagnostic_severity_label(&diagnostic.severity); + global::meter("switchyard") + .u64_counter("switchyard.translation_diagnostics") + .build() + .add( + 1, + &[ + KeyValue::new("code", diagnostic.code.clone()), + KeyValue::new("format", format.as_str()), + KeyValue::new("operation", operation.as_str()), + KeyValue::new("severity", severity), + ], + ); + tracing::warn!( + target: "libsy", + code = %diagnostic.code, + format = format.as_str(), + operation = operation.as_str(), + severity, + diagnostic = %diagnostic.message, + path = diagnostic.path.as_deref().unwrap_or(""), + "LLM protocol translation emitted a diagnostic" + ); + } +} + +const fn diagnostic_severity_label(severity: &DiagnosticSeverity) -> &'static str { + match severity { + DiagnosticSeverity::Info => "info", + DiagnosticSeverity::Warning => "warning", + DiagnosticSeverity::Error => "error", + } +} + /// Records the time needed to produce the routing outcome, including classifier calls, /// target resolution, request rewrites, and decision publishing. pub(crate) fn record_routing_overhead(algorithm: &str, overhead: Duration) { diff --git a/crates/libsy-llm-client/tests/observability.rs b/crates/libsy-llm-client/tests/observability.rs index 4d3921afa..ccd437b62 100644 --- a/crates/libsy-llm-client/tests/observability.rs +++ b/crates/libsy-llm-client/tests/observability.rs @@ -38,6 +38,7 @@ use switchyard_libsy::{ LlmClassifierConfig, LlmTaskClassifier, PickerMode, RoutingOutcome, StageRouter, StageRouterConfig, Step, TaskClassifierConfig, }; +use switchyard_llm_client::metrics::{TranslationOperation, record_translation_diagnostics}; use switchyard_llm_client::{ClientRouter, RunObservation, RunObserver}; use switchyard_protocol::ModelId; use switchyard_protocol::{ @@ -48,6 +49,7 @@ use switchyard_protocol::{ LlmClientError, LlmResponseChunk, LlmResponseStreamEvent, StopReason, text_request, text_response, }; +use switchyard_translation::TranslationDiagnostic; #[derive(Debug, thiserror::Error)] #[error("{0}")] @@ -647,6 +649,66 @@ fn otel_attribute<'a>(span: &'a SpanData, key: &str) -> Option<&'a OtelValue> { .map(|attribute| &attribute.value) } +#[tokio::test] +async fn translation_diagnostics_emit_a_metric_and_structured_warning() { + let _guard = serialize_test().lock().await; + let (store, exporter, provider, _, _) = telemetry(); + let attributes = [ + ("code", "lossy_conversion"), + ("format", "anthropic_messages"), + ("operation", "request_encode"), + ("severity", "warning"), + ]; + let before = u64_counter_value( + &flushed_metrics(exporter, provider), + "switchyard.translation_diagnostics", + &attributes, + ) + .unwrap_or_default(); + let event_count = store.events().len(); + + record_translation_diagnostics( + &[TranslationDiagnostic::warning( + "lossy_conversion", + "Anthropic structured output dropped unsupported JSON Schema constraints", + ) + .at_path("$.response_format")], + TranslationOperation::RequestEncode, + WireFormat::AnthropicMessages, + ); + + let after = u64_counter_value( + &flushed_metrics(exporter, provider), + "switchyard.translation_diagnostics", + &attributes, + ); + assert_eq!(after, Some(before + 1)); + let events = store.events(); + assert!( + events[event_count..].iter().any(|event| { + event.target == "libsy" + && event.level == "WARN" + && event + .fields + .get("code") + .is_some_and(|value| value == "lossy_conversion") + && event + .fields + .get("format") + .is_some_and(|value| value == "anthropic_messages") + && event + .fields + .get("operation") + .is_some_and(|value| value == "request_encode") + && event.fields.get("diagnostic").is_some_and(|value| { + value.contains("dropped unsupported JSON Schema constraints") + }) + }), + "no structured translation warning in {:?}", + &events[event_count..] + ); +} + #[tokio::test] async fn affinity_warns_once_when_request_has_no_usable_identity() -> switchyard_libsy::Result<()> { let _guard = serialize_test().lock().await; diff --git a/crates/switchyard-server/src/lib.rs b/crates/switchyard-server/src/lib.rs index 8b6f06de8..1dd562974 100644 --- a/crates/switchyard-server/src/lib.rs +++ b/crates/switchyard-server/src/lib.rs @@ -37,6 +37,7 @@ use libsy::{Algorithm, LibsyError, RoutingOutcome}; use parking_lot::Mutex; use serde::{Deserialize, Serialize}; use serde_json::{Value, json}; +use switchyard_llm_client::metrics::{TranslationOperation, record_translation_diagnostics}; use switchyard_llm_client::{AuxiliaryOperation, ClientRouter, RunObservation, RunObserver}; use switchyard_protocol::{LlmClientError, Metadata, ModelId, Request, Usage}; use switchyard_runner::{ @@ -46,7 +47,9 @@ use tokio::net::{TcpListener, TcpSocket}; use tokio::task; use tracing::{Instrument, Level}; -use switchyard_translation::{WireFormat, decode_request, encode_aggregated_response}; +use switchyard_translation::{ + WireFormat, decode_request_with_diagnostics, encode_aggregated_response_with_diagnostics, +}; use crate::response::into_http_response; use crate::stats::{StatsAccumulator, StatsSnapshot, prefix_probe, tracking_enabled_from_env}; @@ -738,12 +741,19 @@ async fn decision( // The request moved into the decision run, so its namespace mapping // is gone by here. A Codex tool call in this preview keeps its // qualified name. - match encode_aggregated_response( + match encode_aggregated_response_with_diagnostics( &aggregate, input_format, outcome.selected_model_id().ok().map(ModelId::as_str), ) { - Ok(response) => Some(response), + Ok(encoded) => { + record_translation_diagnostics( + &encoded.diagnostics, + TranslationOperation::ResponseEncode, + input_format, + ); + Some(encoded.body) + } Err(error) => return server_error(error.to_string()), } } @@ -959,8 +969,14 @@ fn resolve_route( body: Value, wire_format: WireFormat, ) -> std::result::Result<(&Route, Request), Response> { - let llm_request = decode_request(wire_format, &body) + let decoded = decode_request_with_diagnostics(wire_format, &body) .map_err(|error| invalid_body_error(StatusCode::BAD_REQUEST, error.to_string()))?; + record_translation_diagnostics( + &decoded.diagnostics, + TranslationOperation::RequestDecode, + wire_format, + ); + let llm_request = decoded.request; let requested_model = llm_request .model .clone() diff --git a/crates/switchyard-server/src/response.rs b/crates/switchyard-server/src/response.rs index 950ec6399..9fdf74edf 100644 --- a/crates/switchyard-server/src/response.rs +++ b/crates/switchyard-server/src/response.rs @@ -7,9 +7,11 @@ use std::error::Error; use axum::Json; use axum::response::{IntoResponse, Response as HttpResponse}; +use switchyard_llm_client::metrics::{TranslationOperation, record_translation_diagnostics}; use switchyard_protocol::{LlmResponse, ProviderExtensions, Response as AlgorithmResponse}; use switchyard_translation::{ - WireFormat, encode_aggregated_response_with_extensions, encode_stream_with_extensions, + WireFormat, encode_aggregated_response_with_extensions_and_diagnostics, + encode_stream_with_extensions, }; use crate::sse::frame_stream; @@ -27,13 +29,18 @@ pub(crate) fn into_http_response( ) -> Result { match response.llm_response { LlmResponse::Agg(response) => { - let body = encode_aggregated_response_with_extensions( + let encoded = encode_aggregated_response_with_extensions_and_diagnostics( &response, target_format, served_model.as_deref(), &request_extensions, )?; - Ok(Json(body).into_response()) + record_translation_diagnostics( + &encoded.diagnostics, + TranslationOperation::ResponseEncode, + target_format, + ); + Ok(Json(encoded.body).into_response()) } LlmResponse::Stream(stream) => { let events = encode_stream_with_extensions( diff --git a/crates/switchyard-server/tests/server.rs b/crates/switchyard-server/tests/server.rs index 0f6567092..ee5ec7ad7 100644 --- a/crates/switchyard-server/tests/server.rs +++ b/crates/switchyard-server/tests/server.rs @@ -344,20 +344,21 @@ async fn upstream_messages_requires_forwarded_oauth( Json(body): Json, ) -> HttpResponse { calls.lock().await.push(body.clone()); - let has_expected_headers = headers - .get("authorization") - .and_then(|value| value.to_str().ok()) - == Some("Bearer claude-oauth-token") - && headers - .get("anthropic-beta") - .and_then(|value| value.to_str().ok()) - == Some("oauth-2025-04-20") - && headers - .get("anthropic-version") + let has_expected_headers = body["model"] == "model/anthropic-diagnostics" + || (headers + .get("authorization") .and_then(|value| value.to_str().ok()) - == Some("2023-06-01") - && !headers.contains_key("chatgpt-account-id") - && !headers.contains_key("x-openai-fedramp"); + == Some("Bearer claude-oauth-token") + && headers + .get("anthropic-beta") + .and_then(|value| value.to_str().ok()) + == Some("oauth-2025-04-20") + && headers + .get("anthropic-version") + .and_then(|value| value.to_str().ok()) + == Some("2023-06-01") + && !headers.contains_key("chatgpt-account-id") + && !headers.contains_key("x-openai-fedramp")); if !has_expected_headers { return ( StatusCode::UNAUTHORIZED, @@ -871,6 +872,89 @@ async fn metrics_exposes_switchyard_otel_instruments() -> TestResult { Ok(()) } +// A successful cross-provider request must expose any contract weakening in runtime telemetry. +#[tokio::test] +async fn metrics_exposes_lossy_outbound_request_translation() -> TestResult { + let upstream = MockUpstream::start().await?; + let state = load_test_config(&format!( + r#" +schema_version = 1 + +[llm_clients.anthropic] +format = "anthropic_messages" +base_url = "{base_url}" +max_retries = 0 + +[targets.anthropic] +id = "model/anthropic-diagnostics" +llm_client = "anthropic" + +[routes.diagnostics] +id = "switchyard/diagnostics" +type = "passthrough" +target = "anthropic" +"#, + base_url = upstream.base_url + ))?; + let app = build_switchyard_router(state); + let before = send(&app, "GET", "/metrics", None) + .await? + .text()? + .to_string(); + + let response = send( + &app, + "POST", + "/v1/chat/completions", + Some(json!({ + "model": "switchyard/diagnostics", + "messages": [{"role": "user", "content": "Return constrained JSON"}], + "response_format": { + "type": "json_schema", + "json_schema": { + "name": "answer", + "strict": true, + "schema": { + "type": "object", + "properties": {"answer": {"type": "string", "minLength": 5}}, + "required": ["answer"] + } + } + } + })), + ) + .await?; + assert_eq!(response.status, StatusCode::OK); + let calls = upstream.calls.lock().await; + assert_eq!(calls.len(), 1); + assert!( + calls[0] + .pointer("/output_config/format/schema/properties/answer/minLength") + .is_none() + ); + drop(calls); + + let after = send(&app, "GET", "/metrics", None) + .await? + .text()? + .to_string(); + assert_eq!( + metric_delta( + &before, + &after, + "switchyard_translation_diagnostics_total", + &[ + ("code", "lossy_conversion"), + ("format", "anthropic_messages"), + ("operation", "request_encode"), + ("severity", "warning"), + ], + ), + Some(1.0) + ); + Ok(()) +} + #[tokio::test] async fn accepts_requests_larger_than_the_axum_default_body_limit() -> TestResult { let (_upstream, app) = test_app(&[(ROUTE_MODEL, &["model/a"])]).await?; diff --git a/crates/switchyard-translation/src/helpers.rs b/crates/switchyard-translation/src/helpers.rs index 061517e24..e4bcff35b 100644 --- a/crates/switchyard-translation/src/helpers.rs +++ b/crates/switchyard-translation/src/helpers.rs @@ -19,8 +19,9 @@ use crate::codecs::stream::encode_response_stream_event; use crate::sse; use crate::{ AggLlmResponse, FormatId, LlmRequest, LlmResponseChunk, LlmResponseStream, - LlmResponseStreamEvent, LlmStreamError, Result, StreamCodecRegistry, StreamTranslationState, - TranslationEngine, TranslationPolicy, WireFormat, + LlmResponseStreamEvent, LlmStreamError, RequestIrOutput, ResponseIrOutput, Result, + StreamCodecRegistry, StreamTranslationState, TranslationEngine, TranslationOutput, + TranslationPolicy, WireFormat, }; static DEFAULT_TRANSLATION_POLICY: LazyLock = @@ -30,23 +31,41 @@ static DEFAULT_TRANSLATION_ENGINE: LazyLock = /// Decodes a `wire_format` request body into the neutral IR. pub fn decode_request(wire_format: WireFormat, body: &Value) -> Result { - Ok(DEFAULT_TRANSLATION_ENGINE - .decode_request(wire_format, body, &DEFAULT_TRANSLATION_POLICY)? - .request) + Ok(decode_request_with_diagnostics(wire_format, body)?.request) +} + +/// Decodes a request and retains any diagnostics emitted by the codec. +pub fn decode_request_with_diagnostics( + wire_format: WireFormat, + body: &Value, +) -> Result { + DEFAULT_TRANSLATION_ENGINE.decode_request(wire_format, body, &DEFAULT_TRANSLATION_POLICY) } /// Encodes a normalized request into `wire_format`'s JSON body. pub fn encode_request(request: &LlmRequest, wire_format: WireFormat) -> Result { - Ok(DEFAULT_TRANSLATION_ENGINE - .encode_request(wire_format, request, &DEFAULT_TRANSLATION_POLICY)? - .body) + Ok(encode_request_with_diagnostics(request, wire_format)?.body) +} + +/// Encodes a request and retains any diagnostics emitted by the codec. +pub fn encode_request_with_diagnostics( + request: &LlmRequest, + wire_format: WireFormat, +) -> Result { + DEFAULT_TRANSLATION_ENGINE.encode_request(wire_format, request, &DEFAULT_TRANSLATION_POLICY) } /// Decodes a buffered `wire_format` response body into the neutral aggregate. pub fn decode_aggregated_response(body: &Value, wire_format: WireFormat) -> Result { - Ok(DEFAULT_TRANSLATION_ENGINE - .decode_response(wire_format, body, &DEFAULT_TRANSLATION_POLICY)? - .response) + Ok(decode_aggregated_response_with_diagnostics(body, wire_format)?.response) +} + +/// Decodes a buffered response and retains any diagnostics emitted by the codec. +pub fn decode_aggregated_response_with_diagnostics( + body: &Value, + wire_format: WireFormat, +) -> Result { + DEFAULT_TRANSLATION_ENGINE.decode_response(wire_format, body, &DEFAULT_TRANSLATION_POLICY) } /// Encodes a buffered aggregate into `wire_format`'s JSON body, stamping @@ -57,7 +76,16 @@ pub fn encode_aggregated_response( wire_format: WireFormat, served_model: Option<&str>, ) -> Result { - encode_aggregated_response_with_extensions( + Ok(encode_aggregated_response_with_diagnostics(agg, wire_format, served_model)?.body) +} + +/// Encodes a buffered response and retains any diagnostics emitted by the codec. +pub fn encode_aggregated_response_with_diagnostics( + agg: &AggLlmResponse, + wire_format: WireFormat, + served_model: Option<&str>, +) -> Result { + encode_aggregated_response_with_extensions_and_diagnostics( agg, wire_format, served_model, @@ -75,18 +103,32 @@ pub fn encode_aggregated_response_with_extensions( served_model: Option<&str>, request_extensions: &switchyard_protocol::ProviderExtensions, ) -> Result { - let mut body = DEFAULT_TRANSLATION_ENGINE - .encode_response_with_extensions( - wire_format, - agg, - request_extensions, - &DEFAULT_TRANSLATION_POLICY, - )? - .body; - if let (Some(model), Value::Object(object)) = (served_model, &mut body) { + Ok(encode_aggregated_response_with_extensions_and_diagnostics( + agg, + wire_format, + served_model, + request_extensions, + )? + .body) +} + +/// Encodes a buffered response with request extensions and retains codec diagnostics. +pub fn encode_aggregated_response_with_extensions_and_diagnostics( + agg: &AggLlmResponse, + wire_format: WireFormat, + served_model: Option<&str>, + request_extensions: &switchyard_protocol::ProviderExtensions, +) -> Result { + let mut output = DEFAULT_TRANSLATION_ENGINE.encode_response_with_extensions( + wire_format, + agg, + request_extensions, + &DEFAULT_TRANSLATION_POLICY, + )?; + if let (Some(model), Value::Object(object)) = (served_model, &mut output.body) { object.insert("model".to_string(), Value::String(model.to_string())); } - Ok(body) + Ok(output) } /// A stream of wire-format event objects in one format — the unframed body of an diff --git a/docs/internal/metrics_reference.md b/docs/internal/metrics_reference.md index 782f99f90..6dea9bdc6 100644 --- a/docs/internal/metrics_reference.md +++ b/docs/internal/metrics_reference.md @@ -75,6 +75,17 @@ Each histogram emits `_bucket`, `_sum`, and `_count` series. Use `upstream_5xx`, `upstream_non_5xx`, `invalid_response`, `parse_error`, `client_error`, or `call_error`. The labels never include request or response text. +## Translation diagnostic counter + +| Metric | Type | Meaning | +|---|---|---| +| `switchyard_translation_diagnostics_total{code,format,operation,severity}` | counter | Buffered protocol translations that preserved service while dropping or degrading request or response data. | + +`operation` identifies the runtime boundary: `request_decode`, `request_encode`, +`response_decode`, or `response_encode`. `format` is the wire format being decoded or +encoded. Diagnostic messages and JSON paths are emitted only in structured logs; they are +never metric labels. + ## Outcome counters for error-rate ratios For HTTP-derived responses and attempts, the `outcome` label takes three values: @@ -161,6 +172,10 @@ into label space. | `tier` | Small enumerated set, optional. | Per-endpoint counters and histograms on algorithms that supply it | | `judge_model` | One per configured judge target. | Classifier fail-open counter | | `reason` | Exactly 8 fixed error categories. | Classifier fail-open counter | +| `operation` | Exactly 4 translation boundaries: request/response decode/encode. | Translation diagnostic counter | +| `format` | Exactly 3 built-in wire formats. | Translation diagnostic counter | +| `severity` | Exactly 3 diagnostic levels: `info`, `warning`, `error`. | Translation diagnostic counter | +| `code` | Translation diagnostic identifiers defined by the built-in codecs. | Translation diagnostic counter | ## Triage cheatsheet @@ -171,4 +186,5 @@ into label space. | `switchyard_routing_overhead_ms_count` stuck at `0` | No successful algorithm run has recorded a successful routed model call. | | `switchyard_algorithms_in_flight` stuck above zero with no traffic | Runs are parked on an internal routing call that never returns. Check the classifier or judge target's upstream. | | `switchyard_classifier_fail_open_total` rising | The judge target is failing or returning a response the classifier cannot parse. Check `judge_model` and `reason`. | +| `switchyard_translation_diagnostics_total` rising | Cross-provider translation is dropping or degrading request or response data. Check the matching structured warning for the diagnostic and JSON path. | | `switchyard_client_responses_total{outcome="retryable_error"}` rising | Either the upstream is genuinely flaky, or retries are exhausting; compare client responses with retryable upstream attempts. | From 84dc2ef24f6ae3d0079501ac8f9e39f82d6acc3d Mon Sep 17 00:00:00 2001 From: Clement Pakkam Isaac Date: Thu, 3 Sep 2026 17:00:53 -0700 Subject: [PATCH 2/3] test(observability): tighten translation diagnostic coverage Signed-off-by: Clement Pakkam Isaac --- .../libsy-llm-client/tests/observability.rs | 41 +++++++-- crates/switchyard-server/tests/server.rs | 87 ++++++++++++++++--- crates/switchyard-translation/src/helpers.rs | 20 +++++ 3 files changed, 131 insertions(+), 17 deletions(-) diff --git a/crates/libsy-llm-client/tests/observability.rs b/crates/libsy-llm-client/tests/observability.rs index ccd437b62..2689354ba 100644 --- a/crates/libsy-llm-client/tests/observability.rs +++ b/crates/libsy-llm-client/tests/observability.rs @@ -649,6 +649,7 @@ fn otel_attribute<'a>(span: &'a SpanData, key: &str) -> Option<&'a OtelValue> { .map(|attribute| &attribute.value) } +// Verifies the required metric labels and exactly one structured warning per diagnostic. #[tokio::test] async fn translation_diagnostics_emit_a_metric_and_structured_warning() { let _guard = serialize_test().lock().await; @@ -677,15 +678,41 @@ async fn translation_diagnostics_emit_a_metric_and_structured_warning() { WireFormat::AnthropicMessages, ); + let snapshots = flushed_metrics(exporter, provider); let after = u64_counter_value( - &flushed_metrics(exporter, provider), + &snapshots, "switchyard.translation_diagnostics", &attributes, ); assert_eq!(after, Some(before + 1)); + let mut metric_attribute_keys = snapshots + .iter() + .flat_map(|snapshot| snapshot.scope_metrics()) + .flat_map(|scope| scope.metrics()) + .filter(|metric| metric.name() == "switchyard.translation_diagnostics") + .filter_map(|metric| match metric.data() { + AggregatedMetrics::U64(MetricData::Sum(sum)) => sum + .data_points() + .find(|point| attributes_match(point.attributes(), &attributes)) + .map(|point| { + point + .attributes() + .map(|attribute| attribute.key.as_str().to_string()) + .collect::>() + }), + _ => None, + }) + .next() + .expect("missing translation diagnostic metric attributes"); + metric_attribute_keys.sort_unstable(); + assert_eq!( + metric_attribute_keys, + ["code", "format", "operation", "severity"] + ); let events = store.events(); - assert!( - events[event_count..].iter().any(|event| { + let matching_warnings = events[event_count..] + .iter() + .filter(|event| { event.target == "libsy" && event.level == "WARN" && event @@ -703,9 +730,11 @@ async fn translation_diagnostics_emit_a_metric_and_structured_warning() { && event.fields.get("diagnostic").is_some_and(|value| { value.contains("dropped unsupported JSON Schema constraints") }) - }), - "no structured translation warning in {:?}", - &events[event_count..] + }) + .count(); + assert_eq!( + matching_warnings, 1, + "expected one structured translation warning" ); } diff --git a/crates/switchyard-server/tests/server.rs b/crates/switchyard-server/tests/server.rs index ee5ec7ad7..b7ac2629d 100644 --- a/crates/switchyard-server/tests/server.rs +++ b/crates/switchyard-server/tests/server.rs @@ -344,6 +344,7 @@ async fn upstream_messages_requires_forwarded_oauth( Json(body): Json, ) -> HttpResponse { calls.lock().await.push(body.clone()); + // The diagnostics fixture intentionally has no OAuth credentials. let has_expected_headers = body["model"] == "model/anthropic-diagnostics" || (headers .get("authorization") @@ -897,11 +898,67 @@ target = "anthropic" base_url = upstream.base_url ))?; let app = build_switchyard_router(state); + let diagnostic_labels = [ + ("code", "lossy_conversion"), + ("format", "anthropic_messages"), + ("operation", "request_encode"), + ("severity", "warning"), + ]; let before = send(&app, "GET", "/metrics", None) .await? .text()? .to_string(); + let lossless_response = send( + &app, + "POST", + "/v1/chat/completions", + Some(json!({ + "model": "switchyard/diagnostics", + "messages": [{"role": "user", "content": "Return JSON"}], + "response_format": { + "type": "json_schema", + "json_schema": { + "name": "answer", + "strict": true, + "schema": { + "type": "object", + "properties": {"answer": {"type": "string"}}, + "required": ["answer"] + } + } + } + })), + ) + .await?; + assert_eq!(lossless_response.status, StatusCode::OK); + assert_eq!( + lossless_response.json()?["choices"][0]["message"]["content"], + "ok" + ); + let calls = upstream.calls.lock().await; + assert_eq!(calls.len(), 1); + assert_eq!( + calls[0].pointer("/output_config/format/schema/properties/answer"), + Some(&json!({"type": "string"})) + ); + drop(calls); + + let after_lossless = send(&app, "GET", "/metrics", None) + .await? + .text()? + .to_string(); + assert_eq!( + metric_delta( + &before, + &after_lossless, + "switchyard_translation_diagnostics_total", + &diagnostic_labels, + ) + .unwrap_or_default(), + 0.0 + ); + let response = send( &app, "POST", @@ -925,33 +982,41 @@ target = "anthropic" ) .await?; assert_eq!(response.status, StatusCode::OK); + assert_eq!(response.json()?["choices"][0]["message"]["content"], "ok"); let calls = upstream.calls.lock().await; - assert_eq!(calls.len(), 1); + assert_eq!(calls.len(), 2); assert!( - calls[0] + calls[1] .pointer("/output_config/format/schema/properties/answer/minLength") .is_none() ); drop(calls); - let after = send(&app, "GET", "/metrics", None) + let after_lossy = send(&app, "GET", "/metrics", None) .await? .text()? .to_string(); assert_eq!( metric_delta( - &before, - &after, + &after_lossless, + &after_lossy, "switchyard_translation_diagnostics_total", - &[ - ("code", "lossy_conversion"), - ("format", "anthropic_messages"), - ("operation", "request_encode"), - ("severity", "warning"), - ], + &diagnostic_labels, ), Some(1.0) ); + let diagnostic_line = metric_line( + &after_lossy, + "switchyard_translation_diagnostics_total", + &diagnostic_labels, + ) + .ok_or("missing translation diagnostic metric")?; + for forbidden_label in ["diagnostic=", "path="] { + assert!( + !diagnostic_line.contains(forbidden_label), + "unexpected high-cardinality label in {diagnostic_line}" + ); + } Ok(()) } diff --git a/crates/switchyard-translation/src/helpers.rs b/crates/switchyard-translation/src/helpers.rs index e4bcff35b..68cb69cbe 100644 --- a/crates/switchyard-translation/src/helpers.rs +++ b/crates/switchyard-translation/src/helpers.rs @@ -35,6 +35,10 @@ pub fn decode_request(wire_format: WireFormat, body: &Value) -> Result Result Resu } /// Decodes a buffered response and retains any diagnostics emitted by the codec. +/// +/// # Errors +/// +/// Returns an error when the response codec cannot decode `body`. pub fn decode_aggregated_response_with_diagnostics( body: &Value, wire_format: WireFormat, @@ -80,6 +92,10 @@ pub fn encode_aggregated_response( } /// Encodes a buffered response and retains any diagnostics emitted by the codec. +/// +/// # Errors +/// +/// Returns an error when the response codec cannot encode `agg`. pub fn encode_aggregated_response_with_diagnostics( agg: &AggLlmResponse, wire_format: WireFormat, @@ -113,6 +129,10 @@ pub fn encode_aggregated_response_with_extensions( } /// Encodes a buffered response with request extensions and retains codec diagnostics. +/// +/// # Errors +/// +/// Returns an error when the response codec cannot encode `agg`. pub fn encode_aggregated_response_with_extensions_and_diagnostics( agg: &AggLlmResponse, wire_format: WireFormat, From 322dc84a386f28ac71c10fd4ef85b074575a497c Mon Sep 17 00:00:00 2001 From: Clement Pakkam Isaac Date: Wed, 9 Sep 2026 13:07:55 -0700 Subject: [PATCH 3/3] refactor(observability): keep translation diagnostics internal Signed-off-by: Clement Pakkam Isaac --- crates/libsy-llm-client/src/client.rs | 41 ++---- crates/libsy-llm-client/src/lib.rs | 1 + crates/libsy-llm-client/src/metrics.rs | 90 +++++++++---- crates/libsy-llm-client/src/translation.rs | 48 +++++++ .../libsy-llm-client/tests/observability.rs | 91 -------------- crates/switchyard-server/src/lib.rs | 28 ++--- crates/switchyard-server/src/metrics.rs | 39 ++++++ crates/switchyard-server/src/response.rs | 16 +-- crates/switchyard-server/src/translation.rs | 36 ++++++ crates/switchyard-server/tests/server.rs | 118 ++++++------------ crates/switchyard-translation/src/helpers.rs | 106 ++++------------ 11 files changed, 270 insertions(+), 344 deletions(-) create mode 100644 crates/libsy-llm-client/src/translation.rs create mode 100644 crates/switchyard-server/src/translation.rs diff --git a/crates/libsy-llm-client/src/client.rs b/crates/libsy-llm-client/src/client.rs index 06a6eaf7f..620504f31 100644 --- a/crates/libsy-llm-client/src/client.rs +++ b/crates/libsy-llm-client/src/client.rs @@ -18,17 +18,14 @@ use switchyard_protocol::{ LlmRequest, LlmResponse, LlmResponseChunk, LlmResponseStreamEvent, Metadata, ModelId, Request, Response, RoutedLlmClient, }; -use switchyard_translation::{ - WireFormat, decode_aggregated_response_with_diagnostics, decode_request_with_diagnostics, - decode_stream, encode_aggregated_response_with_extensions_and_diagnostics, - encode_request_with_diagnostics, encode_stream_with_extensions, -}; +use switchyard_translation::{WireFormat, decode_stream, encode_stream_with_extensions}; use tracing::Instrument; use crate::backend::Backend; use crate::error::{LlmClientError, Result}; use crate::metrics; use crate::raw::RawResponse; +use crate::translation; // Headers this client owns or that are hop-by-hop. Backends apply an explicitly // enabled caller credential after generic metadata forwarding skips these. @@ -241,14 +238,8 @@ impl TranslatingLlmClient { model: &ModelId, endpoint: UpstreamEndpoint, ) -> Result { - let encoded = encode_request_with_diagnostics(&llm_request, wire_format) + let mut body = translation::encode_request(&llm_request, wire_format) .map_err(|error| LlmClientError::RequestEncoding(error.to_string()))?; - metrics::record_translation_diagnostics( - &encoded.diagnostics, - metrics::TranslationOperation::RequestEncode, - wire_format, - ); - let mut body = encoded.body; // `encode_request` round-trips a preserved same-format body verbatim, // which keeps the caller's original `model`; force the resolved model so // the upstream always sees the target id. @@ -508,14 +499,9 @@ impl TranslatingLlmClient { let body = serde_json::from_slice::(&body).map_err(|error| { LlmClientError::ResponseTranslation(format!("invalid upstream JSON: {error}")) })?; - let decoded = decode_aggregated_response_with_diagnostics(&body, wire_format) + let agg = translation::decode_response(&body, wire_format) .map_err(|error| LlmClientError::ResponseTranslation(error.to_string()))?; - metrics::record_translation_diagnostics( - &decoded.diagnostics, - metrics::TranslationOperation::ResponseDecode, - wire_format, - ); - LlmResponse::Agg(decoded.response) + LlmResponse::Agg(agg) } }; @@ -546,14 +532,8 @@ impl TranslatingLlmClient { model: Option<&ModelId>, wire_format: WireFormat, ) -> Result { - let decoded = decode_request_with_diagnostics(wire_format, &raw_http_request) + let llm_request = translation::decode_request(wire_format, &raw_http_request) .map_err(|error| LlmClientError::RequestTranslation(error.to_string()))?; - metrics::record_translation_diagnostics( - &decoded.diagnostics, - metrics::TranslationOperation::RequestDecode, - wire_format, - ); - let llm_request = decoded.request; let request_extensions = llm_request.extensions.clone(); // The model that serves the call — the rewrite target when the caller pinned // one, else the request's own model. Mirrors `call_rewrite_model`'s own @@ -580,19 +560,14 @@ impl TranslatingLlmClient { match response.llm_response { LlmResponse::Agg(agg) => { - let encoded = encode_aggregated_response_with_extensions_and_diagnostics( + let body = translation::encode_response( &agg, wire_format, served_model.as_deref(), &request_extensions, ) .map_err(|error| LlmClientError::ResponseTranslation(error.to_string()))?; - metrics::record_translation_diagnostics( - &encoded.diagnostics, - metrics::TranslationOperation::ResponseEncode, - wire_format, - ); - Ok(RawResponse::Buffered(encoded.body)) + Ok(RawResponse::Buffered(body)) } LlmResponse::Stream(chunks) => { let events = encode_stream_with_extensions( diff --git a/crates/libsy-llm-client/src/lib.rs b/crates/libsy-llm-client/src/lib.rs index d5f579c7d..45d0a2feb 100644 --- a/crates/libsy-llm-client/src/lib.rs +++ b/crates/libsy-llm-client/src/lib.rs @@ -24,6 +24,7 @@ mod observability; mod observation; pub mod raw; pub mod run; +mod translation; pub use backend::{Backend, DEFAULT_MAX_RETRIES, HttpBackendConfig}; pub use client::{AuxiliaryOperation, ModelConfig, TranslatingLlmClient}; diff --git a/crates/libsy-llm-client/src/metrics.rs b/crates/libsy-llm-client/src/metrics.rs index c2cc20445..4006f4729 100644 --- a/crates/libsy-llm-client/src/metrics.rs +++ b/crates/libsy-llm-client/src/metrics.rs @@ -19,27 +19,6 @@ static TOTAL_REQUESTS: AtomicU64 = AtomicU64::new(0); static TOTAL_ERRORS: AtomicU64 = AtomicU64::new(0); static TOTAL_GAUGES: OnceLock<(ObservableGauge, ObservableGauge)> = OnceLock::new(); -/// Runtime boundary at which a buffered translation diagnostic was emitted. -#[derive(Clone, Copy, Debug, Eq, PartialEq)] -pub enum TranslationOperation { - RequestDecode, - RequestEncode, - ResponseDecode, - ResponseEncode, -} - -impl TranslationOperation { - /// Returns the bounded metric-label value for this operation. - pub const fn as_str(self) -> &'static str { - match self { - Self::RequestDecode => "request_decode", - Self::RequestEncode => "request_encode", - Self::ResponseDecode => "response_decode", - Self::ResponseEncode => "response_encode", - } - } -} - /// Registers process-wide compatibility gauges with the installed global meter provider. pub fn initialize() { TOTAL_GAUGES.get_or_init(|| { @@ -126,10 +105,10 @@ pub(crate) fn record_retry_recovered() { .add(1, &[]); } -/// Records translation diagnostics without putting request-derived values in metric labels. -pub fn record_translation_diagnostics( +// Records translation diagnostics without putting request-derived values in metric labels. +pub(crate) fn record_translation_diagnostics( diagnostics: &[TranslationDiagnostic], - operation: TranslationOperation, + operation: &'static str, format: WireFormat, ) { for diagnostic in diagnostics { @@ -142,7 +121,7 @@ pub fn record_translation_diagnostics( &[ KeyValue::new("code", diagnostic.code.clone()), KeyValue::new("format", format.as_str()), - KeyValue::new("operation", operation.as_str()), + KeyValue::new("operation", operation), KeyValue::new("severity", severity), ], ); @@ -150,7 +129,7 @@ pub fn record_translation_diagnostics( target: "libsy", code = %diagnostic.code, format = format.as_str(), - operation = operation.as_str(), + operation, severity, diagnostic = %diagnostic.message, path = diagnostic.path.as_deref().unwrap_or(""), @@ -233,8 +212,25 @@ pub(crate) fn record_routed_request( #[cfg(test)] mod tests { + use std::io::{self, Write}; + use std::sync::{Arc, Mutex}; + use super::*; + #[derive(Default)] + struct LogBuffer(Mutex>); + + impl Write for &LogBuffer { + fn write(&mut self, bytes: &[u8]) -> io::Result { + self.0.lock().expect("log buffer lock").extend(bytes); + Ok(bytes.len()) + } + + fn flush(&mut self) -> io::Result<()> { + Ok(()) + } + } + #[test] fn outcome_labels_match_the_retry_policy() { assert_eq!(http_outcome_label(Some(200)), "ok"); @@ -248,4 +244,46 @@ mod tests { } assert_eq!(http_outcome_label(None), "retryable_error"); } + + // One diagnostic produces one warning containing its structured troubleshooting fields. + #[test] + fn translation_diagnostic_emits_one_structured_warning() { + let output = Arc::new(LogBuffer::default()); + let subscriber = tracing_subscriber::fmt() + .with_ansi(false) + .without_time() + .with_writer(output.clone()) + .finish(); + tracing::subscriber::with_default(subscriber, || { + record_translation_diagnostics( + &[TranslationDiagnostic::warning( + "lossy_conversion", + "dropped unsupported JSON Schema constraints", + ) + .at_path("$.response_format")], + "request_encode", + WireFormat::AnthropicMessages, + ); + }); + + let bytes = output.0.lock().expect("log buffer lock").clone(); + let logs = String::from_utf8(bytes).expect("captured log must be UTF-8"); + let warnings = logs + .lines() + .filter(|line| line.contains("LLM protocol translation emitted a diagnostic")) + .collect::>(); + assert_eq!(warnings.len(), 1, "diagnostic warnings: {logs}"); + for field in [ + "WARN", + "libsy", + "lossy_conversion", + "anthropic_messages", + "request_encode", + "warning", + "dropped unsupported JSON Schema constraints", + "$.response_format", + ] { + assert!(warnings[0].contains(field), "missing {field:?} in {logs}"); + } + } } diff --git a/crates/libsy-llm-client/src/translation.rs b/crates/libsy-llm-client/src/translation.rs new file mode 100644 index 000000000..055d53a16 --- /dev/null +++ b/crates/libsy-llm-client/src/translation.rs @@ -0,0 +1,48 @@ +// SPDX-FileCopyrightText: Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved. +// SPDX-License-Identifier: Apache-2.0 + +//! Buffered translation with runtime diagnostics for the HTTP client. + +use std::sync::LazyLock; + +use serde_json::Value; +use switchyard_protocol::{AggLlmResponse, LlmRequest, ProviderExtensions}; +use switchyard_translation::{Result, TranslationEngine, TranslationPolicy, WireFormat}; + +use crate::metrics; + +static ENGINE: LazyLock = LazyLock::new(TranslationEngine::default); +static POLICY: LazyLock = LazyLock::new(TranslationPolicy::default); + +pub(crate) fn decode_request(format: WireFormat, body: &Value) -> Result { + let decoded = ENGINE.decode_request(format, body, &POLICY)?; + metrics::record_translation_diagnostics(&decoded.diagnostics, "request_decode", format); + Ok(decoded.request) +} + +pub(crate) fn encode_request(request: &LlmRequest, format: WireFormat) -> Result { + let encoded = ENGINE.encode_request(format, request, &POLICY)?; + metrics::record_translation_diagnostics(&encoded.diagnostics, "request_encode", format); + Ok(encoded.body) +} + +pub(crate) fn decode_response(body: &Value, format: WireFormat) -> Result { + let decoded = ENGINE.decode_response(format, body, &POLICY)?; + metrics::record_translation_diagnostics(&decoded.diagnostics, "response_decode", format); + Ok(decoded.response) +} + +pub(crate) fn encode_response( + response: &AggLlmResponse, + format: WireFormat, + served_model: Option<&str>, + request_extensions: &ProviderExtensions, +) -> Result { + let mut encoded = + ENGINE.encode_response_with_extensions(format, response, request_extensions, &POLICY)?; + metrics::record_translation_diagnostics(&encoded.diagnostics, "response_encode", format); + if let (Some(model), Value::Object(body)) = (served_model, &mut encoded.body) { + body.insert("model".to_string(), Value::String(model.to_string())); + } + Ok(encoded.body) +} diff --git a/crates/libsy-llm-client/tests/observability.rs b/crates/libsy-llm-client/tests/observability.rs index 2689354ba..4d3921afa 100644 --- a/crates/libsy-llm-client/tests/observability.rs +++ b/crates/libsy-llm-client/tests/observability.rs @@ -38,7 +38,6 @@ use switchyard_libsy::{ LlmClassifierConfig, LlmTaskClassifier, PickerMode, RoutingOutcome, StageRouter, StageRouterConfig, Step, TaskClassifierConfig, }; -use switchyard_llm_client::metrics::{TranslationOperation, record_translation_diagnostics}; use switchyard_llm_client::{ClientRouter, RunObservation, RunObserver}; use switchyard_protocol::ModelId; use switchyard_protocol::{ @@ -49,7 +48,6 @@ use switchyard_protocol::{ LlmClientError, LlmResponseChunk, LlmResponseStreamEvent, StopReason, text_request, text_response, }; -use switchyard_translation::TranslationDiagnostic; #[derive(Debug, thiserror::Error)] #[error("{0}")] @@ -649,95 +647,6 @@ fn otel_attribute<'a>(span: &'a SpanData, key: &str) -> Option<&'a OtelValue> { .map(|attribute| &attribute.value) } -// Verifies the required metric labels and exactly one structured warning per diagnostic. -#[tokio::test] -async fn translation_diagnostics_emit_a_metric_and_structured_warning() { - let _guard = serialize_test().lock().await; - let (store, exporter, provider, _, _) = telemetry(); - let attributes = [ - ("code", "lossy_conversion"), - ("format", "anthropic_messages"), - ("operation", "request_encode"), - ("severity", "warning"), - ]; - let before = u64_counter_value( - &flushed_metrics(exporter, provider), - "switchyard.translation_diagnostics", - &attributes, - ) - .unwrap_or_default(); - let event_count = store.events().len(); - - record_translation_diagnostics( - &[TranslationDiagnostic::warning( - "lossy_conversion", - "Anthropic structured output dropped unsupported JSON Schema constraints", - ) - .at_path("$.response_format")], - TranslationOperation::RequestEncode, - WireFormat::AnthropicMessages, - ); - - let snapshots = flushed_metrics(exporter, provider); - let after = u64_counter_value( - &snapshots, - "switchyard.translation_diagnostics", - &attributes, - ); - assert_eq!(after, Some(before + 1)); - let mut metric_attribute_keys = snapshots - .iter() - .flat_map(|snapshot| snapshot.scope_metrics()) - .flat_map(|scope| scope.metrics()) - .filter(|metric| metric.name() == "switchyard.translation_diagnostics") - .filter_map(|metric| match metric.data() { - AggregatedMetrics::U64(MetricData::Sum(sum)) => sum - .data_points() - .find(|point| attributes_match(point.attributes(), &attributes)) - .map(|point| { - point - .attributes() - .map(|attribute| attribute.key.as_str().to_string()) - .collect::>() - }), - _ => None, - }) - .next() - .expect("missing translation diagnostic metric attributes"); - metric_attribute_keys.sort_unstable(); - assert_eq!( - metric_attribute_keys, - ["code", "format", "operation", "severity"] - ); - let events = store.events(); - let matching_warnings = events[event_count..] - .iter() - .filter(|event| { - event.target == "libsy" - && event.level == "WARN" - && event - .fields - .get("code") - .is_some_and(|value| value == "lossy_conversion") - && event - .fields - .get("format") - .is_some_and(|value| value == "anthropic_messages") - && event - .fields - .get("operation") - .is_some_and(|value| value == "request_encode") - && event.fields.get("diagnostic").is_some_and(|value| { - value.contains("dropped unsupported JSON Schema constraints") - }) - }) - .count(); - assert_eq!( - matching_warnings, 1, - "expected one structured translation warning" - ); -} - #[tokio::test] async fn affinity_warns_once_when_request_has_no_usable_identity() -> switchyard_libsy::Result<()> { let _guard = serialize_test().lock().await; diff --git a/crates/switchyard-server/src/lib.rs b/crates/switchyard-server/src/lib.rs index 1dd562974..10691805a 100644 --- a/crates/switchyard-server/src/lib.rs +++ b/crates/switchyard-server/src/lib.rs @@ -11,6 +11,7 @@ mod routing_log; mod shutdown; mod sse; mod stats; +mod translation; mod usage_metrics; use std::collections::BTreeMap; @@ -37,9 +38,8 @@ use libsy::{Algorithm, LibsyError, RoutingOutcome}; use parking_lot::Mutex; use serde::{Deserialize, Serialize}; use serde_json::{Value, json}; -use switchyard_llm_client::metrics::{TranslationOperation, record_translation_diagnostics}; use switchyard_llm_client::{AuxiliaryOperation, ClientRouter, RunObservation, RunObserver}; -use switchyard_protocol::{LlmClientError, Metadata, ModelId, Request, Usage}; +use switchyard_protocol::{LlmClientError, Metadata, ModelId, ProviderExtensions, Request, Usage}; use switchyard_runner::{ CallerAuthKind, DecisionTarget, ModelCapabilities, Route, RunOutput, Runner, RunnerError, }; @@ -47,9 +47,7 @@ use tokio::net::{TcpListener, TcpSocket}; use tokio::task; use tracing::{Instrument, Level}; -use switchyard_translation::{ - WireFormat, decode_request_with_diagnostics, encode_aggregated_response_with_diagnostics, -}; +use switchyard_translation::WireFormat; use crate::response::into_http_response; use crate::stats::{StatsAccumulator, StatsSnapshot, prefix_probe, tracking_enabled_from_env}; @@ -741,19 +739,13 @@ async fn decision( // The request moved into the decision run, so its namespace mapping // is gone by here. A Codex tool call in this preview keeps its // qualified name. - match encode_aggregated_response_with_diagnostics( + match translation::encode_response( &aggregate, input_format, outcome.selected_model_id().ok().map(ModelId::as_str), + &ProviderExtensions::default(), ) { - Ok(encoded) => { - record_translation_diagnostics( - &encoded.diagnostics, - TranslationOperation::ResponseEncode, - input_format, - ); - Some(encoded.body) - } + Ok(response) => Some(response), Err(error) => return server_error(error.to_string()), } } @@ -969,14 +961,8 @@ fn resolve_route( body: Value, wire_format: WireFormat, ) -> std::result::Result<(&Route, Request), Response> { - let decoded = decode_request_with_diagnostics(wire_format, &body) + let llm_request = translation::decode_request(wire_format, &body) .map_err(|error| invalid_body_error(StatusCode::BAD_REQUEST, error.to_string()))?; - record_translation_diagnostics( - &decoded.diagnostics, - TranslationOperation::RequestDecode, - wire_format, - ); - let llm_request = decoded.request; let requested_model = llm_request .model .clone() diff --git a/crates/switchyard-server/src/metrics.rs b/crates/switchyard-server/src/metrics.rs index d1d75f327..0879d4710 100644 --- a/crates/switchyard-server/src/metrics.rs +++ b/crates/switchyard-server/src/metrics.rs @@ -9,6 +9,8 @@ use opentelemetry::{KeyValue, global}; use opentelemetry_sdk::metrics::{Aggregation, Instrument, SdkMeterProvider, Stream}; use prometheus::{Encoder, Registry, TextEncoder}; use switchyard_llm_client::metrics::{http_outcome_label, http_status_code_label}; +use switchyard_protocol::WireFormat; +use switchyard_translation::{DiagnosticSeverity, TranslationDiagnostic}; pub(crate) const CONTENT_TYPE: &str = "text/plain; version=0.0.4; charset=utf-8"; @@ -162,6 +164,43 @@ pub(crate) fn record_client_response(status: u16) { ); } +// Records server-boundary diagnostics with bounded metric labels and request details only in logs. +pub(crate) fn record_translation_diagnostics( + diagnostics: &[TranslationDiagnostic], + operation: &'static str, + format: WireFormat, +) { + for diagnostic in diagnostics { + let severity = match diagnostic.severity { + DiagnosticSeverity::Info => "info", + DiagnosticSeverity::Warning => "warning", + DiagnosticSeverity::Error => "error", + }; + global::meter("switchyard") + .u64_counter("switchyard.translation_diagnostics") + .build() + .add( + 1, + &[ + KeyValue::new("code", diagnostic.code.clone()), + KeyValue::new("format", format.as_str()), + KeyValue::new("operation", operation), + KeyValue::new("severity", severity), + ], + ); + tracing::warn!( + target: "libsy", + code = %diagnostic.code, + format = format.as_str(), + operation, + severity, + diagnostic = %diagnostic.message, + path = diagnostic.path.as_deref().unwrap_or(""), + "LLM protocol translation emitted a diagnostic" + ); + } +} + /// Encodes the current cumulative metric values in Prometheus text format. pub(crate) fn encode(registry: &Registry) -> Result, String> { let mut body = Vec::new(); diff --git a/crates/switchyard-server/src/response.rs b/crates/switchyard-server/src/response.rs index 9fdf74edf..5466f6835 100644 --- a/crates/switchyard-server/src/response.rs +++ b/crates/switchyard-server/src/response.rs @@ -7,14 +7,11 @@ use std::error::Error; use axum::Json; use axum::response::{IntoResponse, Response as HttpResponse}; -use switchyard_llm_client::metrics::{TranslationOperation, record_translation_diagnostics}; use switchyard_protocol::{LlmResponse, ProviderExtensions, Response as AlgorithmResponse}; -use switchyard_translation::{ - WireFormat, encode_aggregated_response_with_extensions_and_diagnostics, - encode_stream_with_extensions, -}; +use switchyard_translation::{WireFormat, encode_stream_with_extensions}; use crate::sse::frame_stream; +use crate::translation; type BoxError = Box; @@ -29,18 +26,13 @@ pub(crate) fn into_http_response( ) -> Result { match response.llm_response { LlmResponse::Agg(response) => { - let encoded = encode_aggregated_response_with_extensions_and_diagnostics( + let body = translation::encode_response( &response, target_format, served_model.as_deref(), &request_extensions, )?; - record_translation_diagnostics( - &encoded.diagnostics, - TranslationOperation::ResponseEncode, - target_format, - ); - Ok(Json(encoded.body).into_response()) + Ok(Json(body).into_response()) } LlmResponse::Stream(stream) => { let events = encode_stream_with_extensions( diff --git a/crates/switchyard-server/src/translation.rs b/crates/switchyard-server/src/translation.rs new file mode 100644 index 000000000..cdf92bc7f --- /dev/null +++ b/crates/switchyard-server/src/translation.rs @@ -0,0 +1,36 @@ +// SPDX-FileCopyrightText: Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved. +// SPDX-License-Identifier: Apache-2.0 + +//! Buffered translation with runtime diagnostics for server HTTP boundaries. + +use std::sync::LazyLock; + +use serde_json::Value; +use switchyard_protocol::{AggLlmResponse, LlmRequest, ProviderExtensions}; +use switchyard_translation::{Result, TranslationEngine, TranslationPolicy, WireFormat}; + +use crate::metrics; + +static ENGINE: LazyLock = LazyLock::new(TranslationEngine::default); +static POLICY: LazyLock = LazyLock::new(TranslationPolicy::default); + +pub(crate) fn decode_request(format: WireFormat, body: &Value) -> Result { + let decoded = ENGINE.decode_request(format, body, &POLICY)?; + metrics::record_translation_diagnostics(&decoded.diagnostics, "request_decode", format); + Ok(decoded.request) +} + +pub(crate) fn encode_response( + response: &AggLlmResponse, + format: WireFormat, + served_model: Option<&str>, + request_extensions: &ProviderExtensions, +) -> Result { + let mut encoded = + ENGINE.encode_response_with_extensions(format, response, request_extensions, &POLICY)?; + metrics::record_translation_diagnostics(&encoded.diagnostics, "response_encode", format); + if let (Some(model), Value::Object(body)) = (served_model, &mut encoded.body) { + body.insert("model".to_string(), Value::String(model.to_string())); + } + Ok(encoded.body) +} diff --git a/crates/switchyard-server/tests/server.rs b/crates/switchyard-server/tests/server.rs index b7ac2629d..f309d2f94 100644 --- a/crates/switchyard-server/tests/server.rs +++ b/crates/switchyard-server/tests/server.rs @@ -904,16 +904,12 @@ target = "anthropic" ("operation", "request_encode"), ("severity", "warning"), ]; - let before = send(&app, "GET", "/metrics", None) - .await? - .text()? - .to_string(); - - let lossless_response = send( - &app, - "POST", - "/v1/chat/completions", - Some(json!({ + let request = |min_length| { + let answer = match min_length { + Some(min_length) => json!({"type": "string", "minLength": min_length}), + None => json!({"type": "string"}), + }; + json!({ "model": "switchyard/diagnostics", "messages": [{"role": "user", "content": "Return JSON"}], "response_format": { @@ -923,90 +919,58 @@ target = "anthropic" "strict": true, "schema": { "type": "object", - "properties": {"answer": {"type": "string"}}, + "properties": {"answer": answer}, "required": ["answer"] } } } - })), - ) - .await?; - assert_eq!(lossless_response.status, StatusCode::OK); - assert_eq!( - lossless_response.json()?["choices"][0]["message"]["content"], - "ok" - ); - let calls = upstream.calls.lock().await; - assert_eq!(calls.len(), 1); - assert_eq!( - calls[0].pointer("/output_config/format/schema/properties/answer"), - Some(&json!({"type": "string"})) - ); - drop(calls); - - let after_lossless = send(&app, "GET", "/metrics", None) + }) + }; + let mut metrics = send(&app, "GET", "/metrics", None) .await? .text()? .to_string(); - assert_eq!( - metric_delta( - &before, - &after_lossless, - "switchyard_translation_diagnostics_total", - &diagnostic_labels, + + for (min_length, expected_delta) in [(None, 0.0), (Some(5), 1.0)] { + let response = send( + &app, + "POST", + "/v1/chat/completions", + Some(request(min_length)), ) - .unwrap_or_default(), - 0.0 - ); + .await?; + assert_eq!(response.status, StatusCode::OK); + assert_eq!(response.json()?["choices"][0]["message"]["content"], "ok"); + let after = send(&app, "GET", "/metrics", None) + .await? + .text()? + .to_string(); + assert_eq!( + metric_delta( + &metrics, + &after, + "switchyard_translation_diagnostics_total", + &diagnostic_labels, + ) + .unwrap_or_default(), + expected_delta + ); + metrics = after; + } - let response = send( - &app, - "POST", - "/v1/chat/completions", - Some(json!({ - "model": "switchyard/diagnostics", - "messages": [{"role": "user", "content": "Return constrained JSON"}], - "response_format": { - "type": "json_schema", - "json_schema": { - "name": "answer", - "strict": true, - "schema": { - "type": "object", - "properties": {"answer": {"type": "string", "minLength": 5}}, - "required": ["answer"] - } - } - } - })), - ) - .await?; - assert_eq!(response.status, StatusCode::OK); - assert_eq!(response.json()?["choices"][0]["message"]["content"], "ok"); let calls = upstream.calls.lock().await; assert_eq!(calls.len(), 2); + assert_eq!( + calls[0].pointer("/output_config/format/schema/properties/answer"), + Some(&json!({"type": "string"})) + ); assert!( calls[1] .pointer("/output_config/format/schema/properties/answer/minLength") .is_none() ); - drop(calls); - - let after_lossy = send(&app, "GET", "/metrics", None) - .await? - .text()? - .to_string(); - assert_eq!( - metric_delta( - &after_lossless, - &after_lossy, - "switchyard_translation_diagnostics_total", - &diagnostic_labels, - ), - Some(1.0) - ); let diagnostic_line = metric_line( - &after_lossy, + &metrics, "switchyard_translation_diagnostics_total", &diagnostic_labels, ) diff --git a/crates/switchyard-translation/src/helpers.rs b/crates/switchyard-translation/src/helpers.rs index 68cb69cbe..061517e24 100644 --- a/crates/switchyard-translation/src/helpers.rs +++ b/crates/switchyard-translation/src/helpers.rs @@ -19,9 +19,8 @@ use crate::codecs::stream::encode_response_stream_event; use crate::sse; use crate::{ AggLlmResponse, FormatId, LlmRequest, LlmResponseChunk, LlmResponseStream, - LlmResponseStreamEvent, LlmStreamError, RequestIrOutput, ResponseIrOutput, Result, - StreamCodecRegistry, StreamTranslationState, TranslationEngine, TranslationOutput, - TranslationPolicy, WireFormat, + LlmResponseStreamEvent, LlmStreamError, Result, StreamCodecRegistry, StreamTranslationState, + TranslationEngine, TranslationPolicy, WireFormat, }; static DEFAULT_TRANSLATION_POLICY: LazyLock = @@ -31,53 +30,23 @@ static DEFAULT_TRANSLATION_ENGINE: LazyLock = /// Decodes a `wire_format` request body into the neutral IR. pub fn decode_request(wire_format: WireFormat, body: &Value) -> Result { - Ok(decode_request_with_diagnostics(wire_format, body)?.request) -} - -/// Decodes a request and retains any diagnostics emitted by the codec. -/// -/// # Errors -/// -/// Returns an error when the request codec cannot decode `body`. -pub fn decode_request_with_diagnostics( - wire_format: WireFormat, - body: &Value, -) -> Result { - DEFAULT_TRANSLATION_ENGINE.decode_request(wire_format, body, &DEFAULT_TRANSLATION_POLICY) + Ok(DEFAULT_TRANSLATION_ENGINE + .decode_request(wire_format, body, &DEFAULT_TRANSLATION_POLICY)? + .request) } /// Encodes a normalized request into `wire_format`'s JSON body. pub fn encode_request(request: &LlmRequest, wire_format: WireFormat) -> Result { - Ok(encode_request_with_diagnostics(request, wire_format)?.body) -} - -/// Encodes a request and retains any diagnostics emitted by the codec. -/// -/// # Errors -/// -/// Returns an error when the request codec cannot encode `request`. -pub fn encode_request_with_diagnostics( - request: &LlmRequest, - wire_format: WireFormat, -) -> Result { - DEFAULT_TRANSLATION_ENGINE.encode_request(wire_format, request, &DEFAULT_TRANSLATION_POLICY) + Ok(DEFAULT_TRANSLATION_ENGINE + .encode_request(wire_format, request, &DEFAULT_TRANSLATION_POLICY)? + .body) } /// Decodes a buffered `wire_format` response body into the neutral aggregate. pub fn decode_aggregated_response(body: &Value, wire_format: WireFormat) -> Result { - Ok(decode_aggregated_response_with_diagnostics(body, wire_format)?.response) -} - -/// Decodes a buffered response and retains any diagnostics emitted by the codec. -/// -/// # Errors -/// -/// Returns an error when the response codec cannot decode `body`. -pub fn decode_aggregated_response_with_diagnostics( - body: &Value, - wire_format: WireFormat, -) -> Result { - DEFAULT_TRANSLATION_ENGINE.decode_response(wire_format, body, &DEFAULT_TRANSLATION_POLICY) + Ok(DEFAULT_TRANSLATION_ENGINE + .decode_response(wire_format, body, &DEFAULT_TRANSLATION_POLICY)? + .response) } /// Encodes a buffered aggregate into `wire_format`'s JSON body, stamping @@ -88,20 +57,7 @@ pub fn encode_aggregated_response( wire_format: WireFormat, served_model: Option<&str>, ) -> Result { - Ok(encode_aggregated_response_with_diagnostics(agg, wire_format, served_model)?.body) -} - -/// Encodes a buffered response and retains any diagnostics emitted by the codec. -/// -/// # Errors -/// -/// Returns an error when the response codec cannot encode `agg`. -pub fn encode_aggregated_response_with_diagnostics( - agg: &AggLlmResponse, - wire_format: WireFormat, - served_model: Option<&str>, -) -> Result { - encode_aggregated_response_with_extensions_and_diagnostics( + encode_aggregated_response_with_extensions( agg, wire_format, served_model, @@ -119,36 +75,18 @@ pub fn encode_aggregated_response_with_extensions( served_model: Option<&str>, request_extensions: &switchyard_protocol::ProviderExtensions, ) -> Result { - Ok(encode_aggregated_response_with_extensions_and_diagnostics( - agg, - wire_format, - served_model, - request_extensions, - )? - .body) -} - -/// Encodes a buffered response with request extensions and retains codec diagnostics. -/// -/// # Errors -/// -/// Returns an error when the response codec cannot encode `agg`. -pub fn encode_aggregated_response_with_extensions_and_diagnostics( - agg: &AggLlmResponse, - wire_format: WireFormat, - served_model: Option<&str>, - request_extensions: &switchyard_protocol::ProviderExtensions, -) -> Result { - let mut output = DEFAULT_TRANSLATION_ENGINE.encode_response_with_extensions( - wire_format, - agg, - request_extensions, - &DEFAULT_TRANSLATION_POLICY, - )?; - if let (Some(model), Value::Object(object)) = (served_model, &mut output.body) { + let mut body = DEFAULT_TRANSLATION_ENGINE + .encode_response_with_extensions( + wire_format, + agg, + request_extensions, + &DEFAULT_TRANSLATION_POLICY, + )? + .body; + if let (Some(model), Value::Object(object)) = (served_model, &mut body) { object.insert("model".to_string(), Value::String(model.to_string())); } - Ok(output) + Ok(body) } /// A stream of wire-format event objects in one format — the unframed body of an