feat: request lifecycle logging + metrics capture [DIS-1643]

lib/llm/src/grpc/service/openai.rs

@@ -88,10 +88,12 @@ pub async fn completion_response_stream(
 
     let http_queue_guard = state.metrics_clone().create_http_queue_guard(model);
 
-    let inflight_guard =
-        state
-            .metrics_clone()
-            .create_inflight_guard(model, Endpoint::Completions, streaming);
+    let inflight_guard = state.metrics_clone().create_inflight_guard(
+        model,
+        Endpoint::Completions,
+        streaming,
+        &request_id,
+    );
 
     let mut response_collector = state.metrics_clone().create_response_collector(model);
 

lib/llm/src/grpc/service/tensor.rs

@@ -89,10 +89,12 @@ pub async fn tensor_response_stream(
 
     let http_queue_guard = state.metrics_clone().create_http_queue_guard(model);
 
-    let inflight_guard =
-        state
-            .metrics_clone()
-            .create_inflight_guard(model, Endpoint::Tensor, streaming);
+    let inflight_guard = state.metrics_clone().create_inflight_guard(
+        model,
+        Endpoint::Tensor,
+        streaming,
+        &request_id,
+    );
 
     let mut response_collector = state.metrics_clone().create_response_collector(model);
 

lib/llm/src/http/service/anthropic.rs

@@ -273,6 +273,14 @@ async fn anthropic_messages(
 
     let mut response_collector = state.metrics_clone().create_response_collector(&model);
 
+    // Create inflight_guard early to ensure all errors are counted
+    let mut inflight_guard = state.metrics_clone().create_inflight_guard(
+        &model,
+        Endpoint::AnthropicMessages,
+        streaming,
+        request.id(),
+    );
+
     tracing::trace!("Issuing generate call for Anthropic messages");
 
     let engine_stream = engine.generate(request).await.map_err(|e| {
@@ -305,11 +313,6 @@ async fn anthropic_messages(
         Box<dyn futures::Stream<Item = Annotated<NvCreateChatCompletionStreamResponse>> + Send>,
     > = Box::pin(engine_stream);
 
-    let mut inflight_guard =
-        state
-            .metrics_clone()
-            .create_inflight_guard(&model, Endpoint::AnthropicMessages, streaming);
-
     if streaming {
         stream_handle.arm();
 

lib/llm/src/http/service/disconnect.rs

@@ -135,7 +135,7 @@ async fn connection_monitor(
     match connection_rx.await {
         Err(_) | Ok(ConnectionStatus::ClosedUnexpectedly) => {
             // the client has disconnected, no need to gracefully cancel, just kill the context
-            tracing::trace!("Connection closed unexpectedly; issuing cancellation");
+            tracing::warn!("Connection closed unexpectedly; issuing cancellation");
             if let Some(metrics) = &metrics {
                 metrics.inc_client_disconnect();
                 metrics.inc_cancellation(&cancellation_labels);
@@ -150,7 +150,7 @@ async fn connection_monitor(
 
     match stream_rx.await {
         Err(_) | Ok(ConnectionStatus::ClosedUnexpectedly) => {
-            tracing::trace!("Stream closed unexpectedly; issuing cancellation");
+            tracing::warn!("Stream closed unexpectedly; issuing cancellation");
             if let Some(metrics) = &metrics {
                 metrics.inc_client_disconnect();
                 metrics.inc_cancellation(&cancellation_labels);
@@ -211,9 +211,19 @@ pub fn monitor_for_disconnects(
                     }
                 }
                 _ = context.stopped() => {
-                    tracing::trace!("Context stopped; breaking stream");
                     // Mark as cancelled when context is stopped (client disconnect or timeout)
                     inflight_guard.mark_error(ErrorType::Cancelled);
+                    // Token counts (input_tokens, output_tokens) are recorded on
+                    // the enclosing span by ResponseMetricCollector::Drop.
+                    tracing::warn!(
+                        request_id = %inflight_guard.request_id(),
+                        model = %inflight_guard.model(),
+                        endpoint = %inflight_guard.endpoint(),
+                        request_type = %inflight_guard.request_type(),
+                        error_type = "cancelled",
+                        elapsed_ms = %inflight_guard.elapsed_ms(),
+                        "request cancelled"
+                    );
                     break;
                 }
             }