From c2b2c8696513985d93223668a410e85ee0c6f10b Mon Sep 17 00:00:00 2001 From: Eric Liu Date: Wed, 15 Jul 2026 15:42:58 +0000 Subject: [PATCH 01/10] feat(libsy): add OTel metrics, tracing spans, and structured logging Signed-off-by: Eric Liu --- .../switchyard-codebase-exploration/SKILL.md | 4 +- Cargo.lock | 160 ++++- crates/libsy/Cargo.toml | 7 + crates/libsy/README.md | 40 +- crates/libsy/examples/ensemble.rs | 4 + crates/libsy/src/algorithms/fall_through.rs | 4 + crates/libsy/src/algorithms/llm_class.rs | 4 + crates/libsy/src/algorithms/noop.rs | 4 + crates/libsy/src/algorithms/rand.rs | 4 + .../libsy/src/algorithms/subagent_override.rs | 4 + crates/libsy/src/core/algorithm.rs | 105 ++- crates/libsy/src/lib.rs | 16 + crates/libsy/src/observability.rs | 225 +++++++ crates/libsy/tests/observability.rs | 600 ++++++++++++++++++ 14 files changed, 1154 insertions(+), 27 deletions(-) create mode 100644 crates/libsy/src/observability.rs create mode 100644 crates/libsy/tests/observability.rs diff --git a/.agents/skills/switchyard-codebase-exploration/SKILL.md b/.agents/skills/switchyard-codebase-exploration/SKILL.md index 8057b926..c91609c9 100644 --- a/.agents/skills/switchyard-codebase-exploration/SKILL.md +++ b/.agents/skills/switchyard-codebase-exploration/SKILL.md @@ -9,7 +9,9 @@ description: Use when modifying, debugging, reviewing, refactoring, renaming, re > `crates/switchyard-core`, `crates/switchyard-components`, > `crates/switchyard-server`, `crates/switchyard-translation`, > `crates/switchyard-py`, and `crates/*/tests` -> directly when a change touches Rust. +> directly when a change touches Rust. The libsy algorithm layer lives in +> `crates/libsy` (core + `observability.rs` telemetry), `crates/libsy-protocol` +> (conversation IR), and `crates/libsy-examples` (reference algorithms). > Direct PyO3 bindings for concrete components live under > `crates/switchyard-py/src/component_bindings/` with Python lazy exports in > `switchyard_rust/components.py`. diff --git a/Cargo.lock b/Cargo.lock index 8129d2df..e62283d8 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -580,6 +580,18 @@ dependencies = [ "wasm-bindgen", ] +[[package]] +name = "getrandom" +version = "0.3.4" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "899def5c37c4fd7b2664648c28120ecec138e4d395b459e5ca34f9cce2dd77fd" +dependencies = [ + "cfg-if", + "libc", + "r-efi 5.3.0", + "wasip2", +] + [[package]] name = "getrandom" version = "0.4.3" @@ -589,7 +601,7 @@ dependencies = [ "cfg-if", "js-sys", "libc", - "r-efi", + "r-efi 6.0.0", "rand_core 0.10.1", "wasm-bindgen", ] @@ -983,6 +995,15 @@ dependencies = [ "windows-sys 0.61.2", ] +[[package]] +name = "nu-ansi-term" +version = "0.50.3" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "7957b9740744892f114936ab4a57b3f487491bbeafaf8083688b16841a4240e5" +dependencies = [ + "windows-sys 0.61.2", +] + [[package]] name = "num_cpus" version = "1.17.0" @@ -1011,6 +1032,37 @@ version = "0.2.1" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "7c87def4c32ab89d880effc9e097653c8da5d6ef28e6b539d313baaacfbafcbe" +[[package]] +name = "opentelemetry" +version = "0.32.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "b0142c63252a9e054e68a4c61a5778f7b14f576274d593f8ce883d191a099682" +dependencies = [ + "futures-core", + "futures-sink", + "js-sys", + "pin-project-lite", + "thiserror", + "tracing", +] + +[[package]] +name = "opentelemetry_sdk" +version = "0.32.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "9b59f80e1ac4d5ff7a2db8fb6c80badb7f0f3f858211fba08dd9aaec750894f9" +dependencies = [ + "futures-channel", + "futures-executor", + "futures-util", + "opentelemetry", + "percent-encoding", + "portable-atomic", + "rand 0.9.5", + "thiserror", + "tokio", +] + [[package]] name = "parking_lot" version = "0.12.5" @@ -1242,6 +1294,12 @@ dependencies = [ "proc-macro2", ] +[[package]] +name = "r-efi" +version = "5.3.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "69cdb34c158ceb288df11e18b4bd39de994f6657d83847bdffdbd7f346754b0f" + [[package]] name = "r-efi" version = "6.0.0" @@ -1255,10 +1313,20 @@ source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "22f6172bdec972074665ed81ed53b71da00bfc44b65a753cfde883ec4c702a1a" dependencies = [ "libc", - "rand_chacha", + "rand_chacha 0.3.1", "rand_core 0.6.4", ] +[[package]] +name = "rand" +version = "0.9.5" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "b9ef1d0d795eb7d84685bca4f72f3649f064e6641543d3a8c415898726a57b41" +dependencies = [ + "rand_chacha 0.9.0", + "rand_core 0.9.5", +] + [[package]] name = "rand" version = "0.10.2" @@ -1280,6 +1348,16 @@ dependencies = [ "rand_core 0.6.4", ] +[[package]] +name = "rand_chacha" +version = "0.9.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "d3022b5f1df60f26e1ffddd6c66e8aa15de382ae63b3a0c1bfc0e4d3e3f325cb" +dependencies = [ + "ppv-lite86", + "rand_core 0.9.5", +] + [[package]] name = "rand_core" version = "0.6.4" @@ -1289,6 +1367,15 @@ dependencies = [ "getrandom 0.2.17", ] +[[package]] +name = "rand_core" +version = "0.9.5" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "76afc826de14238e6e8c374ddcc1fa19e374fd8dd986b0d2af0d02377261d83c" +dependencies = [ + "getrandom 0.3.4", +] + [[package]] name = "rand_core" version = "0.10.1" @@ -1591,6 +1678,15 @@ dependencies = [ "serde", ] +[[package]] +name = "sharded-slab" +version = "0.1.7" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "f40ca3c46823713e0d4209592e8d6e826aa57e928f09752619fc696c499637f6" +dependencies = [ + "lazy_static", +] + [[package]] name = "shlex" version = "2.0.1" @@ -1686,12 +1782,16 @@ version = "0.1.0" dependencies = [ "async-trait", "futures", + "opentelemetry", + "opentelemetry_sdk", "rand 0.8.7", "serde_json", "switchyard-llm-client", "switchyard-protocol", "tokio", "tokio-stream", + "tracing", + "tracing-subscriber", ] [[package]] @@ -1858,6 +1958,15 @@ dependencies = [ "syn", ] +[[package]] +name = "thread_local" +version = "1.1.10" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "1ad99c4c6d32803332c548b1af0540b357b3f5fc0be8f6c6bfe8b2e6ae784070" +dependencies = [ + "cfg-if", +] + [[package]] name = "tinystr" version = "0.8.3" @@ -2048,6 +2157,32 @@ source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "db97caf9d906fbde555dd62fa95ddba9eecfd14cb388e4f491a66d74cd5fb79a" dependencies = [ "once_cell", + "valuable", +] + +[[package]] +name = "tracing-log" +version = "0.2.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "ee855f1f400bd0e5c02d150ae5de3840039a3f54b025156404e34c23c03f47c3" +dependencies = [ + "log", + "once_cell", + "tracing-core", +] + +[[package]] +name = "tracing-subscriber" +version = "0.3.23" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "cb7f578e5945fb242538965c2d0b04418d38ec25c79d160cd279bf0731c8d319" +dependencies = [ + "nu-ansi-term", + "sharded-slab", + "smallvec", + "thread_local", + "tracing-core", + "tracing-log", ] [[package]] @@ -2092,6 +2227,12 @@ version = "0.2.2" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "06abde3611657adf66d383f00b093d7faecc7fa57071cce2578660c9f1010821" +[[package]] +name = "valuable" +version = "0.1.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "ba73ea9cf16a25df0c8caa16c51acb937d5712a8429db78a3ee29d5dcacd3a65" + [[package]] name = "want" version = "0.3.1" @@ -2107,6 +2248,15 @@ version = "0.11.1+wasi-snapshot-preview1" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "ccf3ec651a847eb01de73ccad15eb7d99f80485de043efb2f370cd654f4ea44b" +[[package]] +name = "wasip2" +version = "1.0.4+wasi-0.2.12" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "b67efb37e106e55ce722a510d6b5f9c17f083e5fc79afc2badeb12cc313d9487" +dependencies = [ + "wit-bindgen", +] + [[package]] name = "wasm-bindgen" version = "0.2.126" @@ -2312,6 +2462,12 @@ dependencies = [ "url", ] +[[package]] +name = "wit-bindgen" +version = "0.57.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "1ebf944e87a7c253233ad6766e082e3cd714b5d03812acc24c318f549614536e" + [[package]] name = "writeable" version = "0.6.3" diff --git a/crates/libsy/Cargo.toml b/crates/libsy/Cargo.toml index fca1dcf8..a366b121 100644 --- a/crates/libsy/Cargo.toml +++ b/crates/libsy/Cargo.toml @@ -15,11 +15,18 @@ rust-version.workspace = true async-trait = "0.1" serde_json = "1" futures = "0.3" +# Metrics-only OTel API: instruments record through the host-installed global +# meter provider. Pinned to the 0.32 line used across the workspace. +opentelemetry = { version = "0.32", default-features = false, features = ["metrics"] } rand = "0.8" switchyard-protocol = { path = "../protocol" } tokio = { version = "1", features = ["full"] } tokio-stream = "0.1" +tracing = { version = "0.1", default-features = false, features = ["std"] } [dev-dependencies] +# SDK + in-memory exporter to assert what the observability layer records. +opentelemetry_sdk = { version = "0.32", features = ["metrics", "testing"] } switchyard-llm-client = { path = "../libsy-llm-client" } tokio = { version = "1", features = ["macros", "rt"] } +tracing-subscriber = "0.3" diff --git a/crates/libsy/README.md b/crates/libsy/README.md index c687b3af..9b2e6344 100644 --- a/crates/libsy/README.md +++ b/crates/libsy/README.md @@ -64,14 +64,14 @@ pub enum LlmResponse { `switchyard-protocol` owns the conversation IR: `LlmRequest`, the buffered `AggLlmResponse`, the streaming `LlmResponseChunk`, and the building blocks `Message`, -`ContentBlock`, `ResponseOutput`, and `Role`. `libsy` re-exports the envelope +`ContentBlock`, `ResponseOutput`, `Usage`, and `Role`. `libsy` re-exports the envelope (`Request`/`Response`/`Metadata`) plus `LlmRequest`, `LlmResponse`, `AggLlmResponse`, -`LlmResponseChunk`, and `LlmResponseStream`; import the rest (`Message`, `Role`, …) and the -`text_request` / `text_response` / `completion_text` helpers from `switchyard_protocol`. -Construct and inspect the conversation model directly so tools, sampling parameters, -reasoning, and provider extensions stay visible instead of being hidden behind a second -convenience API. `raw_request` remains available when a host needs exact source-body -fidelity. +`LlmResponseChunk`, `LlmResponseStream`, and `Usage`; import the rest (`Message`, `Role`, …) +and the `text_request` / `text_response` / `completion_text` helpers from +`switchyard_protocol`. Construct and inspect the conversation model directly so tools, +sampling parameters, reasoning, and provider extensions stay visible instead of being hidden +behind a second convenience API. `raw_request` remains available when a host needs exact +source-body fidelity. ## Targets and clients @@ -199,6 +199,8 @@ model and *why*. ```rust #[async_trait] pub trait Algorithm: Send + Sync + 'static { + // Stable telemetry label: the `algorithm` attribute on libsy's spans/metrics/logs. + fn name(&self) -> &str; // `self: Arc` (not `&mut`): one algorithm serves requests concurrently — use // interior mutability for state. Offload calls/decisions on `driver`. async fn create_run_task(self: Arc, ctx: Context, driver: Driver, request: Request) @@ -249,6 +251,29 @@ impl Algorithm for LlmClassifier { } ``` +## Observability + +libsy instruments every algorithm at the core — the `Decision` hook plus the offload +boundary — so algorithms carry no telemetry code beyond `Algorithm::name()`, the +`algorithm` attribute everything below is keyed by. + +- **Spans** (`tracing`): one `libsy.run` per request, carrying the `Metadata` + correlation ids and the outcome, with one child `libsy.llm_call` per model call + (selected model, outcome, token counts, latency). +- **Structured logs** (`tracing`, target `libsy`): an info event per published + `Decision` (selected model + reasoning), warn events for failed calls and runs. +- **Metrics** (OpenTelemetry, scope `libsy`, via the global meter provider): counters + `libsy.runs`, `libsy.llm_calls`, `libsy.decisions`, `libsy.input_tokens`, + `libsy.output_tokens`, `libsy.total_tokens`, `libsy.reasoning_tokens`; histograms + `libsy.run_duration_ms`, `libsy.llm_call_duration_ms`. Attributes are `algorithm`, + `selected_model`, and `outcome` (`ok`/`error`) — failure rates are the + `outcome="error"` share of runs/calls. + +libsy owns no exporter. The host installs an OpenTelemetry SDK meter provider +(`opentelemetry::global::set_meter_provider`) and a `tracing` subscriber (bridge spans +into OTel with `tracing-opentelemetry` if desired); with neither installed, all +instrumentation is a no-op. + ## Explore The core crate includes uniform random routing and naive LLM classifier. Runnable @@ -277,5 +302,4 @@ agents live in [`examples`](examples/) folder. - **`Signals` events** — `process_signals` / `Signals` exist but carry nothing yet. - **`Context` fields** — the per-request state carrier is an empty placeholder today. -- **Observability** — spans + a metrics sink (`Decision` is the hook). - **Config-driven construction**, **typed errors** (vs `Box`), **weighted random**. diff --git a/crates/libsy/examples/ensemble.rs b/crates/libsy/examples/ensemble.rs index 3c985376..c82f8b8d 100644 --- a/crates/libsy/examples/ensemble.rs +++ b/crates/libsy/examples/ensemble.rs @@ -381,6 +381,10 @@ fn parse_choice(completion: &str, count: usize) -> usize { #[async_trait] impl Algorithm for EnsembleOrchAlgo { + fn name(&self) -> &str { + "ensemble" + } + async fn create_run_task( self: Arc, ctx: Context, diff --git a/crates/libsy/src/algorithms/fall_through.rs b/crates/libsy/src/algorithms/fall_through.rs index b42de891..49ae385a 100644 --- a/crates/libsy/src/algorithms/fall_through.rs +++ b/crates/libsy/src/algorithms/fall_through.rs @@ -79,6 +79,10 @@ impl FallThrough { } #[async_trait] impl Algorithm for FallThrough { + fn name(&self) -> &str { + "fall_through" + } + async fn create_run_task( self: Arc, ctx: Context, diff --git a/crates/libsy/src/algorithms/llm_class.rs b/crates/libsy/src/algorithms/llm_class.rs index aa10a9fe..14bf43db 100644 --- a/crates/libsy/src/algorithms/llm_class.rs +++ b/crates/libsy/src/algorithms/llm_class.rs @@ -139,6 +139,10 @@ fn trim_messages(messages: &[Message], recent_turn_window: usize) -> Vec &str { + "llm_classifier" + } + async fn create_run_task( self: Arc, ctx: Context, diff --git a/crates/libsy/src/algorithms/noop.rs b/crates/libsy/src/algorithms/noop.rs index a3ec368c..aeddb1e3 100644 --- a/crates/libsy/src/algorithms/noop.rs +++ b/crates/libsy/src/algorithms/noop.rs @@ -35,6 +35,10 @@ impl Decision for NoopDecision { #[async_trait::async_trait] impl Algorithm for Noop { + fn name(&self) -> &str { + "noop" + } + async fn create_run_task( self: Arc, ctx: Context, diff --git a/crates/libsy/src/algorithms/rand.rs b/crates/libsy/src/algorithms/rand.rs index 817fa956..a23690b8 100644 --- a/crates/libsy/src/algorithms/rand.rs +++ b/crates/libsy/src/algorithms/rand.rs @@ -55,6 +55,10 @@ impl Random { #[async_trait] impl Algorithm for Random { + fn name(&self) -> &str { + "random" + } + async fn create_run_task( self: Arc, ctx: Context, diff --git a/crates/libsy/src/algorithms/subagent_override.rs b/crates/libsy/src/algorithms/subagent_override.rs index 599213b1..feca43fb 100644 --- a/crates/libsy/src/algorithms/subagent_override.rs +++ b/crates/libsy/src/algorithms/subagent_override.rs @@ -59,6 +59,10 @@ impl SubagentOverride { #[async_trait] impl Algorithm for SubagentOverride { + fn name(&self) -> &str { + "subagent_override" + } + async fn create_run_task( self: Arc, ctx: Context, diff --git a/crates/libsy/src/core/algorithm.rs b/crates/libsy/src/core/algorithm.rs index d3f42765..1938dd2d 100644 --- a/crates/libsy/src/core/algorithm.rs +++ b/crates/libsy/src/core/algorithm.rs @@ -5,10 +5,11 @@ //! routing/optimization algorithm implements, and the offload channel it makes model //! calls and publishes [`Decision`]s over. See the crate root for the narrative model. -use std::{error::Error, pin::Pin, sync::Arc}; +use std::{error::Error, pin::Pin, sync::Arc, time::Instant}; use async_trait::async_trait; use futures::{Stream, StreamExt}; +use tracing::Instrument; /// The request/response protocol types, re-exported from [`switchyard_protocol`]. /// [`LlmRequest`] is the normalized request; [`AggLlmResponse`] is the buffered response; @@ -16,10 +17,11 @@ use futures::{Stream, StreamExt}; /// (a live [`LlmResponseStream`] or the terminal aggregate). pub use switchyard_protocol::{ AggLlmResponse, Context, Decision, LlmRequest, LlmResponse, LlmResponseChunk, - LlmResponseStream, Metadata, Request, Response, RoutedLlmClient, Signals, + LlmResponseStream, Metadata, Request, Response, RoutedLlmClient, Signals, Usage, }; use super::driver::{DriverRequest, DriverStep, TypeErasedDriver}; +use crate::observability; /// Shorthand for the crate's boxed, thread-safe error type. type BoxErr = Box; @@ -112,24 +114,47 @@ impl CallLlmRequest { #[derive(Clone)] pub struct Driver { driver: TypeErasedDriver, + /// Name of the algorithm this driver serves — the `algorithm` attribute on + /// the telemetry emitted for its calls and decisions. + algorithm: Arc, } impl Driver { - /// Build an empty driver with its step channel ready. Created per call by + /// Build an empty driver with its step channel ready, labeled with the name + /// of the algorithm it serves. Created per call by /// [`run_stream`](Algorithm::run_stream). - pub(crate) fn new() -> Self { + pub(crate) fn new(algorithm: &str) -> Self { Self { driver: TypeErasedDriver::new(), + algorithm: Arc::from(algorithm), } } /// Offload a model call: publish `routed` as a [`Step::CallLlm`] and await the /// consumer's [`Response`]. The call's context travels inside /// [`routed.ctx`](RoutedRequest::ctx). Errors if the stream is closed or the call failed. + /// The await is wrapped in a `libsy.llm_call` span, and the call's latency, + /// outcome, and token usage are recorded when it resolves. pub async fn call_llm(&self, routed: RoutedRequest) -> Result { - self.driver - .fulfill_request::(routed.ctx.clone(), routed) - .await + let selected_model = routed.decision.selected_model().to_string(); + let span = observability::llm_call_span(&self.algorithm, &selected_model); + async { + let started = Instant::now(); + let result = self + .driver + .fulfill_request::(routed.ctx.clone(), routed) + .await; + observability::record_llm_call( + &self.algorithm, + &selected_model, + started.elapsed(), + &result, + &tracing::Span::current(), + ); + result + } + .instrument(span) + .await } /// Offload a call to `target`: pair `request` with `decision` and the target's @@ -154,7 +179,9 @@ impl Driver { } /// Publish a routing [`Decision`] as a [`Step::Decision`] on the stream. + /// Each published decision is counted and logged with its reasoning. pub async fn info(&self, ctx: Context, decision: Arc) -> Result<(), BoxErr> { + observability::record_decision(&self.algorithm, decision.as_ref()); self.driver.info(ctx, decision).await } @@ -190,12 +217,6 @@ impl Driver { } } -impl Default for Driver { - fn default() -> Self { - Self::new() - } -} - /// One item in the stream returned by `Driver::stream` / [`Algorithm::run_stream`]. pub enum Step { /// The algorithm needs this model call performed. The host serves it (optionally @@ -280,6 +301,12 @@ pub trait Algorithm: Send + Sync + 'static where S: Clone + Send + Sync + 'static, { + /// Stable, low-cardinality name identifying this algorithm — the + /// `algorithm` attribute on every span, metric, and log line the crate + /// emits for its runs (see the crate docs' Observability section). + fn name(&self) -> &str; + + /// Run one request to completion: make model calls with [`Driver::call_llm_target`], /// publish [`Decision`]s with [`Driver::info`], and return the final [`Response`]. /// The method an algorithm implements; [`run`](Self::run) / [`run_stream`](Self::run_stream) @@ -306,12 +333,30 @@ where /// Each [`Step::CallLlm`] is an offloaded model call the consumer must serve. /// The stream ends with a [`Step::ReturnToAgent`] on success, or an `Err` item on failure. fn run_stream(self: Arc, ctx: Context, request: Request) -> StepStream { - let driver = Driver::new(); + let driver = Driver::new(self.name()); let task_driver = driver.clone(); let task_ctx = ctx.clone(); let stream = task_driver.stream(); - let handle = - tokio::spawn(async move { self.create_run_task(task_ctx, task_driver, request).await }); + // 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 handle = tokio::spawn( + async move { + let started = Instant::now(); + let outcome = self + .create_run_task(task_ctx, task_driver.clone(), request) + .await; + observability::record_run( + &task_driver.algorithm, + started.elapsed(), + &outcome, + &tracing::Span::current(), + ); + outcome + } + .instrument(span), + ); // Dropping the stream aborts the algorithm task, so it doesn't keep running after the let abort_guard = AbortOnDrop(handle.abort_handle()); @@ -448,6 +493,10 @@ mod tests { #[async_trait] impl Algorithm for TestAlgo { + fn name(&self) -> &str { + "test" + } + async fn create_run_task( self: Arc, ctx: Context, @@ -827,6 +876,10 @@ mod tests { #[async_trait] impl Algorithm for StuckAlgo { + fn name(&self) -> &str { + "stuck" + } + async fn create_run_task( self: Arc, _ctx: Context, @@ -869,6 +922,10 @@ mod tests { #[async_trait] impl Algorithm for Panicky { + fn name(&self) -> &str { + "panicky" + } + async fn create_run_task( self: Arc, _ctx: Context, @@ -907,6 +964,10 @@ mod tests { #[async_trait] impl Algorithm for Panicky { + fn name(&self) -> &str { + "panicky" + } + async fn create_run_task( self: Arc, _ctx: Context, @@ -950,6 +1011,10 @@ mod tests { #[async_trait] impl Algorithm for StuckAlgo { + fn name(&self) -> &str { + "stuck" + } + async fn create_run_task( self: Arc, _ctx: Context, @@ -1053,6 +1118,10 @@ mod tests { #[async_trait] impl Algorithm for Hedge { + fn name(&self) -> &str { + "hedge" + } + async fn create_run_task( self: Arc, ctx: Context, @@ -1176,6 +1245,10 @@ mod tests { #[async_trait] impl Algorithm for FanOutThenError { + fn name(&self) -> &str { + "fan_out_then_error" + } + async fn create_run_task( self: Arc, ctx: Context, diff --git a/crates/libsy/src/lib.rs b/crates/libsy/src/lib.rs index 60221bea..b9642a35 100644 --- a/crates/libsy/src/lib.rs +++ b/crates/libsy/src/lib.rs @@ -52,6 +52,20 @@ //! responsible for its own thread-safety — stateless (like the reference routers) or //! interior mutability over just its own state. //! +//! ## Observability +//! +//! The provided run methods instrument every algorithm from the outside — at the +//! [`Decision`] hook and the offload boundary — so algorithms carry no telemetry +//! code. Each run gets a `libsy.run` tracing span (correlation ids from +//! [`Metadata`] attached) with a child `libsy.llm_call` span per model call; +//! each [`Driver::info`] decision is logged with its reasoning; and OpenTelemetry +//! metrics record run/call counts, latency, token usage, and published +//! decisions, keyed by [`Algorithm::name`] plus `selected_model` and `outcome`. +//! Metrics use the global meter provider and spans/logs the `tracing` facade: a +//! host that installs an OTel SDK and a `tracing` subscriber (bridged with +//! `tracing-opentelemetry` for OTLP spans) gets the full signal set; with +//! neither installed, everything is a no-op. +//! //! ## Algorithms //! //! Concrete algorithms live in [`algorithms`]: @@ -65,3 +79,5 @@ mod core; pub use core::*; pub mod algorithms; + +mod observability; diff --git a/crates/libsy/src/observability.rs b/crates/libsy/src/observability.rs new file mode 100644 index 00000000..3d7f7a73 --- /dev/null +++ b/crates/libsy/src/observability.rs @@ -0,0 +1,225 @@ +// SPDX-FileCopyrightText: Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved. +// SPDX-License-Identifier: Apache-2.0 + +//! OpenTelemetry metrics plus `tracing` spans and structured logs for the +//! algorithm layer. +//! +//! The crate's provided run methods call these helpers around the [`Decision`] +//! hook and the offload boundary, so every algorithm is instrumented from the +//! outside and carries no telemetry code of its own. Metrics record through the +//! OpenTelemetry **global** meter provider under the `libsy` scope — the host +//! installs an SDK provider and exporters; with none installed, recording is a +//! no-op. Spans and logs use the `tracing` facade (the async-native surface the +//! OpenTelemetry ecosystem bridges with `tracing-opentelemetry` / +//! `opentelemetry-appender-tracing`), so the host's subscriber decides where +//! they go. +//! +//! Instrument names use the OTel dotted form with the unit baked into the name +//! (`libsy.run_duration_ms`), matching the switchyard metric surface; a +//! Prometheus exporter sanitizes them to `libsy_run_duration_ms`. Attribute +//! cardinality is bounded: `algorithm` and `selected_model` are small +//! configured sets and `outcome` is `ok`/`error`. Nothing per-request becomes a +//! metric attribute — correlation ids ride on the `libsy.run` span instead. +//! +//! Instruments are resolved from the global provider on every record (an +//! instrument-cache lookup inside the SDK) so recording follows a meter +//! provider installed at any point in the process lifetime; the cost is +//! negligible next to a model call. + +use std::time::Duration; + +use opentelemetry::metrics::Meter; +use opentelemetry::{global, KeyValue}; +use tracing::Span; + +use crate::{Decision, Metadata, Response}; + +/// Shorthand for the crate's boxed, thread-safe error type. +type BoxErr = Box; + +/// Instrumentation scope for every libsy meter, span, and log line. +const SCOPE: &str = "libsy"; + +/// The `libsy`-scoped meter from the globally installed provider. +fn meter() -> Meter { + global::meter(SCOPE) +} + +/// `outcome` attribute value for a result: `ok` or `error`. +fn outcome_value(result: &Result) -> &'static str { + if result.is_ok() { + "ok" + } else { + "error" + } +} + +/// Span covering one algorithm run (the whole `create_run_task` execution). +/// +/// Correlation ids from the request [`Metadata`] are recorded as span fields +/// when present; `outcome` and `error` are filled in by [`record_run`] when the +/// run ends. +pub(crate) fn run_span(algorithm: &str, metadata: Option<&Metadata>) -> Span { + let span = tracing::info_span!( + target: SCOPE, + "libsy.run", + algorithm, + session_id = tracing::field::Empty, + agent_id = tracing::field::Empty, + task_id = tracing::field::Empty, + correlation_id = tracing::field::Empty, + outcome = tracing::field::Empty, + error = tracing::field::Empty, + ); + if let Some(metadata) = metadata { + for (field, value) in [ + ("session_id", &metadata.session_id), + ("agent_id", &metadata.agent_id), + ("task_id", &metadata.task_id), + ("correlation_id", &metadata.correlation_id), + ] { + if let Some(value) = value { + span.record(field, value.as_str()); + } + } + } + span +} + +/// Span covering one offloaded model call, a child of the surrounding +/// `libsy.run` span. `outcome`, `error`, and the token-count fields are filled +/// in by [`record_llm_call`] when the call resolves. +pub(crate) fn llm_call_span(algorithm: &str, selected_model: &str) -> Span { + tracing::info_span!( + target: SCOPE, + "libsy.llm_call", + algorithm, + selected_model, + outcome = tracing::field::Empty, + error = tracing::field::Empty, + input_tokens = tracing::field::Empty, + output_tokens = tracing::field::Empty, + total_tokens = tracing::field::Empty, + reasoning_tokens = tracing::field::Empty, + ) +} + +/// 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. +pub(crate) fn record_run( + algorithm: &str, + duration: Duration, + result: &Result, + span: &Span, +) { + let outcome = outcome_value(result); + span.record("outcome", outcome); + if let Err(error) = result { + span.record("error", tracing::field::display(error)); + tracing::warn!(target: SCOPE, algorithm, error = %error, "algorithm run failed"); + } + + let attributes = [ + KeyValue::new("algorithm", algorithm.to_string()), + KeyValue::new("outcome", outcome), + ]; + let meter = meter(); + meter.u64_counter("libsy.runs").build().add(1, &attributes); + meter + .f64_histogram("libsy.run_duration_ms") + .build() + .record(duration.as_secs_f64() * 1000.0, &attributes); +} + +/// Records the resolution of one offloaded model call: the call counter and +/// latency histogram, token counters from the response usage (absent fields are +/// skipped, not recorded as zero), the `outcome`/`error`/token fields on +/// `span`, and a warn log when the call failed. +pub(crate) fn record_llm_call( + algorithm: &str, + selected_model: &str, + duration: Duration, + result: &Result, + span: &Span, +) { + let outcome = outcome_value(result); + span.record("outcome", outcome); + + let meter = meter(); + let call_attributes = [ + KeyValue::new("algorithm", algorithm.to_string()), + KeyValue::new("selected_model", selected_model.to_string()), + KeyValue::new("outcome", outcome), + ]; + meter + .u64_counter("libsy.llm_calls") + .build() + .add(1, &call_attributes); + meter + .f64_histogram("libsy.llm_call_duration_ms") + .build() + .record(duration.as_secs_f64() * 1000.0, &call_attributes); + + match result { + Ok(response) => { + // Token usage exists only once a response is buffered; a streamed + // response resolves before its usage is known, so none is recorded. + let Some(usage) = response.llm_response.as_agg().map(|agg| &agg.usage) else { + return; + }; + let token_attributes = [ + KeyValue::new("algorithm", algorithm.to_string()), + KeyValue::new("selected_model", selected_model.to_string()), + ]; + for (counter, field, value) in [ + ("libsy.input_tokens", "input_tokens", usage.input_tokens), + ("libsy.output_tokens", "output_tokens", usage.output_tokens), + ("libsy.total_tokens", "total_tokens", usage.total_tokens), + ( + "libsy.reasoning_tokens", + "reasoning_tokens", + usage.reasoning_tokens, + ), + ] { + if let Some(value) = value { + span.record(field, value); + meter + .u64_counter(counter) + .build() + .add(value, &token_attributes); + } + } + } + Err(error) => { + span.record("error", tracing::field::display(error)); + tracing::warn!( + target: SCOPE, + algorithm, + selected_model, + error = %error, + "model call failed" + ); + } + } +} + +/// Records one published routing decision: the decision counter plus a +/// structured info log carrying the decision's reasoning. +pub(crate) fn record_decision(algorithm: &str, decision: &dyn Decision) { + let selected_model = decision.selected_model(); + tracing::info!( + target: SCOPE, + algorithm, + selected_model, + reasoning = decision.reasoning().unwrap_or(""), + "routing decision" + ); + meter().u64_counter("libsy.decisions").build().add( + 1, + &[ + KeyValue::new("algorithm", algorithm.to_string()), + KeyValue::new("selected_model", selected_model.to_string()), + ], + ); +} diff --git a/crates/libsy/tests/observability.rs b/crates/libsy/tests/observability.rs new file mode 100644 index 00000000..15684021 --- /dev/null +++ b/crates/libsy/tests/observability.rs @@ -0,0 +1,600 @@ +// SPDX-FileCopyrightText: Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved. +// SPDX-License-Identifier: Apache-2.0 + +//! Integration tests for the crate's observability layer: OpenTelemetry metrics +//! recorded through the global meter provider, and `tracing` spans/logs captured +//! by a global subscriber. +//! +//! Both telemetry sinks are process-global, so they are installed exactly once +//! for this test binary and every assertion filters by a test-unique algorithm +//! and model name. Counters are cumulative across flushes; the helpers take the +//! latest (max) matching data point. + +use std::collections::BTreeMap; +use std::error::Error; +use std::fmt; +use std::sync::{Arc, Mutex, MutexGuard, OnceLock}; + +use async_trait::async_trait; +use futures::StreamExt; +use opentelemetry_sdk::metrics::data::{AggregatedMetrics, MetricData, ResourceMetrics}; +use opentelemetry_sdk::metrics::{InMemoryMetricExporter, PeriodicReader, SdkMeterProvider}; +use tracing::field::{Field, Visit}; +use tracing::span::{Attributes, Id, Record}; +use tracing::{Event, Subscriber}; +use tracing_subscriber::layer::{Context as LayerContext, SubscriberExt}; +use tracing_subscriber::registry::LookupSpan; +use tracing_subscriber::Layer; + +use switchyard_libsy::{ + AggLlmResponse, Algorithm, Context, Decision, Driver, LlmResponse, LlmTarget, LlmTargetSet, + Metadata, Request, Response, RoutedLlmClient, Step, Usage, +}; +use switchyard_protocol::text_request; + +type BoxErr = Box; + +/// Locks a mutex, recovering the inner value if a panicking test poisoned it. +fn lock(mutex: &Mutex) -> MutexGuard<'_, T> { + match mutex.lock() { + Ok(guard) => guard, + Err(poisoned) => poisoned.into_inner(), + } +} + +/// One captured span: its name, contextual parent span name, and fields +/// (creation-time fields merged with later `Span::record` updates). +#[derive(Clone, Debug, Default)] +struct SpanRecord { + name: String, + parent: Option, + fields: BTreeMap, +} + +/// One captured event (log line): its target and fields, `message` included. +#[derive(Clone, Debug, Default)] +struct EventRecord { + target: String, + fields: BTreeMap, +} + +/// Shared store the capture layer writes into and tests read from. +#[derive(Clone, Default)] +struct CaptureStore { + spans: Arc>>, + events: Arc>>, +} + +impl CaptureStore { + fn spans(&self) -> Vec { + lock(&self.spans).values().cloned().collect() + } + + fn events(&self) -> Vec { + lock(&self.events).clone() + } +} + +/// Renders every field type into a string map so assertions can use `contains`. +struct FieldVisitor<'a>(&'a mut BTreeMap); + +impl Visit for FieldVisitor<'_> { + fn record_debug(&mut self, field: &Field, value: &dyn fmt::Debug) { + self.0 + .insert(field.name().to_string(), format!("{value:?}")); + } + + fn record_str(&mut self, field: &Field, value: &str) { + self.0.insert(field.name().to_string(), value.to_string()); + } + + fn record_u64(&mut self, field: &Field, value: u64) { + self.0.insert(field.name().to_string(), value.to_string()); + } +} + +/// Subscriber layer capturing spans (with contextual parents and recorded +/// fields) and events into a [`CaptureStore`]. +struct CaptureLayer { + store: CaptureStore, +} + +impl Layer for CaptureLayer +where + S: Subscriber + for<'a> LookupSpan<'a>, +{ + fn on_new_span(&self, attrs: &Attributes<'_>, id: &Id, ctx: LayerContext<'_, S>) { + let mut fields = BTreeMap::new(); + attrs.record(&mut FieldVisitor(&mut fields)); + // Resolve the parent the same way tracing does: an explicit parent wins, + // otherwise the contextually current span (if any) at creation time. + let parent = if let Some(parent_id) = attrs.parent() { + ctx.span(parent_id).map(|span| span.name().to_string()) + } else if attrs.is_contextual() { + ctx.lookup_current().map(|span| span.name().to_string()) + } else { + None + }; + lock(&self.store.spans).insert( + id.into_u64(), + SpanRecord { + name: attrs.metadata().name().to_string(), + parent, + fields, + }, + ); + } + + fn on_record(&self, id: &Id, values: &Record<'_>, _ctx: LayerContext<'_, S>) { + if let Some(record) = lock(&self.store.spans).get_mut(&id.into_u64()) { + values.record(&mut FieldVisitor(&mut record.fields)); + } + } + + fn on_event(&self, event: &Event<'_>, _ctx: LayerContext<'_, S>) { + let mut fields = BTreeMap::new(); + event.record(&mut FieldVisitor(&mut fields)); + lock(&self.store.events).push(EventRecord { + target: event.metadata().target().to_string(), + fields, + }); + } +} + +/// 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(); + TELEMETRY.get_or_init(|| { + let exporter = InMemoryMetricExporter::default(); + let reader = PeriodicReader::builder(exporter.clone()).build(); + let provider = SdkMeterProvider::builder().with_reader(reader).build(); + opentelemetry::global::set_meter_provider(provider.clone()); + + let store = CaptureStore::default(); + let subscriber = tracing_subscriber::registry().with(CaptureLayer { + store: store.clone(), + }); + if tracing::subscriber::set_global_default(subscriber).is_err() { + panic!("a global tracing subscriber was already installed in this test binary"); + } + (store, exporter, provider) + }) +} + +/// Flushes the metric pipeline and returns every exported snapshot. +fn flushed_metrics( + exporter: &InMemoryMetricExporter, + provider: &SdkMeterProvider, +) -> Vec { + if let Err(error) = provider.force_flush() { + panic!("force_flush failed: {error}"); + } + match exporter.get_finished_metrics() { + Ok(metrics) => metrics, + Err(error) => panic!("get_finished_metrics failed: {error}"), + } +} + +/// True when the data point carries every wanted `key=value` attribute. +fn attributes_match<'a>( + mut attributes: impl Iterator, + wanted: &[(&str, &str)], +) -> bool { + let present: Vec<(String, String)> = attributes + .by_ref() + .map(|kv| (kv.key.as_str().to_string(), kv.value.as_str().to_string())) + .collect(); + wanted + .iter() + .all(|(key, value)| present.iter().any(|(k, v)| k == key && v == value)) +} + +/// Latest cumulative value of a `u64` counter for the given attribute set. +fn u64_counter_value( + snapshots: &[ResourceMetrics], + name: &str, + wanted: &[(&str, &str)], +) -> Option { + let mut latest = None; + for snapshot in snapshots { + for scope in snapshot.scope_metrics() { + if scope.scope().name() != "libsy" { + continue; + } + for metric in scope.metrics() { + if metric.name() != name { + continue; + } + if let AggregatedMetrics::U64(MetricData::Sum(sum)) = metric.data() { + for point in sum.data_points() { + if attributes_match(point.attributes(), wanted) { + // Cumulative counters only grow; max = most recent. + latest = + Some(latest.map_or(point.value(), |v: u64| v.max(point.value()))); + } + } + } + } + } + } + latest +} + +/// Latest cumulative sample count of an `f64` histogram for the attribute set. +fn f64_histogram_count( + snapshots: &[ResourceMetrics], + name: &str, + wanted: &[(&str, &str)], +) -> Option { + let mut latest = None; + for snapshot in snapshots { + for scope in snapshot.scope_metrics() { + if scope.scope().name() != "libsy" { + continue; + } + for metric in scope.metrics() { + if metric.name() != name { + continue; + } + if let AggregatedMetrics::F64(MetricData::Histogram(histogram)) = metric.data() { + for point in histogram.data_points() { + if attributes_match(point.attributes(), wanted) { + latest = + Some(latest.map_or(point.count(), |v: u64| v.max(point.count()))); + } + } + } + } + } + } + latest +} + +/// Decision with a fixed model and reasoning string. +struct StaticDecision { + model: String, + reasoning: String, +} + +impl Decision for StaticDecision { + fn selected_model(&self) -> &str { + &self.model + } + fn reasoning(&self) -> Option<&str> { + Some(&self.reasoning) + } + fn as_any(&self) -> &dyn std::any::Any { + self + } +} + +/// Client that answers every call with a fixed token [`Usage`]. +struct UsageClient { + usage: Usage, +} + +#[async_trait] +impl RoutedLlmClient for UsageClient { + async fn call( + &self, + _ctx: Context, + _request: Request, + decision: Arc, + ) -> Result { + Ok(Response { + llm_response: LlmResponse::Agg(AggLlmResponse { + model: Some(decision.selected_model().to_string()), + usage: self.usage.clone(), + ..AggLlmResponse::default() + }), + metadata: None, + }) + } +} + +/// Publishes one decision for the first target, then calls it — the smallest +/// algorithm exercising both instrumented driver paths. +struct SingleCallAlgo { + name: String, + target_set: LlmTargetSet, +} + +#[async_trait] +impl Algorithm for SingleCallAlgo { + fn name(&self) -> &str { + &self.name + } + + async fn create_run_task( + self: Arc, + ctx: Context, + driver: Driver, + request: Request, + ) -> Result { + let target = self + .target_set + .targets() + .first() + .ok_or("no targets")? + .clone(); + let decision: Arc = Arc::new(StaticDecision { + reasoning: format!("picked '{}'", target.semantic_name), + model: target.semantic_name.clone(), + }); + driver.info(ctx.clone(), decision.clone()).await?; + driver + .call_llm_target(ctx, &target, request, decision) + .await + } +} + +fn request_with_metadata(session_id: &str, correlation_id: &str) -> Request { + Request { + llm_request: text_request(Some("auto".to_string()), "hi"), + raw_request: None, + metadata: Some(Metadata { + session_id: Some(session_id.to_string()), + correlation_id: Some(correlation_id.to_string()), + ..Metadata::default() + }), + } +} + +fn algo(name: &str, model: &str, client: Option>) -> Arc { + Arc::new(SingleCallAlgo { + name: name.to_string(), + target_set: LlmTargetSet::new(vec![LlmTarget { + semantic_name: model.to_string(), + llm_client: client, + }]), + }) +} + +fn find_span(spans: &[SpanRecord], name: &str, field: &str, value: &str) -> SpanRecord { + match spans + .iter() + .find(|span| span.name == name && span.fields.get(field).map(String::as_str) == Some(value)) + { + Some(span) => span.clone(), + None => panic!("no '{name}' span with {field}={value} in {spans:?}"), + } +} + +#[tokio::test] +async fn successful_run_records_metrics_spans_and_decision_log() -> Result<(), BoxErr> { + let (store, exporter, provider) = telemetry(); + const ALGO: &str = "obs-success-algo"; + const MODEL: &str = "obs-success-model"; + + let client = Arc::new(UsageClient { + usage: Usage { + input_tokens: Some(11), + output_tokens: Some(7), + total_tokens: Some(18), + reasoning_tokens: Some(2), + ..Usage::default() + }, + }) as Arc; + let (trace, _response) = algo(ALGO, MODEL, Some(client)) + .run( + Context::default(), + request_with_metadata("obs-session-1", "obs-corr-1"), + ) + .await?; + assert_eq!(trace.len(), 1); + + // Metrics: run/call counters and latency histograms keyed by algorithm, + // token counters from the response usage, one published decision. + let snapshots = flushed_metrics(exporter, provider); + let run_attrs = [("algorithm", ALGO), ("outcome", "ok")]; + let call_attrs = [ + ("algorithm", ALGO), + ("selected_model", MODEL), + ("outcome", "ok"), + ]; + let token_attrs = [("algorithm", ALGO), ("selected_model", MODEL)]; + assert_eq!( + u64_counter_value(&snapshots, "libsy.runs", &run_attrs), + Some(1) + ); + assert_eq!( + u64_counter_value(&snapshots, "libsy.llm_calls", &call_attrs), + Some(1) + ); + assert_eq!( + f64_histogram_count(&snapshots, "libsy.run_duration_ms", &run_attrs), + Some(1) + ); + assert_eq!( + f64_histogram_count(&snapshots, "libsy.llm_call_duration_ms", &call_attrs), + Some(1) + ); + assert_eq!( + u64_counter_value(&snapshots, "libsy.input_tokens", &token_attrs), + Some(11) + ); + assert_eq!( + u64_counter_value(&snapshots, "libsy.output_tokens", &token_attrs), + Some(7) + ); + assert_eq!( + u64_counter_value(&snapshots, "libsy.total_tokens", &token_attrs), + Some(18) + ); + assert_eq!( + u64_counter_value(&snapshots, "libsy.reasoning_tokens", &token_attrs), + Some(2) + ); + assert_eq!( + u64_counter_value(&snapshots, "libsy.decisions", &token_attrs), + Some(1) + ); + + // Spans: one run span carrying the correlation ids and outcome, one child + // llm_call span carrying the selection, outcome, and token counts. + let spans = store.spans(); + let run_span = find_span(&spans, "libsy.run", "algorithm", ALGO); + assert_eq!(run_span.parent, None); + assert_eq!( + run_span.fields.get("session_id").map(String::as_str), + Some("obs-session-1") + ); + assert_eq!( + run_span.fields.get("correlation_id").map(String::as_str), + Some("obs-corr-1") + ); + assert_eq!( + run_span.fields.get("outcome").map(String::as_str), + Some("ok") + ); + + let call_span = find_span(&spans, "libsy.llm_call", "selected_model", MODEL); + assert_eq!(call_span.parent.as_deref(), Some("libsy.run")); + assert_eq!( + call_span.fields.get("algorithm").map(String::as_str), + Some(ALGO) + ); + assert_eq!( + call_span.fields.get("outcome").map(String::as_str), + Some("ok") + ); + assert_eq!( + call_span.fields.get("input_tokens").map(String::as_str), + Some("11") + ); + assert_eq!( + call_span.fields.get("output_tokens").map(String::as_str), + Some("7") + ); + assert_eq!( + call_span.fields.get("total_tokens").map(String::as_str), + Some("18") + ); + assert_eq!( + call_span.fields.get("reasoning_tokens").map(String::as_str), + Some("2") + ); + + // Structured log: the published decision with its reasoning. + let events = store.events(); + assert!( + events.iter().any(|event| { + event.target == "libsy" + && event.fields.get("selected_model").map(String::as_str) == Some(MODEL) + && event + .fields + .get("reasoning") + .is_some_and(|reasoning| reasoning.contains("picked")) + && event + .fields + .get("message") + .is_some_and(|message| message.contains("routing decision")) + }), + "no routing-decision log event for {MODEL} in {events:?}" + ); + Ok(()) +} + +#[tokio::test] +async fn failed_call_records_error_outcome_and_warn_logs() -> Result<(), BoxErr> { + let (store, exporter, provider) = telemetry(); + const ALGO: &str = "obs-failure-algo"; + const MODEL: &str = "obs-failure-model"; + + // Client-less target: the call is offloaded and we fail it by hand. + let stream = algo(ALGO, MODEL, None).run_stream( + Context::default(), + request_with_metadata("obs-session-2", "obs-corr-2"), + ); + tokio::pin!(stream); + + let mut saw_error_step = false; + while let Some(step) = stream.next().await { + match step { + Ok(Step::CallLlm(call)) => { + call.respond(Err("synthetic upstream failure".into()))?; + } + Ok(Step::Decision(_)) => {} + Ok(Step::ReturnToAgent(_)) => { + return Err("expected the failed call to fail the run".into()); + } + Err(_) => saw_error_step = true, + } + } + assert!( + saw_error_step, + "expected an error step from the failed call" + ); + + // Metrics: the call and the run both count under outcome=error, and no + // token counters exist for the failed model. + let snapshots = flushed_metrics(exporter, provider); + let run_attrs = [("algorithm", ALGO), ("outcome", "error")]; + let call_attrs = [ + ("algorithm", ALGO), + ("selected_model", MODEL), + ("outcome", "error"), + ]; + assert_eq!( + u64_counter_value(&snapshots, "libsy.runs", &run_attrs), + Some(1) + ); + assert_eq!( + u64_counter_value(&snapshots, "libsy.llm_calls", &call_attrs), + Some(1) + ); + assert_eq!( + u64_counter_value( + &snapshots, + "libsy.input_tokens", + &[("algorithm", ALGO), ("selected_model", MODEL)], + ), + None + ); + + // Spans: both spans carry outcome=error and the propagated error text. + let spans = store.spans(); + let run_span = find_span(&spans, "libsy.run", "algorithm", ALGO); + assert_eq!( + run_span.fields.get("outcome").map(String::as_str), + Some("error") + ); + assert!(run_span + .fields + .get("error") + .is_some_and(|error| error.contains("synthetic upstream failure"))); + let call_span = find_span(&spans, "libsy.llm_call", "selected_model", MODEL); + assert_eq!( + call_span.fields.get("outcome").map(String::as_str), + Some("error") + ); + + // Structured logs: a warn per failed call and per failed run. + let events = store.events(); + assert!( + events.iter().any(|event| { + event.target == "libsy" + && event.fields.get("selected_model").map(String::as_str) == Some(MODEL) + && event + .fields + .get("message") + .is_some_and(|message| message.contains("model call failed")) + }), + "no call-failure log for {MODEL} in {events:?}" + ); + assert!( + events.iter().any(|event| { + event.target == "libsy" + && event.fields.get("algorithm").map(String::as_str) == Some(ALGO) + && event + .fields + .get("message") + .is_some_and(|message| message.contains("algorithm run failed")) + }), + "no run-failure log for {ALGO} in {events:?}" + ); + Ok(()) +} From 6d33a6c2bc2157090810927f2e2b9ea013c14f75 Mon Sep 17 00:00:00 2001 From: Eric Liu Date: Wed, 15 Jul 2026 23:44:01 +0000 Subject: [PATCH 02/10] docs(libsy): implement name() in the README classifier example Signed-off-by: Eric Liu --- crates/libsy/README.md | 2 ++ 1 file changed, 2 insertions(+) diff --git a/crates/libsy/README.md b/crates/libsy/README.md index 9b2e6344..835db99d 100644 --- a/crates/libsy/README.md +++ b/crates/libsy/README.md @@ -224,6 +224,8 @@ Example — the LLM classifier (classify, then route; full version in ```rust #[async_trait] impl Algorithm for LlmClassifier { + fn name(&self) -> &str { "llm_classifier" } + async fn create_run_task(self: Arc, ctx: Context, driver: Driver, request: Request) -> Result> { // Thread `ctx` into every offloaded call and decision — it carries the request's From d9f8597652a83eb18a05463612e9a9ce244de940 Mon Sep 17 00:00:00 2001 From: Eric Liu Date: Wed, 15 Jul 2026 23:44:20 +0000 Subject: [PATCH 03/10] fix(libsy): record decisions only after the driver accepts them Signed-off-by: Eric Liu --- crates/libsy/src/core/algorithm.rs | 6 ++++-- 1 file changed, 4 insertions(+), 2 deletions(-) diff --git a/crates/libsy/src/core/algorithm.rs b/crates/libsy/src/core/algorithm.rs index 1938dd2d..d2259be0 100644 --- a/crates/libsy/src/core/algorithm.rs +++ b/crates/libsy/src/core/algorithm.rs @@ -179,10 +179,12 @@ impl Driver { } /// Publish a routing [`Decision`] as a [`Step::Decision`] on the stream. - /// Each published decision is counted and logged with its reasoning. + /// Each successfully published decision is counted and logged with its + /// reasoning; a decision the stream never accepted is not recorded. pub async fn info(&self, ctx: Context, decision: Arc) -> Result<(), BoxErr> { + self.driver.info(ctx, decision.clone()).await?; observability::record_decision(&self.algorithm, decision.as_ref()); - self.driver.info(ctx, decision).await + Ok(()) } /// Emit the terminal step: [`Step::ReturnToAgent`] on `Ok`, or an `Err` stream From 56382179d598286bb30c6570660f73d0f5a07ebf Mon Sep 17 00:00:00 2001 From: Eric Liu Date: Wed, 15 Jul 2026 23:45:52 +0000 Subject: [PATCH 04/10] refactor(libsy): carry the algorithm telemetry label on Context Signed-off-by: Eric Liu --- crates/libsy/README.md | 2 +- crates/libsy/src/core/algorithm.rs | 42 ++++++++++++++++++------------ crates/libsy/src/observability.rs | 14 +++++++++- 3 files changed, 40 insertions(+), 18 deletions(-) diff --git a/crates/libsy/README.md b/crates/libsy/README.md index 835db99d..99e675b7 100644 --- a/crates/libsy/README.md +++ b/crates/libsy/README.md @@ -303,5 +303,5 @@ agents live in [`examples`](examples/) folder. ## Not yet built - **`Signals` events** — `process_signals` / `Signals` exist but carry nothing yet. -- **`Context` fields** — the per-request state carrier is an empty placeholder today. +- **`Context` fields** — carries the algorithm telemetry label today; correlation ids, budgets, and deadlines still to come. - **Config-driven construction**, **typed errors** (vs `Box`), **weighted random**. diff --git a/crates/libsy/src/core/algorithm.rs b/crates/libsy/src/core/algorithm.rs index d2259be0..7cfdbeb7 100644 --- a/crates/libsy/src/core/algorithm.rs +++ b/crates/libsy/src/core/algorithm.rs @@ -114,19 +114,14 @@ impl CallLlmRequest { #[derive(Clone)] pub struct Driver { driver: TypeErasedDriver, - /// Name of the algorithm this driver serves — the `algorithm` attribute on - /// the telemetry emitted for its calls and decisions. - algorithm: Arc, } impl Driver { - /// Build an empty driver with its step channel ready, labeled with the name - /// of the algorithm it serves. Created per call by + /// Build an empty driver with its step channel ready. Created per call by /// [`run_stream`](Algorithm::run_stream). - pub(crate) fn new(algorithm: &str) -> Self { + pub(crate) fn new() -> Self { Self { driver: TypeErasedDriver::new(), - algorithm: Arc::from(algorithm), } } @@ -136,8 +131,10 @@ impl Driver { /// The await is wrapped in a `libsy.llm_call` span, and the call's latency, /// outcome, and token usage are recorded when it resolves. pub async fn call_llm(&self, routed: RoutedRequest) -> Result { + let ctx = routed.ctx.clone(); let selected_model = routed.decision.selected_model().to_string(); - let span = observability::llm_call_span(&self.algorithm, &selected_model); + let span = + observability::llm_call_span(observability::algorithm_label(&ctx), &selected_model); async { let started = Instant::now(); let result = self @@ -145,7 +142,7 @@ impl Driver { .fulfill_request::(routed.ctx.clone(), routed) .await; observability::record_llm_call( - &self.algorithm, + observability::algorithm_label(&ctx), &selected_model, started.elapsed(), &result, @@ -182,8 +179,8 @@ impl Driver { /// Each successfully published decision is counted and logged with its /// reasoning; a decision the stream never accepted is not recorded. pub async fn info(&self, ctx: Context, decision: Arc) -> Result<(), BoxErr> { - self.driver.info(ctx, decision.clone()).await?; - observability::record_decision(&self.algorithm, decision.as_ref()); + self.driver.info(ctx.clone(), decision.clone()).await?; + observability::record_decision(observability::algorithm_label(&ctx), decision.as_ref()); Ok(()) } @@ -219,6 +216,12 @@ impl Driver { } } +impl Default for Driver { + fn default() -> Self { + Self::new() + } +} + /// One item in the stream returned by `Driver::stream` / [`Algorithm::run_stream`]. pub enum Step { /// The algorithm needs this model call performed. The host serves it (optionally @@ -308,11 +311,11 @@ where /// emits for its runs (see the crate docs' Observability section). fn name(&self) -> &str; - /// Run one request to completion: make model calls with [`Driver::call_llm_target`], /// publish [`Decision`]s with [`Driver::info`], and return the final [`Response`]. /// The method an algorithm implements; [`run`](Self::run) / [`run_stream`](Self::run_stream) - /// drive it. `ctx` carries the request's cross-cutting values and per-session `state`. + /// drive it. `ctx` carries the request's cross-cutting values (today: the + /// algorithm's telemetry label in [`Context::values`]) and per-session `state`. async fn create_run_task( self: Arc, ctx: Context, @@ -335,7 +338,14 @@ where /// Each [`Step::CallLlm`] is an offloaded model call the consumer must serve. /// The stream ends with a [`Step::ReturnToAgent`] on success, or an `Err` item on failure. fn run_stream(self: Arc, ctx: Context, request: Request) -> StepStream { - let driver = Driver::new(self.name()); + // Stamp the algorithm's telemetry label into the request context; the + // context rides on every driver call, so its telemetry is attributed. + let mut ctx = ctx; + ctx.values.insert( + observability::ALGORITHM_KEY.to_string(), + self.name().to_string(), + ); + let driver = Driver::new(); let task_driver = driver.clone(); let task_ctx = ctx.clone(); let stream = task_driver.stream(); @@ -347,10 +357,10 @@ where async move { let started = Instant::now(); let outcome = self - .create_run_task(task_ctx, task_driver.clone(), request) + .create_run_task(task_ctx.clone(), task_driver, request) .await; observability::record_run( - &task_driver.algorithm, + observability::algorithm_label(&task_ctx), started.elapsed(), &outcome, &tracing::Span::current(), diff --git a/crates/libsy/src/observability.rs b/crates/libsy/src/observability.rs index 3d7f7a73..cadf5f4d 100644 --- a/crates/libsy/src/observability.rs +++ b/crates/libsy/src/observability.rs @@ -32,7 +32,7 @@ use opentelemetry::metrics::Meter; use opentelemetry::{global, KeyValue}; use tracing::Span; -use crate::{Decision, Metadata, Response}; +use crate::{Context, Decision, Metadata, Response}; /// Shorthand for the crate's boxed, thread-safe error type. type BoxErr = Box; @@ -40,6 +40,18 @@ type BoxErr = Box; /// Instrumentation scope for every libsy meter, span, and log line. const SCOPE: &str = "libsy"; +/// [`Context::values`] key under which `run_stream` stamps the algorithm's +/// telemetry label ([`Algorithm::name`](crate::Algorithm::name)). +pub(crate) const ALGORITHM_KEY: &str = "algorithm"; + +/// The algorithm label carried by a request context; empty until stamped. +pub(crate) fn algorithm_label(ctx: &Context) -> &str { + ctx.values + .get(ALGORITHM_KEY) + .map(String::as_str) + .unwrap_or("") +} + /// The `libsy`-scoped meter from the globally installed provider. fn meter() -> Meter { global::meter(SCOPE) From 774f42fd21057f732c7fa916f7722403d12d6596 Mon Sep 17 00:00:00 2001 From: Eric Liu Date: Wed, 15 Jul 2026 23:47:34 +0000 Subject: [PATCH 05/10] refactor(libsy): hoist telemetry plumbing into observe_run and observe_llm_call Signed-off-by: Eric Liu --- crates/libsy/src/core/algorithm.rs | 44 ++++++-------------- crates/libsy/src/observability.rs | 65 ++++++++++++++++++++++++------ 2 files changed, 66 insertions(+), 43 deletions(-) diff --git a/crates/libsy/src/core/algorithm.rs b/crates/libsy/src/core/algorithm.rs index 7cfdbeb7..66841ec3 100644 --- a/crates/libsy/src/core/algorithm.rs +++ b/crates/libsy/src/core/algorithm.rs @@ -5,7 +5,7 @@ //! routing/optimization algorithm implements, and the offload channel it makes model //! calls and publishes [`Decision`]s over. See the crate root for the narrative model. -use std::{error::Error, pin::Pin, sync::Arc, time::Instant}; +use std::{error::Error, pin::Pin, sync::Arc}; use async_trait::async_trait; use futures::{Stream, StreamExt}; @@ -133,24 +133,12 @@ impl Driver { pub async fn call_llm(&self, routed: RoutedRequest) -> Result { let ctx = routed.ctx.clone(); let selected_model = routed.decision.selected_model().to_string(); - let span = - observability::llm_call_span(observability::algorithm_label(&ctx), &selected_model); - async { - let started = Instant::now(); - let result = self - .driver - .fulfill_request::(routed.ctx.clone(), routed) - .await; - observability::record_llm_call( - observability::algorithm_label(&ctx), - &selected_model, - started.elapsed(), - &result, - &tracing::Span::current(), - ); - result - } - .instrument(span) + observability::observe_llm_call( + &ctx, + &selected_model, + self.driver + .fulfill_request::(routed.ctx.clone(), routed), + ) .await } @@ -180,7 +168,7 @@ impl Driver { /// reasoning; a decision the stream never accepted is not recorded. pub async fn info(&self, ctx: Context, decision: Arc) -> Result<(), BoxErr> { self.driver.info(ctx.clone(), decision.clone()).await?; - observability::record_decision(observability::algorithm_label(&ctx), decision.as_ref()); + observability::record_decision(&ctx, decision.as_ref()); Ok(()) } @@ -355,17 +343,11 @@ where let span = observability::run_span(self.name(), request.metadata.as_ref()); let handle = tokio::spawn( async move { - let started = Instant::now(); - let outcome = self - .create_run_task(task_ctx.clone(), task_driver, request) - .await; - observability::record_run( - observability::algorithm_label(&task_ctx), - started.elapsed(), - &outcome, - &tracing::Span::current(), - ); - outcome + observability::observe_run( + task_ctx.clone(), + self.create_run_task(task_ctx, task_driver, request), + ) + .await } .instrument(span), ); diff --git a/crates/libsy/src/observability.rs b/crates/libsy/src/observability.rs index cadf5f4d..361fe033 100644 --- a/crates/libsy/src/observability.rs +++ b/crates/libsy/src/observability.rs @@ -26,11 +26,12 @@ //! provider installed at any point in the process lifetime; the cost is //! negligible next to a model call. -use std::time::Duration; +use std::future::Future; +use std::time::{Duration, Instant}; use opentelemetry::metrics::Meter; use opentelemetry::{global, KeyValue}; -use tracing::Span; +use tracing::{Instrument, Span}; use crate::{Context, Decision, Metadata, Response}; @@ -45,7 +46,7 @@ const SCOPE: &str = "libsy"; pub(crate) const ALGORITHM_KEY: &str = "algorithm"; /// The algorithm label carried by a request context; empty until stamped. -pub(crate) fn algorithm_label(ctx: &Context) -> &str { +fn algorithm_label(ctx: &Context) -> &str { ctx.values .get(ALGORITHM_KEY) .map(String::as_str) @@ -98,10 +99,54 @@ pub(crate) fn run_span(algorithm: &str, metadata: Option<&Metadata>) -> Span { span } +/// Runs one algorithm task to completion, recording the run counter, duration +/// histogram, span outcome, and failure log when it resolves. Executes inside +/// the `libsy.run` span its caller instruments the task with. +pub(crate) async fn observe_run( + ctx: Context, + run: impl Future>, +) -> Result { + let started = Instant::now(); + let result = run.await; + record_run( + algorithm_label(&ctx), + started.elapsed(), + &result, + &Span::current(), + ); + result +} + +/// Drives one offloaded model call inside its own `libsy.llm_call` span, +/// recording the call counter, latency histogram, token usage, span fields, +/// and failure log when the call resolves. +pub(crate) async fn observe_llm_call( + ctx: &Context, + selected_model: &str, + call: impl Future>, +) -> Result { + let algorithm = algorithm_label(ctx); + let span = llm_call_span(algorithm, selected_model); + async { + let started = Instant::now(); + let result = call.await; + record_llm_call( + algorithm, + selected_model, + started.elapsed(), + &result, + &Span::current(), + ); + result + } + .instrument(span) + .await +} + /// Span covering one offloaded model call, a child of the surrounding /// `libsy.run` span. `outcome`, `error`, and the token-count fields are filled /// in by [`record_llm_call`] when the call resolves. -pub(crate) fn llm_call_span(algorithm: &str, selected_model: &str) -> Span { +fn llm_call_span(algorithm: &str, selected_model: &str) -> Span { tracing::info_span!( target: SCOPE, "libsy.llm_call", @@ -119,12 +164,7 @@ pub(crate) fn llm_call_span(algorithm: &str, selected_model: &str) -> Span { /// 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. -pub(crate) fn record_run( - algorithm: &str, - duration: Duration, - result: &Result, - span: &Span, -) { +fn record_run(algorithm: &str, duration: Duration, result: &Result, span: &Span) { let outcome = outcome_value(result); span.record("outcome", outcome); if let Err(error) = result { @@ -148,7 +188,7 @@ pub(crate) fn record_run( /// latency histogram, token counters from the response usage (absent fields are /// skipped, not recorded as zero), the `outcome`/`error`/token fields on /// `span`, and a warn log when the call failed. -pub(crate) fn record_llm_call( +fn record_llm_call( algorithm: &str, selected_model: &str, duration: Duration, @@ -218,7 +258,8 @@ pub(crate) fn record_llm_call( /// Records one published routing decision: the decision counter plus a /// structured info log carrying the decision's reasoning. -pub(crate) fn record_decision(algorithm: &str, decision: &dyn Decision) { +pub(crate) fn record_decision(ctx: &Context, decision: &dyn Decision) { + let algorithm = algorithm_label(ctx); let selected_model = decision.selected_model(); tracing::info!( target: SCOPE, From a0d0169f1d140f0492489dcfd417e465c8a13143 Mon Sep 17 00:00:00 2001 From: Eric Liu Date: Wed, 15 Jul 2026 23:49:24 +0000 Subject: [PATCH 06/10] feat(libsy): span the default-client API call and clarify llm_call timing Signed-off-by: Eric Liu --- crates/libsy/README.md | 6 +++++- crates/libsy/src/core/algorithm.rs | 27 ++++++++++++++++------- crates/libsy/src/lib.rs | 4 +++- crates/libsy/src/observability.rs | 33 +++++++++++++++++++++++++++-- crates/libsy/tests/observability.rs | 11 ++++++++++ 5 files changed, 69 insertions(+), 12 deletions(-) diff --git a/crates/libsy/README.md b/crates/libsy/README.md index 99e675b7..70346ac6 100644 --- a/crates/libsy/README.md +++ b/crates/libsy/README.md @@ -261,7 +261,11 @@ boundary — so algorithms carry no telemetry code beyond `Algorithm::name()`, t - **Spans** (`tracing`): one `libsy.run` per request, carrying the `Metadata` correlation ids and the outcome, with one child `libsy.llm_call` per model call - (selected model, outcome, token counts, latency). + (selected model, outcome, token counts; measures *fulfillment* as the algorithm + observes it, host serving included — a streamed response resolves when its stream + handle arrives). When `run` serves a call via the target's default client, the + actual API call gets its own `libsy.client_call` span — hosts serving calls over + their own transport should span their `RoutedLlmClient` equivalently. - **Structured logs** (`tracing`, target `libsy`): an info event per published `Decision` (selected model + reasoning), warn events for failed calls and runs. - **Metrics** (OpenTelemetry, scope `libsy`, via the global meter provider): counters diff --git a/crates/libsy/src/core/algorithm.rs b/crates/libsy/src/core/algorithm.rs index 66841ec3..03c54a92 100644 --- a/crates/libsy/src/core/algorithm.rs +++ b/crates/libsy/src/core/algorithm.rs @@ -128,8 +128,11 @@ impl Driver { /// Offload a model call: publish `routed` as a [`Step::CallLlm`] and await the /// consumer's [`Response`]. The call's context travels inside /// [`routed.ctx`](RoutedRequest::ctx). Errors if the stream is closed or the call failed. - /// The await is wrapped in a `libsy.llm_call` span, and the call's latency, - /// outcome, and token usage are recorded when it resolves. + /// The await is wrapped in a `libsy.llm_call` span measuring *fulfillment* as + /// the algorithm observes it (host queueing/serving included; a streamed + /// response resolves when its stream handle arrives); latency, outcome, and + /// token usage are recorded when it resolves. The provider call itself gets a + /// `libsy.client_call` span when [`Algorithm::run`] serves it. pub async fn call_llm(&self, routed: RoutedRequest) -> Result { let ctx = routed.ctx.clone(); let selected_model = routed.decision.selected_model().to_string(); @@ -394,11 +397,11 @@ where routed.decision.selected_model() ) })?; - call.respond( - client - .call(routed.ctx, routed.request, routed.decision) - .await, - ) + let result = client + .call(routed.ctx, routed.request, routed.decision) + .await; + observability::record_client_call(&result); + call.respond(result) } let stream = self.run_stream(ctx, request); @@ -418,7 +421,15 @@ where match step { None => break, // stream has ended, no more steps Some(item) => match item? { - Step::CallLlm(call) => in_flight.push(serve(*call)), + Step::CallLlm(call) => { + // `serve` makes the one API call libsy itself + // performs; give it its own client-call span. + let span = observability::client_call_span( + &call.get_routed().ctx, + call.get_decision().selected_model(), + ); + in_flight.push(serve(*call).instrument(span)); + } Step::Decision(decision) => trace.push(decision), Step::ReturnToAgent(response) => { final_response = Some(*response); diff --git a/crates/libsy/src/lib.rs b/crates/libsy/src/lib.rs index b9642a35..4edb2a65 100644 --- a/crates/libsy/src/lib.rs +++ b/crates/libsy/src/lib.rs @@ -57,7 +57,9 @@ //! The provided run methods instrument every algorithm from the outside — at the //! [`Decision`] hook and the offload boundary — so algorithms carry no telemetry //! code. Each run gets a `libsy.run` tracing span (correlation ids from -//! [`Metadata`] attached) with a child `libsy.llm_call` span per model call; +//! [`Metadata`] attached) with a child `libsy.llm_call` span per model call +//! (fulfillment time as the algorithm observes it) plus a `libsy.client_call` +//! span around the actual API call when [`run`](Algorithm::run) serves it; //! each [`Driver::info`] decision is logged with its reasoning; and OpenTelemetry //! metrics record run/call counts, latency, token usage, and published //! decisions, keyed by [`Algorithm::name`] plus `selected_model` and `outcome`. diff --git a/crates/libsy/src/observability.rs b/crates/libsy/src/observability.rs index 361fe033..a4e6fac1 100644 --- a/crates/libsy/src/observability.rs +++ b/crates/libsy/src/observability.rs @@ -143,9 +143,38 @@ pub(crate) async fn observe_llm_call( .await } +/// Span covering one *actual* provider API call the crate itself performs — +/// the default-client serve path inside [`Algorithm::run`](crate::Algorithm::run). +/// `libsy.llm_call` measures fulfillment as the algorithm observes it; this +/// span isolates the client call that fulfills it. A host serving calls over +/// its own transport should emit an equivalent span in its `LlmClient`. +pub(crate) fn client_call_span(ctx: &Context, selected_model: &str) -> Span { + tracing::info_span!( + target: SCOPE, + "libsy.client_call", + algorithm = algorithm_label(ctx), + selected_model, + outcome = tracing::field::Empty, + error = tracing::field::Empty, + ) +} + +/// 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) { + let span = Span::current(); + span.record("outcome", outcome_value(result)); + if let Err(error) = result { + span.record("error", tracing::field::display(error)); + } +} + /// Span covering one offloaded model call, a child of the surrounding -/// `libsy.run` span. `outcome`, `error`, and the token-count fields are filled -/// in by [`record_llm_call`] when the call resolves. +/// `libsy.run` span. It measures *fulfillment* as the algorithm observes it — +/// host queueing and serving included, not just the provider call. `outcome`, +/// `error`, and the token-count fields are filled in by [`record_llm_call`] +/// when the call resolves. fn llm_call_span(algorithm: &str, selected_model: &str) -> Span { tracing::info_span!( target: SCOPE, diff --git a/crates/libsy/tests/observability.rs b/crates/libsy/tests/observability.rs index 15684021..c99ea802 100644 --- a/crates/libsy/tests/observability.rs +++ b/crates/libsy/tests/observability.rs @@ -451,6 +451,17 @@ async fn successful_run_records_metrics_spans_and_decision_log() -> Result<(), B Some("ok") ); + // The default-client serve inside `run` gets its own client-call span. + let client_span = find_span(&spans, "libsy.client_call", "selected_model", MODEL); + assert_eq!( + client_span.fields.get("algorithm").map(String::as_str), + Some(ALGO) + ); + assert_eq!( + client_span.fields.get("outcome").map(String::as_str), + Some("ok") + ); + let call_span = find_span(&spans, "libsy.llm_call", "selected_model", MODEL); assert_eq!(call_span.parent.as_deref(), Some("libsy.run")); assert_eq!( From e7a507933a37c675422ff7b4c8577ba1ec2ddae0 Mon Sep 17 00:00:00 2001 From: Eric Liu Date: Wed, 15 Jul 2026 23:50:40 +0000 Subject: [PATCH 07/10] feat(libsy): record Metadata.extra_metadata as generic run-span labels Signed-off-by: Eric Liu --- crates/libsy/README.md | 3 ++- crates/libsy/src/observability.rs | 10 ++++++++-- crates/libsy/tests/observability.rs | 9 +++++++++ 3 files changed, 19 insertions(+), 3 deletions(-) diff --git a/crates/libsy/README.md b/crates/libsy/README.md index 70346ac6..60013aac 100644 --- a/crates/libsy/README.md +++ b/crates/libsy/README.md @@ -260,7 +260,8 @@ boundary — so algorithms carry no telemetry code beyond `Algorithm::name()`, t `algorithm` attribute everything below is keyed by. - **Spans** (`tracing`): one `libsy.run` per request, carrying the `Metadata` - correlation ids and the outcome, with one child `libsy.llm_call` per model call + correlation ids, any host-defined `extra_metadata` labels, and the outcome, + with one child `libsy.llm_call` per model call (selected model, outcome, token counts; measures *fulfillment* as the algorithm observes it, host serving included — a streamed response resolves when its stream handle arrives). When `run` serves a call via the target's default client, the diff --git a/crates/libsy/src/observability.rs b/crates/libsy/src/observability.rs index a4e6fac1..6e4c6aa0 100644 --- a/crates/libsy/src/observability.rs +++ b/crates/libsy/src/observability.rs @@ -70,8 +70,10 @@ 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 -/// when present; `outcome` and `error` are filled in by [`record_run`] when the -/// run ends. +/// when present. `tracing` spans cannot grow field names at runtime, so +/// arbitrary host labels ride in via [`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 { let span = tracing::info_span!( target: SCOPE, @@ -81,6 +83,7 @@ pub(crate) fn run_span(algorithm: &str, metadata: Option<&Metadata>) -> Span { agent_id = tracing::field::Empty, task_id = tracing::field::Empty, correlation_id = tracing::field::Empty, + extra_metadata = tracing::field::Empty, outcome = tracing::field::Empty, error = tracing::field::Empty, ); @@ -95,6 +98,9 @@ pub(crate) fn run_span(algorithm: &str, metadata: Option<&Metadata>) -> Span { span.record(field, value.as_str()); } } + if let Some(extra) = &metadata.extra_metadata { + span.record("extra_metadata", tracing::field::debug(extra)); + } } span } diff --git a/crates/libsy/tests/observability.rs b/crates/libsy/tests/observability.rs index c99ea802..84347b93 100644 --- a/crates/libsy/tests/observability.rs +++ b/crates/libsy/tests/observability.rs @@ -338,6 +338,10 @@ fn request_with_metadata(session_id: &str, correlation_id: &str) -> Request { metadata: Some(Metadata { session_id: Some(session_id.to_string()), correlation_id: Some(correlation_id.to_string()), + extra_metadata: Some(BTreeMap::from([( + "tenant".to_string(), + "obs-tenant-1".to_string(), + )])), ..Metadata::default() }), } @@ -450,6 +454,11 @@ async fn successful_run_records_metrics_spans_and_decision_log() -> Result<(), B run_span.fields.get("outcome").map(String::as_str), Some("ok") ); + // Host-defined labels ride in generically via Metadata.extra_metadata. + assert!(run_span + .fields + .get("extra_metadata") + .is_some_and(|extra| extra.contains("tenant") && extra.contains("obs-tenant-1"))); // The default-client serve inside `run` gets its own client-call span. let client_span = find_span(&spans, "libsy.client_call", "selected_model", MODEL); From d2005a30e413bccb7eba8a603dc91a51c75c3543 Mon Sep 17 00:00:00 2001 From: Eric Liu Date: Wed, 15 Jul 2026 23:51:20 +0000 Subject: [PATCH 08/10] test(libsy): dedupe metric filtering with a shared helper Signed-off-by: Eric Liu --- crates/libsy/tests/observability.rs | 80 +++++++++++++---------------- 1 file changed, 35 insertions(+), 45 deletions(-) diff --git a/crates/libsy/tests/observability.rs b/crates/libsy/tests/observability.rs index 84347b93..2b5cac52 100644 --- a/crates/libsy/tests/observability.rs +++ b/crates/libsy/tests/observability.rs @@ -192,35 +192,39 @@ fn attributes_match<'a>( .all(|(key, value)| present.iter().any(|(k, v)| k == key && v == value)) } +/// Latest (max) value for a `libsy`-scoped metric across snapshots, with +/// `extract` pulling the matching data points' values out of one metric's +/// aggregated data. Counters and histogram counts are cumulative, so the max +/// across snapshots is the most recent value. +fn latest_metric_value( + snapshots: &[ResourceMetrics], + name: &str, + extract: impl Fn(&AggregatedMetrics) -> Vec, +) -> Option { + snapshots + .iter() + .flat_map(|snapshot| snapshot.scope_metrics()) + .filter(|scope| scope.scope().name() == "libsy") + .flat_map(|scope| scope.metrics()) + .filter(|metric| metric.name() == name) + .flat_map(|metric| extract(metric.data())) + .max() +} + /// Latest cumulative value of a `u64` counter for the given attribute set. fn u64_counter_value( snapshots: &[ResourceMetrics], name: &str, wanted: &[(&str, &str)], ) -> Option { - let mut latest = None; - for snapshot in snapshots { - for scope in snapshot.scope_metrics() { - if scope.scope().name() != "libsy" { - continue; - } - for metric in scope.metrics() { - if metric.name() != name { - continue; - } - if let AggregatedMetrics::U64(MetricData::Sum(sum)) = metric.data() { - for point in sum.data_points() { - if attributes_match(point.attributes(), wanted) { - // Cumulative counters only grow; max = most recent. - latest = - Some(latest.map_or(point.value(), |v: u64| v.max(point.value()))); - } - } - } - } - } - } - latest + latest_metric_value(snapshots, name, |data| match data { + AggregatedMetrics::U64(MetricData::Sum(sum)) => sum + .data_points() + .filter(|point| attributes_match(point.attributes(), wanted)) + .map(|point| point.value()) + .collect(), + _ => Vec::new(), + }) } /// Latest cumulative sample count of an `f64` histogram for the attribute set. @@ -229,28 +233,14 @@ fn f64_histogram_count( name: &str, wanted: &[(&str, &str)], ) -> Option { - let mut latest = None; - for snapshot in snapshots { - for scope in snapshot.scope_metrics() { - if scope.scope().name() != "libsy" { - continue; - } - for metric in scope.metrics() { - if metric.name() != name { - continue; - } - if let AggregatedMetrics::F64(MetricData::Histogram(histogram)) = metric.data() { - for point in histogram.data_points() { - if attributes_match(point.attributes(), wanted) { - latest = - Some(latest.map_or(point.count(), |v: u64| v.max(point.count()))); - } - } - } - } - } - } - latest + latest_metric_value(snapshots, name, |data| match data { + AggregatedMetrics::F64(MetricData::Histogram(histogram)) => histogram + .data_points() + .filter(|point| attributes_match(point.attributes(), wanted)) + .map(|point| point.count()) + .collect(), + _ => Vec::new(), + }) } /// Decision with a fixed model and reasoning string. From 5ed660e022df652638a6fa0d5dbdbebaa2d46ac6 Mon Sep 17 00:00:00 2001 From: Eric Liu Date: Tue, 21 Jul 2026 03:03:22 +0000 Subject: [PATCH 09/10] test(libsy): pin client_call span parentage against entered-guard leakage Signed-off-by: Eric Liu --- crates/libsy/src/observability.rs | 5 ++++- crates/libsy/tests/observability.rs | 5 +++++ 2 files changed, 9 insertions(+), 1 deletion(-) diff --git a/crates/libsy/src/observability.rs b/crates/libsy/src/observability.rs index 6e4c6aa0..3854f6e0 100644 --- a/crates/libsy/src/observability.rs +++ b/crates/libsy/src/observability.rs @@ -12,7 +12,10 @@ //! no-op. Spans and logs use the `tracing` facade (the async-native surface the //! OpenTelemetry ecosystem bridges with `tracing-opentelemetry` / //! `opentelemetry-appender-tracing`), so the host's subscriber decides where -//! they go. +//! they go. Spans are attached to futures with [`Instrument`], never a +//! [`Span::enter`] guard held across an `.await`: a suspended task would leave +//! the span entered on its executor thread, mis-parenting every span other +//! tasks create there (see the `tracing` docs on spans in asynchronous code). //! //! Instrument names use the OTel dotted form with the unit baked into the name //! (`libsy.run_duration_ms`), matching the switchyard metric surface; a diff --git a/crates/libsy/tests/observability.rs b/crates/libsy/tests/observability.rs index 2b5cac52..e345e079 100644 --- a/crates/libsy/tests/observability.rs +++ b/crates/libsy/tests/observability.rs @@ -452,6 +452,11 @@ async fn successful_run_records_metrics_spans_and_decision_log() -> Result<(), B // The default-client serve inside `run` gets its own client-call span. let client_span = find_span(&spans, "libsy.client_call", "selected_model", MODEL); + // Host-side span: `run`'s serve loop creates it outside the algorithm's + // spans, so it has no libsy parent. This pins the `Future::instrument` + // idiom — an `Entered` guard held across the offload `.await` would leave + // `libsy.llm_call` entered on the thread and leak it in as the parent. + assert_eq!(client_span.parent.as_deref(), None); assert_eq!( client_span.fields.get("algorithm").map(String::as_str), Some(ALGO) From 7e25cc74a98c9866b0cb3599e14ac0f84f2d4744 Mon Sep 17 00:00:00 2001 From: Eric Liu Date: Tue, 21 Jul 2026 21:11:07 +0000 Subject: [PATCH 10/10] refactor(libsy): adopt #[tracing::instrument] for method spans Signed-off-by: Eric Liu --- Cargo.lock | 12 +++++ crates/libsy/Cargo.toml | 2 +- crates/libsy/src/core/algorithm.rs | 60 +++++++++++++++++------- crates/libsy/src/observability.rs | 73 +++-------------------------- crates/libsy/tests/observability.rs | 7 +-- 5 files changed, 66 insertions(+), 88 deletions(-) diff --git a/Cargo.lock b/Cargo.lock index e62283d8..dd3b302f 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -2147,9 +2147,21 @@ checksum = "63e71662fa4b2a2c3a26f570f037eb95bb1f85397f3cd8076caed2f026a6d100" dependencies = [ "log", "pin-project-lite", + "tracing-attributes", "tracing-core", ] +[[package]] +name = "tracing-attributes" +version = "0.1.31" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "7490cfa5ec963746568740651ac6781f701c9c5ea257c58e057f3ba8cf69e8da" +dependencies = [ + "proc-macro2", + "quote", + "syn", +] + [[package]] name = "tracing-core" version = "0.1.36" diff --git a/crates/libsy/Cargo.toml b/crates/libsy/Cargo.toml index a366b121..cc911d40 100644 --- a/crates/libsy/Cargo.toml +++ b/crates/libsy/Cargo.toml @@ -22,7 +22,7 @@ rand = "0.8" switchyard-protocol = { path = "../protocol" } tokio = { version = "1", features = ["full"] } tokio-stream = "0.1" -tracing = { version = "0.1", default-features = false, features = ["std"] } +tracing = { version = "0.1", default-features = false, features = ["std", "attributes"] } [dev-dependencies] # SDK + in-memory exporter to assert what the observability layer records. diff --git a/crates/libsy/src/core/algorithm.rs b/crates/libsy/src/core/algorithm.rs index 03c54a92..b54623d4 100644 --- a/crates/libsy/src/core/algorithm.rs +++ b/crates/libsy/src/core/algorithm.rs @@ -5,7 +5,7 @@ //! routing/optimization algorithm implements, and the offload channel it makes model //! calls and publishes [`Decision`]s over. See the crate root for the narrative model. -use std::{error::Error, pin::Pin, sync::Arc}; +use std::{error::Error, pin::Pin, sync::Arc, time::Instant}; use async_trait::async_trait; use futures::{Stream, StreamExt}; @@ -133,16 +133,37 @@ impl Driver { /// response resolves when its stream handle arrives); latency, outcome, and /// token usage are recorded when it resolves. The provider call itself gets a /// `libsy.client_call` span when [`Algorithm::run`] serves it. + #[tracing::instrument( + target = "libsy", + name = "libsy.llm_call", + skip_all, + fields( + algorithm = observability::algorithm_label(&routed.ctx), + selected_model = routed.decision.selected_model(), + outcome = tracing::field::Empty, + error = tracing::field::Empty, + input_tokens = tracing::field::Empty, + output_tokens = tracing::field::Empty, + total_tokens = tracing::field::Empty, + reasoning_tokens = tracing::field::Empty, + ) + )] pub async fn call_llm(&self, routed: RoutedRequest) -> Result { - let ctx = routed.ctx.clone(); + let algorithm = observability::algorithm_label(&routed.ctx).to_string(); let selected_model = routed.decision.selected_model().to_string(); - observability::observe_llm_call( - &ctx, + let started = Instant::now(); + let result = self + .driver + .fulfill_request::(routed.ctx.clone(), routed) + .await; + observability::record_llm_call( + &algorithm, &selected_model, - self.driver - .fulfill_request::(routed.ctx.clone(), routed), - ) - .await + started.elapsed(), + &result, + &tracing::Span::current(), + ); + result } /// Offload a call to `target`: pair `request` with `decision` and the target's @@ -389,6 +410,19 @@ where // Serve one offloaded call with its target's default client. A failed *model* // call is forwarded to the algorithm via `respond`; this errors only on an // infrastructure failure (no default client, or the promise was dropped). + // `serve` makes the one API call libsy itself performs, so it gets its + // own `libsy.client_call` span. + #[tracing::instrument( + target = "libsy", + name = "libsy.client_call", + skip_all, + fields( + algorithm = observability::algorithm_label(&call.get_routed().ctx), + selected_model = call.get_decision().selected_model(), + outcome = tracing::field::Empty, + error = tracing::field::Empty, + ) + )] async fn serve(call: CallLlmRequest) -> Result<(), Box> { let routed = call.get_routed().clone(); let client = routed.default_client.clone().ok_or_else(|| { @@ -421,15 +455,7 @@ where match step { None => break, // stream has ended, no more steps Some(item) => match item? { - Step::CallLlm(call) => { - // `serve` makes the one API call libsy itself - // performs; give it its own client-call span. - let span = observability::client_call_span( - &call.get_routed().ctx, - call.get_decision().selected_model(), - ); - in_flight.push(serve(*call).instrument(span)); - } + Step::CallLlm(call) => in_flight.push(serve(*call)), Step::Decision(decision) => trace.push(decision), Step::ReturnToAgent(response) => { final_response = Some(*response); diff --git a/crates/libsy/src/observability.rs b/crates/libsy/src/observability.rs index 3854f6e0..3e0ade39 100644 --- a/crates/libsy/src/observability.rs +++ b/crates/libsy/src/observability.rs @@ -12,8 +12,9 @@ //! no-op. Spans and logs use the `tracing` facade (the async-native surface the //! OpenTelemetry ecosystem bridges with `tracing-opentelemetry` / //! `opentelemetry-appender-tracing`), so the host's subscriber decides where -//! they go. Spans are attached to futures with [`Instrument`], never a -//! [`Span::enter`] guard held across an `.await`: a suspended task would leave +//! they go. Method spans use `#[tracing::instrument]`; the `libsy.run` span is +//! attached to the spawned run task with [`tracing::Instrument`]. Neither holds +//! a [`Span::enter`] guard across an `.await` — a suspended task would leave //! the span entered on its executor thread, mis-parenting every span other //! tasks create there (see the `tracing` docs on spans in asynchronous code). //! @@ -34,7 +35,7 @@ use std::time::{Duration, Instant}; use opentelemetry::metrics::Meter; use opentelemetry::{global, KeyValue}; -use tracing::{Instrument, Span}; +use tracing::Span; use crate::{Context, Decision, Metadata, Response}; @@ -49,7 +50,7 @@ const SCOPE: &str = "libsy"; pub(crate) const ALGORITHM_KEY: &str = "algorithm"; /// The algorithm label carried by a request context; empty until stamped. -fn algorithm_label(ctx: &Context) -> &str { +pub(crate) fn algorithm_label(ctx: &Context) -> &str { ctx.values .get(ALGORITHM_KEY) .map(String::as_str) @@ -126,48 +127,6 @@ pub(crate) async fn observe_run( result } -/// Drives one offloaded model call inside its own `libsy.llm_call` span, -/// recording the call counter, latency histogram, token usage, span fields, -/// and failure log when the call resolves. -pub(crate) async fn observe_llm_call( - ctx: &Context, - selected_model: &str, - call: impl Future>, -) -> Result { - let algorithm = algorithm_label(ctx); - let span = llm_call_span(algorithm, selected_model); - async { - let started = Instant::now(); - let result = call.await; - record_llm_call( - algorithm, - selected_model, - started.elapsed(), - &result, - &Span::current(), - ); - result - } - .instrument(span) - .await -} - -/// Span covering one *actual* provider API call the crate itself performs — -/// the default-client serve path inside [`Algorithm::run`](crate::Algorithm::run). -/// `libsy.llm_call` measures fulfillment as the algorithm observes it; this -/// span isolates the client call that fulfills it. A host serving calls over -/// its own transport should emit an equivalent span in its `LlmClient`. -pub(crate) fn client_call_span(ctx: &Context, selected_model: &str) -> Span { - tracing::info_span!( - target: SCOPE, - "libsy.client_call", - algorithm = algorithm_label(ctx), - selected_model, - outcome = tracing::field::Empty, - error = tracing::field::Empty, - ) -} - /// 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. @@ -179,26 +138,6 @@ pub(crate) fn record_client_call(result: &Result) { } } -/// Span covering one offloaded model call, a child of the surrounding -/// `libsy.run` span. It measures *fulfillment* as the algorithm observes it — -/// host queueing and serving included, not just the provider call. `outcome`, -/// `error`, and the token-count fields are filled in by [`record_llm_call`] -/// when the call resolves. -fn llm_call_span(algorithm: &str, selected_model: &str) -> Span { - tracing::info_span!( - target: SCOPE, - "libsy.llm_call", - algorithm, - selected_model, - outcome = tracing::field::Empty, - error = tracing::field::Empty, - input_tokens = tracing::field::Empty, - output_tokens = tracing::field::Empty, - total_tokens = tracing::field::Empty, - reasoning_tokens = tracing::field::Empty, - ) -} - /// 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. @@ -226,7 +165,7 @@ fn record_run(algorithm: &str, duration: Duration, result: &Result Result<(), B // The default-client serve inside `run` gets its own client-call span. let client_span = find_span(&spans, "libsy.client_call", "selected_model", MODEL); // Host-side span: `run`'s serve loop creates it outside the algorithm's - // spans, so it has no libsy parent. This pins the `Future::instrument` - // idiom — an `Entered` guard held across the offload `.await` would leave - // `libsy.llm_call` entered on the thread and leak it in as the parent. + // spans, so it has no libsy parent. This pins the instrument idiom + // (`#[tracing::instrument]` / `Future::instrument`) — an `Entered` guard + // held across the offload `.await` would leave `libsy.llm_call` entered + // on the thread and leak it in as the parent. assert_eq!(client_span.parent.as_deref(), None); assert_eq!( client_span.fields.get("algorithm").map(String::as_str),