diff --git a/Cargo.lock b/Cargo.lock index c1664fc838..ec5215206c 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -3362,6 +3362,8 @@ dependencies = [ "tempfile", "tokio", "toml 1.1.3+spec-1.1.0", + "tracing", + "tracing-subscriber", "wiremock", ] diff --git a/README.md b/README.md index ab16fee2cd..9bad346903 100644 --- a/README.md +++ b/README.md @@ -24,7 +24,7 @@ Evaluate the complete agent, model pool, and routing configuration against your | --- | --- | --- | | **OpenRouter** | [Use Switchyard through OpenRouter](https://openrouter.ai/docs/guides/routing/routers/switchyard-router) | Hosted by OpenRouter. Set `model` to `nvidia/switchyard` in API requests. | | **LiteLLM** | [Run the Switchyard routing-plugin example](examples/litellm/README.md#quick-start-with-the-local-proxy) | Experimental and checkout-only. The example pins LiteLLM 1.102.0 and supports Stage routing based on request history, plus Random routing. It cannot service intermediate model calls required by classifier or escalation algorithms. | -| **NeMo Relay** | [Build and configure the native plugin](crates/switchyard-nemo-relay-plugin/README.md#build-from-source) | Requires Relay `>=0.8.0, <1.0.0`. The source-build path requires a Rust toolchain and Python 3. | +| **NeMo Relay** | [Build and configure the native plugin](crates/switchyard-nemo-relay-plugin/README.md#build-from-source) | Requires Relay `>=0.10.0, <1.0.0`. The source-build path requires a Rust toolchain and Python 3. | Relay 0.8.x and 0.9.0 can lose upstream error status and details when the plugin is enabled, even for models outside its routes. Review the diff --git a/crates/libsy-llm-client/src/run.rs b/crates/libsy-llm-client/src/run.rs index ab8a3e9cb2..8ce6e8ceca 100644 --- a/crates/libsy-llm-client/src/run.rs +++ b/crates/libsy-llm-client/src/run.rs @@ -232,7 +232,14 @@ async fn serve( let algorithm = call.algorithm.clone(); let candidates = call.models.len(); call.respond(call_one( - &clients, &target, request, &algorithm, &observe, 0, candidates, true, + &clients, + &target, + request, + &algorithm, + &observe, + 0, + candidates, + CallPhase::Routing, )) .await } @@ -255,7 +262,7 @@ async fn call_first_available( observe, index, models.len(), - false, + CallPhase::Completion, ) .await { @@ -275,6 +282,25 @@ async fn call_first_available( Err(LibsyError::NoTargets) } +/// The stage of the Switchyard run in which a model call was issued. +/// +/// A routing-phase call may still supply the final answer. This is deliberately +/// distinct from the `routing`/`answer` role assigned after the outcome is known. +#[derive(Clone, Copy)] +enum CallPhase { + Routing, + Completion, +} + +impl CallPhase { + const fn as_str(self) -> &'static str { + match self { + Self::Routing => "routing", + Self::Completion => "completion", + } + } +} + /// Call one candidate model and record its observation and span. #[allow(clippy::too_many_arguments)] #[tracing::instrument( @@ -286,6 +312,7 @@ async fn call_first_available( switchyard.algorithm = algorithm, switchyard.candidate = index + 1, switchyard.candidate_count = count, + switchyard.call_phase = phase.as_str(), selected_model = %model_id, otel.kind = "client", otel.name = %format_args!("chat {model_id}"), @@ -325,8 +352,9 @@ async fn call_one( index: usize, // count is for span log count: usize, - buffer: bool, + phase: CallPhase, ) -> Result { + let is_routing = matches!(phase, CallPhase::Routing); let span = tracing::Span::current(); observability::record_gen_ai_request(&span, &request.llm_request); if let Some(session_id) = request @@ -342,7 +370,7 @@ async fn call_one( let started = Instant::now(); let result = async { let mut response = client?.call(request).await?; - if buffer && let LlmResponse::Stream(chunks) = response.llm_response { + if is_routing && let LlmResponse::Stream(chunks) = response.llm_response { response.llm_response = LlmResponse::Stream(buffer_routing_stream(chunks).await?); } Ok(response) @@ -356,10 +384,10 @@ async fn call_one( response }); let result = observability::observe_client_call(result); - let result = if !buffer { - metrics::observe_routed_request(algorithm, model_id, Some(duration), result) - } else { + let result = if is_routing { result + } else { + metrics::observe_routed_request(algorithm, model_id, Some(duration), result) }; observe(ModelCallObservation { selected_model: model_id.clone(), diff --git a/crates/libsy-llm-client/tests/observability.rs b/crates/libsy-llm-client/tests/observability.rs index 37a71f3dbe..46ff1b13df 100644 --- a/crates/libsy-llm-client/tests/observability.rs +++ b/crates/libsy-llm-client/tests/observability.rs @@ -1387,6 +1387,35 @@ async fn observed_run_reports_one_successful_routed_call() -> switchyard_libsy:: Ok(()) } +#[tokio::test] +async fn answered_routing_call_keeps_its_execution_phase() -> switchyard_libsy::Result<()> { + let _guard = serialize_test().lock().await; + let (store, _, _, _, _) = telemetry(); + const ALGO: &str = "answered-routing-call"; + const MODEL: &str = "answer-from-routing"; + + switchyard_llm_client::run( + Arc::new(RoutingCallAlgo { + name: ALGO.to_string(), + target: MODEL.into(), + }), + ClientRouter::single(Arc::new(UsageClient { + usage: Usage::default(), + })), + request_with_metadata("answered-routing-session", "answered-routing-correlation"), + Arc::new(RuntimeModels::default()), + None, + ) + .await?; + + let span = find_span(&store.spans(), "libsy.client_call", "selected_model", MODEL); + assert_eq!( + span.fields.get("switchyard.call_phase").map(String::as_str), + Some("routing") + ); + Ok(()) +} + /// A streamed response keeps the client span available until terminal usage arrives. struct StreamingUsageClient; @@ -1839,7 +1868,7 @@ async fn failed_call_records_metrics_without_error_details() -> switchyard_libsy #[tokio::test] async fn classifier_metrics_count_routing_and_answer_calls_once() -> 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(); @@ -1854,6 +1883,13 @@ async fn classifier_metrics_count_routing_and_answer_calls_once() -> switchyard_ assert_eq!(selected_model, "weak"); + let spans = store.spans(); + let span = find_span(&spans, "libsy.client_call", "selected_model", "weak"); + assert_eq!( + span.fields.get("switchyard.call_phase").map(String::as_str), + Some("completion") + ); + let snapshots = flushed_metrics(exporter, provider); assert_eq!( u64_counter_value( diff --git a/crates/switchyard-nemo-relay-plugin/Cargo.toml b/crates/switchyard-nemo-relay-plugin/Cargo.toml index ac9099f7f7..1e893593df 100644 --- a/crates/switchyard-nemo-relay-plugin/Cargo.toml +++ b/crates/switchyard-nemo-relay-plugin/Cargo.toml @@ -27,6 +27,8 @@ switchyard-runner.workspace = true switchyard-translation.workspace = true tokio.workspace = true toml = "1.1" +tracing.workspace = true +tracing-subscriber.workspace = true [dev-dependencies] tempfile = "3" diff --git a/crates/switchyard-nemo-relay-plugin/README.md b/crates/switchyard-nemo-relay-plugin/README.md index 208de4386a..34fa3dbab2 100644 --- a/crates/switchyard-nemo-relay-plugin/README.md +++ b/crates/switchyard-nemo-relay-plugin/README.md @@ -231,6 +231,12 @@ marks. Call marks distinguish routing from answer calls; decisions distinguish selected from served models. Token metrics cover both call roles, while Relay retains ownership of the outer LLM lifecycle. +The plugin also captures the existing `libsy.client_call` and +`libsy.upstream_attempt` spans for each routed request. It replays completed +spans beneath Relay's outer LLM scope with their original timestamps. The +projection includes allowlisted, payload-free routing metadata only; it excludes +prompts, responses, provider error bodies, and credentials. + ## Observability When Relay is configured with OTLP logs and metrics exporters, the plugin emits diff --git a/crates/switchyard-nemo-relay-plugin/src/lib.rs b/crates/switchyard-nemo-relay-plugin/src/lib.rs index ba8db5de67..6f0c543119 100644 --- a/crates/switchyard-nemo-relay-plugin/src/lib.rs +++ b/crates/switchyard-nemo-relay-plugin/src/lib.rs @@ -3,17 +3,20 @@ mod config; mod runtime; +mod span_capture; mod translation; use std::sync::Arc; -use nemo_relay_plugin::{ - ConfigDiagnostic, DiagnosticLevel, Json, NativePlugin, PluginContext, PluginRuntime, -}; +use nemo_relay_plugin::{ConfigDiagnostic, DiagnosticLevel, Json, NativePlugin, PluginContext}; use serde_json::Map; use crate::config::{SwitchyardConfig, protocol_from_call}; -use crate::runtime::{SwitchyardRuntime, emit_event, emit_events}; +use crate::runtime::{ + RoutingEvent, RoutingEventEmitter, SwitchyardRuntime, emit_event as emit_routing_event, + emit_events, +}; +use crate::span_capture::SpanCapture; #[derive(Default)] struct SwitchyardPlugin; @@ -50,8 +53,24 @@ impl NativePlugin for SwitchyardPlugin { let priority = config.priority; let runtime = Arc::new(SwitchyardRuntime::new(config)?); let plugin_runtime = ctx.runtime(); - register_buffered(ctx, priority, Arc::clone(&runtime), plugin_runtime.clone())?; - register_stream(ctx, priority, runtime, plugin_runtime)?; + let event_runtime = Arc::clone(&runtime); + let emit_event: RoutingEventEmitter = Arc::new(move |event| { + emit_routing_event(&plugin_runtime, event_runtime.redactor(), event) + }); + let span_runtime = Arc::clone(&runtime); + let span_emit_event = Arc::clone(&emit_event); + let span_capture = SpanCapture::new(move |mut span| { + span.sanitize(span_runtime.redactor()); + span_emit_event(RoutingEvent::Span(span)); + }); + register_buffered( + ctx, + priority, + Arc::clone(&runtime), + Arc::clone(&emit_event), + span_capture.clone(), + )?; + register_stream(ctx, priority, runtime, emit_event, span_capture)?; Ok(()) } } @@ -60,14 +79,16 @@ fn register_buffered( ctx: &mut PluginContext<'_>, priority: i32, runtime: Arc, - plugin_runtime: PluginRuntime, + emit_event: RoutingEventEmitter, + span_capture: SpanCapture, ) -> Result<(), String> { ctx.register_llm_execution_intercept( "switchyard.runner.buffered", priority, move |name, request, _context, next| { let runtime = Arc::clone(&runtime); - let plugin_runtime = plugin_runtime.clone(); + let emit_event = Arc::clone(&emit_event); + let span_capture = span_capture.clone(); async move { let Some(inbound) = protocol_from_call(&name) else { return next.call(request).await; @@ -76,8 +97,10 @@ fn register_buffered( return next.call(request).await; } let decoded = runtime.decode_request(inbound, request, false)?; - let execution = runtime.execute_buffered(inbound, decoded).await; - emit_events(&plugin_runtime, runtime.redactor(), execution.events); + let execution = span_capture + .capture(runtime.execute_buffered(inbound, decoded)) + .await; + emit_events(&emit_event, execution.events); execution.result } }, @@ -88,14 +111,16 @@ fn register_stream( ctx: &mut PluginContext<'_>, priority: i32, runtime: Arc, - plugin_runtime: PluginRuntime, + emit_event: RoutingEventEmitter, + span_capture: SpanCapture, ) -> Result<(), String> { ctx.register_llm_stream_execution_intercept( "switchyard.runner.streaming", priority, move |name, request, _context, next| { let runtime = Arc::clone(&runtime); - let plugin_runtime = plugin_runtime.clone(); + let emit_event = Arc::clone(&emit_event); + let span_capture = span_capture.clone(); async move { let Some(inbound) = protocol_from_call(&name) else { return next.call(request).await; @@ -104,18 +129,10 @@ fn register_stream( return next.call(request).await; } let decoded = runtime.decode_request(inbound, request, true)?; - let stream_plugin_runtime = plugin_runtime.clone(); - let stream_runtime = Arc::clone(&runtime); - let execution = runtime - .execute_stream( - inbound, - decoded, - Arc::new(move |event| { - emit_event(&stream_plugin_runtime, stream_runtime.redactor(), event) - }), - ) + let execution = span_capture + .capture(runtime.execute_stream(inbound, decoded, Arc::clone(&emit_event))) .await; - emit_events(&plugin_runtime, runtime.redactor(), execution.events); + emit_events(&emit_event, execution.events); execution.result } }, diff --git a/crates/switchyard-nemo-relay-plugin/src/redaction_tests.rs b/crates/switchyard-nemo-relay-plugin/src/redaction_tests.rs index e25e0ad7de..d3a42c72e0 100644 --- a/crates/switchyard-nemo-relay-plugin/src/redaction_tests.rs +++ b/crates/switchyard-nemo-relay-plugin/src/redaction_tests.rs @@ -8,6 +8,8 @@ use std::process::Command; use wiremock::matchers::{header, method, path}; use wiremock::{Mock, MockServer, ResponseTemplate}; +use crate::span_capture::SpanCapture; + use super::*; const KEY_ENV: &str = "SWITCHYARD_RELAY_REDACTION_TEST_KEY"; @@ -122,16 +124,22 @@ async fn reflected_response(format: WireFormat, streaming: bool, fail: bool) { }), }; let captured = Arc::new(Mutex::new(Vec::new())); + let span_observed = Arc::clone(&captured); + let redactor = Arc::clone(&runtime.redactor); + let span_capture = SpanCapture::new(move |mut span| { + span.sanitize(&redactor); + span_observed.lock().unwrap().push(RoutingEvent::Span(span)); + }); let output = if streaming { let observed = Arc::clone(&captured); - let execution = runtime - .execute_stream( + let execution = span_capture + .capture(runtime.execute_stream( format, request, Arc::new(move |event| { observed.lock().unwrap().push(event); }), - ) + )) .await; assert!( captured.lock().unwrap().is_empty(), @@ -142,7 +150,9 @@ async fn reflected_response(format: WireFormat, streaming: bool, fail: bool) { assert_eq!(events.iter().any(Result::is_err), fail); format!("{events:?}") } else { - let execution = runtime.execute_buffered(format, request).await; + let execution = span_capture + .capture(runtime.execute_buffered(format, request)) + .await; captured.lock().unwrap().extend(execution.events); assert_eq!(execution.result.is_err(), fail); format!("{:?}", execution.result) @@ -155,7 +165,14 @@ async fn reflected_response(format: WireFormat, streaming: bool, fail: bool) { assert!(output.contains("ordinary diagnostics"), "{output}"); assert!(output.contains("[REDACTED]"), "{output}"); } - let telemetry = format!("{:?}", captured.lock().unwrap()); + let captured = captured.lock().unwrap(); + assert!( + captured + .iter() + .any(|event| matches!(event, RoutingEvent::Span(_))), + "client span was not emitted" + ); + let telemetry = format!("{captured:?}"); assert!(!telemetry.contains(KEY), "{telemetry}"); assert!(telemetry.contains("[REDACTED]"), "{telemetry}"); } diff --git a/crates/switchyard-nemo-relay-plugin/src/runtime.rs b/crates/switchyard-nemo-relay-plugin/src/runtime.rs index 4c1b7ab057..0870348323 100644 --- a/crates/switchyard-nemo-relay-plugin/src/runtime.rs +++ b/crates/switchyard-nemo-relay-plugin/src/runtime.rs @@ -22,6 +22,7 @@ use switchyard_runner::{ use switchyard_translation::{TranslationEngine, encode_stream_with_extensions}; use crate::config::SwitchyardConfig; +use crate::span_capture::CapturedSpan; use crate::translation; #[cfg(test)] @@ -59,6 +60,7 @@ pub(crate) struct RoutingMetric { pub(crate) enum RoutingEvent { Mark(RoutingMark), Metric(RoutingMetric), + Span(CapturedSpan), } struct MetricDescriptor<'a> { @@ -170,21 +172,12 @@ impl SwitchyardRuntime { request: Request, emit_event: RoutingEventEmitter, ) -> Execution { - let redactor = Arc::clone(&self.redactor); - let emit_event: RoutingEventEmitter = Arc::new(move |event| { - emit_event(sanitize_event(&redactor, event)); - }); + let emit_event = self.sanitized_emitter(emit_event); let request_extensions = request.llm_request.extensions.clone(); let Execution { result, mut events } = self.execute(inbound, request).await; let (result, finalization_failed) = match result { Ok(routed) => { - let metadata = events - .iter() - .find_map(|event| match event { - RoutingEvent::Mark(mark) => Some(mark.metadata.clone()), - RoutingEvent::Metric(_) => None, - }) - .unwrap_or_else(|| Json::Object(Map::new())); + let metadata = event_metadata(&events).unwrap_or_else(|| Json::Object(Map::new())); let result = returned_events( routed.response, inbound, @@ -224,6 +217,11 @@ impl SwitchyardRuntime { } } + fn sanitized_emitter(&self, emit_event: RoutingEventEmitter) -> RoutingEventEmitter { + let redactor = Arc::clone(&self.redactor); + Arc::new(move |event| emit_event(sanitize_event(&redactor, event))) + } + async fn execute(&self, inbound: WireFormat, request: Request) -> Execution { let Some(route) = self.route(&request) else { return Execution { @@ -465,13 +463,9 @@ fn evidence_for_mark(evidence: Option) -> Option { (!evidence.is_empty()).then_some(Json::Object(evidence)) } -pub(crate) fn emit_events( - runtime: &PluginRuntime, - redactor: &ProviderKeyRedactor, - events: Vec, -) { +pub(crate) fn emit_events(emit_event: &RoutingEventEmitter, events: Vec) { for event in events { - emit_event(runtime, redactor, event); + emit_event(event); } } @@ -496,6 +490,9 @@ pub(crate) fn emit_event( RoutingEvent::Metric(metric) => runtime .emit_metric(&metric.name, metric.measurements, Some(&metric.metadata)) .map_err(|error| ("routing metric", metric.name, error)), + RoutingEvent::Span(span) => span + .emit(runtime) + .map_err(|error| ("routing span", "libsy".into(), error)), }; if let Err((kind, name, error)) = result { let message = redactor.text(format!( @@ -521,6 +518,7 @@ fn sanitize_event(redactor: &ProviderKeyRedactor, mut event: RoutingEvent) -> Ro .map(|value| redactor.value(value)); } } + RoutingEvent::Span(span) => span.sanitize(redactor), } event } @@ -690,7 +688,7 @@ fn route_execution_error_events( fn event_metadata(events: &[RoutingEvent]) -> Option { events.iter().find_map(|event| match event { RoutingEvent::Mark(mark) => Some(mark.metadata.clone()), - RoutingEvent::Metric(_) => None, + _ => None, }) } @@ -1057,7 +1055,7 @@ mod tests { RoutingEvent::Mark(mark) if mark.name == "switchyard.routing.decision" => { Some(&mark.data) } - RoutingEvent::Mark(_) | RoutingEvent::Metric(_) => None, + _ => None, }) .expect("decision mark should be emitted"); assert_eq!(decision["selected_model"], "target/model"); @@ -1141,7 +1139,7 @@ mod tests { RoutingEvent::Mark(mark) if mark.name == "switchyard.routing.error" => { Some(&mark.data) } - RoutingEvent::Mark(_) | RoutingEvent::Metric(_) => None, + _ => None, }) .expect("error mark should be emitted"); assert!(error["outcome_id"].is_string()); diff --git a/crates/switchyard-nemo-relay-plugin/src/span_capture.rs b/crates/switchyard-nemo-relay-plugin/src/span_capture.rs new file mode 100644 index 0000000000..10fc1c4734 --- /dev/null +++ b/crates/switchyard-nemo-relay-plugin/src/span_capture.rs @@ -0,0 +1,436 @@ +// SPDX-FileCopyrightText: Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved. +// SPDX-License-Identifier: Apache-2.0 + +use std::fmt; +use std::future::Future; +use std::pin::Pin; +use std::task::{Context as TaskContext, Poll}; +use std::time::{Instant, SystemTime}; + +use nemo_relay_plugin::{Json, PluginRuntime, ScopeType}; +use serde_json::{Map, Value}; +use switchyard_runner::ProviderKeyRedactor; +use tracing::field::{Field, Visit}; +use tracing::span::{Attributes, Id, Record}; +use tracing::{Metadata, Subscriber}; +use tracing_subscriber::Layer; +use tracing_subscriber::filter::{LevelFilter, filter_fn}; +use tracing_subscriber::layer::Context; +use tracing_subscriber::prelude::*; +use tracing_subscriber::registry::LookupSpan; + +const CLIENT_CALL: &str = "libsy.client_call"; +const UPSTREAM_ATTEMPT: &str = "libsy.upstream_attempt"; + +type SpanEmitter = Box; + +#[derive(Debug)] +pub(crate) struct CapturedSpan { + name: &'static str, + fields: Map, + started_at: SystemTime, + ended_at: SystemTime, + children: Vec, +} + +struct OpenSpan { + name: &'static str, + fields: Map, + started_at: SystemTime, + started: Instant, + children: Vec, +} + +struct CaptureLayer { + emit: SpanEmitter, +} + +impl Layer for CaptureLayer +where + S: Subscriber + for<'lookup> LookupSpan<'lookup>, +{ + fn on_new_span(&self, attrs: &Attributes<'_>, id: &Id, ctx: Context<'_, S>) { + let name = attrs.metadata().name(); + let mut fields = Map::new(); + attrs.record(&mut FieldVisitor(&mut fields)); + if let Some(span) = ctx.span(id) { + span.extensions_mut().insert(OpenSpan { + name, + fields, + started_at: SystemTime::now(), + started: Instant::now(), + children: Vec::new(), + }); + } + } + + fn on_record(&self, id: &Id, values: &Record<'_>, ctx: Context<'_, S>) { + let Some(span) = ctx.span(id) else { + return; + }; + let mut extensions = span.extensions_mut(); + if let Some(open) = extensions.get_mut::() { + values.record(&mut FieldVisitor(&mut open.fields)); + } + } + + fn on_close(&self, id: Id, ctx: Context<'_, S>) { + let Some(span) = ctx.span(&id) else { + return; + }; + let parent_id = span.parent().map(|parent| parent.id().clone()); + let Some(open) = span.extensions_mut().remove::() else { + return; + }; + let OpenSpan { + name, + mut fields, + started_at, + started, + children, + } = open; + // A span that closes before its terminal bookkeeping ran was cancelled. + fields + .entry("outcome") + .or_insert_with(|| Value::String("cancelled".into())); + let captured = CapturedSpan { + name, + fields, + started_at, + ended_at: started_at + .checked_add(started.elapsed()) + .unwrap_or_else(SystemTime::now), + children, + }; + + if captured.name == UPSTREAM_ATTEMPT + && let Some(parent) = parent_id.and_then(|parent| ctx.span(&parent)) + && let Some(open) = parent.extensions_mut().get_mut::() + { + open.children.push(captured); + return; + } + (self.emit)(captured); + } +} + +impl CapturedSpan { + pub(crate) fn sanitize(&mut self, redactor: &ProviderKeyRedactor) { + if let Json::Object(fields) = redactor.value(Json::Object(std::mem::take(&mut self.fields))) + { + self.fields = fields; + } + for child in &mut self.children { + child.sanitize(redactor); + } + } + + pub(crate) fn emit(self, runtime: &PluginRuntime) -> Result<(), String> { + let metadata = Json::Object(self.fields); + let status = span_status(&metadata); + let mut scope = runtime.scope_at( + self.name, + relay_scope_type(self.name), + None, + Some(&metadata), + None, + self.started_at, + )?; + let mut child_result = Ok(()); + for child in self.children { + child_result = child_result.and(child.emit(runtime)); + } + let close_result = scope.close_at(None, status.as_ref(), self.ended_at); + child_result.and(close_result) + } +} + +/// Captures Switchyard client spans for one plugin registration. +#[derive(Clone)] +pub(crate) struct SpanCapture { + dispatch: tracing::Dispatch, +} + +impl SpanCapture { + pub(crate) fn new(emit: impl Fn(CapturedSpan) + Send + Sync + 'static) -> Self { + let layer = CaptureLayer { + emit: Box::new(emit), + } + .with_filter(filter_fn(captures).with_max_level_hint(LevelFilter::DEBUG)); + Self { + dispatch: tracing::Dispatch::new(tracing_subscriber::registry().with(layer)), + } + } + + /// Polls a future with the capture subscriber as its default. + pub(crate) async fn capture(&self, future: impl Future) -> T { + CaptureFuture { + inner: Some(Box::pin(future)), + dispatch: self.dispatch.clone(), + } + .await + } +} + +// Standard subscriber wrappers restore the ambient dispatch before drop. This +// wrapper also drops the inner future under capture so cancelled spans close here. +struct CaptureFuture { + inner: Option>>, + dispatch: tracing::Dispatch, +} + +impl Future for CaptureFuture { + type Output = F::Output; + + fn poll(self: Pin<&mut Self>, cx: &mut TaskContext<'_>) -> Poll { + let this = self.get_mut(); + let Some(inner) = this.inner.as_mut() else { + return Poll::Pending; + }; + tracing::dispatcher::with_default(&this.dispatch, || inner.as_mut().poll(cx)) + } +} + +impl Drop for CaptureFuture { + fn drop(&mut self) { + let Some(inner) = self.inner.take() else { + return; + }; + tracing::dispatcher::with_default(&self.dispatch, || drop(inner)); + } +} + +struct FieldVisitor<'a>(&'a mut Map); + +impl FieldVisitor<'_> { + fn insert(&mut self, field: &Field, value: impl FnOnce() -> Value) { + if is_safe_field(field.name()) { + self.0.insert(field.name().into(), value()); + } + } +} + +impl Visit for FieldVisitor<'_> { + fn record_debug(&mut self, field: &Field, value: &dyn fmt::Debug) { + self.insert(field, || Value::String(format!("{value:?}"))); + } + + fn record_str(&mut self, field: &Field, value: &str) { + self.insert(field, || Value::String(value.into())); + } + + fn record_bool(&mut self, field: &Field, value: bool) { + self.insert(field, || Value::Bool(value)); + } + + fn record_i64(&mut self, field: &Field, value: i64) { + self.insert(field, || Value::from(value)); + } + + fn record_u64(&mut self, field: &Field, value: u64) { + self.insert(field, || Value::from(value)); + } + + fn record_f64(&mut self, field: &Field, value: f64) { + self.insert(field, || Value::from(value)); + } +} + +fn captures(metadata: &Metadata<'_>) -> bool { + metadata.target() == "libsy" && matches!(metadata.name(), CLIENT_CALL | UPSTREAM_ATTEMPT) +} + +fn relay_scope_type(name: &str) -> ScopeType { + if name == CLIENT_CALL { + ScopeType::Llm + } else { + ScopeType::Custom + } +} + +fn is_safe_field(name: &str) -> bool { + matches!( + name, + "algorithm" + | "switchyard.candidate" + | "switchyard.candidate_count" + | "switchyard.call_phase" + | "selected_model" + | "model" + | "wire_format" + | "attempt" + | "max_attempts" + | "retry" + | "outcome" + | "status_code" + | "will_retry" + | "retry_delay_ms" + | "otel.status_code" + | "error.type" + ) +} + +fn span_status(metadata: &Json) -> Option { + let outcome = metadata.get("outcome").and_then(Json::as_str); + let status = metadata.get("otel.status_code").and_then(Json::as_str); + match (status, outcome) { + (Some("ERROR"), _) | (_, Some("error" | "cancelled")) => { + Some(serde_json::json!({"otel.status_code": "ERROR"})) + } + (Some("OK"), _) | (_, Some("ok")) => Some(serde_json::json!({"otel.status_code": "OK"})), + _ => None, + } +} + +#[cfg(test)] +mod tests { + use std::fmt; + use std::sync::mpsc; + + use tracing::Instrument; + + use super::*; + + struct PanicOnDebug; + + impl fmt::Debug for PanicOnDebug { + fn fmt(&self, _formatter: &mut fmt::Formatter<'_>) -> fmt::Result { + panic!("disallowed fields must not be formatted"); + } + } + + fn recorder() -> (SpanCapture, mpsc::Receiver) { + let (sender, receiver) = mpsc::channel(); + let capture = SpanCapture::new(move |span| sender.send(span).unwrap()); + (capture, receiver) + } + + #[tokio::test] + async fn captures_only_allowlisted_client_and_attempt_fields() { + let (capture, captured) = recorder(); + capture + .capture(async { + drop(tracing::info_span!(target: "other", CLIENT_CALL)); + let call = tracing::info_span!( + target: "libsy", + CLIENT_CALL, + selected_model = "target/model", + switchyard.call_phase = "routing", + outcome = tracing::field::Empty, + error = ?PanicOnDebug, + ); + { + let _entered = call.enter(); + drop(tracing::debug_span!( + target: "libsy", + UPSTREAM_ATTEMPT, + model = "target/model", + attempt = 1_u64, + outcome = "error", + error = "provider body", + )); + call.record("outcome", "ok"); + } + drop(call); + }) + .await; + + let mut call = captured.try_recv().expect("expected one client call"); + assert!(captured.try_recv().is_err()); + assert_eq!(call.name, CLIENT_CALL); + assert!(matches!(relay_scope_type(call.name), ScopeType::Llm)); + assert_eq!(call.fields["selected_model"], "target/model"); + assert_eq!(call.fields["switchyard.call_phase"], "routing"); + assert_eq!(call.fields["outcome"], "ok"); + assert!(!call.fields.contains_key("error")); + let [attempt] = call.children.as_slice() else { + panic!("expected one upstream attempt: {call:?}"); + }; + assert_eq!(attempt.name, UPSTREAM_ATTEMPT); + assert!(matches!(relay_scope_type(attempt.name), ScopeType::Custom)); + assert_eq!(attempt.fields["model"], "target/model"); + assert_eq!(attempt.fields["attempt"], 1); + assert_eq!(attempt.fields["outcome"], "error"); + assert!(!attempt.fields.contains_key("error")); + assert!(call.started_at <= attempt.started_at); + assert!(attempt.ended_at <= call.ended_at); + + call.sanitize(&ProviderKeyRedactor::new(&["target/model".into()])); + assert_eq!(call.fields["selected_model"], "[REDACTED]"); + assert_eq!(call.children[0].fields["model"], "[REDACTED]"); + } + + #[tokio::test] + async fn captures_spans_that_close_after_the_scoped_future_returns() { + let (capture, captured) = recorder(); + let call = capture + .capture(async { + let call = tracing::info_span!(target: "libsy", CLIENT_CALL); + drop(tracing::debug_span!(target: "libsy", parent: &call, UPSTREAM_ATTEMPT)); + call + }) + .await; + + assert!(captured.try_recv().is_err()); + drop(capture); + drop(call); + + let call = captured + .try_recv() + .expect("expected one deferred client call"); + assert_eq!(call.children.len(), 1); + } + + #[tokio::test] + async fn one_capture_handles_concurrent_futures() { + let (capture, captured) = recorder(); + + tokio::join!( + capture.capture(async { + tokio::task::yield_now().await; + drop(tracing::info_span!(target: "libsy", CLIENT_CALL, selected_model = "first")); + }), + capture.capture(async { + tokio::task::yield_now().await; + drop(tracing::info_span!(target: "libsy", CLIENT_CALL, selected_model = "second")); + }), + ); + + let mut models = captured + .try_iter() + .map(|span| span.fields["selected_model"].as_str().unwrap().to_owned()) + .collect::>(); + models.sort_unstable(); + assert_eq!(models, ["first", "second"]); + } + + #[tokio::test] + async fn dropped_in_flight_spans_are_cancelled() { + let (capture, captured) = recorder(); + + let mut future = Box::pin(capture.capture(async move { + let call = tracing::info_span!(target: "libsy", CLIENT_CALL); + let attempt = tracing::debug_span!(target: "libsy", parent: &call, UPSTREAM_ATTEMPT); + std::future::pending::<()>() + .instrument(attempt) + .instrument(call) + .await; + })); + + assert!(futures_util::poll!(&mut future).is_pending()); + drop(future); + + let call = captured + .try_recv() + .expect("expected one cancelled client call"); + assert_eq!(call.fields["outcome"], "cancelled"); + assert_eq!( + span_status(&Json::Object(call.fields.clone())), + Some(serde_json::json!({"otel.status_code": "ERROR"})) + ); + let [attempt] = call.children.as_slice() else { + panic!("expected one cancelled upstream attempt: {call:?}"); + }; + assert_eq!(attempt.fields["outcome"], "cancelled"); + } +} diff --git a/docs/integrations/nemo_relay.md b/docs/integrations/nemo_relay.md index baa6bbec18..83bba4022a 100644 --- a/docs/integrations/nemo_relay.md +++ b/docs/integrations/nemo_relay.md @@ -305,17 +305,29 @@ can receive these records, but each output format presents them differently. | Output | What the current integration shows | | --- | --- | | ATIF | The request and response seen by the agent. Internal routing is not added as separate steps or included in `final_metrics`. Relay-managed ATIF files may retain the raw records under `extra.observed_events`. | -| OpenTelemetry traces, including OpenInference | Relay's LLM span plus eligible routing marks, depending on the projection. Marks are point-in-time records, not duration spans. | +| OpenTelemetry traces, including OpenInference | Relay's outer LLM span, nested `libsy.client_call` and `libsy.upstream_attempt` duration scopes, plus eligible routing marks. | | OpenTelemetry metrics | Switchyard request, routing-call, latency, token, and failure measurements. Combine the reported routing usage with Relay's answer cost to estimate the observed routed cost. | | OTLP logs | Non-metric Switchyard marks that meet the configured severity threshold. | -Switchyard creates `libsy.run`, `libsy.llm_call`, `libsy.client_call`, and -`libsy.upstream_attempt` spans internally. Together they cover the algorithm -run, model calls requested by the algorithm, candidate attempts including -fallbacks, and individual HTTP attempts. They are not one connected hierarchy -today, and the plugin does not send them to Relay. As a result, Relay traces do -not show those internal operations as nested duration spans or give an internal -stream its own cancellation lifecycle. +The plugin captures two existing Switchyard spans. Each routing or answer LLM +call becomes a `libsy.client_call` scope. Physical HTTP attempts and retries +become `libsy.upstream_attempt` child scopes. Buffered calls close with the response; +streaming calls close when the stream finishes, fails, or is dropped. The +projection contains allowlisted, payload-free routing metadata only and does not +copy prompts, responses, provider error bodies, or credentials into Relay. + +`switchyard.call_phase` records when Switchyard issued a client call: +`routing` while the algorithm was still deciding, or `completion` after a +target was selected. A routing-phase response can later become the final +answer, so this is intentionally different from the `call_role` on +`switchyard.routing.llm_call` marks, which records the post-outcome +classification as `routing` or `answer`. + +| Switchyard signal | Relay record | Parent | Lifetime | +| --- | --- | --- | --- | +| Routing or answer model call | `libsy.client_call` scope | Relay's managed request scope | Starts with the Switchyard call and closes with the buffered response or terminal stream event. | +| Physical provider request or retry | `libsy.upstream_attempt` scope | Its `libsy.client_call` scope | Starts before the HTTP attempt and closes on response, error, or cancellation. | +| Routing request, decision, overhead, or failure | Existing `switchyard.routing.*` mark | Active Relay scope at emission | Point-in-time mark using the existing outcome fields. | ### How Routing Appears in Traces @@ -334,12 +346,16 @@ omits marks. With `mark_projection = "tool"`, the trace has this shape: flowchart LR agent["Relay agent scope"] llm["LLM call"] + call["libsy.client_call
routing or completion"] + attempt["libsy.upstream_attempt
one per HTTP attempt"] requested["mark:switchyard.routing.requested
zero duration"] calls["mark:switchyard.routing.llm_call
zero duration"] overhead["mark:switchyard.routing.overhead
zero duration"] decision["mark:switchyard.routing.decision
zero duration"] error["mark:switchyard.routing.error
zero duration"] agent --> llm + llm --> call + call --> attempt agent --> requested agent --> calls agent --> overhead