diff --git a/crates/libsy-llm-client/src/client.rs b/crates/libsy-llm-client/src/client.rs index de7f2ca6c..620504f31 100644 --- a/crates/libsy-llm-client/src/client.rs +++ b/crates/libsy-llm-client/src/client.rs @@ -18,16 +18,14 @@ use switchyard_protocol::{ LlmRequest, LlmResponse, LlmResponseChunk, LlmResponseStreamEvent, Metadata, ModelId, Request, Response, RoutedLlmClient, }; -use switchyard_translation::{ - WireFormat, decode_aggregated_response, decode_request, decode_stream, - encode_aggregated_response_with_extensions, encode_request, 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. @@ -240,7 +238,7 @@ impl TranslatingLlmClient { model: &ModelId, endpoint: UpstreamEndpoint, ) -> Result { - let mut body = encode_request(&llm_request, wire_format) + let mut body = translation::encode_request(&llm_request, wire_format) .map_err(|error| LlmClientError::RequestEncoding(error.to_string()))?; // `encode_request` round-trips a preserved same-format body verbatim, // which keeps the caller's original `model`; force the resolved model so @@ -501,7 +499,7 @@ 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 agg = translation::decode_response(&body, wire_format) .map_err(|error| LlmClientError::ResponseTranslation(error.to_string()))?; LlmResponse::Agg(agg) } @@ -534,7 +532,7 @@ impl TranslatingLlmClient { model: Option<&ModelId>, wire_format: WireFormat, ) -> Result { - let llm_request = decode_request(wire_format, &raw_http_request) + let llm_request = translation::decode_request(wire_format, &raw_http_request) .map_err(|error| LlmClientError::RequestTranslation(error.to_string()))?; let request_extensions = llm_request.extensions.clone(); // The model that serves the call — the rewrite target when the caller pinned @@ -562,7 +560,7 @@ impl TranslatingLlmClient { match response.llm_response { LlmResponse::Agg(agg) => { - let body = encode_aggregated_response_with_extensions( + let body = translation::encode_response( &agg, wire_format, served_model.as_deref(), 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 76730ebfe..4006f4729 100644 --- a/crates/libsy-llm-client/src/metrics.rs +++ b/crates/libsy-llm-client/src/metrics.rs @@ -12,7 +12,8 @@ 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); @@ -104,6 +105,47 @@ pub(crate) fn record_retry_recovered() { .add(1, &[]); } +// Records translation diagnostics without putting request-derived values in metric labels. +pub(crate) fn record_translation_diagnostics( + diagnostics: &[TranslationDiagnostic], + operation: &'static str, + 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), + 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" + ); + } +} + +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) { @@ -170,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"); @@ -185,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/switchyard-server/src/lib.rs b/crates/switchyard-server/src/lib.rs index 8b6f06de8..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; @@ -38,7 +39,7 @@ 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, ProviderExtensions, Request, Usage}; use switchyard_runner::{ CallerAuthKind, DecisionTarget, ModelCapabilities, Route, RunOutput, Runner, RunnerError, }; @@ -46,7 +47,7 @@ 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; use crate::response::into_http_response; use crate::stats::{StatsAccumulator, StatsSnapshot, prefix_probe, tracking_enabled_from_env}; @@ -738,10 +739,11 @@ 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 translation::encode_response( &aggregate, input_format, outcome.selected_model_id().ok().map(ModelId::as_str), + &ProviderExtensions::default(), ) { Ok(response) => Some(response), Err(error) => return server_error(error.to_string()), @@ -959,7 +961,7 @@ fn resolve_route( body: Value, wire_format: WireFormat, ) -> std::result::Result<(&Route, Request), Response> { - let llm_request = decode_request(wire_format, &body) + let llm_request = translation::decode_request(wire_format, &body) .map_err(|error| invalid_body_error(StatusCode::BAD_REQUEST, error.to_string()))?; let requested_model = llm_request .model 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 950ec6399..5466f6835 100644 --- a/crates/switchyard-server/src/response.rs +++ b/crates/switchyard-server/src/response.rs @@ -8,11 +8,10 @@ use std::error::Error; use axum::Json; use axum::response::{IntoResponse, Response as HttpResponse}; use switchyard_protocol::{LlmResponse, ProviderExtensions, Response as AlgorithmResponse}; -use switchyard_translation::{ - WireFormat, encode_aggregated_response_with_extensions, encode_stream_with_extensions, -}; +use switchyard_translation::{WireFormat, encode_stream_with_extensions}; use crate::sse::frame_stream; +use crate::translation; type BoxError = Box; @@ -27,7 +26,7 @@ pub(crate) fn into_http_response( ) -> Result { match response.llm_response { LlmResponse::Agg(response) => { - let body = encode_aggregated_response_with_extensions( + let body = translation::encode_response( &response, target_format, served_model.as_deref(), 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 0f6567092..f309d2f94 100644 --- a/crates/switchyard-server/tests/server.rs +++ b/crates/switchyard-server/tests/server.rs @@ -344,20 +344,22 @@ 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") + // The diagnostics fixture intentionally has no OAuth credentials. + 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 +873,117 @@ 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 diagnostic_labels = [ + ("code", "lossy_conversion"), + ("format", "anthropic_messages"), + ("operation", "request_encode"), + ("severity", "warning"), + ]; + 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": { + "type": "json_schema", + "json_schema": { + "name": "answer", + "strict": true, + "schema": { + "type": "object", + "properties": {"answer": answer}, + "required": ["answer"] + } + } + } + }) + }; + let mut metrics = send(&app, "GET", "/metrics", None) + .await? + .text()? + .to_string(); + + 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)), + ) + .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 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() + ); + let diagnostic_line = metric_line( + &metrics, + "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(()) +} + #[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/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. |