From 370ffc3dfe5e3fdf02300d6e02a72a60ad1e0499 Mon Sep 17 00:00:00 2001 From: nachiketb Date: Thu, 30 Jul 2026 19:11:26 -0700 Subject: [PATCH 1/3] feat(observability): add GenAI semantic attributes Signed-off-by: nachiketb --- crates/libsy-llm-client/src/backend.rs | 8 ++ crates/libsy-llm-client/src/client.rs | 16 +++ crates/libsy/src/core/algorithm.rs | 36 ++++- crates/libsy/src/observability.rs | 119 +++++++++++++++-- crates/libsy/tests/observability.rs | 124 +++++++++++++++++- crates/switchyard-server/src/lib.rs | 15 ++- crates/switchyard-server/src/observability.rs | 68 ++++++++++ 7 files changed, 365 insertions(+), 21 deletions(-) diff --git a/crates/libsy-llm-client/src/backend.rs b/crates/libsy-llm-client/src/backend.rs index f4c4f4fe0..5b01a2a72 100644 --- a/crates/libsy-llm-client/src/backend.rs +++ b/crates/libsy-llm-client/src/backend.rs @@ -144,6 +144,14 @@ impl Backend { self.config().max_retries } + /// Stable GenAI semantic-convention provider name for this wire protocol. + pub(crate) fn provider_name(&self) -> &'static str { + match self { + Backend::OpenAiChat(_) | Backend::OpenAiResponses(_) => "openai", + Backend::Anthropic(_) => "anthropic", + } + } + /// Whether this backend speaks the Anthropic Messages wire format — the only /// one with a `count_tokens` endpoint. pub fn is_anthropic(&self) -> bool { diff --git a/crates/libsy-llm-client/src/client.rs b/crates/libsy-llm-client/src/client.rs index ed8c5b7bc..ca068d6c1 100644 --- a/crates/libsy-llm-client/src/client.rs +++ b/crates/libsy-llm-client/src/client.rs @@ -178,6 +178,7 @@ impl TranslatingLlmClient { let streaming = endpoint.allows_streaming() && body.get("stream").and_then(Value::as_bool).unwrap_or(false); let url = endpoint.url(backend); + record_gen_ai_request(backend, &url, model, streaming); let max_retries = u64::from(backend.max_retries()); let max_attempts = max_retries + 1; @@ -601,6 +602,21 @@ fn duration_millis(duration: Duration) -> u64 { u64::try_from(duration.as_millis()).unwrap_or(u64::MAX) } +fn record_gen_ai_request(backend: &Backend, url: &str, model: &str, streaming: bool) { + let span = tracing::Span::current(); + span.record("gen_ai.provider.name", backend.provider_name()); + span.record("gen_ai.request.model", model); + span.record("gen_ai.request.stream", streaming); + if let Ok(url) = reqwest::Url::parse(url) { + if let Some(host) = url.host_str() { + span.record("server.address", host); + } + if let Some(port) = url.port_or_known_default() { + span.record("server.port", u64::from(port)); + } + } +} + fn convert_reqwest_error(error: reqwest::Error) -> LlmClientError { // Reqwest labels truncated or otherwise unreadable response bodies as decode // errors, so distinguish them from serde JSON failures at the call site. diff --git a/crates/libsy/src/core/algorithm.rs b/crates/libsy/src/core/algorithm.rs index 84eb41d55..c1d4c73af 100644 --- a/crates/libsy/src/core/algorithm.rs +++ b/crates/libsy/src/core/algorithm.rs @@ -618,7 +618,7 @@ pub trait Algorithm: Send + Sync + 'static { // One `libsy.run` span covers the whole algorithm task; the driver's // `libsy.llm_call` spans and decision logs nest inside it via `tracing`'s // contextual parenting. - let span = observability::run_span(self.name(), request.metadata.as_ref()); + let span = observability::run_span(self.name(), &request); let observed_driver = task_driver.clone(); let handle = tokio::spawn( async move { @@ -684,12 +684,44 @@ pub trait Algorithm: Send + Sync + 'static { skip_all, fields( algorithm = observability::algorithm_label(&call.get_routed().ctx), + switchyard.algorithm = observability::algorithm_label(&call.get_routed().ctx), + switchyard.routing.tier = tracing::field::Empty, selected_model = call.get_decision().selected_model(), + otel.kind = "client", + gen_ai.operation.name = "chat", + gen_ai.request.model = call.get_decision().selected_model(), + gen_ai.request.stream = call.get_routed().request.llm_request.stream, + gen_ai.conversation.id = tracing::field::Empty, + gen_ai.provider.name = tracing::field::Empty, + server.address = tracing::field::Empty, + server.port = tracing::field::Empty, + gen_ai.response.id = tracing::field::Empty, + gen_ai.response.model = tracing::field::Empty, + gen_ai.usage.input_tokens = tracing::field::Empty, + gen_ai.usage.output_tokens = tracing::field::Empty, + gen_ai.usage.cache_read.input_tokens = tracing::field::Empty, + gen_ai.usage.cache_creation.input_tokens = tracing::field::Empty, + gen_ai.usage.reasoning.output_tokens = tracing::field::Empty, outcome = tracing::field::Empty, + otel.status_code = tracing::field::Empty, + error.type = tracing::field::Empty, error = tracing::field::Empty, ) )] async fn serve(call: CallLlmRequest) -> Result<()> { + let span = tracing::Span::current(); + if let Some(tier) = call.get_decision().routing_tier() { + span.record("switchyard.routing.tier", tier); + } + if let Some(session_id) = call + .get_routed() + .request + .metadata + .as_ref() + .and_then(|metadata| metadata.session_id.as_deref()) + { + span.record("gen_ai.conversation.id", session_id); + } let routed = call.get_routed().clone(); let target = routed.decision.selected_model().to_string(); let client = @@ -703,7 +735,7 @@ pub trait Algorithm: Send + Sync + 'static { .call(routed.ctx, routed.request, routed.decision) .await .map_err(|source| LibsyError::client_call(target, source)); - observability::record_client_call(&result); + let result = observability::observe_client_call(result); call.respond(result) } diff --git a/crates/libsy/src/observability.rs b/crates/libsy/src/observability.rs index fd5ef5839..a1cba8dcf 100644 --- a/crates/libsy/src/observability.rs +++ b/crates/libsy/src/observability.rs @@ -35,11 +35,15 @@ use std::sync::atomic::{AtomicU64, Ordering}; use std::sync::OnceLock; use std::time::{Duration, Instant}; +use futures::StreamExt; use opentelemetry::metrics::{Meter, ObservableGauge}; use opentelemetry::{global, KeyValue}; use tracing::Span; -use crate::{Context, Decision, Driver, Metadata, Response, Result}; +use crate::{ + AggLlmResponse, Context, Decision, Driver, LlmResponse, LlmResponseChunk, LlmResponseStream, + Request, Response, Result, Usage, +}; const METRICS_SCOPE: &str = "switchyard"; const TRACING_TARGET: &str = "libsy"; @@ -96,17 +100,20 @@ fn outcome_value(result: &Result) -> &'static str { /// Span covering one algorithm run (the whole `create_run_task` execution). /// -/// Correlation ids from the request [`Metadata`] are recorded as span fields +/// Correlation ids from the request [`crate::Metadata`] are recorded as span fields /// when present. `tracing` spans cannot grow field names at runtime, so -/// arbitrary host labels ride in via [`Metadata::extra_metadata`], recorded +/// arbitrary host labels ride in via [`crate::Metadata::extra_metadata`], recorded /// whole into the `extra_metadata` field. `outcome` and `error` are filled in /// by [`record_run`] when the run ends. -pub(crate) fn run_span(algorithm: &str, metadata: Option<&Metadata>) -> Span { +pub(crate) fn run_span(algorithm: &str, request: &Request) -> Span { let span = tracing::info_span!( target: TRACING_TARGET, "libsy.run", algorithm, + switchyard.algorithm = algorithm, + switchyard.route = tracing::field::Empty, session_id = tracing::field::Empty, + session.id = tracing::field::Empty, agent_id = tracing::field::Empty, task_id = tracing::field::Empty, correlation_id = tracing::field::Empty, @@ -114,7 +121,10 @@ pub(crate) fn run_span(algorithm: &str, metadata: Option<&Metadata>) -> Span { outcome = tracing::field::Empty, error = tracing::field::Empty, ); - if let Some(metadata) = metadata { + if let Some(route) = request.requested_model() { + span.record("switchyard.route", route); + } + if let Some(metadata) = &request.metadata { for (field, value) in [ ("session_id", &metadata.session_id), ("agent_id", &metadata.agent_id), @@ -125,6 +135,9 @@ pub(crate) fn run_span(algorithm: &str, metadata: Option<&Metadata>) -> Span { span.record(field, value.as_str()); } } + if let Some(session_id) = &metadata.session_id { + span.record("session.id", session_id.as_str()); + } if let Some(extra) = &metadata.extra_metadata { span.record("extra_metadata", tracing::field::debug(extra)); } @@ -156,17 +169,99 @@ pub(crate) async fn observe_run( result } -/// Records the outcome fields on the enclosing `libsy.client_call` span. The -/// failure itself is not logged here — it propagates to the algorithm, where -/// the `libsy.llm_call` recording logs it once. -pub(crate) fn record_client_call(result: &Result) { +/// Adds terminal response and usage fields to the enclosing `libsy.client_call` +/// span without consuming or buffering a streaming response. +pub(crate) fn observe_client_call(result: Result) -> Result { let span = Span::current(); - span.record("outcome", outcome_value(result)); - if let Err(error) = result { - span.record("error", tracing::field::display(error)); + match result { + Ok(mut response) => { + span.record("outcome", "ok"); + match response.llm_response { + LlmResponse::Agg(agg) => { + record_gen_ai_response(&span, &agg); + response.llm_response = LlmResponse::Agg(agg); + } + LlmResponse::Stream(stream) => { + response.llm_response = + LlmResponse::Stream(observe_client_stream(stream, span)); + } + } + Ok(response) + } + Err(error) => { + record_client_error(&span, "client_call", &error); + Err(error) + } } } +fn observe_client_stream(stream: LlmResponseStream, span: Span) -> LlmResponseStream { + Box::pin(stream.inspect(move |item| match item { + Ok(LlmResponseChunk::MessageStart { id, model }) => { + record_optional(&span, "gen_ai.response.id", id.as_deref()); + record_optional(&span, "gen_ai.response.model", model.as_deref()); + } + Ok(LlmResponseChunk::Usage(usage)) => { + record_gen_ai_usage(&span, usage); + } + Ok(LlmResponseChunk::DecodeError { message }) => { + record_client_error(&span, "decode_error", message); + } + Ok(LlmResponseChunk::StreamError { message }) => { + record_client_error(&span, "stream_error", message); + } + Err(error) => { + record_client_error(&span, "client_stream", error); + } + _ => {} + })) +} + +fn record_gen_ai_response(span: &Span, response: &AggLlmResponse) { + record_optional(span, "gen_ai.response.id", response.id.as_deref()); + record_optional(span, "gen_ai.response.model", response.model.as_deref()); + record_gen_ai_usage(span, &response.usage); +} + +fn record_gen_ai_usage(span: &Span, usage: &Usage) { + let cache_read = usage.cached_input_tokens(); + let cache_creation = usage.cache_creation_input_tokens(); + if usage.input_tokens.is_some() || cache_read.is_some() || cache_creation.is_some() { + let input_tokens = usage + .input_tokens + .unwrap_or_default() + .saturating_add(cache_read.unwrap_or_default()) + .saturating_add(cache_creation.unwrap_or_default()); + span.record("gen_ai.usage.input_tokens", input_tokens); + } + for (field, value) in [ + ("gen_ai.usage.output_tokens", usage.output_tokens), + ("gen_ai.usage.cache_read.input_tokens", cache_read), + ("gen_ai.usage.cache_creation.input_tokens", cache_creation), + ( + "gen_ai.usage.reasoning.output_tokens", + usage.reasoning_tokens, + ), + ] { + if let Some(value) = value { + span.record(field, value); + } + } +} + +fn record_optional(span: &Span, field: &str, value: Option<&str>) { + if let Some(value) = value { + span.record(field, value); + } +} + +fn record_client_error(span: &Span, error_type: &str, error: &dyn std::fmt::Display) { + span.record("outcome", "error"); + span.record("otel.status_code", "ERROR"); + span.record("error.type", error_type); + span.record("error", tracing::field::display(error)); +} + /// Records the end of one algorithm run: the run counter and duration /// histogram, the `outcome`/`error` fields on `span`, and a warn log when the /// run failed. diff --git a/crates/libsy/tests/observability.rs b/crates/libsy/tests/observability.rs index 896929a48..fca1be926 100644 --- a/crates/libsy/tests/observability.rs +++ b/crates/libsy/tests/observability.rs @@ -29,8 +29,9 @@ use tracing_subscriber::Layer; use switchyard_libsy::algorithms::{LlmTaskClassifier, TaskClassifierConfig}; use switchyard_libsy::{ - AggLlmResponse, Algorithm, Context, Decision, Driver, LibsyError, LlmResponse, LlmTarget, - LlmTargetSet, Metadata, Request, Response, RoutedLlmClient, Step, Usage, + AggLlmResponse, Algorithm, Context, Decision, Driver, LibsyError, LlmResponse, + LlmResponseChunk, LlmTarget, LlmTargetSet, Metadata, Request, Response, RoutedLlmClient, Step, + Usage, }; use switchyard_protocol::{text_request, text_response, LlmClientError}; @@ -167,7 +168,7 @@ fn telemetry() -> &'static (CaptureStore, InMemoryMetricExporter, SdkMeterProvid }) } -/// The three tests in this file must not overlap because metrics are global. +/// The tests in this file must not overlap because metrics are global. /// There is no Rust/cargo-native way of saying this (people use `serial_test` crate) so use a /// lock. /// Each file in `tests/` (integration tests) runs as a separate test process, so we are not @@ -345,6 +346,7 @@ impl RoutedLlmClient for UsageClient { ) -> Result { Ok(Response { llm_response: LlmResponse::Agg(AggLlmResponse { + id: Some("obs-response-1".to_string()), model: Some(decision.selected_model().to_string()), usage: self.usage.clone(), ..AggLlmResponse::default() @@ -442,9 +444,9 @@ async fn successful_run_records_metrics_spans_and_decision_log() -> switchyard_l usage: Usage { input_tokens: Some(11), output_tokens: Some(7), - total_tokens: Some(18), + total_tokens: Some(25), reasoning_tokens: Some(2), - ..Usage::default() + cache: Usage::cache_details(Some(3), Some(4)), }, }) as Arc; let (trace, _response) = algo(ALGO, MODEL, Some(client)) @@ -525,6 +527,21 @@ async fn successful_run_records_metrics_spans_and_decision_log() -> switchyard_l run_span.fields.get("session_id").map(String::as_str), Some("obs-session-1") ); + assert_eq!( + run_span.fields.get("session.id").map(String::as_str), + Some("obs-session-1") + ); + assert_eq!( + run_span + .fields + .get("switchyard.algorithm") + .map(String::as_str), + Some(ALGO) + ); + assert_eq!( + run_span.fields.get("switchyard.route").map(String::as_str), + Some("auto") + ); assert_eq!( run_span.fields.get("correlation_id").map(String::as_str), Some("obs-corr-1") @@ -555,6 +572,27 @@ async fn successful_run_records_metrics_spans_and_decision_log() -> switchyard_l client_span.fields.get("outcome").map(String::as_str), Some("ok") ); + for (field, value) in [ + ("otel.kind", "client"), + ("switchyard.algorithm", ALGO), + ("gen_ai.operation.name", "chat"), + ("gen_ai.request.model", MODEL), + ("gen_ai.request.stream", "false"), + ("gen_ai.conversation.id", "obs-session-1"), + ("gen_ai.response.id", "obs-response-1"), + ("gen_ai.response.model", MODEL), + ("gen_ai.usage.input_tokens", "18"), + ("gen_ai.usage.output_tokens", "7"), + ("gen_ai.usage.cache_read.input_tokens", "3"), + ("gen_ai.usage.cache_creation.input_tokens", "4"), + ("gen_ai.usage.reasoning.output_tokens", "2"), + ] { + assert_eq!( + client_span.fields.get(field).map(String::as_str), + Some(value), + "unexpected {field}" + ); + } let call_span = find_span(&spans, "libsy.llm_call", "selected_model", MODEL); assert_eq!(call_span.parent.as_deref(), Some("libsy.run")); @@ -576,7 +614,7 @@ async fn successful_run_records_metrics_spans_and_decision_log() -> switchyard_l ); assert_eq!( call_span.fields.get("total_tokens").map(String::as_str), - Some("18") + Some("25") ); assert_eq!( call_span.fields.get("reasoning_tokens").map(String::as_str), @@ -604,6 +642,80 @@ async fn successful_run_records_metrics_spans_and_decision_log() -> switchyard_l Ok(()) } +/// A streamed response keeps the client span available until terminal usage arrives. +struct StreamingUsageClient; + +#[async_trait] +impl RoutedLlmClient for StreamingUsageClient { + async fn call( + &self, + _ctx: Context, + _request: Request, + decision: Arc, + ) -> Result { + let usage = Usage { + input_tokens: Some(13), + output_tokens: Some(5), + cache: Usage::cache_details(Some(8), None), + ..Usage::default() + }; + let chunks = vec![ + Ok(LlmResponseChunk::MessageStart { + id: Some("obs-stream-response".to_string()), + model: Some(decision.selected_model().to_string()), + }), + Ok(LlmResponseChunk::Usage(usage)), + Ok(LlmResponseChunk::MessageStop { + reason: Some("end_turn".to_string()), + }), + ]; + Ok(Response { + llm_response: LlmResponse::Stream(Box::pin(futures::stream::iter(chunks))), + metadata: None, + }) + } +} + +#[tokio::test] +async fn streamed_usage_updates_the_client_call_span() -> switchyard_libsy::Result<()> { + let _guard = serialize_test().lock().await; + let (store, _, _) = telemetry(); + const ALGO: &str = "obs-stream-algo"; + const MODEL: &str = "obs-stream-model"; + let client = Arc::new(StreamingUsageClient) as Arc; + let mut request = request_with_metadata("obs-stream-session", "obs-stream-corr"); + request.llm_request.stream = true; + let (_, response) = algo(ALGO, MODEL, Some(client)) + .run(Context::default(), request) + .await?; + let LlmResponse::Stream(mut stream) = response.llm_response else { + return Err(test_error("expected a streamed response")); + }; + while let Some(item) = stream.next().await { + if let Err(error) = item { + panic!("unexpected stream error: {error}"); + } + } + + let spans = store.spans(); + let client_span = find_span(&spans, "libsy.client_call", "selected_model", MODEL); + for (field, value) in [ + ("gen_ai.request.stream", "true"), + ("gen_ai.response.id", "obs-stream-response"), + ("gen_ai.response.model", MODEL), + ("gen_ai.usage.input_tokens", "21"), + ("gen_ai.usage.output_tokens", "5"), + ("gen_ai.usage.cache_read.input_tokens", "8"), + ] { + assert_eq!( + client_span.fields.get(field).map(String::as_str), + Some(value), + "unexpected {field}" + ); + } + Ok(()) +} + #[tokio::test] async fn failed_call_records_error_outcome_and_warn_logs() -> 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 e97de8f12..0f2f14273 100644 --- a/crates/switchyard-server/src/lib.rs +++ b/crates/switchyard-server/src/lib.rs @@ -34,7 +34,7 @@ use libsy::{ }; use serde_json::{json, Value}; use tokio::net::{TcpListener, TcpSocket}; -use tracing::Level; +use tracing::{Instrument, Level}; use switchyard_translation::{decode_request, WireFormat}; @@ -416,6 +416,19 @@ async fn handle_endpoint( headers: HeaderMap, body: std::result::Result, JsonRejection>, wire_format: WireFormat, +) -> Response { + let span = observability::request_span(&headers); + handle_endpoint_inner(state, started, headers, body, wire_format) + .instrument(span) + .await +} + +async fn handle_endpoint_inner( + state: ServerState, + started: RequestStart, + headers: HeaderMap, + body: std::result::Result, JsonRejection>, + wire_format: WireFormat, ) -> Response { let metadata = metadata_from_headers(&headers); let request_log = RequestLogContext { diff --git a/crates/switchyard-server/src/observability.rs b/crates/switchyard-server/src/observability.rs index 6f9735bbf..fe3036cf2 100644 --- a/crates/switchyard-server/src/observability.rs +++ b/crates/switchyard-server/src/observability.rs @@ -6,9 +6,13 @@ use std::env; use std::sync::OnceLock; +use axum::http::HeaderMap; +use opentelemetry::propagation::{Extractor, TextMapPropagator}; use opentelemetry::trace::TracerProvider as _; +use opentelemetry_sdk::propagation::TraceContextPropagator; use opentelemetry_sdk::trace::SdkTracerProvider; use opentelemetry_sdk::Resource; +use tracing_opentelemetry::OpenTelemetrySpanExt; use tracing_subscriber::layer::SubscriberExt as _; use tracing_subscriber::util::SubscriberInitExt as _; use tracing_subscriber::{EnvFilter, Layer as _}; @@ -44,6 +48,30 @@ pub fn flush_observability() { metrics::flush(); } +/// Creates the server request span with any incoming W3C trace context as its parent. +pub(crate) fn request_span(headers: &HeaderMap) -> tracing::Span { + let parent = TraceContextPropagator::new().extract(&HeaderExtractor(headers)); + let span = tracing::info_span!( + target: "switchyard_server", + "switchyard.request", + otel.kind = "server", + ); + let _ = span.set_parent(parent); + span +} + +struct HeaderExtractor<'a>(&'a HeaderMap); + +impl Extractor for HeaderExtractor<'_> { + fn get(&self, key: &str) -> Option<&str> { + self.0.get(key).and_then(|value| value.to_str().ok()) + } + + fn keys(&self) -> Vec<&str> { + self.0.keys().map(|name| name.as_str()).collect() + } +} + pub(crate) fn otlp_enabled(signal: &str) -> bool { if env_var_is_true("OTEL_SDK_DISABLED") { return false; @@ -130,3 +158,43 @@ fn build_tracer_provider() -> Result { fn env_var_is_true(name: &str) -> bool { env::var(name).is_ok_and(|value| matches!(value.to_ascii_lowercase().as_str(), "true" | "1")) } + +#[cfg(test)] +mod tests { + use axum::http::{HeaderMap, HeaderValue}; + use opentelemetry::trace::{TraceContextExt, TracerProvider as _}; + use opentelemetry_sdk::trace::SdkTracerProvider; + use tracing_opentelemetry::OpenTelemetrySpanExt as _; + use tracing_subscriber::layer::SubscriberExt as _; + + use super::request_span; + + #[test] + fn request_span_continues_incoming_w3c_trace_context() { + let provider = SdkTracerProvider::builder().build(); + let tracer = provider.tracer("request-span-test"); + let subscriber = + tracing_subscriber::registry().with(tracing_opentelemetry::layer().with_tracer(tracer)); + let mut headers = HeaderMap::new(); + headers.insert( + "traceparent", + HeaderValue::from_static("00-4bf92f3577b34da6a3ce929d0e0e4736-00f067aa0ba902b7-01"), + ); + headers.insert( + "tracestate", + HeaderValue::from_static("vendor=opaque-value"), + ); + + tracing::subscriber::with_default(subscriber, || { + let span = request_span(&headers); + let context = span.context(); + let current = context.span(); + let span_context = current.span_context(); + assert_eq!( + span_context.trace_id().to_string(), + "4bf92f3577b34da6a3ce929d0e0e4736" + ); + assert_eq!(span_context.trace_state().header(), "vendor=opaque-value"); + }); + } +} From 70dbbda50975443bece599183a0c03e8de8905c9 Mon Sep 17 00:00:00 2001 From: nachiketb Date: Fri, 31 Jul 2026 09:43:56 -0700 Subject: [PATCH 2/3] fix(observability): stop conflating provider and wire format Signed-off-by: nachiketb --- crates/libsy-llm-client/src/backend.rs | 8 -------- crates/libsy-llm-client/src/client.rs | 5 ++--- crates/libsy/src/core/algorithm.rs | 1 - 3 files changed, 2 insertions(+), 12 deletions(-) diff --git a/crates/libsy-llm-client/src/backend.rs b/crates/libsy-llm-client/src/backend.rs index 5b01a2a72..f4c4f4fe0 100644 --- a/crates/libsy-llm-client/src/backend.rs +++ b/crates/libsy-llm-client/src/backend.rs @@ -144,14 +144,6 @@ impl Backend { self.config().max_retries } - /// Stable GenAI semantic-convention provider name for this wire protocol. - pub(crate) fn provider_name(&self) -> &'static str { - match self { - Backend::OpenAiChat(_) | Backend::OpenAiResponses(_) => "openai", - Backend::Anthropic(_) => "anthropic", - } - } - /// Whether this backend speaks the Anthropic Messages wire format — the only /// one with a `count_tokens` endpoint. pub fn is_anthropic(&self) -> bool { diff --git a/crates/libsy-llm-client/src/client.rs b/crates/libsy-llm-client/src/client.rs index ca068d6c1..f63d3ed2d 100644 --- a/crates/libsy-llm-client/src/client.rs +++ b/crates/libsy-llm-client/src/client.rs @@ -178,7 +178,7 @@ impl TranslatingLlmClient { let streaming = endpoint.allows_streaming() && body.get("stream").and_then(Value::as_bool).unwrap_or(false); let url = endpoint.url(backend); - record_gen_ai_request(backend, &url, model, streaming); + record_gen_ai_request(&url, model, streaming); let max_retries = u64::from(backend.max_retries()); let max_attempts = max_retries + 1; @@ -602,9 +602,8 @@ fn duration_millis(duration: Duration) -> u64 { u64::try_from(duration.as_millis()).unwrap_or(u64::MAX) } -fn record_gen_ai_request(backend: &Backend, url: &str, model: &str, streaming: bool) { +fn record_gen_ai_request(url: &str, model: &str, streaming: bool) { let span = tracing::Span::current(); - span.record("gen_ai.provider.name", backend.provider_name()); span.record("gen_ai.request.model", model); span.record("gen_ai.request.stream", streaming); if let Ok(url) = reqwest::Url::parse(url) { diff --git a/crates/libsy/src/core/algorithm.rs b/crates/libsy/src/core/algorithm.rs index c1d4c73af..c405b8ea9 100644 --- a/crates/libsy/src/core/algorithm.rs +++ b/crates/libsy/src/core/algorithm.rs @@ -692,7 +692,6 @@ pub trait Algorithm: Send + Sync + 'static { gen_ai.request.model = call.get_decision().selected_model(), gen_ai.request.stream = call.get_routed().request.llm_request.stream, gen_ai.conversation.id = tracing::field::Empty, - gen_ai.provider.name = tracing::field::Empty, server.address = tracing::field::Empty, server.port = tracing::field::Empty, gen_ai.response.id = tracing::field::Empty, From 879b54d7640382a1dac7098f457de895c02560dd Mon Sep 17 00:00:00 2001 From: nachiketb Date: Fri, 31 Jul 2026 10:39:41 -0700 Subject: [PATCH 3/3] fix(observability): complete GenAI client span attributes Signed-off-by: nachiketb --- Cargo.lock | 1 + crates/libsy-llm-client/src/client.rs | 6 +- crates/libsy/Cargo.toml | 3 +- crates/libsy/src/core/algorithm.rs | 10 +- crates/libsy/src/observability.rs | 239 +++++++++++++++++++++++--- crates/libsy/tests/observability.rs | 226 +++++++++++++++++++++--- 6 files changed, 428 insertions(+), 57 deletions(-) diff --git a/Cargo.lock b/Cargo.lock index a4ccd77b0..67e410ec5 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -2009,6 +2009,7 @@ dependencies = [ "tokio", "tokio-stream", "tracing", + "tracing-opentelemetry", "tracing-subscriber", ] diff --git a/crates/libsy-llm-client/src/client.rs b/crates/libsy-llm-client/src/client.rs index f63d3ed2d..df6d3b91b 100644 --- a/crates/libsy-llm-client/src/client.rs +++ b/crates/libsy-llm-client/src/client.rs @@ -605,13 +605,15 @@ fn duration_millis(duration: Duration) -> u64 { fn record_gen_ai_request(url: &str, model: &str, streaming: bool) { let span = tracing::Span::current(); span.record("gen_ai.request.model", model); - span.record("gen_ai.request.stream", streaming); + if streaming { + span.record("gen_ai.request.stream", true); + } if let Ok(url) = reqwest::Url::parse(url) { if let Some(host) = url.host_str() { span.record("server.address", host); } if let Some(port) = url.port_or_known_default() { - span.record("server.port", u64::from(port)); + span.record("server.port", i64::from(port)); } } } diff --git a/crates/libsy/Cargo.toml b/crates/libsy/Cargo.toml index 350c6031b..dc2be5249 100644 --- a/crates/libsy/Cargo.toml +++ b/crates/libsy/Cargo.toml @@ -26,10 +26,11 @@ thiserror.workspace = true tokio.workspace = true tokio-stream = "0.1" tracing.workspace = true +tracing-opentelemetry = "0.33" [dev-dependencies] # SDK + in-memory exporter to assert what the observability layer records. -opentelemetry_sdk = { version = "0.32", features = ["metrics", "testing"] } +opentelemetry_sdk = { version = "0.32", features = ["metrics", "testing", "trace"] } switchyard-llm-client.workspace = true tokio.workspace = true tracing-subscriber = "0.3" diff --git a/crates/libsy/src/core/algorithm.rs b/crates/libsy/src/core/algorithm.rs index c405b8ea9..9339d482b 100644 --- a/crates/libsy/src/core/algorithm.rs +++ b/crates/libsy/src/core/algorithm.rs @@ -688,9 +688,16 @@ pub trait Algorithm: Send + Sync + 'static { switchyard.routing.tier = tracing::field::Empty, selected_model = call.get_decision().selected_model(), otel.kind = "client", + otel.name = %format_args!("chat {}", call.get_decision().selected_model()), gen_ai.operation.name = "chat", gen_ai.request.model = call.get_decision().selected_model(), - gen_ai.request.stream = call.get_routed().request.llm_request.stream, + gen_ai.request.stream = tracing::field::Empty, + gen_ai.request.temperature = tracing::field::Empty, + gen_ai.request.top_p = tracing::field::Empty, + gen_ai.request.top_k = tracing::field::Empty, + gen_ai.request.max_tokens = tracing::field::Empty, + gen_ai.request.reasoning.level = tracing::field::Empty, + gen_ai.output.type = tracing::field::Empty, gen_ai.conversation.id = tracing::field::Empty, server.address = tracing::field::Empty, server.port = tracing::field::Empty, @@ -709,6 +716,7 @@ pub trait Algorithm: Send + Sync + 'static { )] async fn serve(call: CallLlmRequest) -> Result<()> { let span = tracing::Span::current(); + observability::record_gen_ai_request(&span, &call.get_routed().request.llm_request); if let Some(tier) = call.get_decision().routing_tier() { span.record("switchyard.routing.tier", tier); } diff --git a/crates/libsy/src/observability.rs b/crates/libsy/src/observability.rs index a1cba8dcf..96fed3d3f 100644 --- a/crates/libsy/src/observability.rs +++ b/crates/libsy/src/observability.rs @@ -30,19 +30,24 @@ //! provider installed at any point in the process lifetime; the cost is //! negligible next to a model call. +use std::borrow::Cow; use std::future::Future; +use std::pin::Pin; use std::sync::atomic::{AtomicU64, Ordering}; use std::sync::OnceLock; +use std::task::{Context as TaskContext, Poll}; use std::time::{Duration, Instant}; -use futures::StreamExt; +use futures::Stream; use opentelemetry::metrics::{Meter, ObservableGauge}; -use opentelemetry::{global, KeyValue}; +use opentelemetry::{global, Array as OtelArray, KeyValue, StringValue, Value as OtelValue}; +use switchyard_protocol::StopReason; use tracing::Span; +use tracing_opentelemetry::OpenTelemetrySpanExt; use crate::{ - AggLlmResponse, Context, Decision, Driver, LlmResponse, LlmResponseChunk, LlmResponseStream, - Request, Response, Result, Usage, + AggLlmResponse, Context, Decision, Driver, LibsyError, LlmClientError, LlmRequest, LlmResponse, + LlmResponseChunk, LlmResponseStream, Request, Response, Result, Usage, }; const METRICS_SCOPE: &str = "switchyard"; @@ -169,15 +174,45 @@ pub(crate) async fn observe_run( result } +/// Records request parameters represented directly by the neutral IR. +pub(crate) fn record_gen_ai_request(span: &Span, request: &LlmRequest) { + if request.stream { + span.record("gen_ai.request.stream", true); + } + if let Some(value) = request.sampling.temperature { + span.record("gen_ai.request.temperature", value); + } + if let Some(value) = request.sampling.top_p { + span.record("gen_ai.request.top_p", value); + } + if let Some(value) = request.sampling.top_k { + span.record("gen_ai.request.top_k", value); + } + if let Some(value) = request.output.max_output_tokens { + span.record("gen_ai.request.max_tokens", otel_int(value)); + } + if let Some(value) = request.reasoning.effort.as_deref() { + span.record("gen_ai.request.reasoning.level", value); + } + if let Some(value) = request + .output + .response_format + .as_ref() + .and_then(gen_ai_output_type) + { + span.record("gen_ai.output.type", value); + } +} + /// Adds terminal response and usage fields to the enclosing `libsy.client_call` /// span without consuming or buffering a streaming response. pub(crate) fn observe_client_call(result: Result) -> Result { let span = Span::current(); match result { Ok(mut response) => { - span.record("outcome", "ok"); match response.llm_response { LlmResponse::Agg(agg) => { + span.record("outcome", "ok"); record_gen_ai_response(&span, &agg); response.llm_response = LlmResponse::Agg(agg); } @@ -189,37 +224,34 @@ pub(crate) fn observe_client_call(result: Result) -> Result Ok(response) } Err(error) => { - record_client_error(&span, "client_call", &error); + let error_type = client_call_error_type(&error); + record_client_error(&span, &error_type, &error); Err(error) } } } fn observe_client_stream(stream: LlmResponseStream, span: Span) -> LlmResponseStream { - Box::pin(stream.inspect(move |item| match item { - Ok(LlmResponseChunk::MessageStart { id, model }) => { - record_optional(&span, "gen_ai.response.id", id.as_deref()); - record_optional(&span, "gen_ai.response.model", model.as_deref()); - } - Ok(LlmResponseChunk::Usage(usage)) => { - record_gen_ai_usage(&span, usage); - } - Ok(LlmResponseChunk::DecodeError { message }) => { - record_client_error(&span, "decode_error", message); - } - Ok(LlmResponseChunk::StreamError { message }) => { - record_client_error(&span, "stream_error", message); - } - Err(error) => { - record_client_error(&span, "client_stream", error); - } - _ => {} - })) + Box::pin(ObservedClientStream { + stream, + observer: Some(ClientStreamObserver { + span, + terminal: false, + }), + }) } fn record_gen_ai_response(span: &Span, response: &AggLlmResponse) { record_optional(span, "gen_ai.response.id", response.id.as_deref()); record_optional(span, "gen_ai.response.model", response.model.as_deref()); + record_finish_reasons( + span, + response + .outputs + .iter() + .filter_map(|output| output.stop_reason) + .map(stop_reason_name), + ); record_gen_ai_usage(span, &response.usage); } @@ -232,7 +264,7 @@ fn record_gen_ai_usage(span: &Span, usage: &Usage) { .unwrap_or_default() .saturating_add(cache_read.unwrap_or_default()) .saturating_add(cache_creation.unwrap_or_default()); - span.record("gen_ai.usage.input_tokens", input_tokens); + span.record("gen_ai.usage.input_tokens", otel_int(input_tokens)); } for (field, value) in [ ("gen_ai.usage.output_tokens", usage.output_tokens), @@ -244,17 +276,170 @@ fn record_gen_ai_usage(span: &Span, usage: &Usage) { ), ] { if let Some(value) = value { - span.record(field, value); + span.record(field, otel_int(value)); } } } +// OpenTelemetry integer attributes are signed; token counts are unsigned in the IR. +fn otel_int(value: u64) -> i64 { + value.min(i64::MAX as u64) as i64 +} + fn record_optional(span: &Span, field: &str, value: Option<&str>) { if let Some(value) = value { span.record(field, value); } } +// `tracing` fields cannot preserve a typed string array, so write this attribute directly. +fn record_finish_reasons(span: &Span, reasons: impl IntoIterator>) { + let reasons = reasons + .into_iter() + .map(|reason| StringValue::from(reason.into())) + .collect::>(); + if !reasons.is_empty() { + span.set_attribute( + "gen_ai.response.finish_reasons", + OtelValue::Array(OtelArray::String(reasons)), + ); + } +} + +fn stop_reason_name(reason: StopReason) -> &'static str { + match reason { + StopReason::EndTurn => "end_turn", + StopReason::MaxTokens => "max_tokens", + StopReason::ToolUse => "tool_use", + StopReason::ContentFilter => "content_filter", + StopReason::Error => "error", + StopReason::Unknown => "unknown", + } +} + +fn gen_ai_output_type(response_format: &serde_json::Value) -> Option<&'static str> { + match response_format + .get("type") + .and_then(serde_json::Value::as_str) + { + Some("json" | "json_object" | "json_schema") => Some("json"), + Some("text") => Some("text"), + _ => None, + } +} + +fn client_call_error_type(error: &LibsyError) -> Cow<'static, str> { + match error { + LibsyError::ClientCall { source, .. } => llm_client_error_type(source), + LibsyError::TargetNotFound { .. } => Cow::Borrowed("target_not_found"), + LibsyError::NoTargets => Cow::Borrowed("no_targets"), + LibsyError::AlgorithmError { .. } => Cow::Borrowed("algorithm_error"), + LibsyError::MissingClient { .. } => Cow::Borrowed("client_configuration_error"), + LibsyError::Driver(_) => Cow::Borrowed("driver_error"), + LibsyError::AlgorithmTask { .. } => Cow::Borrowed("algorithm_task_error"), + LibsyError::MissingFinalResponse => Cow::Borrowed("missing_final_response"), + LibsyError::AllTargetsExcluded => Cow::Borrowed("context_window_exceeded"), + LibsyError::External { .. } => Cow::Borrowed("_OTHER"), + } +} + +fn llm_client_error_type(error: &LlmClientError) -> Cow<'static, str> { + match error { + LlmClientError::InvalidRequest { .. } => Cow::Borrowed("invalid_request"), + LlmClientError::RequestTranslation(_) => Cow::Borrowed("request_translation"), + LlmClientError::RequestEncoding(_) => Cow::Borrowed("request_encoding"), + LlmClientError::ResponseTranslation(_) => Cow::Borrowed("response_translation"), + LlmClientError::Configuration { .. } => Cow::Borrowed("configuration"), + LlmClientError::Transport { .. } => Cow::Borrowed("transport"), + LlmClientError::Timeout { .. } => Cow::Borrowed("timeout"), + LlmClientError::ContextWindowExceeded { .. } => Cow::Borrowed("context_window_exceeded"), + LlmClientError::UpstreamHttp { status, .. } => Cow::Owned(status.to_string()), + LlmClientError::InvalidResponse { .. } => Cow::Borrowed("invalid_response"), + LlmClientError::Ffi { .. } => Cow::Borrowed("ffi"), + _ => Cow::Borrowed("_OTHER"), + } +} + +// Keep the client span alive until the response is drained, errors, or is abandoned. +struct ObservedClientStream { + stream: LlmResponseStream, + observer: Option, +} + +impl Stream for ObservedClientStream { + type Item = std::result::Result; + + fn poll_next(mut self: Pin<&mut Self>, cx: &mut TaskContext<'_>) -> Poll> { + match self.stream.as_mut().poll_next(cx) { + Poll::Ready(Some(item)) => { + let failed = self + .observer + .as_mut() + .is_some_and(|observer| observer.observe(&item)); + if failed { + self.observer.take(); + } + Poll::Ready(Some(item)) + } + Poll::Ready(None) => { + if let Some(mut observer) = self.observer.take() { + observer.complete(); + } + Poll::Ready(None) + } + Poll::Pending => Poll::Pending, + } + } +} + +struct ClientStreamObserver { + span: Span, + terminal: bool, +} + +impl ClientStreamObserver { + fn observe(&mut self, item: &std::result::Result) -> bool { + match item { + Ok(LlmResponseChunk::MessageStart { id, model }) => { + record_optional(&self.span, "gen_ai.response.id", id.as_deref()); + record_optional(&self.span, "gen_ai.response.model", model.as_deref()); + } + Ok(LlmResponseChunk::Usage(usage)) => record_gen_ai_usage(&self.span, usage), + Ok(LlmResponseChunk::MessageStop { reason }) => { + record_finish_reasons(&self.span, reason.iter().cloned()); + } + Ok(LlmResponseChunk::DecodeError { message }) => { + record_client_error(&self.span, "response_translation", message); + self.terminal = true; + } + Ok(LlmResponseChunk::StreamError { message }) => { + record_client_error(&self.span, "502", message); + self.terminal = true; + } + Err(error) => { + let error_type = llm_client_error_type(error); + record_client_error(&self.span, &error_type, error); + self.terminal = true; + } + _ => {} + } + self.terminal + } + + fn complete(&mut self) { + self.span.record("outcome", "ok"); + self.terminal = true; + } +} + +impl Drop for ClientStreamObserver { + fn drop(&mut self) { + if !self.terminal { + self.span.record("outcome", "cancelled"); + } + } +} + fn record_client_error(span: &Span, error_type: &str, error: &dyn std::fmt::Display) { span.record("outcome", "error"); span.record("otel.status_code", "ERROR"); diff --git a/crates/libsy/tests/observability.rs b/crates/libsy/tests/observability.rs index fca1be926..a7a218104 100644 --- a/crates/libsy/tests/observability.rs +++ b/crates/libsy/tests/observability.rs @@ -17,23 +17,27 @@ use std::time::Duration; use async_trait::async_trait; use futures::StreamExt; +use opentelemetry::trace::TracerProvider as _; +use opentelemetry::{Array as OtelArray, Value as OtelValue}; use opentelemetry_sdk::metrics::data::{AggregatedMetrics, MetricData, ResourceMetrics}; use opentelemetry_sdk::metrics::{InMemoryMetricExporter, PeriodicReader, SdkMeterProvider}; +use opentelemetry_sdk::trace::{InMemorySpanExporter, SdkTracerProvider, SpanData}; use parking_lot::Mutex; +use serde_json::json; use tracing::field::{Field, Visit}; use tracing::span::{Attributes, Id, Record}; use tracing::{Event, Subscriber}; +use tracing_opentelemetry::OpenTelemetryLayer; use tracing_subscriber::layer::{Context as LayerContext, SubscriberExt}; use tracing_subscriber::registry::LookupSpan; use tracing_subscriber::Layer; use switchyard_libsy::algorithms::{LlmTaskClassifier, TaskClassifierConfig}; use switchyard_libsy::{ - AggLlmResponse, Algorithm, Context, Decision, Driver, LibsyError, LlmResponse, - LlmResponseChunk, LlmTarget, LlmTargetSet, Metadata, Request, Response, RoutedLlmClient, Step, - Usage, + Algorithm, Context, Decision, Driver, LibsyError, LlmResponse, LlmResponseChunk, LlmTarget, + LlmTargetSet, Metadata, Request, Response, RoutedLlmClient, Step, Usage, }; -use switchyard_protocol::{text_request, text_response, LlmClientError}; +use switchyard_protocol::{text_request, text_response, LlmClientError, StopReason}; #[derive(Debug, thiserror::Error)] #[error("{0}")] @@ -93,6 +97,14 @@ impl Visit for FieldVisitor<'_> { fn record_u64(&mut self, field: &Field, value: u64) { self.0.insert(field.name().to_string(), value.to_string()); } + + fn record_i64(&mut self, field: &Field, value: i64) { + self.0.insert(field.name().to_string(), value.to_string()); + } + + fn record_f64(&mut self, field: &Field, value: f64) { + self.0.insert(field.name().to_string(), value.to_string()); + } } /// Subscriber layer capturing spans (with contextual parents and recorded @@ -147,9 +159,16 @@ where /// Installs the process-global telemetry sinks once: an in-memory OTel metric /// pipeline behind the global meter provider, and the capture layer as the /// global tracing subscriber. -fn telemetry() -> &'static (CaptureStore, InMemoryMetricExporter, SdkMeterProvider) { - static TELEMETRY: OnceLock<(CaptureStore, InMemoryMetricExporter, SdkMeterProvider)> = - OnceLock::new(); +type Telemetry = ( + CaptureStore, + InMemoryMetricExporter, + SdkMeterProvider, + InMemorySpanExporter, + SdkTracerProvider, +); + +fn telemetry() -> &'static Telemetry { + static TELEMETRY: OnceLock = OnceLock::new(); TELEMETRY.get_or_init(|| { let exporter = InMemoryMetricExporter::default(); let reader = PeriodicReader::builder(exporter.clone()).build(); @@ -157,14 +176,23 @@ fn telemetry() -> &'static (CaptureStore, InMemoryMetricExporter, SdkMeterProvid opentelemetry::global::set_meter_provider(provider.clone()); switchyard_libsy::initialize_metrics(); + let span_exporter = InMemorySpanExporter::default(); + let tracer_provider = SdkTracerProvider::builder() + .with_simple_exporter(span_exporter.clone()) + .build(); + let tracer = tracer_provider.tracer("switchyard-observability-test"); let store = CaptureStore::default(); - let subscriber = tracing_subscriber::registry().with(CaptureLayer { - store: store.clone(), - }); + let otel_layer: OpenTelemetryLayer<_, _> = + tracing_opentelemetry::layer().with_tracer(tracer); + let subscriber = tracing_subscriber::registry() + .with(CaptureLayer { + store: store.clone(), + }) + .with(otel_layer); if tracing::subscriber::set_global_default(subscriber).is_err() { panic!("a global tracing subscriber was already installed in this test binary"); } - (store, exporter, provider) + (store, exporter, provider, span_exporter, tracer_provider) }) } @@ -344,13 +372,15 @@ impl RoutedLlmClient for UsageClient { _request: Request, decision: Arc, ) -> Result { + let mut response = text_response( + Some(decision.selected_model().to_string()), + "observed response", + ); + response.id = Some("obs-response-1".to_string()); + response.usage = self.usage.clone(); + response.outputs[0].stop_reason = Some(StopReason::EndTurn); Ok(Response { - llm_response: LlmResponse::Agg(AggLlmResponse { - id: Some("obs-response-1".to_string()), - model: Some(decision.selected_model().to_string()), - usage: self.usage.clone(), - ..AggLlmResponse::default() - }), + llm_response: LlmResponse::Agg(response), metadata: None, }) } @@ -428,10 +458,47 @@ fn find_span(spans: &[SpanRecord], name: &str, field: &str, value: &str) -> Span } } +fn find_otel_span(exporter: &InMemorySpanExporter, name: &str, model: &str) -> SpanData { + let spans = match exporter.get_finished_spans() { + Ok(spans) => spans, + Err(error) => panic!("failed to read exported spans: {error}"), + }; + match spans.iter().find(|span| { + span.name == name + && span.attributes.iter().any(|attribute| { + attribute.key.as_str() == "gen_ai.request.model" + && attribute.value.as_str() == model + }) + }) { + Some(span) => span.clone(), + None => { + let available = spans + .iter() + .map(|span| { + let model = span + .attributes + .iter() + .find(|attribute| attribute.key.as_str() == "gen_ai.request.model") + .map(|attribute| attribute.value.as_str().into_owned()); + (span.name.to_string(), model) + }) + .collect::>(); + panic!("no exported '{name}' span for model {model}; available: {available:?}") + } + } +} + +fn otel_attribute<'a>(span: &'a SpanData, key: &str) -> Option<&'a OtelValue> { + span.attributes + .iter() + .find(|attribute| attribute.key.as_str() == key) + .map(|attribute| &attribute.value) +} + #[tokio::test] async fn successful_run_records_metrics_spans_and_decision_log() -> switchyard_libsy::Result<()> { let _guard = serialize_test().lock().await; - let (store, exporter, provider) = telemetry(); + let (store, exporter, provider, span_exporter, _) = telemetry(); const ALGO: &str = "obs-success-algo"; const MODEL: &str = "obs-success-model"; let before = flushed_metrics(exporter, provider); @@ -449,11 +516,15 @@ async fn successful_run_records_metrics_spans_and_decision_log() -> switchyard_l cache: Usage::cache_details(Some(3), Some(4)), }, }) as Arc; + let mut request = request_with_metadata("obs-session-1", "obs-corr-1"); + request.llm_request.sampling.temperature = Some(0.25); + request.llm_request.sampling.top_p = Some(0.9); + request.llm_request.sampling.top_k = Some(40); + request.llm_request.output.max_output_tokens = Some(512); + request.llm_request.output.response_format = Some(json!({"type": "json_schema"})); + request.llm_request.reasoning.effort = Some("high".to_string()); let (trace, _response) = algo(ALGO, MODEL, Some(client)) - .run( - Context::default(), - request_with_metadata("obs-session-1", "obs-corr-1"), - ) + .run(Context::default(), request) .await?; assert_eq!(trace.len(), 1); @@ -575,9 +646,15 @@ async fn successful_run_records_metrics_spans_and_decision_log() -> switchyard_l for (field, value) in [ ("otel.kind", "client"), ("switchyard.algorithm", ALGO), + ("otel.name", "chat obs-success-model"), ("gen_ai.operation.name", "chat"), ("gen_ai.request.model", MODEL), - ("gen_ai.request.stream", "false"), + ("gen_ai.request.temperature", "0.25"), + ("gen_ai.request.top_p", "0.9"), + ("gen_ai.request.top_k", "40"), + ("gen_ai.request.max_tokens", "512"), + ("gen_ai.request.reasoning.level", "high"), + ("gen_ai.output.type", "json"), ("gen_ai.conversation.id", "obs-session-1"), ("gen_ai.response.id", "obs-response-1"), ("gen_ai.response.model", MODEL), @@ -593,6 +670,22 @@ async fn successful_run_records_metrics_spans_and_decision_log() -> switchyard_l "unexpected {field}" ); } + assert_eq!(client_span.fields.get("gen_ai.request.stream"), None); + + let otel_span = find_otel_span(span_exporter, "chat obs-success-model", MODEL); + assert!(matches!( + otel_attribute(&otel_span, "gen_ai.response.finish_reasons"), + Some(OtelValue::Array(OtelArray::String(reasons))) + if reasons.len() == 1 && reasons[0].as_str() == "end_turn" + )); + assert_eq!( + otel_attribute(&otel_span, "gen_ai.request.max_tokens"), + Some(&OtelValue::I64(512)) + ); + assert_eq!( + otel_attribute(&otel_span, "gen_ai.usage.input_tokens"), + Some(&OtelValue::I64(18)) + ); let call_span = find_span(&spans, "libsy.llm_call", "selected_model", MODEL); assert_eq!(call_span.parent.as_deref(), Some("libsy.run")); @@ -676,10 +769,26 @@ impl RoutedLlmClient for StreamingUsageClient { } } +struct TimeoutClient; + +#[async_trait] +impl RoutedLlmClient for TimeoutClient { + async fn call( + &self, + _ctx: Context, + _request: Request, + _decision: Arc, + ) -> Result { + Err(LlmClientError::Timeout { + source: Box::new(TestError("upstream timed out")), + }) + } +} + #[tokio::test] async fn streamed_usage_updates_the_client_call_span() -> switchyard_libsy::Result<()> { let _guard = serialize_test().lock().await; - let (store, _, _) = telemetry(); + let (store, _, _, span_exporter, _) = telemetry(); const ALGO: &str = "obs-stream-algo"; const MODEL: &str = "obs-stream-model"; let client = Arc::new(StreamingUsageClient) as Arc; @@ -700,6 +809,7 @@ async fn streamed_usage_updates_the_client_call_span() -> switchyard_libsy::Resu let spans = store.spans(); let client_span = find_span(&spans, "libsy.client_call", "selected_model", MODEL); for (field, value) in [ + ("otel.name", "chat obs-stream-model"), ("gen_ai.request.stream", "true"), ("gen_ai.response.id", "obs-stream-response"), ("gen_ai.response.model", MODEL), @@ -713,13 +823,77 @@ async fn streamed_usage_updates_the_client_call_span() -> switchyard_libsy::Resu "unexpected {field}" ); } + let otel_span = find_otel_span(span_exporter, "chat obs-stream-model", MODEL); + assert!(matches!( + otel_attribute(&otel_span, "gen_ai.response.finish_reasons"), + Some(OtelValue::Array(OtelArray::String(reasons))) + if reasons.len() == 1 && reasons[0].as_str() == "end_turn" + )); Ok(()) } +#[tokio::test] +async fn dropped_stream_records_cancelled_outcome() -> switchyard_libsy::Result<()> { + let _guard = serialize_test().lock().await; + let (store, _, _, _, _) = telemetry(); + const ALGO: &str = "obs-cancelled-stream-algo"; + const MODEL: &str = "obs-cancelled-stream-model"; + let client = Arc::new(StreamingUsageClient) as Arc; + let mut request = request_with_metadata("obs-cancelled-session", "obs-cancelled-corr"); + request.llm_request.stream = true; + let (_, response) = algo(ALGO, MODEL, Some(client)) + .run(Context::default(), request) + .await?; + let LlmResponse::Stream(stream) = response.llm_response else { + return Err(test_error("expected a streamed response")); + }; + drop(stream); + + let spans = store.spans(); + let client_span = find_span(&spans, "libsy.client_call", "selected_model", MODEL); + assert_eq!( + client_span.fields.get("outcome").map(String::as_str), + Some("cancelled") + ); + Ok(()) +} + +#[tokio::test] +async fn typed_client_failure_records_semantic_error_type() { + let _guard = serialize_test().lock().await; + let (store, _, _, _, _) = telemetry(); + const ALGO: &str = "obs-timeout-algo"; + const MODEL: &str = "obs-timeout-model"; + let result = algo( + ALGO, + MODEL, + Some(Arc::new(TimeoutClient) as Arc), + ) + .run( + Context::default(), + request_with_metadata("obs-timeout-session", "obs-timeout-corr"), + ) + .await; + assert!(matches!( + result, + Err(LibsyError::ClientCall { + source: LlmClientError::Timeout { .. }, + .. + }) + )); + + let spans = store.spans(); + let client_span = find_span(&spans, "libsy.client_call", "selected_model", MODEL); + assert_eq!( + client_span.fields.get("error.type").map(String::as_str), + Some("timeout") + ); +} + #[tokio::test] async fn failed_call_records_error_outcome_and_warn_logs() -> switchyard_libsy::Result<()> { let _guard = serialize_test().lock().await; - let (store, exporter, provider) = telemetry(); + let (store, exporter, provider, _, _) = telemetry(); const ALGO: &str = "obs-failure-algo"; const MODEL: &str = "obs-failure-model"; let before = flushed_metrics(exporter, provider); @@ -841,7 +1015,7 @@ async fn failed_call_records_error_outcome_and_warn_logs() -> switchyard_libsy:: #[tokio::test] async fn classifier_metrics_count_only_the_final_routed_call() -> switchyard_libsy::Result<()> { let _guard = serialize_test().lock().await; - let (_store, exporter, provider) = telemetry(); + let (_store, exporter, provider, _, _) = telemetry(); let before = flushed_metrics(exporter, provider); let total_requests_before = u64_gauge_value(&before, "switchyard.total_requests").unwrap_or_default();