LCORE-3755: emit eval turn attrs on llm.inference vs root

Split turn-summary OTEL fields on top of the unified span hierarchy:
JSON tool/RAG payloads, model/provider, and tokens go on llm.inference;
input/output/session/compacted stay on the root span.

Co-authored-by: Cursor <cursoragent@cursor.com>

Author: asimurka

src/app/endpoints/a2a.py

@@ -158,11 +158,11 @@ def _record_model_span(span: trace.Span, model_id: str) -> None:
         span: The active OpenTelemetry span.
         model_id: Full model identifier in "provider/model" format.
     """
-    provider_id, _ = extract_provider_and_model_from_model_id(model_id)
+    provider_id, bare_model_id = extract_provider_and_model_from_model_id(model_id)
     set_span_attributes(
         span,
         {
-            SpanAttributes.LLM_MODEL_ID: model_id,
+            SpanAttributes.LLM_MODEL_ID: bare_model_id,
             SpanAttributes.LLM_PROVIDER_ID: provider_id,
         },
     )
@@ -172,13 +172,17 @@ def _record_execution_span(
     span: trace.Span,
     tool_call_names: list[str],
     run_result: Optional[AgentRunResult[str]],
+    compacted: bool,
+    inference_time: float,
 ) -> None:
     """Record tool-call metrics, token usage, and output on an a2a.execute span.
 
     Parameters:
         span: The active OpenTelemetry span.
         tool_call_names: Tool names collected during streaming.
         run_result: Completed agent run result, or None.
+        compacted: Whether the turn used compacted conversation context.
+        inference_time: Request processing duration in seconds.
     """
     if tool_call_names:
         set_span_attributes(
@@ -205,6 +209,9 @@ def _record_execution_span(
         if output_text:
             span.set_attribute(SpanAttributes.OUTPUT, output_text)
 
+    span.set_attribute(SpanAttributes.COMPACTED, compacted)
+    span.set_attribute(SpanAttributes.INFERENCE_TIME, inference_time)
+
 
 async def _persist_compacted_a2a_turn(
     client: Any,
@@ -385,7 +392,7 @@ async def execute(
                     "Failed to publish failure event: %s", enqueue_error, exc_info=True
                 )
 
-    async def _process_task_streaming(  # pylint: disable=too-many-locals
+    async def _process_task_streaming(  # pylint: disable=too-many-locals,too-many-statements
         self,
         context: RequestContext,
         task_updater: TaskUpdater,
@@ -405,6 +412,7 @@ async def _process_task_streaming(  # pylint: disable=too-many-locals
 
         with tracer.start_as_current_span("a2a.execute") as span:
             span.set_attribute(SpanAttributes.SESSION_ID, context_id)
+            started_at = datetime.now(UTC)
 
             # Extract user input using SDK utility
             user_input = context.get_user_input()
@@ -578,7 +586,13 @@ async def _process_task_streaming(  # pylint: disable=too-many-locals
                 client, responses_params, compaction, agent, task_id
             )
 
-            _record_execution_span(span, self._tool_call_names, self._run_result)
+            _record_execution_span(
+                span,
+                self._tool_call_names,
+                self._run_result,
+                compacted=compaction.compacted,
+                inference_time=(datetime.now(UTC) - started_at).total_seconds(),
+            )
 
             # Publish the final task result event
             if aggregator.task_state == TaskState.working:

src/app/endpoints/info.py

@@ -20,7 +20,7 @@
 )
 from models.api.responses.successful import InfoResponse
 from models.config import Action
-from utils.otel_tracing import set_span_attributes
+from utils.otel_tracing import SpanAttributes, set_span_attributes
 from utils.types import Responses
 from version import __version__
 
@@ -85,8 +85,8 @@ async def info_endpoint_handler(
             set_span_attributes(
                 span,
                 {
-                    "service.name": configuration.configuration.name,
-                    "service.version": __version__,
+                    SpanAttributes.SERVICE_NAME: configuration.configuration.name,
+                    SpanAttributes.SERVICE_VERSION: __version__,
                 },
             )
             return InfoResponse(

src/app/endpoints/query.py

@@ -44,6 +44,7 @@
     SpanEvents,
     add_span_event,
     anonymize_value,
+    root_span_turn_attributes,
     set_span_attributes,
 )
 from utils.query import (
@@ -153,7 +154,8 @@ async def _handle_query_with_tracing(
     """
     check_configuration_loaded(configuration)
 
-    started_at = datetime.datetime.now(datetime.UTC).strftime("%Y-%m-%dT%H:%M:%SZ")
+    started_at_dt = datetime.datetime.now(datetime.UTC)
+    started_at = started_at_dt.strftime("%Y-%m-%dT%H:%M:%SZ")
     user_id, _, _skip_userid_check, token = auth
 
     # Set initial span attributes
@@ -275,20 +277,21 @@ async def _handle_query_with_tracing(
         shield_ids=query_request.shield_ids,
         no_tools=bool(query_request.no_tools),
         image_attachments=image_attachments,
+        extra_rag_chunks=(
+            inline_rag_context.rag_chunks
+            if moderation_result.decision == "passed"
+            else None
+        ),
     )
 
     if moderation_result.decision == "passed":
-        # Combine inline RAG results (BYOK + Solr) with tool-based RAG results for the transcript
-        rag_chunks = inline_rag_context.rag_chunks
-        tool_rag_chunks = turn_summary.rag_chunks
-        logger.info("RAG as a tool retrieved %d chunks", len(tool_rag_chunks))
-        turn_summary.rag_chunks = rag_chunks + tool_rag_chunks
-
-        # Add tool-based RAG documents and chunks
-        rag_documents = inline_rag_context.referenced_documents
-        tool_rag_documents = turn_summary.referenced_documents
+        # Combine inline RAG documents with tool-based RAG documents for the transcript
+        logger.info(
+            "RAG as a tool retrieved %d chunks",
+            len(turn_summary.rag_chunks) - len(inline_rag_context.rag_chunks),
+        )
         turn_summary.referenced_documents = deduplicate_referenced_documents(
-            rag_documents + tool_rag_documents
+            inline_rag_context.referenced_documents + turn_summary.referenced_documents
         )
 
     # Get topic summary for new conversation
@@ -314,7 +317,8 @@ async def _handle_query_with_tracing(
         quota_limiters=configuration.quota_limiters, user_id=user_id
     )
 
-    completed_at = datetime.datetime.now(datetime.UTC).strftime("%Y-%m-%dT%H:%M:%SZ")
+    completed_at_dt = datetime.datetime.now(datetime.UTC)
+    completed_at = completed_at_dt.strftime("%Y-%m-%dT%H:%M:%SZ")
     conversation_id = normalize_conversation_id(responses_params.conversation)
 
     logger.info("Storing query results")
@@ -335,13 +339,14 @@ async def _handle_query_with_tracing(
 
     logger.info("Building final response")
 
-    # Set final span attributes
+    # Set final root-span attributes (llm.* attrs live on the llm.inference span)
     set_span_attributes(
         root_span,
-        {
-            SpanAttributes.SESSION_ID: conversation_id,
-            SpanAttributes.OUTPUT: turn_summary.llm_response,
-        },
+        root_span_turn_attributes(
+            turn_summary,
+            session_id=conversation_id,
+            compacted=compaction.compacted,
+        ),
     )
 
     # Emit LLM response completed event

src/app/endpoints/responses.py

@@ -79,7 +79,9 @@
     SpanEvents,
     add_span_event,
     anonymize_value,
+    llm_inference_span_attributes,
     record_exception,
+    root_span_turn_attributes,
     set_span_attributes,
 )
 from utils.prompts import get_system_prompt
@@ -158,18 +160,18 @@ def _count_request_attachments(response_input: ResponseInput) -> int:
 def _finalize_responses_root_span(
     root_span: trace.Span,
     turn_summary: TurnSummary,
+    compacted: bool = False,
 ) -> None:
     """Set final root-span attributes and completion events for /responses.
 
     Args:
         root_span: OpenTelemetry root span for the request.
-        turn_summary: Completed turn summary with output text.
+        turn_summary: Completed turn summary with the LLM response text.
+        compacted: Whether the turn used compacted conversation context.
     """
     set_span_attributes(
         root_span,
-        {
-            SpanAttributes.OUTPUT: turn_summary.llm_response,
-        },
+        root_span_turn_attributes(turn_summary, compacted=compacted),
     )
     add_span_event(root_span, SpanEvents.LLM_RESPONSE_COMPLETED)
 
@@ -206,33 +208,29 @@ def _start_llm_inference_span(
 def _complete_llm_inference_span(
     span: trace.Span,
     turn_summary: TurnSummary,
+    model: str,
+    inference_time: float,
 ) -> None:
-    """Record usage/tool attrs and completion event, then end an inference span.
+    """Record turn-summary attrs and completion event, then end an inference span.
 
     Args:
         span: The ``llm.inference`` span to finalize.
-        turn_summary: Completed turn summary providing token usage and tool calls.
+        turn_summary: Completed turn summary with tools, RAG, and tokens.
+        model: Composite model identifier in ``provider/model`` format.
+        inference_time: Inference duration in seconds.
     """
+    provider_id, bare_model_id = extract_provider_and_model_from_model_id(model)
     set_span_attributes(
         span,
-        {
-            SpanAttributes.LLM_USAGE_INPUT_TOKENS: (
-                turn_summary.token_usage.input_tokens
-            ),
-            SpanAttributes.LLM_USAGE_OUTPUT_TOKENS: (
-                turn_summary.token_usage.output_tokens
-            ),
-        },
+        llm_inference_span_attributes(
+            turn_summary,
+            bare_model_id,
+            provider_id,
+            inference_time,
+        ),
     )
-    tool_names = [tc.name for tc in turn_summary.tool_calls]
-    if tool_names:
-        set_span_attributes(
-            span,
-            {
-                SpanAttributes.TOOL_CALLS_COUNT: len(tool_names),
-                SpanAttributes.TOOL_CALLS_NAMES: tool_names,
-            },
-        )
+    if turn_summary.tool_calls:
+        tool_names = [tc.name for tc in turn_summary.tool_calls]
         add_span_event(
             span,
             SpanEvents.TOOL_EXECUTION_COMPLETED,
@@ -570,7 +568,7 @@ async def handle_responses_with_tracing(  # pylint: disable=too-many-locals
     )
     attachments_count = _count_request_attachments(original_request.input)
 
-    span_attributes: dict[str, Any] = {
+    span_attributes: dict[SpanAttributes, Any] = {
         SpanAttributes.USER_ID: anonymize_value(user_id),
         SpanAttributes.INPUT: input_text,
         SpanAttributes.REQUEST_ATTACHMENTS_COUNT: attachments_count,
@@ -1231,7 +1229,8 @@ async def response_generator(
             )
         raise
 
-    # Populate tools before closing llm.inference so tool attrs land on that span.
+    # Extract response metadata from final response object before closing
+    # the inference span so tool/RAG attrs can be recorded on it.
     if latest_response_object:
         _populate_turn_summary(
             latest_response_object,
@@ -1243,6 +1242,8 @@ async def response_generator(
     _complete_llm_inference_span(
         inference_span,
         turn_summary,
+        api_params.model,
+        time.monotonic() - inference_start_time,
     )
 
     # Explicitly append the turn to conversation if context passed by previous response
@@ -1302,7 +1303,11 @@ async def generate_response(
             completed_at,
             turn_summary.llm_response,
         )
-        _finalize_responses_root_span(root_span, turn_summary)
+        _finalize_responses_root_span(
+            root_span,
+            turn_summary,
+            context.compacted_original_input is not None,
+        )
         # Persist conversation state before clients can close the stream.
         yield "data: [DONE]\n\n"
     finally:
@@ -1325,9 +1330,10 @@ async def handle_non_streaming_response(
     """
     root_span = context.root_span
     user_id = context.auth[0]
+    inference_span: Optional[trace.Span] = None
+    inference_start_time: Optional[float] = None
 
     # Fork: Get response object (blocked vs normal)
-    inference_span: Optional[trace.Span] = None
     if context.moderation_result.decision == "blocked":
         output_text = context.moderation_result.message
         api_response = OpenAIResponseObject.model_construct(
@@ -1367,6 +1373,8 @@ async def handle_non_streaming_response(
             token_usage = extract_token_usage(
                 api_response.usage, api_params.model, context.endpoint_path
             )
+            # Keep inference span open until turn_summary is built below so
+            # tool/RAG attributes can be recorded on llm.inference.
             logger.info("Consuming tokens")
             consume_query_tokens(
                 user_id=user_id,
@@ -1411,11 +1419,13 @@ async def handle_non_streaming_response(
     )
     turn_summary.rag_chunks.extend(context.inline_rag_context.rag_chunks)
 
-    # Close llm.inference after tools are known so usage + tool attrs share one span.
-    if inference_span is not None:
+    # Close llm.inference after tools/RAG are known so usage + eval attrs share one span.
+    if inference_span is not None and inference_start_time is not None:
         _complete_llm_inference_span(
             inference_span,
             turn_summary,
+            api_params.model,
+            time.monotonic() - inference_start_time,
         )
 
     # Get available quotas
@@ -1446,7 +1456,11 @@ async def handle_non_streaming_response(
         completed_at,
         output_text,
     )
-    _finalize_responses_root_span(root_span, turn_summary)
+    _finalize_responses_root_span(
+        root_span,
+        turn_summary,
+        context.compacted_original_input is not None,
+    )
     configured_mcp_labels = {s.name for s in configuration.mcp_servers}
     response_dict = (
         api_response.model_dump(exclude_none=True)