diff --git a/crates/libsy-llm-client/src/run.rs b/crates/libsy-llm-client/src/run.rs index 01e1c36fc..73a541f1c 100644 --- a/crates/libsy-llm-client/src/run.rs +++ b/crates/libsy-llm-client/src/run.rs @@ -185,7 +185,7 @@ async fn serve( }; let target = call.models.first().ok_or(LibsyError::NoTargets)?; let request = clients.prepare_routing_request(call.request.clone(), target); - let response = call_one( + match call_one( &clients, target, request, @@ -195,8 +195,11 @@ async fn serve( call.models.len(), true, ) - .await?; - call.respond(Ok(response)) + .await + { + Ok(response) => call.respond(Ok(response)), + Err(error) => call.fail(error), + } } /// Try candidates in order until one succeeds or a failure stops fallback. diff --git a/crates/libsy-llm-client/tests/observability.rs b/crates/libsy-llm-client/tests/observability.rs index 8150564f9..7622efe81 100644 --- a/crates/libsy-llm-client/tests/observability.rs +++ b/crates/libsy-llm-client/tests/observability.rs @@ -1718,6 +1718,23 @@ async fn classifier_stops_on_client_errors_and_records_verdict_fallback() } let snapshots = flushed_metrics(exporter, provider); + let outcome = match client.outcome { + JudgeOutcome::CallFailure | JudgeOutcome::StreamDecodeFailure => "error", + JudgeOutcome::Reply(_) => "ok", + }; + assert_eq!( + u64_counter_value( + &snapshots, + "switchyard.llm_calls", + &[ + ("algorithm", "llm_task_classifier"), + ("selected_model", judge_model), + ("outcome", outcome), + ], + ), + Some(1), + "logical call accounting for {judge_model}" + ); match expected_reason { Some(reason) => assert_eq!( u64_counter_value( diff --git a/crates/libsy/src/core/algorithm.rs b/crates/libsy/src/core/algorithm.rs index 119809278..959621db3 100644 --- a/crates/libsy/src/core/algorithm.rs +++ b/crates/libsy/src/core/algorithm.rs @@ -109,19 +109,51 @@ pub struct CallModel { pub request: Request, /// Candidate models, tried in order until one answers. Never empty. pub models: Vec, - // How to send the response back to the algorithm - reply: oneshot::Sender>, + /// How to send the response back to the algorithm. `None` once the call is recorded. + reply: Option>>, + started: Instant, } impl CallModel { /// Fulfill the promise with the caller's model-call result. Pass `Err(..)` to /// propagate a failed model call back to the algorithm. Consumes the promise: it /// can only be fulfilled once. - pub fn respond(self, result: Result) -> Result<()> { + pub fn respond(mut self, result: Result) -> Result<()> { + self.record(result.is_ok()); self.reply + .take() + .ok_or(DriverError::ResponseDropped)? .send(result) .map_err(|_| DriverError::ResponseDropped.into()) } + + /// Record a failed call and return its error to stop [`drive`]. + /// Leaves the promise unfulfilled so the driver can cancel the algorithm. + pub fn fail(mut self, error: LibsyError) -> Result<()> { + self.reply = None; + self.record(false); + Err(error) + } + + fn record(&self, is_ok: bool) { + observability::record_llm_call( + &self.algorithm, + self.models + .first() + .map(ModelId::as_str) + .unwrap_or("NoTargets"), + self.started.elapsed(), + is_ok, + ); + } +} + +impl Drop for CallModel { + fn drop(&mut self) { + if self.reply.is_some() { + self.record(false); + } + } } /// The terminal result of routing. @@ -239,8 +271,9 @@ impl Driver { /// 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 is the + /// response resolves when its stream handle arrives). The host records call metrics + /// through [`CallModel::respond`] or [`CallModel::fail`]; outcome and token usage + /// are recorded on the span when the promise resolves. The provider call itself is the /// host's, and is instrumented by whoever makes it. #[tracing::instrument( target = "libsy", @@ -258,7 +291,7 @@ impl Driver { ) )] pub async fn call_model(&self, mut request: Request, models: Vec) -> Result { - let Some(selected_model_id) = models.first().cloned() else { + let Some(selected_model_id) = models.first() else { return Err(LibsyError::NoTargets); }; request.llm_request.model = Some(selected_model_id.to_string()); @@ -268,7 +301,8 @@ impl Driver { algorithm: self.algorithm.clone(), request, models, - reply, + reply: Some(reply), + started, }; let result = async { self.step_tx @@ -280,14 +314,7 @@ impl Driver { .map_err(|_| LibsyError::from(DriverError::ResponseDropped))? } .await; - let elapsed = started.elapsed(); - observability::record_llm_call( - &self.algorithm, - selected_model_id.as_str(), - elapsed, - &result, - &tracing::Span::current(), - ); + observability::record_llm_call_span(&result, &tracing::Span::current()); result } @@ -363,9 +390,10 @@ pub enum Step { /// Returns the final [`RoutingOutcome`]. /// `serve` owns the call: it performs it however the host likes and must fulfill the promise /// with [`CallModel::respond`]. A failed *model* call belongs in `respond` — the -/// algorithm may route around it. Returning `Err` from `serve` aborts the whole run, so -/// reserve it for infrastructure failures. Calls are served concurrently, so an algorithm -/// that offloads several at once (hedging, fan-out) gets real parallelism. +/// algorithm may route around it. To stop routing on a model-call failure, return +/// [`CallModel::fail`] instead. Returning `Err` from `serve` aborts the whole run. +/// Calls are served concurrently, so an algorithm that offloads several at once (hedging, fan-out) +/// gets real parallelism. /// /// libsy performs no I/O; this is only the mechanics of consuming its own step stream, kept /// here so every host does not reimplement the same loop. `switchyard-llm-client`'s `run` diff --git a/crates/libsy/src/observability.rs b/crates/libsy/src/observability.rs index 69c6d9295..4779540da 100644 --- a/crates/libsy/src/observability.rs +++ b/crates/libsy/src/observability.rs @@ -232,18 +232,14 @@ pub(crate) fn record_classifier_fail_open(judge_model: &str, reason: &'static st ); } -/// Records the resolution of one offloaded model call: the call counter and -/// latency histogram and the outcome/token fields on `span`, without error details. +/// Records the call counter and latency histogram for a completed offloaded call. pub(crate) fn record_llm_call( algorithm: &str, selected_model: &str, duration: Duration, - result: &Result, - span: &Span, + is_ok: bool, ) { - let outcome = outcome_value(result); - span.record("outcome", outcome); - + let outcome = if is_ok { "ok" } else { "error" }; let meter = meter(); let call_attributes = [ KeyValue::new("algorithm", algorithm.to_string()), @@ -258,7 +254,11 @@ pub(crate) fn record_llm_call( .f64_histogram("switchyard.llm_call_duration_ms") .build() .record(duration.as_secs_f64() * 1000.0, &call_attributes); +} +/// Records the outcome and token fields on the algorithm's call span. +pub(crate) fn record_llm_call_span(result: &Result, span: &Span) { + span.record("outcome", outcome_value(result)); if let Ok(response) = result { // Token usage exists only once a response is buffered; a streamed // response resolves before its usage is known, so none is recorded.