diff --git a/src/strands_evals/mappers/__init__.py b/src/strands_evals/mappers/__init__.py index 1d192b90..15c55f16 100644 --- a/src/strands_evals/mappers/__init__.py +++ b/src/strands_evals/mappers/__init__.py @@ -1,5 +1,6 @@ """Converters for transforming telemetry data to Session format.""" +from .adk_otel_session_mapper import ADKOtelSessionMapper from .cloudwatch_parser import CloudWatchLogsParser, parse_cloudwatch_logs from .cloudwatch_session_mapper import CloudWatchSessionMapper from .langchain_otel_session_mapper import LangChainOtelSessionMapper @@ -10,6 +11,7 @@ from .utils import detect_otel_mapper, get_scope_name, readable_spans_to_dicts __all__ = [ + "ADKOtelSessionMapper", "CloudWatchLogsParser", "CloudWatchSessionMapper", "GenAIConventionVersion", diff --git a/src/strands_evals/mappers/adk_otel_session_mapper.py b/src/strands_evals/mappers/adk_otel_session_mapper.py new file mode 100644 index 00000000..63c78e74 --- /dev/null +++ b/src/strands_evals/mappers/adk_otel_session_mapper.py @@ -0,0 +1,637 @@ +"""Google ADK session mapper - converts ADK OTel spans to Session format. + +Google ADK (Agent Development Kit) produces OpenTelemetry spans with `gen_ai.*` and +`gcp.vertex.agent.*` attributes under the instrumentation scope `gcp.vertex.agent`. +Detection uses scope name as the primary signal. + +Span hierarchy (single tool-use turn): + invocation (root) + invoke_agent + call_llm + generate_content + execute_tool + call_llm + generate_content + +The `call_llm` spans carry the full serialized LLM request/response as JSON strings in +`gcp.vertex.agent.llm_request` and `gcp.vertex.agent.llm_response`. These are the primary +data source for reconstructing messages, system prompts, and tool definitions. + +Limitations: + - tool_call_id on InferenceSpan messages requires Gemini 3+; Gemini 2.x + yields None. ToolExecutionSpan always has the ID from gen_ai.tool.call.id. + - The skip_summarization agent_response fallback exposes raw tool_response + JSON (e.g. '{"result": "555"}'), not display output. ADK telemetry has no + separate display-output field. +""" + +import json +import logging +from collections import defaultdict +from typing import Any + +from ..types.trace import ( + AgentInvocationSpan, + AssistantMessage, + InferenceSpan, + Session, + SpanInfo, + TextContent, + ToolCall, + ToolCallContent, + ToolConfig, + ToolExecutionSpan, + ToolResult, + ToolResultContent, + Trace, + UserMessage, +) +from .constants import SCOPE_ADK +from .session_mapper import SessionMapper +from .utils import get_scope_name, safe_json_parse + +logger = logging.getLogger(__name__) + + +class ADKOtelSessionMapper(SessionMapper): + """Maps Google ADK OTel spans to Session format. + + This mapper handles traces produced by the Google ADK framework. ADK spans use: + - `gen_ai.operation.name` for span type detection (`invoke_agent`, `call_llm`, + `generate_content`, `execute_tool`) + - `gcp.vertex.agent.llm_request` / `gcp.vertex.agent.llm_response` for full + request/response payloads (JSON strings on `call_llm` spans) + - `gcp.vertex.agent.tool_call_args` / `gcp.vertex.agent.tool_response` for tool I/O + """ + + def map_to_session(self, data: Any, session_id: str) -> Session: + """Map ADK spans to Session format. + + Args: + data: Trace data in various formats: + - Flat list of spans: [{"trace_id": "x", "span_id": "y", ...}, ...] + - Grouped by trace_id: {"trace_1": [spans], "trace_2": [spans]} + - List of trace objects: [{"trace_id": "x", "spans": [...]}, ...] + session_id: Session identifier. + + Returns: + Session object ready for evaluation. + """ + spans = self._normalize_to_flat_spans(data) + + # Filter to only spans from the ADK instrumentation scope. + # Include spans with no scope (e.g. to_json format) since this mapper + # was explicitly selected for ADK traces. + spans = [s for s in spans if get_scope_name(s) in (SCOPE_ADK, "")] + + # Group spans by trace_id + grouped: dict[str, list[dict]] = defaultdict(list) + for span in spans: + trace_id = self._extract_trace_id(span) + if trace_id: + grouped[trace_id].append(span) + + result_traces: list[Trace] = [] + for trace_id, trace_spans in grouped.items(): + traces = self._build_traces(trace_id, trace_spans, session_id) + result_traces.extend(t for t in traces if t.spans) + + # Sort traces chronologically by earliest span start_time + result_traces.sort(key=lambda t: min(s.span_info.start_time for s in t.spans)) + + return Session(session_id=session_id, traces=result_traces) + + def _build_traces(self, trace_id: str, spans: list[dict], session_id: str) -> list[Trace]: + """Build Trace objects from spans sharing the same trace_id. + + In multi-agent scenarios (e.g. coordinator -> specialist), a single OTel + trace contains multiple ``invoke_agent`` spans. The TraceExtractor expects + one AgentInvocationSpan per Trace so that each tool call is evaluated + against the correct agent's available_tools. This method splits the spans + into one Trace per agent invocation, grouping each agent's descendant + inference and tool execution spans with it. + + When only 0 or 1 ``invoke_agent`` spans exist, the original single-Trace + behavior is preserved. + """ + # Index spans for parent-child lookups + spans_by_id: dict[str, dict] = {} + children_by_parent: dict[str, list[dict]] = defaultdict(list) + + for span in spans: + span_id = self._extract_span_id(span) + if span_id: + spans_by_id[span_id] = span + parent_span_id = self._extract_parent_span_id(span) + if parent_span_id: + children_by_parent[parent_span_id].append(span) + + # Identify invoke_agent spans + agent_spans_raw = [s for s in spans if self._get_operation_name(s) == "invoke_agent"] + + # If 0 or 1 agent invocations, build a single trace (original behavior) + if len(agent_spans_raw) <= 1: + trace = self._build_single_trace(trace_id, spans, session_id, spans_by_id, children_by_parent) + return [trace] + + # Multiple agent invocations — split into per-agent traces. + # Sort agent spans by start_time so child agents come after parents. + agent_spans_raw.sort(key=lambda s: self.parse_timestamp(s.get("start_time"))) + + # Build a mapping: span_id -> owning invoke_agent span_id. + # Process agent spans from innermost (latest start) to outermost so + # that a tool span nested under a child agent is assigned to that child, + # not the parent coordinator. + agent_span_ids = {self._extract_span_id(s) for s in agent_spans_raw} + span_to_agent: dict[str, str] = {} + + for agent_raw in reversed(agent_spans_raw): + agent_id = self._extract_span_id(agent_raw) + descendants = self._get_descendants(agent_id, children_by_parent) + for desc in descendants: + desc_id = self._extract_span_id(desc) + # Skip other invoke_agent spans — they form their own traces + if desc_id in agent_span_ids: + continue + # Only assign if not already claimed by a more specific (inner) agent + if desc_id and desc_id not in span_to_agent: + span_to_agent[desc_id] = agent_id + + # Group non-agent spans by their owning agent + agent_groups: dict[str, list[dict]] = {self._extract_span_id(a): [] for a in agent_spans_raw} + first_agent_id = self._extract_span_id(agent_spans_raw[0]) + for span in spans: + span_id = self._extract_span_id(span) + if span_id in agent_span_ids: + continue + owning_agent = span_to_agent.get(span_id) + if owning_agent and owning_agent in agent_groups: + agent_groups[owning_agent].append(span) + else: + # Unclaimed spans go to the earliest agent rather than being dropped + agent_groups[first_agent_id].append(span) + + # Build one Trace per agent invocation + traces: list[Trace] = [] + for agent_raw in agent_spans_raw: + agent_id = self._extract_span_id(agent_raw) + group_spans = [agent_raw] + agent_groups.get(agent_id, []) + trace = self._build_single_trace(trace_id, group_spans, session_id, spans_by_id, children_by_parent) + traces.append(trace) + + return traces + + def _build_single_trace( + self, + trace_id: str, + spans: list[dict], + session_id: str, + spans_by_id: dict[str, dict], + children_by_parent: dict[str, list[dict]], + ) -> Trace: + """Build a single Trace from a set of spans.""" + converted_spans: list[InferenceSpan | ToolExecutionSpan | AgentInvocationSpan] = [] + + for span in spans: + operation_name = self._get_operation_name(span) + + try: + if operation_name == "invoke_agent": + agent_span = self._convert_agent_invocation_span(span, session_id, children_by_parent) + if agent_span: + converted_spans.append(agent_span) + elif operation_name == "generate_content": + inference_span = self._convert_inference_span(span, session_id, spans_by_id) + if inference_span: + converted_spans.append(inference_span) + elif operation_name == "execute_tool": + converted_tool = self._convert_tool_execution_span(span, session_id) + if converted_tool: + converted_spans.append(converted_tool) + # Skip `call_llm` and `invocation` — data sources, not evals span types + except Exception as e: + span_id = self._extract_span_id(span) or "unknown" + logger.warning("span_id=<%s>, error=<%s> | failed to convert ADK span", span_id, e) + + # Sort spans chronologically by start_time so downstream consumers + # that treat list order as chronology get correct results. + converted_spans.sort(key=lambda s: s.span_info.start_time) + + return Trace(spans=converted_spans, trace_id=trace_id, session_id=session_id) + + # ========================================================================= + # Span Type Detection + # ========================================================================= + + def _get_operation_name(self, span: dict) -> str: + """Get gen_ai.operation.name from span attributes.""" + attrs = span.get("attributes") or {} + return attrs.get("gen_ai.operation.name", "") + + def _is_call_llm_span(self, span: dict) -> bool: + """Check if a span is a `call_llm` span. + + ADK `call_llm` spans don't set `gen_ai.operation.name`; they are identified + by their name field or having a non-empty `gcp.vertex.agent.llm_request`. + """ + name = span.get("name", "") + if name == "call_llm" or name.startswith("call_llm "): + return True + attrs = span.get("attributes", {}) + llm_request = attrs.get("gcp.vertex.agent.llm_request", "") + return bool(llm_request) and llm_request != "{}" + + # ========================================================================= + # Span Conversion + # ========================================================================= + + def _convert_agent_invocation_span( + self, + span: dict, + session_id: str, + children_by_parent: dict[str, list[dict]], + ) -> AgentInvocationSpan | None: + """Convert an ADK `invoke_agent` span to AgentInvocationSpan. + + User prompt, agent response, system prompt, and available tools are + extracted from child `call_llm` spans. + """ + span_info = self._create_span_info(span, session_id) + attrs = span.get("attributes", {}) + + span_id = self._extract_span_id(span) + call_llm_spans = sorted( + [child for child in children_by_parent.get(span_id, []) if self._is_call_llm_span(child)], + key=lambda s: self.parse_timestamp(s.get("start_time")), + ) + + user_prompt = "" + system_prompt: str | None = None + available_tools: list[ToolConfig] = [] + agent_response = "" + + # System prompt and tools from first call_llm; user_prompt from last (full history). + if call_llm_spans: + first_request = self._parse_llm_request(call_llm_spans[0]) + if first_request: + system_prompt = self._extract_system_prompt_from_request(first_request) + available_tools = self._extract_tools_from_request(first_request) + + last_request = self._parse_llm_request(call_llm_spans[-1]) + if last_request: + user_prompt = self._extract_user_prompt_from_request(last_request) + + # Skip preamble text when the response contains a function_call. + if call_llm_spans: + llm_response = self._parse_llm_response(call_llm_spans[-1]) + if llm_response: + response_parts = llm_response.get("content", {}).get("parts", []) + has_function_call = any("function_call" in part for part in response_parts) + if not has_function_call: + agent_response = self._extract_text_from_response(llm_response) + + # Fallback: use the last tool result from the invocation subtree. + if not agent_response: + tool_descendants = sorted( + [ + desc + for desc in self._get_descendants(span_id, children_by_parent, stop_at_agents=True) + if self._get_operation_name(desc) == "execute_tool" + and desc.get("attributes", {}).get("gen_ai.tool.name") != "(merged tools)" + ], + key=lambda s: self.parse_timestamp(s.get("start_time")), + ) + if tool_descendants: + last_tool_attrs = tool_descendants[-1].get("attributes", {}) + agent_response = last_tool_attrs.get("gcp.vertex.agent.tool_response", "") + + if not user_prompt and not agent_response: + return None + + metadata: dict[str, Any] = {} + if attrs.get("gen_ai.agent.name"): + metadata["agent_name"] = attrs["gen_ai.agent.name"] + if attrs.get("gen_ai.agent.description"): + metadata["agent_description"] = attrs["gen_ai.agent.description"] + + return AgentInvocationSpan( + span_info=span_info, + user_prompt=user_prompt, + agent_response=agent_response, + available_tools=available_tools, + system_prompt=system_prompt, + metadata=metadata, + ) + + def _convert_inference_span( + self, + span: dict, + session_id: str, + spans_by_id: dict[str, dict], + ) -> InferenceSpan | None: + """Convert an ADK `generate_content` span to InferenceSpan. + + Messages are reconstructed from the parent `call_llm` span's + llm_request/llm_response attributes. + """ + span_info = self._create_span_info(span, session_id) + attrs = span.get("attributes", {}) + + # Find parent call_llm span + parent_span_id = self._extract_parent_span_id(span) + parent_span = spans_by_id.get(parent_span_id, {}) if parent_span_id else {} + + messages: list[UserMessage | AssistantMessage] = [] + + if parent_span: + llm_request = self._parse_llm_request(parent_span) + llm_response = self._parse_llm_response(parent_span) + + if llm_request: + messages.extend(self._extract_messages_from_request(llm_request)) + + if llm_response: + assistant_msg = self._extract_assistant_message_from_response(llm_response) + if assistant_msg: + messages.append(assistant_msg) + + if not messages: + return None + + metadata: dict[str, Any] = {} + if attrs.get("gen_ai.system"): + metadata["gen_ai.system"] = attrs["gen_ai.system"] + if attrs.get("gen_ai.request.model"): + metadata["model"] = attrs["gen_ai.request.model"] + if attrs.get("gen_ai.agent.name"): + metadata["agent_name"] = attrs["gen_ai.agent.name"] + if attrs.get("gen_ai.usage.input_tokens") is not None: + metadata["input_tokens"] = attrs["gen_ai.usage.input_tokens"] + if attrs.get("gen_ai.usage.output_tokens") is not None: + metadata["output_tokens"] = attrs["gen_ai.usage.output_tokens"] + if attrs.get("gen_ai.usage.reasoning.output_tokens") is not None: + metadata["reasoning_tokens"] = attrs["gen_ai.usage.reasoning.output_tokens"] + if attrs.get("gen_ai.response.finish_reasons"): + metadata["finish_reasons"] = list(attrs["gen_ai.response.finish_reasons"]) + if attrs.get("gcp.vertex.agent.invocation_id"): + metadata["invocation_id"] = attrs["gcp.vertex.agent.invocation_id"] + if attrs.get("gcp.vertex.agent.event_id"): + metadata["event_id"] = attrs["gcp.vertex.agent.event_id"] + + return InferenceSpan(span_info=span_info, messages=messages, metadata=metadata) + + def _convert_tool_execution_span(self, span: dict, session_id: str) -> ToolExecutionSpan | None: + """Convert an ADK `execute_tool` span to ToolExecutionSpan.""" + span_info = self._create_span_info(span, session_id) + attrs = span.get("attributes", {}) + + tool_name = attrs.get("gen_ai.tool.name", "") + tool_call_id = attrs.get("gen_ai.tool.call.id") + + tool_parameters = safe_json_parse(attrs.get("gcp.vertex.agent.tool_call_args", "{}")) + if not isinstance(tool_parameters, dict): + tool_parameters = {} + + tool_response_raw = attrs.get("gcp.vertex.agent.tool_response", "") + tool_output_content = tool_response_raw if isinstance(tool_response_raw, str) else str(tool_response_raw) + + if not tool_name: + return None + + if tool_name == "(merged tools)": + return None + + tool_error: str | None = attrs.get("error.type") or None + if not tool_error: + span_status = span.get("status", {}) + if isinstance(span_status, dict) and span_status.get("code") == "ERROR": + tool_error = span_status.get("description") or "error" + + tool_call = ToolCall(name=tool_name, arguments=tool_parameters, tool_call_id=tool_call_id) + tool_result = ToolResult(content=tool_output_content, error=tool_error, tool_call_id=tool_call_id) + + metadata: dict[str, Any] = {} + if attrs.get("gen_ai.tool.description"): + metadata["description"] = attrs["gen_ai.tool.description"] + if attrs.get("gen_ai.tool.type"): + metadata["tool_type"] = attrs["gen_ai.tool.type"] + if attrs.get("gcp.vertex.agent.event_id"): + metadata["event_id"] = attrs["gcp.vertex.agent.event_id"] + + return ToolExecutionSpan(span_info=span_info, tool_call=tool_call, tool_result=tool_result, metadata=metadata) + + # ========================================================================= + # Data Extraction Helpers + # ========================================================================= + + def _parse_llm_request(self, call_llm_span: dict) -> dict | None: + """Parse the gcp.vertex.agent.llm_request JSON attribute from a call_llm span.""" + attrs = call_llm_span.get("attributes", {}) + raw = attrs.get("gcp.vertex.agent.llm_request", "") + if not raw or raw == "{}": + return None + return safe_json_parse(raw) if isinstance(raw, str) else None + + def _parse_llm_response(self, call_llm_span: dict) -> dict | None: + """Parse the gcp.vertex.agent.llm_response JSON attribute from a call_llm span.""" + attrs = call_llm_span.get("attributes", {}) + raw = attrs.get("gcp.vertex.agent.llm_response", "") + if not raw or raw == "{}": + return None + return safe_json_parse(raw) if isinstance(raw, str) else None + + def _extract_user_prompt_from_request(self, llm_request: dict) -> str: + """Extract the latest user text from llm_request.contents. + + ADK requests carry accumulated conversation history; the last user + message is the prompt that triggered this invocation. + """ + for content_item in reversed(llm_request.get("contents", [])): + if content_item.get("role") == "user": + texts = [ + part["text"] for part in content_item.get("parts", []) if "text" in part and not part.get("thought") + ] + if texts: + return "".join(texts) + return "" + + def _extract_system_prompt_from_request(self, llm_request: dict) -> str | None: + """Extract system_instruction from llm_request.config.""" + config = llm_request.get("config", {}) + return config.get("system_instruction") + + def _extract_tools_from_request(self, llm_request: dict) -> list[ToolConfig]: + """Extract tool definitions from llm_request.config.tools.""" + available_tools: list[ToolConfig] = [] + config = llm_request.get("config", {}) + for tool_group in config.get("tools", []): + for func_decl in tool_group.get("function_declarations", []): + available_tools.append( + ToolConfig( + name=func_decl.get("name", ""), + description=func_decl.get("description"), + parameters=func_decl.get("parameters_json_schema") or func_decl.get("parameters"), + ) + ) + return available_tools + + def _extract_text_from_response(self, llm_response: dict) -> str: + """Extract visible text from llm_response, filtering out thought parts.""" + content = llm_response.get("content", {}) + texts = [part["text"] for part in content.get("parts", []) if "text" in part and not part.get("thought")] + return "".join(texts) + + def _extract_messages_from_request( + self, + llm_request: dict, + ) -> list[UserMessage | AssistantMessage]: + """Extract typed messages from llm_request.contents (Gemini format). + + Reads function_call.id / function_response.id when present (Gemini 3+); + for Gemini 2.x models these fields are absent and tool_call_id is None. + """ + messages: list[UserMessage | AssistantMessage] = [] + + for content_item in llm_request.get("contents", []): + role = content_item.get("role", "") + parts = content_item.get("parts", []) + + if role == "user": + user_content: list[TextContent | ToolResultContent] = [] + for part in parts: + if "text" in part and not part.get("thought"): + user_content.append(TextContent(text=part["text"])) + elif "function_response" in part: + func_resp = part["function_response"] + user_content.append( + ToolResultContent( + content=json.dumps(func_resp.get("response", {})), + tool_call_id=func_resp.get("id"), + ) + ) + if user_content: + messages.append(UserMessage(content=user_content)) + + elif role == "model": + assistant_content: list[TextContent | ToolCallContent] = [] + for part in parts: + if "text" in part and not part.get("thought"): + assistant_content.append(TextContent(text=part["text"])) + elif "function_call" in part: + func_call = part["function_call"] + assistant_content.append( + ToolCallContent( + name=func_call.get("name", ""), + arguments=func_call.get("args", {}), + tool_call_id=func_call.get("id"), + ) + ) + if assistant_content: + messages.append(AssistantMessage(content=assistant_content)) + + return messages + + def _extract_assistant_message_from_response( + self, + llm_response: dict, + ) -> AssistantMessage | None: + """Extract assistant message from llm_response.content.parts.""" + content = llm_response.get("content", {}) + parts = content.get("parts", []) + + assistant_content: list[TextContent | ToolCallContent] = [] + for part in parts: + if "text" in part and not part.get("thought"): + assistant_content.append(TextContent(text=part["text"])) + elif "function_call" in part: + func_call = part["function_call"] + assistant_content.append( + ToolCallContent( + name=func_call.get("name", ""), + arguments=func_call.get("args", {}), + tool_call_id=func_call.get("id"), + ) + ) + + return AssistantMessage(content=assistant_content) if assistant_content else None + + # ========================================================================= + # Common Helpers + # ========================================================================= + + def _create_span_info(self, span: dict, session_id: str) -> SpanInfo: + """Create SpanInfo from an ADK span dict.""" + return SpanInfo( + trace_id=self._extract_trace_id(span), + span_id=self._extract_span_id(span), + session_id=session_id, + parent_span_id=self._extract_parent_span_id(span), + start_time=self.parse_timestamp(span.get("start_time")), + end_time=self.parse_timestamp(span.get("end_time")), + ) + + def _extract_trace_id(self, span: dict) -> str: + """Extract trace_id from span dict. + + Falls back to span["context"]["trace_id"] for the to_json export format. + """ + trace_id = span.get("trace_id", "") + if not trace_id: + context = span.get("context", {}) + if isinstance(context, dict): + trace_id = context.get("trace_id", "") + return self._strip_hex_prefix(trace_id) + + def _get_descendants( + self, + span_id: str, + children_by_parent: dict[str, list[dict]], + stop_at_agents: bool = False, + ) -> list[dict]: + """Get all descendants of a span by traversing the parent-child hierarchy. + + Args: + span_id: Root span to start traversal from. + children_by_parent: Full parent-child index. + stop_at_agents: If True, stop traversal at nested invoke_agent spans + (don't include them or their descendants). Used in multi-agent + scenarios to scope results to a single agent's subtree. + """ + descendants: list[dict] = [] + queue = list(children_by_parent.get(span_id, [])) + while queue: + child = queue.pop(0) + if stop_at_agents and self._get_operation_name(child) == "invoke_agent": + continue + descendants.append(child) + child_id = self._extract_span_id(child) + if child_id: + queue.extend(children_by_parent.get(child_id, [])) + return descendants + + def _extract_span_id(self, span: dict) -> str: + """Extract span_id from span dict. + + Falls back to span["context"]["span_id"] for the to_json export format. + """ + span_id = span.get("span_id", "") + if not span_id: + context = span.get("context", {}) + if isinstance(context, dict): + span_id = context.get("span_id", "") + return self._strip_hex_prefix(span_id) + + def _extract_parent_span_id(self, span: dict) -> str | None: + """Extract parent_span_id, falling back to span["parent_id"].""" + parent = span.get("parent_span_id") or span.get("parent_id") + if parent is None: + return None + return self._strip_hex_prefix(str(parent)) + + @staticmethod + def _strip_hex_prefix(value: Any) -> str: + """Strip `0x` prefix from hex IDs if present.""" + value = str(value) + if value.startswith("0x"): + return value[2:] + return value diff --git a/src/strands_evals/mappers/constants.py b/src/strands_evals/mappers/constants.py index 4f201eb1..13086e34 100644 --- a/src/strands_evals/mappers/constants.py +++ b/src/strands_evals/mappers/constants.py @@ -8,6 +8,7 @@ SCOPE_LANGCHAIN_OTEL = "opentelemetry.instrumentation.langchain" SCOPE_OPENINFERENCE = "openinference.instrumentation.langchain" SCOPE_OPENINFERENCE_SMOLAGENTS = "openinference.instrumentation.smolagents" +SCOPE_ADK = "gcp.vertex.agent" SCOPE_STRANDS = "strands.telemetry.tracer" # All scopes that should route to OpenInferenceSessionMapper diff --git a/src/strands_evals/mappers/langchain_otel_session_mapper.py b/src/strands_evals/mappers/langchain_otel_session_mapper.py index 1bb3ecea..f91d66e7 100644 --- a/src/strands_evals/mappers/langchain_otel_session_mapper.py +++ b/src/strands_evals/mappers/langchain_otel_session_mapper.py @@ -12,7 +12,6 @@ import json import logging from collections import defaultdict -from datetime import datetime, timezone from typing import Any from ..types.trace import ( @@ -47,6 +46,7 @@ SCOPE_LANGCHAIN_OTEL, ) from .session_mapper import SessionMapper +from .utils import safe_json_parse logger = logging.getLogger(__name__) @@ -248,7 +248,7 @@ def _parse_adot_body(self, span: dict) -> Any: result = None else: in_content = input_messages[0].get("content", "") - result = self._safe_json_parse(in_content) if isinstance(in_content, str) else in_content + result = safe_json_parse(in_content) if isinstance(in_content, str) else in_content if span_id: self._adot_body_cache[span_id] = result @@ -327,7 +327,7 @@ def _convert_tool_execution_span(self, span: dict, session_id: str) -> ToolExecu # Direct inputs dict: {"inputs": {"a": 1, "b": 2}} tool_parameters = inputs if tool_parameters is None and ADOT_INPUT_STR_KEY in in_parsed: - params_parsed = self._safe_json_parse(in_parsed.get(ADOT_INPUT_STR_KEY, "")) + params_parsed = safe_json_parse(in_parsed.get(ADOT_INPUT_STR_KEY, "")) if isinstance(params_parsed, dict): tool_parameters = params_parsed @@ -351,17 +351,17 @@ def _convert_tool_execution_span(self, span: dict, session_id: str) -> ToolExecu entity_output = attrs.get(ATTR_TRACELOOP_ENTITY_OUTPUT, "") if entity_input: - parsed = self._safe_json_parse(entity_input) + parsed = safe_json_parse(entity_input) if isinstance(parsed, dict): if "inputs" in parsed and isinstance(parsed.get("inputs"), dict): tool_parameters = parsed.get("inputs") elif ADOT_INPUT_STR_KEY in parsed: - params_parsed = self._safe_json_parse(parsed.get(ADOT_INPUT_STR_KEY, "")) + params_parsed = safe_json_parse(parsed.get(ADOT_INPUT_STR_KEY, "")) if isinstance(params_parsed, dict): tool_parameters = params_parsed if entity_output: - parsed = self._safe_json_parse(entity_output) + parsed = safe_json_parse(entity_output) lc_kwargs = self._extract_lc_kwargs(parsed, "output") if isinstance(parsed, dict) else None if lc_kwargs: tool_output_content = str(lc_kwargs.get("content", "")) @@ -410,14 +410,14 @@ def _convert_agent_invocation_span( entity_output = attrs.get(ATTR_TRACELOOP_ENTITY_OUTPUT, "") if entity_input: - parsed = self._safe_json_parse(entity_input) + parsed = safe_json_parse(entity_input) if isinstance(parsed, dict) and "inputs" in parsed: inputs = parsed["inputs"] if isinstance(inputs, dict) and "messages" in inputs: user_query = self._get_last_message_text(inputs["messages"]) if entity_output: - parsed = self._safe_json_parse(entity_output) + parsed = safe_json_parse(entity_output) if isinstance(parsed, dict) and "outputs" in parsed: outputs = parsed["outputs"] if isinstance(outputs, dict) and "messages" in outputs: @@ -451,8 +451,8 @@ def _get_scope_name(self, span: dict) -> str: def _create_span_info(self, span: dict, session_id: str) -> SpanInfo: """Create SpanInfo from span dict.""" - start_time = self._parse_timestamp(span.get("start_time")) - end_time = self._parse_timestamp(span.get("end_time")) + start_time = self.parse_timestamp(span.get("start_time")) + end_time = self.parse_timestamp(span.get("end_time")) return SpanInfo( trace_id=span.get("trace_id"), @@ -463,44 +463,13 @@ def _create_span_info(self, span: dict, session_id: str) -> SpanInfo: end_time=end_time, ) - def _parse_timestamp(self, value: Any) -> datetime: - """Parse timestamp from various formats.""" - if value is None: - return datetime.now(timezone.utc) - if isinstance(value, datetime): - return value - if isinstance(value, str): - try: - if value.endswith("Z"): - value = value[:-1] + "+00:00" - return datetime.fromisoformat(value) - except ValueError: - return datetime.now(timezone.utc) - if isinstance(value, (int, float)): - # Handle nanoseconds - if value > 1e12: - value = value / 1e9 - return datetime.fromtimestamp(value, tz=timezone.utc) - return datetime.now(timezone.utc) - - def _safe_json_parse(self, content: Any) -> Any: - """Safely parse JSON content.""" - if isinstance(content, dict): - return content - if isinstance(content, str): - try: - return json.loads(content) - except json.JSONDecodeError: - return content - return content - def _parse_adot_tool_content(self, input_messages: list[dict], output_messages: list[dict]) -> tuple[Any, Any]: """Parse and return (input_parsed, output_parsed) from ADOT body messages.""" in_content = input_messages[-1].get("content", "") - in_parsed = self._safe_json_parse(in_content) if isinstance(in_content, str) else in_content + in_parsed = safe_json_parse(in_content) if isinstance(in_content, str) else in_content out_content = output_messages[-1].get("content", "") - out_parsed = self._safe_json_parse(out_content) if isinstance(out_content, str) else out_content + out_parsed = safe_json_parse(out_content) if isinstance(out_content, str) else out_content return in_parsed, out_parsed @@ -639,7 +608,7 @@ def _extract_user_message(self, message: dict, index: int, attrs: dict) -> UserM msg_content = message.get("content", "") if isinstance(msg_content, str): # ADOT double-encodes strings — decode outer JSON quotes if present - text = self._safe_json_parse(msg_content) if msg_content.startswith('"') else msg_content + text = safe_json_parse(msg_content) if msg_content.startswith('"') else msg_content if not isinstance(text, str): text = msg_content if is_tool_msg: @@ -658,7 +627,7 @@ def _extract_assistant_message(self, message: dict, index: int, attrs: dict) -> msg_content = message.get("content", "") if isinstance(msg_content, str): # ADOT double-encodes empty strings as '""' — decode and skip if empty - text = self._safe_json_parse(msg_content) if msg_content.startswith('"') else msg_content + text = safe_json_parse(msg_content) if msg_content.startswith('"') else msg_content if not isinstance(text, str): text = msg_content if text: @@ -716,7 +685,7 @@ def _extract_user_prompt_from_input(self, input_messages: list[dict]) -> str | N msg = input_messages[-1] content = msg.get("content", "") if isinstance(content, str): - parsed = self._safe_json_parse(content) + parsed = safe_json_parse(content) if isinstance(parsed, dict) and "inputs" in parsed: inputs = parsed["inputs"] if isinstance(inputs, dict) and "messages" in inputs: @@ -741,7 +710,7 @@ def _extract_agent_response_from_output(self, output_messages: list[dict]) -> st if not isinstance(content, str): return None - parsed = self._safe_json_parse(content) + parsed = safe_json_parse(content) if not isinstance(parsed, dict): return None diff --git a/src/strands_evals/mappers/openinference_session_mapper.py b/src/strands_evals/mappers/openinference_session_mapper.py index 47286c3f..0890612d 100644 --- a/src/strands_evals/mappers/openinference_session_mapper.py +++ b/src/strands_evals/mappers/openinference_session_mapper.py @@ -16,7 +16,6 @@ import json import logging from collections import defaultdict -from datetime import datetime, timezone from typing import Any from ..types.trace import ( @@ -37,6 +36,7 @@ ) from .constants import SCOPE_OPENINFERENCE_SMOLAGENTS, SCOPES_OPENINFERENCE_FAMILY from .session_mapper import SessionMapper +from .utils import safe_json_parse logger = logging.getLogger(__name__) @@ -315,7 +315,7 @@ def _is_tool_execution_span(self, span: dict) -> bool: input_messages, _ = self._get_messages_from_span_events(span) if input_messages: in_content = input_messages[0].get("content", "") - in_parsed = self._safe_json_parse(in_content) if isinstance(in_content, str) else in_content + in_parsed = safe_json_parse(in_content) if isinstance(in_content, str) else in_content if isinstance(in_parsed, dict) and in_parsed.get("__type") == "tool_call_with_context": return False return True @@ -361,7 +361,7 @@ def _is_agent_invocation_span(self, span: dict) -> bool: input_messages, _ = self._get_messages_from_span_events(span) if input_messages: in_content = input_messages[0].get("content", "") - in_parsed = self._safe_json_parse(in_content) if isinstance(in_content, str) else in_content + in_parsed = safe_json_parse(in_content) if isinstance(in_content, str) else in_content if isinstance(in_parsed, dict) and "messages" in in_parsed and "remaining_steps" not in in_parsed: out_parsed = self._parse_adot_output(span) if isinstance(out_parsed, dict) and "messages" in out_parsed: @@ -599,8 +599,8 @@ def _get_scope_name(self, span: dict) -> str: def _create_span_info(self, span: dict, session_id: str) -> SpanInfo: """Create SpanInfo from span dict.""" - start_time = self._parse_timestamp(span.get("start_time")) - end_time = self._parse_timestamp(span.get("end_time")) + start_time = self.parse_timestamp(span.get("start_time")) + end_time = self.parse_timestamp(span.get("end_time")) return SpanInfo( trace_id=span.get("trace_id"), @@ -611,36 +611,6 @@ def _create_span_info(self, span: dict, session_id: str) -> SpanInfo: end_time=end_time, ) - def _parse_timestamp(self, value: Any) -> datetime: - """Parse timestamp from various formats.""" - if value is None: - return datetime.now(timezone.utc) - if isinstance(value, datetime): - return value - if isinstance(value, str): - try: - if value.endswith("Z"): - value = value[:-1] + "+00:00" - return datetime.fromisoformat(value) - except ValueError: - return datetime.now(timezone.utc) - if isinstance(value, (int, float)): - if value > 1e12: - value = value / 1e9 - return datetime.fromtimestamp(value, tz=timezone.utc) - return datetime.now(timezone.utc) - - def _safe_json_parse(self, content: Any) -> Any: - """Safely parse JSON content.""" - if isinstance(content, dict): - return content - if isinstance(content, str): - try: - return json.loads(content) - except json.JSONDecodeError: - return content - return content - def _parse_adot_output(self, span: dict) -> Any: """Parse the output content from the first ADOT body message. @@ -656,7 +626,7 @@ def _parse_adot_output(self, span: dict) -> Any: result = None else: out_content = output_messages[0].get("content", "") - result = self._safe_json_parse(out_content) if isinstance(out_content, str) else out_content + result = safe_json_parse(out_content) if isinstance(out_content, str) else out_content if span_id: self._adot_output_cache[span_id] = result diff --git a/src/strands_evals/mappers/session_mapper.py b/src/strands_evals/mappers/session_mapper.py index 982108e5..f66fb379 100644 --- a/src/strands_evals/mappers/session_mapper.py +++ b/src/strands_evals/mappers/session_mapper.py @@ -3,6 +3,7 @@ """ from abc import ABC, abstractmethod +from datetime import datetime, timezone from typing_extensions import Any @@ -72,3 +73,39 @@ def _normalize_to_flat_spans(self, data: Any) -> list[dict]: # Fallback for unexpected types return [] + + def parse_timestamp(self, value: Any) -> datetime: + """Parse timestamp from various formats. + + Handles: + - None → current UTC time + - datetime → passthrough + - ISO 8601 string (with optional trailing Z) → parsed datetime + - Numeric (int/float) nanosecond epoch → datetime + - String-encoded nanosecond epoch → datetime + + Args: + value: Raw timestamp value from a span dict. + + Returns: + Timezone-aware datetime in UTC. + """ + if value is None: + return datetime.now(timezone.utc) + if isinstance(value, datetime): + return value + if isinstance(value, str): + if value.isdigit(): + return datetime.fromtimestamp(int(value) / 1e9, tz=timezone.utc) + try: + if value.endswith("Z"): + value = value[:-1] + "+00:00" + return datetime.fromisoformat(value) + except ValueError: + return datetime.now(timezone.utc) + if isinstance(value, (int, float)): + # Handle nanoseconds + if value > 1e12: + value = value / 1e9 + return datetime.fromtimestamp(value, tz=timezone.utc) + return datetime.now(timezone.utc) diff --git a/src/strands_evals/mappers/utils.py b/src/strands_evals/mappers/utils.py index 3a07cdbd..d7d5ab92 100644 --- a/src/strands_evals/mappers/utils.py +++ b/src/strands_evals/mappers/utils.py @@ -6,12 +6,35 @@ import logging from typing import Any -from .constants import SCOPE_LANGCHAIN_OTEL, SCOPE_STRANDS, SCOPES_OPENINFERENCE_FAMILY +from .constants import SCOPE_ADK, SCOPE_LANGCHAIN_OTEL, SCOPE_STRANDS, SCOPES_OPENINFERENCE_FAMILY from .session_mapper import SessionMapper logger = logging.getLogger(__name__) +def safe_json_parse(content: Any) -> Any: + """Safely parse JSON content, returning the original value on failure. + + If content is already a dict, returns it as-is. If it's a string, attempts + JSON parsing and falls back to returning the raw string on decode error. + For all other types, returns the value unchanged. + + Args: + content: Value to parse — typically a str or dict from span attributes. + + Returns: + Parsed dict/list on success, or the original value if parsing fails or is unnecessary. + """ + if isinstance(content, dict): + return content + if isinstance(content, str): + try: + return json.loads(content) + except json.JSONDecodeError: + return content + return content + + def join_tool_result_content(content: Any) -> str: """Join all blocks in a Bedrock-style toolResult content list into one string. @@ -89,6 +112,7 @@ def detect_otel_mapper(spans: list[Any]) -> SessionMapper: >>> session = mapper.map_to_session(spans, "session-123") """ # Import here to avoid circular imports + from .adk_otel_session_mapper import ADKOtelSessionMapper from .cloudwatch_session_mapper import CloudWatchSessionMapper from .langchain_otel_session_mapper import LangChainOtelSessionMapper from .openinference_session_mapper import OpenInferenceSessionMapper @@ -107,6 +131,9 @@ def detect_otel_mapper(spans: list[Any]) -> SessionMapper: if scope_name in SCOPES_OPENINFERENCE_FAMILY: return OpenInferenceSessionMapper() + if scope_name == SCOPE_ADK: + return ADKOtelSessionMapper() + if scope_name == SCOPE_STRANDS: # CloudWatch split format puts body on a separate entry from # the scoped metadata entry. Break here and let the fallback diff --git a/tests/strands_evals/mappers/fixtures/adk_live_spans.json b/tests/strands_evals/mappers/fixtures/adk_live_spans.json new file mode 100644 index 00000000..8ca5e0b6 --- /dev/null +++ b/tests/strands_evals/mappers/fixtures/adk_live_spans.json @@ -0,0 +1,193 @@ +[ + { + "trace_id": "97847cb33dff13e46483685930f4e236", + "span_id": "771ce4dcbd861745", + "parent_span_id": "b5aaea6a8c6e066f", + "name": "execute_tool calculator", + "start_time": 1784995767655639000, + "end_time": 1784995767656160000, + "attributes": { + "gen_ai.operation.name": "execute_tool", + "gen_ai.tool.description": "Evaluate a mathematical expression and return the result.\n\nArgs:\n expression: A mathematical expression to evaluate, e.g. \"2 + 2\" or \"15 * 37\".\n\nReturns:\n The result of the expression as a string.", + "gen_ai.tool.name": "calculator", + "gen_ai.tool.type": "FunctionTool", + "gcp.vertex.agent.llm_request": "{}", + "gcp.vertex.agent.llm_response": "{}", + "gcp.vertex.agent.tool_call_args": "{\"expression\": \"15 * 37\"}", + "gen_ai.tool.call.id": "uwpsprd2", + "gcp.vertex.agent.event_id": "71bae700-48f8-4f1b-9779-c1259c12e5a5", + "gcp.vertex.agent.tool_response": "{\"result\": \"555\"}" + }, + "scope": { + "name": "gcp.vertex.agent", + "version": "2.5.0" + }, + "status": { + "code": "UNSET" + }, + "span_events": [] + }, + { + "trace_id": "97847cb33dff13e46483685930f4e236", + "span_id": "b5aaea6a8c6e066f", + "parent_span_id": "f402c1e62b85df9b", + "name": "generate_content gemini-3.5-flash", + "start_time": 1784995766557686000, + "end_time": 1784995767656646000, + "attributes": { + "gen_ai.system": "gemini", + "gen_ai.operation.name": "generate_content", + "gen_ai.request.model": "gemini-3.5-flash", + "gen_ai.agent.name": "math_agent", + "gen_ai.conversation.id": "b4866a2a-a3d4-4d2f-8f40-6bb0d03d135c", + "gcp.vertex.agent.event_id": "375fe362-48af-4b49-9191-e3bcbe1ebd9c", + "gcp.vertex.agent.invocation_id": "e-5742fbee-9bb9-4445-b224-7ae0041d6631", + "gen_ai.response.finish_reasons": [ + "stop" + ], + "gen_ai.usage.input_tokens": 157, + "gen_ai.usage.output_tokens": 51, + "gen_ai.usage.reasoning.output_tokens": 32 + }, + "scope": { + "name": "gcp.vertex.agent", + "version": "2.5.0" + }, + "status": { + "code": "UNSET" + }, + "span_events": [] + }, + { + "trace_id": "97847cb33dff13e46483685930f4e236", + "span_id": "f402c1e62b85df9b", + "parent_span_id": "67654e241edd27e4", + "name": "call_llm", + "start_time": 1784995766557571000, + "end_time": 1784995767656731000, + "attributes": { + "gen_ai.system": "gcp.vertex.agent", + "gen_ai.request.model": "gemini-3.5-flash", + "gcp.vertex.agent.invocation_id": "e-5742fbee-9bb9-4445-b224-7ae0041d6631", + "gcp.vertex.agent.session_id": "b4866a2a-a3d4-4d2f-8f40-6bb0d03d135c", + "gcp.vertex.agent.event_id": "375fe362-48af-4b49-9191-e3bcbe1ebd9c", + "gcp.vertex.agent.llm_request": "{\"model\": \"gemini-3.5-flash\", \"config\": {\"http_options\": {\"headers\": {\"x-goog-api-client\": \"google-adk/2.5.0 gl-python/3.10.0\", \"user-agent\": \"google-adk/2.5.0 gl-python/3.10.0\"}}, \"system_instruction\": \"You are a math assistant. Use the calculator tool to evaluate mathematical expressions.\\n\\nYou are an agent. Your internal name is \\\"math_agent\\\". The description about you is \\\"A helpful math assistant\\\".\", \"tools\": [{\"function_declarations\": [{\"description\": \"Evaluate a mathematical expression and return the result.\\n\\nArgs:\\n expression: A mathematical expression to evaluate, e.g. \\\"2 + 2\\\" or \\\"15 * 37\\\".\\n\\nReturns:\\n The result of the expression as a string.\", \"name\": \"calculator\", \"parameters_json_schema\": {\"properties\": {\"expression\": {\"title\": \"Expression\", \"type\": \"string\"}}, \"required\": [\"expression\"], \"title\": \"calculatorParams\", \"type\": \"object\"}}]}]}, \"contents\": [{\"parts\": [{\"text\": \"What is 15 multiplied by 37?\"}], \"role\": \"user\"}]}", + "gcp.vertex.agent.llm_response": "{\"model_version\":\"gemini-3.5-flash\",\"content\":{\"parts\":[{\"function_call\":{\"id\":\"uwpsprd2\",\"args\":{\"expression\":\"15 * 37\"},\"name\":\"calculator\"},\"thought_signature\":\"MOCK_SIGNATURE\"}],\"role\":\"model\"},\"finish_reason\":\"STOP\",\"usage_metadata\":{\"candidates_token_count\":19,\"prompt_token_count\":157,\"prompt_tokens_details\":[{\"modality\":\"TEXT\",\"token_count\":157}],\"thoughts_token_count\":32,\"total_token_count\":208}}", + "gen_ai.usage.input_tokens": 157, + "gen_ai.usage.output_tokens": 51, + "gen_ai.usage.reasoning.output_tokens": 32, + "gen_ai.response.finish_reasons": [ + "stop" + ] + }, + "scope": { + "name": "gcp.vertex.agent", + "version": "2.5.0" + }, + "status": { + "code": "UNSET" + }, + "span_events": [] + }, + { + "trace_id": "97847cb33dff13e46483685930f4e236", + "span_id": "44872427376a4cb5", + "parent_span_id": "49b19fa1ee559677", + "name": "generate_content gemini-3.5-flash", + "start_time": 1784995767662764000, + "end_time": 1784995768671922000, + "attributes": { + "gen_ai.system": "gemini", + "gen_ai.operation.name": "generate_content", + "gen_ai.request.model": "gemini-3.5-flash", + "gen_ai.agent.name": "math_agent", + "gen_ai.conversation.id": "b4866a2a-a3d4-4d2f-8f40-6bb0d03d135c", + "gcp.vertex.agent.event_id": "d1dd03f4-d8ba-4f1b-bbf4-fc8ebd4b39f3", + "gcp.vertex.agent.invocation_id": "e-5742fbee-9bb9-4445-b224-7ae0041d6631", + "gen_ai.response.finish_reasons": [ + "stop" + ], + "gen_ai.usage.input_tokens": 222, + "gen_ai.usage.output_tokens": 33, + "gen_ai.usage.reasoning.output_tokens": 20 + }, + "scope": { + "name": "gcp.vertex.agent", + "version": "2.5.0" + }, + "status": { + "code": "UNSET" + }, + "span_events": [] + }, + { + "trace_id": "97847cb33dff13e46483685930f4e236", + "span_id": "49b19fa1ee559677", + "parent_span_id": "67654e241edd27e4", + "name": "call_llm", + "start_time": 1784995767662531000, + "end_time": 1784995768672096000, + "attributes": { + "gen_ai.system": "gcp.vertex.agent", + "gen_ai.request.model": "gemini-3.5-flash", + "gcp.vertex.agent.invocation_id": "e-5742fbee-9bb9-4445-b224-7ae0041d6631", + "gcp.vertex.agent.session_id": "b4866a2a-a3d4-4d2f-8f40-6bb0d03d135c", + "gcp.vertex.agent.event_id": "d1dd03f4-d8ba-4f1b-bbf4-fc8ebd4b39f3", + "gcp.vertex.agent.llm_request": "{\"model\": \"gemini-3.5-flash\", \"config\": {\"http_options\": {\"headers\": {\"x-goog-api-client\": \"google-adk/2.5.0 gl-python/3.10.0\", \"user-agent\": \"google-adk/2.5.0 gl-python/3.10.0\"}}, \"system_instruction\": \"You are a math assistant. Use the calculator tool to evaluate mathematical expressions.\\n\\nYou are an agent. Your internal name is \\\"math_agent\\\". The description about you is \\\"A helpful math assistant\\\".\", \"tools\": [{\"function_declarations\": [{\"description\": \"Evaluate a mathematical expression and return the result.\\n\\nArgs:\\n expression: A mathematical expression to evaluate, e.g. \\\"2 + 2\\\" or \\\"15 * 37\\\".\\n\\nReturns:\\n The result of the expression as a string.\", \"name\": \"calculator\", \"parameters_json_schema\": {\"properties\": {\"expression\": {\"title\": \"Expression\", \"type\": \"string\"}}, \"required\": [\"expression\"], \"title\": \"calculatorParams\", \"type\": \"object\"}}]}]}, \"contents\": [{\"parts\": [{\"text\": \"What is 15 multiplied by 37?\"}], \"role\": \"user\"}, {\"parts\": [{\"function_call\": {\"id\": \"uwpsprd2\", \"args\": {\"expression\": \"15 * 37\"}, \"name\": \"calculator\"}, \"thought_signature\": \"MOCK_SIGNATURE\"}], \"role\": \"model\"}, {\"parts\": [{\"function_response\": {\"id\": \"uwpsprd2\", \"name\": \"calculator\", \"response\": {\"result\": \"555\"}}}], \"role\": \"user\"}]}", + "gcp.vertex.agent.llm_response": "{\"model_version\":\"gemini-3.5-flash\",\"content\":{\"parts\":[{\"text\":\"15 multiplied by 37 is 555.\",\"thought_signature\":\"MOCK_SIGNATURE\"}],\"role\":\"model\"},\"finish_reason\":\"STOP\",\"usage_metadata\":{\"candidates_token_count\":13,\"prompt_token_count\":222,\"prompt_tokens_details\":[{\"modality\":\"TEXT\",\"token_count\":222}],\"thoughts_token_count\":20,\"total_token_count\":255}}", + "gen_ai.usage.input_tokens": 222, + "gen_ai.usage.output_tokens": 33, + "gen_ai.usage.reasoning.output_tokens": 20, + "gen_ai.response.finish_reasons": [ + "stop" + ] + }, + "scope": { + "name": "gcp.vertex.agent", + "version": "2.5.0" + }, + "status": { + "code": "UNSET" + }, + "span_events": [] + }, + { + "trace_id": "97847cb33dff13e46483685930f4e236", + "span_id": "67654e241edd27e4", + "parent_span_id": "6fd93d29bfd4a522", + "name": "invoke_agent math_agent", + "start_time": 1784995765229514000, + "end_time": 1784995768672315000, + "attributes": { + "gen_ai.operation.name": "invoke_agent", + "gen_ai.agent.description": "A helpful math assistant", + "gen_ai.agent.name": "math_agent", + "gen_ai.conversation.id": "b4866a2a-a3d4-4d2f-8f40-6bb0d03d135c" + }, + "scope": { + "name": "gcp.vertex.agent", + "version": "2.5.0" + }, + "status": { + "code": "UNSET" + }, + "span_events": [] + }, + { + "trace_id": "97847cb33dff13e46483685930f4e236", + "span_id": "6fd93d29bfd4a522", + "parent_span_id": null, + "name": "invocation", + "start_time": 1784995765227135000, + "end_time": 1784995768672840000, + "attributes": {}, + "scope": { + "name": "gcp.vertex.agent", + "version": "2.5.0" + }, + "status": { + "code": "UNSET" + }, + "span_events": [] + } +] diff --git a/tests/strands_evals/mappers/test_adk_otel_session_mapper.py b/tests/strands_evals/mappers/test_adk_otel_session_mapper.py new file mode 100644 index 00000000..7d0e0971 --- /dev/null +++ b/tests/strands_evals/mappers/test_adk_otel_session_mapper.py @@ -0,0 +1,1037 @@ +"""Tests for ADKOtelSessionMapper - ADK OTel spans → Session conversion.""" + +import json +from datetime import datetime, timezone +from pathlib import Path + +from strands_evals.mappers import ADKOtelSessionMapper +from strands_evals.types.trace import ( + AgentInvocationSpan, + AssistantMessage, + InferenceSpan, + ToolCallContent, + ToolExecutionSpan, +) + +SESSION_ID = "test-session-1" +_FIXTURES_DIR = Path(__file__).resolve().parent / "fixtures" +_LIVE_SPANS_FILE = _FIXTURES_DIR / "adk_live_spans.json" + + +# ============================================================================ +# Fixture Helpers +# ============================================================================ + + +def make_span( + trace_id="trace-1", + span_id="span-1", + parent_span_id=None, + name="test-span", + attributes=None, + start_time=1700000000000000000, + end_time=1700000001000000000, +): + """Build a normalized ADK span dict.""" + return { + "trace_id": trace_id, + "span_id": span_id, + "parent_span_id": parent_span_id, + "name": name, + "start_time": start_time, + "end_time": end_time, + "attributes": attributes or {}, + "scope": {"name": "gcp.vertex.agent", "version": "2.5.0"}, + "status": {"code": "UNSET"}, + "span_events": [], + } + + +def make_llm_request( + user_text="What is 15 multiplied by 37?", + system_instruction="You are a math assistant.", + tools=None, + history=None, +): + """Build a gcp.vertex.agent.llm_request JSON string.""" + contents = list(history) if history else [{"parts": [{"text": user_text}], "role": "user"}] + config = {"system_instruction": system_instruction} + if tools: + config["tools"] = [{"function_declarations": tools}] + return json.dumps({"model": "gemini-2.5-flash", "config": config, "contents": contents}) + + +def make_llm_response_text(text="15 multiplied by 37 is 555."): + """Build a text-only llm_response JSON string.""" + return json.dumps({"content": {"parts": [{"text": text}], "role": "model"}, "finish_reason": "STOP"}) + + +def make_llm_response_tool_call(name="calculator", args=None, call_id="fc-001"): + """Build a tool-call llm_response (Gemini 3+ style with id).""" + args = args or {"expression": "15 * 37"} + func_call = {"name": name, "args": args} + if call_id is not None: + func_call["id"] = call_id + return json.dumps({"content": {"parts": [{"function_call": func_call}], "role": "model"}, "finish_reason": "STOP"}) + + +def load_full_trace(): + """Load the full ADK trace fixture from JSON.""" + return json.loads(_LIVE_SPANS_FILE.read_text()) + + +# ============================================================================ +# Tests: Full Trace (fixture-based) +# ============================================================================ + + +class TestFullTrace: + """End-to-end session mapping with a complete ADK trace loaded from fixture.""" + + def setup_method(self): + self.mapper = ADKOtelSessionMapper() + self.spans = load_full_trace() + + def test_session_structure(self): + """Full trace produces correct session/trace/span counts and types.""" + session = self.mapper.map_to_session(self.spans, SESSION_ID) + assert session.session_id == SESSION_ID + assert len(session.traces) == 1 + trace = session.traces[0] + assert len(trace.spans) == 4 + types = {type(s).__name__ for s in trace.spans} + assert types == {"AgentInvocationSpan", "InferenceSpan", "ToolExecutionSpan"} + + def test_agent_invocation_span(self): + """Agent span extracts prompt, response, system prompt, tools, and metadata.""" + session = self.mapper.map_to_session(self.spans, SESSION_ID) + agent = [s for s in session.traces[0].spans if isinstance(s, AgentInvocationSpan)][0] + assert agent.user_prompt == "What is 15 multiplied by 37?" + assert agent.agent_response == "15 multiplied by 37 is 555." + assert "math assistant" in agent.system_prompt + assert len(agent.available_tools) == 1 + assert agent.available_tools[0].name == "calculator" + assert agent.metadata["agent_name"] == "math_agent" + + def test_tool_execution_span(self): + """Tool span extracts name, arguments, call_id, and response.""" + session = self.mapper.map_to_session(self.spans, SESSION_ID) + tool = [s for s in session.traces[0].spans if isinstance(s, ToolExecutionSpan)][0] + assert tool.tool_call.name == "calculator" + assert tool.tool_call.arguments == {"expression": "15 * 37"} + assert tool.tool_call.tool_call_id == "uwpsprd2" + assert "555" in tool.tool_result.content + + def test_inference_spans(self): + """Inference spans extract messages and metadata from parent call_llm.""" + session = self.mapper.map_to_session(self.spans, SESSION_ID) + inf_spans = [s for s in session.traces[0].spans if isinstance(s, InferenceSpan)] + assert len(inf_spans) == 2 + first = inf_spans[0] + assert len(first.messages) >= 2 + assert first.metadata["model"] == "gemini-3.5-flash" + assert first.metadata["input_tokens"] == 157 + assert first.metadata["reasoning_tokens"] == 32 + assert first.metadata["invocation_id"] == "e-5742fbee-9bb9-4445-b224-7ae0041d6631" + + def test_tool_call_id_in_messages(self): + """tool_call_id is read from function_call.id in Gemini 3+ payloads.""" + session = self.mapper.map_to_session(self.spans, SESSION_ID) + inf_spans = [s for s in session.traces[0].spans if isinstance(s, InferenceSpan)] + first = inf_spans[0] + assistant_msgs = [m for m in first.messages if isinstance(m, AssistantMessage)] + tool_calls = [c for m in assistant_msgs for c in m.content if isinstance(c, ToolCallContent)] + assert tool_calls[0].tool_call_id == "uwpsprd2" + + +# ============================================================================ +# Tests: Tool Execution Span Conversion +# ============================================================================ + + +class TestToolExecutionSpan: + """Tests for tool execution span conversion.""" + + def setup_method(self): + self.mapper = ADKOtelSessionMapper() + + def test_basic_tool_span(self): + """Tool span with standard attributes is converted correctly.""" + spans = [ + make_span( + span_id="tool-1", + name="execute_tool calculator", + attributes={ + "gen_ai.operation.name": "execute_tool", + "gen_ai.tool.name": "calculator", + "gen_ai.tool.call.id": "call-123", + "gcp.vertex.agent.tool_call_args": '{"expression": "2+2"}', + "gcp.vertex.agent.tool_response": '{"result": "4"}', + }, + ) + ] + session = self.mapper.map_to_session(spans, SESSION_ID) + tool = session.traces[0].spans[0] + assert isinstance(tool, ToolExecutionSpan) + assert tool.tool_call.name == "calculator" + assert tool.tool_call.arguments == {"expression": "2+2"} + assert tool.tool_result.content == '{"result": "4"}' + assert tool.tool_result.error is None + + def test_missing_tool_name_skipped(self): + """Tool span without gen_ai.tool.name is skipped.""" + spans = [ + make_span( + name="execute_tool", + attributes={ + "gen_ai.operation.name": "execute_tool", + "gcp.vertex.agent.tool_call_args": "{}", + "gcp.vertex.agent.tool_response": "ok", + }, + ) + ] + session = self.mapper.map_to_session(spans, SESSION_ID) + assert session.traces == [] + + def test_metadata_populated(self): + """Tool span metadata includes description, type, and event_id.""" + spans = [ + make_span( + name="execute_tool calculator", + attributes={ + "gen_ai.operation.name": "execute_tool", + "gen_ai.tool.name": "calculator", + "gen_ai.tool.description": "Does math", + "gen_ai.tool.type": "FunctionTool", + "gen_ai.tool.call.id": "call-1", + "gcp.vertex.agent.tool_call_args": "{}", + "gcp.vertex.agent.tool_response": "result", + "gcp.vertex.agent.event_id": "evt-123", + }, + ) + ] + session = self.mapper.map_to_session(spans, SESSION_ID) + tool = session.traces[0].spans[0] + assert tool.metadata["description"] == "Does math" + assert tool.metadata["tool_type"] == "FunctionTool" + assert tool.metadata["event_id"] == "evt-123" + + def test_tool_error_preserved(self): + """Tool span with ERROR status populates ToolResult.error.""" + spans = [ + { + "trace_id": "trace-1", + "span_id": "tool-err", + "parent_span_id": None, + "name": "execute_tool failing_tool", + "start_time": 1700000000000000000, + "end_time": 1700000001000000000, + "attributes": { + "gen_ai.operation.name": "execute_tool", + "gen_ai.tool.name": "failing_tool", + "gen_ai.tool.call.id": "call-err", + "gcp.vertex.agent.tool_call_args": "{}", + "gcp.vertex.agent.tool_response": "", + "error.type": "MCP_TOOL_ERROR", + }, + "scope": {"name": "gcp.vertex.agent", "version": "2.5.0"}, + "status": {"code": "ERROR", "description": "Tool execution failed"}, + "span_events": [], + } + ] + session = self.mapper.map_to_session(spans, SESSION_ID) + tool = session.traces[0].spans[0] + assert tool.tool_result.error == "MCP_TOOL_ERROR" + + +# ============================================================================ +# Tests: Inference Span Conversion +# ============================================================================ + + +class TestInferenceSpan: + """Tests for inference span conversion.""" + + def setup_method(self): + self.mapper = ADKOtelSessionMapper() + + def test_simple_text_response(self): + """generate_content with parent call_llm produces inference span.""" + spans = [ + make_span( + span_id="callllm-1", + name="call_llm", + attributes={ + "gcp.vertex.agent.llm_request": make_llm_request(), + "gcp.vertex.agent.llm_response": make_llm_response_text("Hello!"), + }, + ), + make_span( + span_id="gen-1", + parent_span_id="callllm-1", + name="generate_content gemini-2.5-flash", + attributes={"gen_ai.operation.name": "generate_content", "gen_ai.request.model": "gemini-2.5-flash"}, + ), + ] + session = self.mapper.map_to_session(spans, SESSION_ID) + inf = [s for s in session.traces[0].spans if isinstance(s, InferenceSpan)] + assert len(inf) == 1 + assert len(inf[0].messages) == 2 + + def test_orphan_generate_content_skipped(self): + """generate_content without parent call_llm is skipped.""" + spans = [ + make_span( + name="generate_content gemini-2.5-flash", + attributes={"gen_ai.operation.name": "generate_content"}, + ) + ] + session = self.mapper.map_to_session(spans, SESSION_ID) + assert session.traces == [] + + def test_empty_llm_request_skipped(self): + """call_llm with empty llm_request ('{}') produces no inference span.""" + spans = [ + make_span( + span_id="callllm-1", + name="call_llm", + attributes={"gcp.vertex.agent.llm_request": "{}", "gcp.vertex.agent.llm_response": "{}"}, + ), + make_span( + span_id="gen-1", + parent_span_id="callllm-1", + name="generate_content gemini-2.5-flash", + attributes={"gen_ai.operation.name": "generate_content"}, + ), + ] + session = self.mapper.map_to_session(spans, SESSION_ID) + assert session.traces == [] + + def test_thought_parts_filtered(self): + """Thought parts (thought=True) are excluded from the assistant message.""" + llm_response = json.dumps( + { + "content": { + "parts": [ + {"thought": True, "text": "Let me think..."}, + {"text": "The answer is 42."}, + ], + "role": "model", + }, + "finish_reason": "STOP", + } + ) + spans = [ + make_span( + span_id="callllm-1", + name="call_llm", + attributes={ + "gcp.vertex.agent.llm_request": make_llm_request(user_text="Question?"), + "gcp.vertex.agent.llm_response": llm_response, + }, + ), + make_span( + span_id="gen-1", + parent_span_id="callllm-1", + name="generate_content gemini-3.5-flash", + attributes={"gen_ai.operation.name": "generate_content"}, + ), + ] + session = self.mapper.map_to_session(spans, SESSION_ID) + inf = [s for s in session.traces[0].spans if isinstance(s, InferenceSpan)] + assistant_msgs = [m for m in inf[0].messages if isinstance(m, AssistantMessage)] + assert len(assistant_msgs[0].content) == 1 + assert assistant_msgs[0].content[0].text == "The answer is 42." + + def test_tool_call_id_none_for_gemini2(self): + """tool_call_id is None when function_call lacks id field (Gemini 2.x).""" + history = [ + {"parts": [{"text": "What is 2+2?"}], "role": "user"}, + {"parts": [{"function_call": {"name": "calc", "args": {"x": "2+2"}}}], "role": "model"}, + {"parts": [{"function_response": {"name": "calc", "response": {"r": "4"}}}], "role": "user"}, + ] + spans = [ + make_span( + span_id="callllm-1", + name="call_llm", + attributes={ + "gcp.vertex.agent.llm_request": make_llm_request(history=history), + "gcp.vertex.agent.llm_response": make_llm_response_tool_call(call_id=None), + }, + ), + make_span( + span_id="gen-1", + parent_span_id="callllm-1", + name="generate_content gemini-2.5-flash", + attributes={"gen_ai.operation.name": "generate_content"}, + ), + ] + session = self.mapper.map_to_session(spans, SESSION_ID) + inf = [s for s in session.traces[0].spans if isinstance(s, InferenceSpan)] + assistant_msgs = [m for m in inf[0].messages if isinstance(m, AssistantMessage)] + tool_calls = [c for m in assistant_msgs for c in m.content if isinstance(c, ToolCallContent)] + assert all(tc.tool_call_id is None for tc in tool_calls) + + def test_multiple_tool_calls_in_single_response(self): + """LLM response with two function_calls produces two ToolCallContent items.""" + llm_response = json.dumps( + { + "content": { + "parts": [ + {"function_call": {"id": "c-1", "name": "get_weather", "args": {"city": "Seattle"}}}, + {"function_call": {"id": "c-2", "name": "get_weather", "args": {"city": "Tokyo"}}}, + ], + "role": "model", + }, + "finish_reason": "STOP", + } + ) + spans = [ + make_span( + span_id="callllm-1", + name="call_llm", + attributes={ + "gcp.vertex.agent.llm_request": make_llm_request(user_text="Weather?"), + "gcp.vertex.agent.llm_response": llm_response, + }, + ), + make_span( + span_id="gen-1", + parent_span_id="callllm-1", + name="generate_content gemini-3.5-flash", + attributes={"gen_ai.operation.name": "generate_content"}, + ), + ] + session = self.mapper.map_to_session(spans, SESSION_ID) + inf = [s for s in session.traces[0].spans if isinstance(s, InferenceSpan)] + assistant_msgs = [m for m in inf[0].messages if isinstance(m, AssistantMessage)] + tool_calls = [c for m in assistant_msgs for c in m.content if isinstance(c, ToolCallContent)] + assert len(tool_calls) == 2 + assert tool_calls[0].tool_call_id == "c-1" + assert tool_calls[1].tool_call_id == "c-2" + + +# ============================================================================ +# Tests: Agent Invocation Span Conversion +# ============================================================================ + + +class TestAgentInvocationSpan: + """Tests for agent invocation span conversion.""" + + def setup_method(self): + self.mapper = ADKOtelSessionMapper() + + def test_no_child_call_llm_skipped(self): + """invoke_agent without child call_llm spans is skipped.""" + spans = [ + make_span( + name="invoke_agent math_agent", + attributes={"gen_ai.operation.name": "invoke_agent", "gen_ai.agent.name": "math_agent"}, + ) + ] + session = self.mapper.map_to_session(spans, SESSION_ID) + assert session.traces == [] + + def test_with_single_call_llm(self): + """invoke_agent with one call_llm child extracts prompt, response, system prompt.""" + spans = [ + make_span( + span_id="agent-1", + name="invoke_agent x", + attributes={"gen_ai.operation.name": "invoke_agent", "gen_ai.agent.name": "x"}, + ), + make_span( + span_id="callllm-1", + parent_span_id="agent-1", + name="call_llm", + attributes={ + "gcp.vertex.agent.llm_request": make_llm_request( + user_text="Hello", system_instruction="Be helpful." + ), + "gcp.vertex.agent.llm_response": make_llm_response_text("Hi there!"), + }, + ), + ] + session = self.mapper.map_to_session(spans, SESSION_ID) + agent = [s for s in session.traces[0].spans if isinstance(s, AgentInvocationSpan)][0] + assert agent.user_prompt == "Hello" + assert agent.agent_response == "Hi there!" + assert agent.system_prompt == "Be helpful." + + def test_skip_summarization_fallback(self): + """When last call_llm has no text response, agent_response uses the last tool result.""" + llm_response_tool_only = json.dumps( + { + "content": { + "parts": [{"function_call": {"id": "fc-1", "name": "lookup", "args": {"q": "x"}}}], + "role": "model", + }, + "finish_reason": "STOP", + } + ) + spans = [ + make_span( + span_id="agent-1", + name="invoke_agent helper", + attributes={"gen_ai.operation.name": "invoke_agent", "gen_ai.agent.name": "helper"}, + ), + make_span( + span_id="callllm-1", + parent_span_id="agent-1", + name="call_llm", + attributes={ + "gcp.vertex.agent.llm_request": make_llm_request(user_text="Find it"), + "gcp.vertex.agent.llm_response": llm_response_tool_only, + }, + start_time=1700000001000000000, + end_time=1700000002000000000, + ), + make_span( + span_id="tool-1", + parent_span_id="callllm-1", + name="execute_tool lookup", + attributes={ + "gen_ai.operation.name": "execute_tool", + "gen_ai.tool.name": "lookup", + "gen_ai.tool.call.id": "fc-1", + "gcp.vertex.agent.tool_call_args": '{"q": "x"}', + "gcp.vertex.agent.tool_response": "THE ANSWER", + }, + start_time=1700000002000000000, + end_time=1700000003000000000, + ), + ] + session = self.mapper.map_to_session(spans, SESSION_ID) + agent = [s for s in session.traces[0].spans if isinstance(s, AgentInvocationSpan)][0] + assert agent.agent_response == "THE ANSWER" + assert agent.user_prompt == "Find it" + + def test_preamble_text_ignored_when_function_call_present(self): + """When response has both text and function_call, agent_response comes from tool result.""" + llm_response = json.dumps( + { + "content": { + "parts": [ + {"text": "I will check."}, + {"function_call": {"id": "fc-1", "name": "lookup", "args": {"q": "x"}}}, + ], + "role": "model", + }, + "finish_reason": "STOP", + } + ) + spans = [ + make_span( + span_id="agent-1", + name="invoke_agent helper", + attributes={"gen_ai.operation.name": "invoke_agent", "gen_ai.agent.name": "helper"}, + ), + make_span( + span_id="callllm-1", + parent_span_id="agent-1", + name="call_llm", + attributes={ + "gcp.vertex.agent.llm_request": make_llm_request(user_text="Answer?"), + "gcp.vertex.agent.llm_response": llm_response, + }, + start_time=1700000001000000000, + end_time=1700000002000000000, + ), + make_span( + span_id="tool-1", + parent_span_id="callllm-1", + name="execute_tool lookup", + attributes={ + "gen_ai.operation.name": "execute_tool", + "gen_ai.tool.name": "lookup", + "gen_ai.tool.call.id": "fc-1", + "gcp.vertex.agent.tool_call_args": '{"q": "x"}', + "gcp.vertex.agent.tool_response": '{"result": "A"}', + }, + start_time=1700000002000000000, + end_time=1700000003000000000, + ), + ] + session = self.mapper.map_to_session(spans, SESSION_ID) + agent = [s for s in session.traces[0].spans if isinstance(s, AgentInvocationSpan)][0] + assert agent.agent_response == '{"result": "A"}' + + def test_user_prompt_extracts_latest_message(self): + """In multi-turn conversations, user_prompt is the latest user message.""" + history = [ + {"parts": [{"text": "First question"}], "role": "user"}, + {"parts": [{"function_call": {"name": "calc", "args": {}}}], "role": "model"}, + {"parts": [{"function_response": {"name": "calc", "response": {}}}], "role": "user"}, + {"parts": [{"text": "Second question"}], "role": "user"}, + ] + spans = [ + make_span( + span_id="agent-1", + name="invoke_agent x", + attributes={"gen_ai.operation.name": "invoke_agent", "gen_ai.agent.name": "x"}, + ), + make_span( + span_id="callllm-1", + parent_span_id="agent-1", + name="call_llm", + attributes={ + "gcp.vertex.agent.llm_request": make_llm_request(), + "gcp.vertex.agent.llm_response": make_llm_response_tool_call(), + }, + start_time=1700000001000000000, + end_time=1700000002000000000, + ), + make_span( + span_id="callllm-2", + parent_span_id="agent-1", + name="call_llm", + attributes={ + "gcp.vertex.agent.llm_request": make_llm_request(history=history), + "gcp.vertex.agent.llm_response": make_llm_response_text("30"), + }, + start_time=1700000003000000000, + end_time=1700000004000000000, + ), + ] + session = self.mapper.map_to_session(spans, SESSION_ID) + agent = [s for s in session.traces[0].spans if isinstance(s, AgentInvocationSpan)][0] + assert agent.user_prompt == "Second question" + assert agent.agent_response == "30" + + +# ============================================================================ +# Tests: Session-Level Behavior +# ============================================================================ + + +class TestSessionBehavior: + """Tests for session-level grouping and edge cases.""" + + def setup_method(self): + self.mapper = ADKOtelSessionMapper() + + def test_empty_spans_returns_empty_session(self): + """Empty spans list produces empty session.""" + session = self.mapper.map_to_session([], SESSION_ID) + assert session.session_id == SESSION_ID + assert session.traces == [] + + def test_multiple_traces_grouped(self): + """Spans with different trace_ids grouped into separate traces.""" + spans = [ + make_span( + trace_id="trace-1", + span_id="agent-1", + name="invoke_agent a", + attributes={"gen_ai.operation.name": "invoke_agent", "gen_ai.agent.name": "a"}, + ), + make_span( + trace_id="trace-1", + span_id="callllm-1", + parent_span_id="agent-1", + name="call_llm", + attributes={ + "gcp.vertex.agent.llm_request": make_llm_request(user_text="Hi"), + "gcp.vertex.agent.llm_response": make_llm_response_text("Hello"), + }, + ), + make_span( + trace_id="trace-2", + span_id="agent-2", + name="invoke_agent b", + attributes={"gen_ai.operation.name": "invoke_agent", "gen_ai.agent.name": "b"}, + ), + make_span( + trace_id="trace-2", + span_id="callllm-2", + parent_span_id="agent-2", + name="call_llm", + attributes={ + "gcp.vertex.agent.llm_request": make_llm_request(user_text="Bye"), + "gcp.vertex.agent.llm_response": make_llm_response_text("Goodbye"), + }, + ), + ] + session = self.mapper.map_to_session(spans, SESSION_ID) + assert {t.trace_id for t in session.traces} == {"trace-1", "trace-2"} + + +# ============================================================================ +# Tests: Data Format Compatibility +# ============================================================================ + + +class TestDataFormatCompatibility: + """Tests for handling different span dict formats.""" + + def setup_method(self): + self.mapper = ADKOtelSessionMapper() + + def test_to_json_format_with_hex_prefix(self): + """Spans in to_json format (context.trace_id, parent_id with 0x prefix) are parsed.""" + spans = [ + { + "name": "execute_tool calculator", + "context": {"trace_id": "0xabc123", "span_id": "0xdef456"}, + "parent_id": "0x789abc", + "start_time": "2026-07-22T16:34:19.917561Z", + "end_time": "2026-07-22T16:34:19.917765Z", + "attributes": { + "gen_ai.operation.name": "execute_tool", + "gen_ai.tool.name": "calculator", + "gen_ai.tool.call.id": "call-1", + "gcp.vertex.agent.tool_call_args": '{"x": 1}', + "gcp.vertex.agent.tool_response": "result", + }, + } + ] + session = self.mapper.map_to_session(spans, SESSION_ID) + tool = session.traces[0].spans[0] + assert tool.span_info.trace_id == "abc123" + assert tool.span_info.span_id == "def456" + assert tool.span_info.parent_span_id == "789abc" + + +# ============================================================================ +# Tests: Timestamp Parsing +# ============================================================================ + + +class TestTimestampParsing: + """Tests for parse_timestamp handling various formats.""" + + def setup_method(self): + self.mapper = ADKOtelSessionMapper() + + def test_iso_string_with_z(self): + ts = self.mapper.parse_timestamp("2026-07-22T16:34:19.917561Z") + assert ts.year == 2026 and ts.month == 7 + + def test_nanosecond_epoch(self): + """Nanosecond epoch integer is converted to datetime.""" + ts_int = self.mapper.parse_timestamp(1700000000000000000) + assert ts_int.year == 2023 + + def test_string_nanosecond_epoch(self): + """String-encoded nanosecond epoch (OTLP JSON uint64) is correctly parsed.""" + ts_str = self.mapper.parse_timestamp("1700000000000000000") + assert ts_str.year == 2023 and ts_str.month == 11 + + def test_none_returns_now(self): + assert self.mapper.parse_timestamp(None) is not None + + def test_datetime_passthrough(self): + dt = datetime(2024, 1, 1, tzinfo=timezone.utc) + assert self.mapper.parse_timestamp(dt) == dt + + +# ============================================================================ +# Tests: Error Handling +# ============================================================================ + + +class TestErrorHandling: + """Tests for graceful error handling.""" + + def setup_method(self): + self.mapper = ADKOtelSessionMapper() + + def test_malformed_llm_request_json(self): + """Malformed JSON in llm_request does not crash the mapper.""" + spans = [ + make_span( + span_id="agent-1", + name="invoke_agent x", + attributes={"gen_ai.operation.name": "invoke_agent", "gen_ai.agent.name": "x"}, + ), + make_span( + span_id="callllm-1", + parent_span_id="agent-1", + name="call_llm", + attributes={ + "gcp.vertex.agent.llm_request": "not-valid-json{{{", + "gcp.vertex.agent.llm_response": "also-invalid", + }, + ), + ] + session = self.mapper.map_to_session(spans, SESSION_ID) + assert session.traces == [] + + def test_missing_attributes_key(self): + """Span without attributes key does not crash.""" + spans = [{"trace_id": "t1", "span_id": "s1", "name": "test"}] + session = self.mapper.map_to_session(spans, SESSION_ID) + assert session.traces == [] or session.traces[0].spans == [] + + def test_tool_call_args_as_dict(self): + """tool_call_args provided as dict (not JSON string) is handled.""" + spans = [ + make_span( + name="execute_tool calc", + attributes={ + "gen_ai.operation.name": "execute_tool", + "gen_ai.tool.name": "calc", + "gen_ai.tool.call.id": "c1", + "gcp.vertex.agent.tool_call_args": {"x": 42}, + "gcp.vertex.agent.tool_response": "result", + }, + ) + ] + session = self.mapper.map_to_session(spans, SESSION_ID) + assert session.traces[0].spans[0].tool_call.arguments == {"x": 42} + + +# ============================================================================ +# Tests: Multi-Agent Trace Splitting +# ============================================================================ + + +class TestMultiAgentSplitting: + """Tests for per-agent trace splitting in multi-agent scenarios.""" + + def setup_method(self): + self.mapper = ADKOtelSessionMapper() + + def test_coordinator_specialist_produces_separate_traces(self): + """Two invoke_agent spans in one trace produce one Trace per agent with correct tools.""" + spans = [ + # Coordinator agent + make_span( + span_id="coordinator", + name="invoke_agent coordinator", + attributes={"gen_ai.operation.name": "invoke_agent", "gen_ai.agent.name": "coordinator"}, + start_time=1700000001000000000, + end_time=1700000009000000000, + ), + make_span( + span_id="coord-callllm", + parent_span_id="coordinator", + name="call_llm", + attributes={ + "gcp.vertex.agent.llm_request": make_llm_request( + user_text="Book a flight", + system_instruction="You coordinate tasks.", + tools=[{"name": "delegate", "description": "Delegate to specialist"}], + ), + "gcp.vertex.agent.llm_response": make_llm_response_tool_call( + name="delegate", args={"task": "book"}, call_id="fc-coord" + ), + }, + start_time=1700000002000000000, + end_time=1700000003000000000, + ), + # Specialist agent (nested under coordinator's call_llm) + make_span( + span_id="specialist", + parent_span_id="coord-callllm", + name="invoke_agent specialist", + attributes={"gen_ai.operation.name": "invoke_agent", "gen_ai.agent.name": "specialist"}, + start_time=1700000004000000000, + end_time=1700000008000000000, + ), + make_span( + span_id="spec-callllm", + parent_span_id="specialist", + name="call_llm", + attributes={ + "gcp.vertex.agent.llm_request": make_llm_request( + user_text="Book a flight", + system_instruction="You book flights.", + tools=[{"name": "book_flight", "description": "Book a flight"}], + ), + "gcp.vertex.agent.llm_response": make_llm_response_text("Booked seat 4A."), + }, + start_time=1700000005000000000, + end_time=1700000006000000000, + ), + make_span( + span_id="spec-tool", + parent_span_id="spec-callllm", + name="execute_tool book_flight", + attributes={ + "gen_ai.operation.name": "execute_tool", + "gen_ai.tool.name": "book_flight", + "gen_ai.tool.call.id": "fc-spec", + "gcp.vertex.agent.tool_call_args": '{"destination": "NYC"}', + "gcp.vertex.agent.tool_response": '{"result": "SPECIALIST_TOOL_OUTPUT"}', + }, + start_time=1700000006000000000, + end_time=1700000007000000000, + ), + ] + + session = self.mapper.map_to_session(spans, SESSION_ID) + + # Should produce two traces — one per agent + assert len(session.traces) == 2 + + # Find each agent's trace + coord_trace = next( + t + for t in session.traces + if any( + isinstance(s, AgentInvocationSpan) and s.metadata.get("agent_name") == "coordinator" for s in t.spans + ) + ) + spec_trace = next( + t + for t in session.traces + if any(isinstance(s, AgentInvocationSpan) and s.metadata.get("agent_name") == "specialist" for s in t.spans) + ) + + # Coordinator should have its own tools, not the specialist's + coord_agent = [s for s in coord_trace.spans if isinstance(s, AgentInvocationSpan)][0] + assert coord_agent.available_tools[0].name == "delegate" + # Coordinator should NOT leak the specialist's tool response + assert "SPECIALIST_TOOL_OUTPUT" not in coord_agent.agent_response + + # Specialist should have its own tools and response + spec_agent = [s for s in spec_trace.spans if isinstance(s, AgentInvocationSpan)][0] + assert spec_agent.available_tools[0].name == "book_flight" + assert spec_agent.agent_response == "Booked seat 4A." + + # Tool execution span should only appear in the specialist's trace + coord_tools = [s for s in coord_trace.spans if isinstance(s, ToolExecutionSpan)] + spec_tools = [s for s in spec_trace.spans if isinstance(s, ToolExecutionSpan)] + assert len(coord_tools) == 0 + assert len(spec_tools) == 1 + assert spec_tools[0].tool_call.name == "book_flight" + + def test_unclaimed_spans_assigned_to_earliest_agent(self): + """Spans not nested under any invoke_agent go to the earliest agent's trace.""" + spans = [ + # Orphan tool span — not parented to either agent + make_span( + span_id="orphan-tool", + name="execute_tool audit", + attributes={ + "gen_ai.operation.name": "execute_tool", + "gen_ai.tool.name": "audit", + "gen_ai.tool.call.id": "fc-audit", + "gcp.vertex.agent.tool_call_args": "{}", + "gcp.vertex.agent.tool_response": "audited", + }, + start_time=1700000001000000000, + end_time=1700000002000000000, + ), + # Two minimal agent spans (each needs a call_llm child to be non-empty) + make_span( + span_id="agent-1", + name="invoke_agent alpha", + attributes={"gen_ai.operation.name": "invoke_agent", "gen_ai.agent.name": "alpha"}, + start_time=1700000003000000000, + end_time=1700000005000000000, + ), + make_span( + span_id="callllm-1", + parent_span_id="agent-1", + name="call_llm", + attributes={ + "gcp.vertex.agent.llm_request": make_llm_request(user_text="Hello"), + "gcp.vertex.agent.llm_response": make_llm_response_text("Hi"), + }, + start_time=1700000004000000000, + end_time=1700000005000000000, + ), + make_span( + span_id="agent-2", + name="invoke_agent beta", + attributes={"gen_ai.operation.name": "invoke_agent", "gen_ai.agent.name": "beta"}, + start_time=1700000006000000000, + end_time=1700000008000000000, + ), + make_span( + span_id="callllm-2", + parent_span_id="agent-2", + name="call_llm", + attributes={ + "gcp.vertex.agent.llm_request": make_llm_request(user_text="Bye"), + "gcp.vertex.agent.llm_response": make_llm_response_text("Goodbye"), + }, + start_time=1700000007000000000, + end_time=1700000008000000000, + ), + ] + + session = self.mapper.map_to_session(spans, SESSION_ID) + assert len(session.traces) == 2 + + # The orphan tool should appear in the earliest agent's trace (alpha) + alpha_trace = next( + t + for t in session.traces + if any(isinstance(s, AgentInvocationSpan) and s.metadata.get("agent_name") == "alpha" for s in t.spans) + ) + beta_trace = next( + t + for t in session.traces + if any(isinstance(s, AgentInvocationSpan) and s.metadata.get("agent_name") == "beta" for s in t.spans) + ) + + alpha_tools = [s for s in alpha_trace.spans if isinstance(s, ToolExecutionSpan)] + beta_tools = [s for s in beta_trace.spans if isinstance(s, ToolExecutionSpan)] + + assert len(alpha_tools) == 1 + assert alpha_tools[0].tool_call.name == "audit" + assert len(beta_tools) == 0 + + +# ============================================================================ +# Tests: Scope Filtering +# ============================================================================ + + +class TestScopeFiltering: + """Tests for instrumentation scope filtering.""" + + def setup_method(self): + self.mapper = ADKOtelSessionMapper() + + def test_foreign_scope_spans_are_dropped(self): + """Spans from non-ADK instrumentation scopes are excluded from the session. + + Uses a foreign-scope execute_tool span that would convert unconditionally + if not filtered, ensuring the scope check is the only barrier. + """ + spans = [ + # ADK span — should be kept + make_span( + span_id="tool-1", + name="execute_tool calculator", + attributes={ + "gen_ai.operation.name": "execute_tool", + "gen_ai.tool.name": "calculator", + "gen_ai.tool.call.id": "c1", + "gcp.vertex.agent.tool_call_args": '{"x": 1}', + "gcp.vertex.agent.tool_response": "2", + }, + ), + # Foreign scope span with full tool attributes — would convert if not filtered + { + "trace_id": "trace-1", + "span_id": "foreign-1", + "parent_span_id": None, + "name": "execute_tool foreign_tool", + "start_time": 1700000000000000000, + "end_time": 1700000001000000000, + "attributes": { + "gen_ai.operation.name": "execute_tool", + "gen_ai.tool.name": "foreign_tool", + "gen_ai.tool.call.id": "c2", + "gcp.vertex.agent.tool_call_args": '{"q": "test"}', + "gcp.vertex.agent.tool_response": "foreign result", + }, + "scope": {"name": "opentelemetry.instrumentation.vertexai", "version": "1.0.0"}, + "status": {"code": "UNSET"}, + "span_events": [], + }, + ] + + session = self.mapper.map_to_session(spans, SESSION_ID) + + # Only the ADK tool span should survive + assert len(session.traces) == 1 + assert len(session.traces[0].spans) == 1 + assert isinstance(session.traces[0].spans[0], ToolExecutionSpan) + assert session.traces[0].spans[0].tool_call.name == "calculator"