fix(observability): observe routing streams before buffering

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;
             }
             _ => {}

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 {

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<Response, LlmClientError> {
+        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]