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..dd3b302f 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" @@ -2038,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" @@ -2048,6 +2169,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 +2239,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 +2260,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 +2474,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..cc911d40 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", "attributes"] } [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..60013aac 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) @@ -222,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 @@ -249,6 +253,34 @@ 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, 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 + 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 + `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 @@ -276,6 +308,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. -- **Observability** — spans + a metrics sink (`Decision` is the hook). +- **`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/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..b54623d4 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; @@ -126,10 +128,42 @@ 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 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. + #[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 { - self.driver + let algorithm = observability::algorithm_label(&routed.ctx).to_string(); + let selected_model = routed.decision.selected_model().to_string(); + let started = Instant::now(); + let result = self + .driver .fulfill_request::(routed.ctx.clone(), routed) - .await + .await; + observability::record_llm_call( + &algorithm, + &selected_model, + started.elapsed(), + &result, + &tracing::Span::current(), + ); + result } /// Offload a call to `target`: pair `request` with `decision` and the target's @@ -154,8 +188,12 @@ impl Driver { } /// Publish a routing [`Decision`] as a [`Step::Decision`] on the stream. + /// 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).await + self.driver.info(ctx.clone(), decision.clone()).await?; + observability::record_decision(&ctx, decision.as_ref()); + Ok(()) } /// Emit the terminal step: [`Step::ReturnToAgent`] on `Ok`, or an `Err` stream @@ -280,10 +318,16 @@ 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) - /// 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, @@ -306,12 +350,31 @@ 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 { + // 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(); - 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 { + observability::observe_run( + task_ctx.clone(), + self.create_run_task(task_ctx, task_driver, request), + ) + .await + } + .instrument(span), + ); // Dropping the stream aborts the algorithm task, so it doesn't keep running after the let abort_guard = AbortOnDrop(handle.abort_handle()); @@ -347,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(|| { @@ -355,11 +431,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); @@ -448,6 +524,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 +907,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 +953,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 +995,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 +1042,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 +1149,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 +1276,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..4edb2a65 100644 --- a/crates/libsy/src/lib.rs +++ b/crates/libsy/src/lib.rs @@ -52,6 +52,22 @@ //! 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 +//! (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`. +//! 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 +81,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..3e0ade39 --- /dev/null +++ b/crates/libsy/src/observability.rs @@ -0,0 +1,255 @@ +// 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. 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). +//! +//! 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::future::Future; +use std::time::{Duration, Instant}; + +use opentelemetry::metrics::Meter; +use opentelemetry::{global, KeyValue}; +use tracing::Span; + +use crate::{Context, 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"; + +/// [`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) +} + +/// `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. `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, + "libsy.run", + algorithm, + session_id = tracing::field::Empty, + 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, + ); + 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()); + } + } + if let Some(extra) = &metadata.extra_metadata { + span.record("extra_metadata", tracing::field::debug(extra)); + } + } + 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 +} + +/// 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)); + } +} + +/// 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. +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(ctx: &Context, decision: &dyn Decision) { + let algorithm = algorithm_label(ctx); + 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..56565f7f --- /dev/null +++ b/crates/libsy/tests/observability.rs @@ -0,0 +1,616 @@ +// 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 (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 { + 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. +fn f64_histogram_count( + snapshots: &[ResourceMetrics], + name: &str, + wanted: &[(&str, &str)], +) -> Option { + 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. +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()), + extra_metadata: Some(BTreeMap::from([( + "tenant".to_string(), + "obs-tenant-1".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") + ); + // 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); + // Host-side span: `run`'s serve loop creates it outside the algorithm's + // 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), + 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!( + 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(()) +}