diff --git a/crates/libsy-llm-client/src/observability.rs b/crates/libsy-llm-client/src/observability.rs index 26a916343..e1b92bc13 100644 --- a/crates/libsy-llm-client/src/observability.rs +++ b/crates/libsy-llm-client/src/observability.rs @@ -296,6 +296,11 @@ impl ClientStreamObserver { let error_type = llm_client_error_type(error); match error { LlmClientError::Host { .. } => record_host_error(&self.span), + LlmClientError::UpstreamHttp { status, .. } => record_client_error( + &self.span, + &error_type, + &format_args!("upstream HTTP {}", status.as_u16()), + ), _ => record_client_error(&self.span, &error_type, error), } self.outcome = Outcome::Failed; @@ -319,8 +324,8 @@ impl ClientStreamObserver { record_client_error(&self.span, "response_translation", message); self.outcome = Outcome::Failed; } - LlmResponseChunk::StreamError { message } => { - record_client_error(&self.span, "502", message); + LlmResponseChunk::StreamError { .. } => { + record_client_error(&self.span, "502", &"upstream HTTP 502"); self.outcome = Outcome::Failed; } _ => {} diff --git a/crates/libsy-llm-client/src/run.rs b/crates/libsy-llm-client/src/run.rs index 5646344aa..132819641 100644 --- a/crates/libsy-llm-client/src/run.rs +++ b/crates/libsy-llm-client/src/run.rs @@ -456,23 +456,27 @@ async fn call_one( let started = Instant::now(); let mut routing_usage = None; let result = async { - let mut response = client?.call(request).await?; + let response = async { client?.call(request).await } + .await + .map_err(|source| LibsyError::client_call(model_id.clone(), source)); + // Observe dispatch and live I/O once; buffering must not re-observe replay. + let mut response = observability::observe_client_call(response)?; if is_routing && let LlmResponse::Stream(chunks) = response.llm_response { - let (chunks, usage) = buffer_routing_stream(chunks).await?; + let (chunks, usage) = buffer_routing_stream(chunks) + .await + .map_err(|source| LibsyError::client_call(model_id.clone(), source))?; routing_usage = usage; response.llm_response = LlmResponse::Stream(chunks); } Ok(response) } - .await - .map_err(|source| LibsyError::client_call(model_id.clone(), source)); + .await; let duration = started.elapsed(); let result = result.map(|mut response| { response.set_served_model(model_id); response }); - let result = observability::observe_client_call(result); let result = if is_routing { result } else { diff --git a/crates/libsy-llm-client/tests/observability.rs b/crates/libsy-llm-client/tests/observability.rs index 1177a2532..fdf191ea6 100644 --- a/crates/libsy-llm-client/tests/observability.rs +++ b/crates/libsy-llm-client/tests/observability.rs @@ -1457,6 +1457,152 @@ impl RoutedLlmClient for StreamingUsageClient { } } +#[tokio::test] +async fn buffered_routing_closes_observation_before_replay() -> switchyard_libsy::Result<()> { + use tracing::Instrument; + + let _guard = serialize_test().lock().await; + let (_, _, _, exporter, _) = telemetry(); + for drain in [false, true] { + let model = if drain { + "routing-replay-drained" + } else { + "routing-replay-dropped" + }; + let mut request = request_with_metadata(model, model); + request.llm_request.stream = true; + let (_, response) = run( + Arc::new(RoutingCallAlgo { + name: model.into(), + target: model.into(), + }), + Arc::new(StreamingUsageClient), + request, + ) + .instrument(tracing::info_span!( + "buffered-workflow", + gen_ai.request.model = model + )) + .await?; + + // Upstream I/O is complete, so usage and parentage must export before replay. + let call = find_otel_span(exporter, &format!("chat {model}"), model); + let workflow = find_otel_span(exporter, "buffered-workflow", model); + assert_eq!(call.parent_span_id, workflow.span_context.span_id()); + for (field, tokens) in [ + ("gen_ai.usage.input_tokens", 21), + ("gen_ai.usage.output_tokens", 5), + ] { + assert_eq!(otel_attribute(&call, field), Some(&OtelValue::I64(tokens))); + } + let count = exporter.get_finished_spans().expect("exported spans").len(); + if drain { + let replay = response + .llm_response + .into_agg() + .await + .expect("valid replay"); + assert_eq!( + replay.outputs[0].content, + vec![ContentBlock::Text { + text: "hello".into() + }] + ); + } else { + drop(response); + } + assert_eq!( + exporter.get_finished_spans().expect("exported spans").len(), + count, + "consuming or dropping replay must not export another observation" + ); + } + Ok(()) +} + +struct FailingStreamClient(bool); + +#[async_trait] +impl RoutedLlmClient for FailingStreamClient { + async fn call(&self, _request: Request) -> Result { + let failure = if self.0 { + Err(LlmClientError::UpstreamHttp { + status: http::StatusCode::BAD_GATEWAY, + body: LEAKED_CONTENT.into(), + }) + } else { + Ok(LlmResponseChunk::StreamError { + message: LEAKED_CONTENT.into(), + } + .into()) + }; + Ok(Response { + llm_response: LlmResponse::Stream( + futures::stream::iter([ + Ok(LlmResponseChunk::Usage(Usage { + input_tokens: Some(11), + output_tokens: Some(3), + ..Usage::default() + }) + .into()), + failure, + ]) + .boxed(), + ), + metadata: None, + upstream_headers: http::HeaderMap::new(), + }) + } +} + +#[tokio::test] +async fn routing_stream_error_is_recorded_once_without_upstream_body() { + let _guard = serialize_test().lock().await; + let (_, _, _, exporter, _) = telemetry(); + for http_error in [false, true] { + let model = if http_error { + "routing-http-stream-error" + } else { + "routing-in-band-error" + }; + let mut request = request_with_metadata(model, model); + request.llm_request.stream = true; + assert!( + run( + Arc::new(RoutingCallAlgo { + name: model.into(), + target: model.into() + }), + Arc::new(FailingStreamClient(http_error)), + request, + ) + .await + .is_err() + ); + let call = find_otel_span(exporter, &format!("chat {model}"), model); + for (key, expected) in [ + ("outcome", OtelValue::String("error".into())), + ("error.type", OtelValue::String("502".into())), + ("error", OtelValue::String("upstream HTTP 502".into())), + ("gen_ai.usage.input_tokens", OtelValue::I64(11)), + ("gen_ai.usage.output_tokens", OtelValue::I64(3)), + ] { + let values: Vec<_> = call + .attributes + .iter() + .filter(|attr| attr.key.as_str() == key) + .map(|attr| attr.value.clone()) + .collect(); + assert_eq!( + values, + vec![expected], + "{key} must be exported exactly once" + ); + } + assert!(!format!("{:?}", call.attributes).contains(LEAKED_CONTENT)); + } +} + struct TimeoutClient; #[async_trait]