diff --git a/src/agentevals/api/otlp_processing.py b/src/agentevals/api/otlp_processing.py index 948e31b..5aefedd 100644 --- a/src/agentevals/api/otlp_processing.py +++ b/src/agentevals/api/otlp_processing.py @@ -15,7 +15,10 @@ ) from ..extraction import flatten_otlp_attributes +from ..otlp_anyvalue import decode_any_value from ..trace_attrs import ( + AGENTEVALS_EVAL_SET_ID, + AGENTEVALS_SESSION_NAME, OTEL_GENAI_CONVERSATION_ID, OTEL_GENAI_INPUT_MESSAGES, OTEL_GENAI_OUTPUT_MESSAGES, @@ -30,9 +33,6 @@ logger = logging.getLogger(__name__) -AGENTEVALS_EVAL_SET_ID = "agentevals.eval_set_id" -AGENTEVALS_SESSION_NAME = "agentevals.session_name" - async def process_traces(body: dict, manager: StreamingTraceManager) -> None: """Parse ExportTraceServiceRequest and feed spans to the pipeline.""" @@ -242,7 +242,13 @@ def _normalize_span(span_data: dict, scope_name: str, scope_version: str, schema def _extract_agentevals_metadata(resource_attrs: list[dict]) -> dict: - """Extract agentevals-specific metadata from OTLP resource attributes.""" + """Extract agentevals-specific metadata from OTLP resource attributes. + + ``eval_set_id`` and ``session_name`` are used as dict keys downstream + (``_active_session_for_name``) and typed ``str | None`` on the session + models. Neither is in ``SPEC_CONTAINER_ATTRS``, so the shared decoder never + hands them a list or dict in the first place. + """ flat = flatten_otlp_attributes(resource_attrs) return { "eval_set_id": flat.get(AGENTEVALS_EVAL_SET_ID), @@ -310,37 +316,12 @@ def _convert_otlp_log_record(log_record: dict) -> dict | None: return result -def _parse_otlp_any_value(value_obj: dict): - """Recursively parse an OTLP AnyValue to native Python types. - - Handles the full AnyValue union: stringValue, intValue, doubleValue, - boolValue, kvlistValue (→ dict), arrayValue (→ list), bytesValue. - """ - if "stringValue" in value_obj: - return value_obj["stringValue"] - if "intValue" in value_obj: - return int(value_obj["intValue"]) - if "doubleValue" in value_obj: - return float(value_obj["doubleValue"]) - if "boolValue" in value_obj: - return value_obj["boolValue"] - if "kvlistValue" in value_obj: - kv = value_obj["kvlistValue"] - return {item.get("key", ""): _parse_otlp_any_value(item.get("value", {})) for item in kv.get("values", [])} - if "arrayValue" in value_obj: - arr = value_obj["arrayValue"] - return [_parse_otlp_any_value(v) for v in arr.get("values", [])] - if "bytesValue" in value_obj: - return value_obj["bytesValue"] - return value_obj - - def _parse_otlp_body(body_raw: dict) -> dict | str: """Parse OTLP log record body value. Top-level stringValue bodies are JSON-decoded (Strands-style logs store message content as JSON strings). All other AnyValue types are parsed - recursively via ``_parse_otlp_any_value`` (handles the nested kvlistValue / + recursively via ``decode_any_value`` (handles the nested kvlistValue / arrayValue structures used by the OpenAI instrumentor). """ if "stringValue" in body_raw: @@ -351,4 +332,4 @@ def _parse_otlp_body(body_raw: dict) -> dict | str: return json.loads(raw) except (json.JSONDecodeError, TypeError): return raw - return _parse_otlp_any_value(body_raw) + return decode_any_value(body_raw) diff --git a/src/agentevals/extraction.py b/src/agentevals/extraction.py index 141f230..efcc9d6 100644 --- a/src/agentevals/extraction.py +++ b/src/agentevals/extraction.py @@ -17,6 +17,7 @@ from typing import Any, Protocol, TypedDict, TypeVar from .loader.base import Span, Trace +from .otlp_anyvalue import decode_attributes from .trace_attrs import ( ADK_LLM_REQUEST, ADK_LLM_RESPONSE, @@ -523,20 +524,12 @@ def is_invocation_span(span: Span) -> bool: def flatten_otlp_attributes(attrs_list: list[dict]) -> dict[str, Any]: - """Convert OTLP attributes array [{key, value: {stringValue|...}}] to flat dict.""" - result: dict[str, Any] = {} - for attr in attrs_list: - key = attr.get("key", "") - value_obj = attr.get("value", {}) - if "stringValue" in value_obj: - result[key] = value_obj["stringValue"] - elif "intValue" in value_obj: - result[key] = int(value_obj["intValue"]) - elif "doubleValue" in value_obj: - result[key] = float(value_obj["doubleValue"]) - elif "boolValue" in value_obj: - result[key] = value_obj["boolValue"] - return result + """Convert OTLP attributes array [{key, value: {stringValue|...}}] to flat dict. + + Delegates to the shared ``AnyValue`` decoder so array/kvlist/bytes + attributes survive instead of being dropped. + """ + return decode_attributes(attrs_list) # --------------------------------------------------------------------------- diff --git a/src/agentevals/loader/otlp.py b/src/agentevals/loader/otlp.py index ef26cb2..de680fd 100644 --- a/src/agentevals/loader/otlp.py +++ b/src/agentevals/loader/otlp.py @@ -5,6 +5,7 @@ import json import logging +from ..otlp_anyvalue import decode_attribute, decode_attributes, is_any_value from ..trace_attrs import ( OTEL_GENAI_INPUT_MESSAGES, OTEL_GENAI_OUTPUT_MESSAGES, @@ -175,8 +176,10 @@ def _promote_genai_event_attributes(self, span_data: dict, attributes: dict) -> key = attr.get("key", "") if key in self._GENAI_EVENT_KEYS and key not in attributes: value_obj = attr.get("value", {}) - if "stringValue" in value_obj: - attributes[key] = value_obj["stringValue"] + if is_any_value(value_obj): + keep, value = decode_attribute(key, value_obj) + if keep: + attributes[key] = value def _extract_attributes(self, attrs) -> dict: """Convert attributes to a flat ``{key: value}`` dict. @@ -192,25 +195,7 @@ def _extract_attributes(self, attrs) -> dict: if isinstance(attrs, dict): return self._flatten_nested_dict(attrs) - result = {} - for attr in attrs: - key = attr.get("key", "") - value_obj = attr.get("value", {}) - - if "stringValue" in value_obj: - result[key] = value_obj["stringValue"] - elif "intValue" in value_obj: - result[key] = int(value_obj["intValue"]) - elif "doubleValue" in value_obj: - result[key] = float(value_obj["doubleValue"]) - elif "boolValue" in value_obj: - result[key] = value_obj["boolValue"] - elif "arrayValue" in value_obj: - result[key] = json.dumps(value_obj["arrayValue"]) - elif "kvlistValue" in value_obj: - result[key] = json.dumps(value_obj["kvlistValue"]) - - return result + return decode_attributes(attrs) @staticmethod def _flatten_nested_dict(d: dict, prefix: str = "") -> dict: diff --git a/src/agentevals/otlp_anyvalue.py b/src/agentevals/otlp_anyvalue.py new file mode 100644 index 0000000..923a6a7 --- /dev/null +++ b/src/agentevals/otlp_anyvalue.py @@ -0,0 +1,118 @@ +"""Shared decoder for the OTLP ``AnyValue`` union. + +OTLP encodes every attribute value, log body and nested element as an +``AnyValue``: a one-of wrapper such as ``{"stringValue": "chat"}`` or +``{"arrayValue": {"values": [...]}}``. The protobuf receiver (via +``MessageToDict``) and OTLP/JSON payloads deliver that same dict shape, so +every consumer needs identical decoding rules. + +This module depends only on the standard library and ``trace_attrs`` (a leaf +constants module), so ``extraction``, ``loader.otlp`` and ``api.otlp_processing`` +can all use it without creating an import cycle. +""" + +from __future__ import annotations + +import logging +from typing import Any + +from .trace_attrs import SPEC_CONTAINER_ATTRS + +logger = logging.getLogger(__name__) + +ANY_VALUE_FIELDS = ( + "stringValue", + "intValue", + "doubleValue", + "boolValue", + "kvlistValue", + "arrayValue", + "bytesValue", +) + + +def decode_any_value(value_obj: dict) -> Any: + """Recursively decode an OTLP ``AnyValue`` to a native Python value. + + Handles the full union: stringValue, intValue (OTLP sends it as a + string), doubleValue, boolValue, kvlistValue (→ dict), arrayValue + (→ list), bytesValue. + + ``bytesValue`` is returned unchanged. ``MessageToDict`` base64-encodes + protobuf bytes fields and OTLP/JSON does the same, so callers already + receive a str; decoding it here would change the value they see today. + + A value carrying none of the union fields is returned as-is. + """ + if "stringValue" in value_obj: + return value_obj["stringValue"] + if "intValue" in value_obj: + return int(value_obj["intValue"]) + if "doubleValue" in value_obj: + return float(value_obj["doubleValue"]) + if "boolValue" in value_obj: + return value_obj["boolValue"] + if "kvlistValue" in value_obj: + kv = value_obj["kvlistValue"] + return {item.get("key", ""): decode_any_value(item.get("value", {})) for item in kv.get("values", [])} + if "arrayValue" in value_obj: + arr = value_obj["arrayValue"] + return [decode_any_value(v) for v in arr.get("values", [])] + if "bytesValue" in value_obj: + return value_obj["bytesValue"] + return value_obj + + +def is_any_value(value_obj: dict) -> bool: + """Return True when *value_obj* carries one of the ``AnyValue`` fields.""" + for field in ANY_VALUE_FIELDS: + if field in value_obj: + return True + return False + + +def decode_attribute(key: str, value_obj: dict) -> tuple[bool, Any]: + """Decode one attribute, applying the container allowlist. + + Returns ``(keep, value)``. Scalars are always kept. A list or dict is kept + only when *key* is in :data:`SPEC_CONTAINER_ATTRS`; otherwise it is dropped, + which is what ``extraction.py`` did with containers before this decoder was + shared. + + Dropping rather than serialising is deliberate: JSON-dumping the value would + put a blob back into user-visible output, which is the symptom #173 is + about. What the default *should* be is tracked in #208. + """ + value = decode_any_value(value_obj) + if isinstance(value, (list, dict)) and key not in SPEC_CONTAINER_ATTRS: + logger.warning( + "Dropping container value for %s (got %s); only spec container attributes are kept", + key, + type(value).__name__, + ) + return False, None + return True, value + + +def decode_attributes(attrs_list: list[dict]) -> dict[str, Any]: + """Decode an OTLP attributes array to a flat ``{key: value}`` dict. + + Entries whose value carries no ``AnyValue`` field are skipped, matching the + behaviour every call site had before they shared this decoder. + + Container values survive only for the keys in + :data:`~agentevals.trace_attrs.SPEC_CONTAINER_ATTRS`. Keeping the allowlist + here rather than narrowing per consumer means an attribute nobody has + thought about cannot become an unhashable dict key downstream - there is + nothing to remember, because it was never widened in the first place. + """ + result: dict[str, Any] = {} + for attr in attrs_list: + value_obj = attr.get("value", {}) + if not is_any_value(value_obj): + continue + key = attr.get("key", "") + keep, value = decode_attribute(key, value_obj) + if keep: + result[key] = value + return result diff --git a/src/agentevals/trace_attrs.py b/src/agentevals/trace_attrs.py index e4cd16a..bb430e8 100644 --- a/src/agentevals/trace_attrs.py +++ b/src/agentevals/trace_attrs.py @@ -84,3 +84,33 @@ # agentevals custom attributes (repository-specific, outside OTel semconv) AGENTEVALS_SESSION_ID = "agentevals.session_id" +AGENTEVALS_EVAL_SET_ID = "agentevals.eval_set_id" +AGENTEVALS_SESSION_NAME = "agentevals.session_name" + +# Attributes the GenAI semconv types as arrays or structured values. These are +# the only keys the shared AnyValue decoder returns as a native list or dict; +# every other key is decoded to a scalar or dropped, which is what +# ``extraction.py`` did before the decoder was shared. +# +# The set is an allowlist on purpose. A missing entry drops one value, matching +# the pre-existing extraction behaviour; a denylist would instead let an +# unhashable value reach a dict key or set member on an unauthenticated +# receiver port, which crashes ingestion. See the discussion on #187. +# +# Hand-maintained: opentelemetry-semantic-conventions exposes constants and +# prose docstrings only, so attribute value types are not machine-readable and +# this set cannot be derived from the package. +# +# What should happen to container values on keys outside this set is not +# settled; today they are dropped. Tracked in #208. +SPEC_CONTAINER_ATTRS: frozenset[str] = frozenset( + { + OTEL_GENAI_RESPONSE_FINISH_REASONS, + OTEL_GENAI_INPUT_MESSAGES, + OTEL_GENAI_OUTPUT_MESSAGES, + OTEL_GENAI_TOOL_DEFINITIONS, + OTEL_GENAI_SYSTEM_INSTRUCTIONS, + OTEL_GENAI_TOOL_CALL_ARGUMENTS, + OTEL_GENAI_TOOL_CALL_RESULT, + } +) diff --git a/tests/test_extraction.py b/tests/test_extraction.py index 96544f7..8733262 100644 --- a/tests/test_extraction.py +++ b/tests/test_extraction.py @@ -6,6 +6,7 @@ import pytest +from agentevals import trace_attrs from agentevals.extraction import ( AdkExtractor, GenAIExtractor, @@ -400,6 +401,129 @@ def test_mixed_types(self): ) assert result == {"str": "hello", "num": 3.14, "flag": True} + def test_array_value(self): + result = flatten_otlp_attributes( + [ + { + "key": "gen_ai.response.finish_reasons", + "value": {"arrayValue": {"values": [{"stringValue": "stop"}]}}, + }, + ] + ) + assert result == {"gen_ai.response.finish_reasons": ["stop"]} + + def test_kvlist_value(self): + result = flatten_otlp_attributes( + [ + { + "key": "gen_ai.tool.call.arguments", + "value": { + "kvlistValue": { + "values": [ + {"key": "city", "value": {"stringValue": "Berlin"}}, + {"key": "metric", "value": {"boolValue": False}}, + ] + } + }, + }, + ] + ) + assert result == {"gen_ai.tool.call.arguments": {"city": "Berlin", "metric": False}} + + def test_array_of_kvlist(self): + """Messages arrive as an arrayValue of kvlistValue.""" + result = flatten_otlp_attributes( + [ + { + "key": "gen_ai.input.messages", + "value": { + "arrayValue": { + "values": [ + {"kvlistValue": {"values": [{"key": "role", "value": {"stringValue": "user"}}]}}, + ] + } + }, + }, + ] + ) + assert result == {"gen_ai.input.messages": [{"role": "user"}]} + + def test_finish_reasons_survive_to_extracted_model_info(self): + """The symptom #173 names: gen_ai.response.finish_reasons reaching the + consumer as ["stop"] rather than a literal blob or nothing. + + Asserting through extract_extended_model_info_from_attrs rather than at + the decoder keeps the whole path covered - decoding it correctly is not + the same as it arriving correctly. + """ + attrs = flatten_otlp_attributes( + [ + { + "key": "gen_ai.response.finish_reasons", + "value": {"arrayValue": {"values": [{"stringValue": "stop"}]}}, + }, + {"key": "gen_ai.response.model", "value": {"stringValue": "claude-opus-5"}}, + ] + ) + info = extract_extended_model_info_from_attrs(attrs) + assert info["finish_reasons"] == ["stop"] + assert info["response_model"] == "claude-opus-5" + + def test_multiple_finish_reasons_survive(self): + attrs = flatten_otlp_attributes( + [ + { + "key": "gen_ai.response.finish_reasons", + "value": {"arrayValue": {"values": [{"stringValue": "stop"}, {"stringValue": "length"}]}}, + } + ] + ) + assert extract_extended_model_info_from_attrs(attrs)["finish_reasons"] == [ + "stop", + "length", + ] + + def test_unlisted_key_drops_container_value(self): + """Containers survive only for SPEC_CONTAINER_ATTRS. Everything else is + dropped, which is what extraction did before the decoder was shared.""" + attrs = flatten_otlp_attributes( + [ + { + "key": "gen_ai.response.model", + "value": {"arrayValue": {"values": [{"stringValue": "claude-opus-5"}]}}, + }, + {"key": "gen_ai.request.model", "value": {"stringValue": "claude-sonnet-5"}}, + ] + ) + assert "gen_ai.response.model" not in attrs + info = extract_extended_model_info_from_attrs(attrs) + assert info["response_model"] is None + assert info["request_model"] == "claude-sonnet-5" + + def test_no_unlisted_key_can_yield_an_unhashable_value(self): + """The property the allowlist exists for: nothing outside + SPEC_CONTAINER_ATTRS can reach a consumer as a dict key or set member + and raise TypeError. Covers every attribute constant we declare, so a + new one cannot quietly reopen the hazard.""" + container = {"arrayValue": {"values": [{"stringValue": "x"}]}} + for name in dir(trace_attrs): + if not name.isupper(): + continue + key = getattr(trace_attrs, name) + if not isinstance(key, str): + continue + value = flatten_otlp_attributes([{"key": key, "value": container}]).get(key) + if key in trace_attrs.SPEC_CONTAINER_ATTRS: + assert value == ["x"], f"{key} should keep its container" + else: + assert value is None, f"{key} leaked a container" + hash(value) + + def test_bytes_value(self): + """MessageToDict base64-encodes bytes fields, so the decoder sees a str.""" + result = flatten_otlp_attributes([{"key": "payload", "value": {"bytesValue": "AP9oaQ=="}}]) + assert result == {"payload": "AP9oaQ=="} + def test_empty(self): assert flatten_otlp_attributes([]) == {} diff --git a/tests/test_otlp_loader.py b/tests/test_otlp_loader.py index c3a3428..2b2f27e 100644 --- a/tests/test_otlp_loader.py +++ b/tests/test_otlp_loader.py @@ -7,6 +7,7 @@ import pytest +from agentevals.extraction import extract_user_text_from_attrs from agentevals.loader.otlp import OtlpJsonLoader @@ -258,6 +259,161 @@ def test_load_from_dict_empty_resource_spans(self): assert traces == [] +class TestAnyValueAttributes: + """Attributes carrying the full OTLP AnyValue union (array / kvlist / bytes).""" + + @staticmethod + def _load_span_with(attribute): + loader = OtlpJsonLoader() + data = { + "resourceSpans": [ + { + "resource": {"attributes": []}, + "scopeSpans": [ + { + "scope": {"name": "test-scope"}, + "spans": [ + { + "traceId": "t1", + "spanId": "s1", + "name": "test", + "startTimeUnixNano": "1000000000", + "endTimeUnixNano": "2000000000", + "attributes": [attribute], + } + ], + } + ], + } + ], + } + return loader.load_from_dict(data)[0].all_spans[0] + + def test_array_value(self): + span = self._load_span_with( + { + "key": "gen_ai.response.finish_reasons", + "value": {"arrayValue": {"values": [{"stringValue": "stop"}]}}, + } + ) + assert span.tags["gen_ai.response.finish_reasons"] == ["stop"] + + def test_kvlist_value(self): + span = self._load_span_with( + { + "key": "gen_ai.tool.call.arguments", + "value": { + "kvlistValue": { + "values": [ + {"key": "temperature", "value": {"doubleValue": 0.7}}, + {"key": "stream", "value": {"boolValue": False}}, + ] + } + }, + } + ) + assert span.tags["gen_ai.tool.call.arguments"] == {"temperature": 0.7, "stream": False} + + def test_bytes_value(self): + span = self._load_span_with({"key": "payload", "value": {"bytesValue": "AP9oaQ=="}}) + assert span.tags["payload"] == "AP9oaQ==" + + @staticmethod + def _load_span_with_event_attribute(attribute): + """Span carrying a GenAI event attribute in OTLP array format.""" + loader = OtlpJsonLoader() + data = { + "resourceSpans": [ + { + "resource": {"attributes": []}, + "scopeSpans": [ + { + "scope": {"name": "test-scope"}, + "spans": [ + { + "traceId": "t1", + "spanId": "s1", + "name": "chat", + "startTimeUnixNano": "0", + "endTimeUnixNano": "0", + "attributes": [], + "events": [ + { + "timeUnixNano": "0", + "name": "gen_ai.client.inference.operation.details", + "attributes": [attribute], + } + ], + } + ], + } + ], + } + ], + } + return loader.load_from_dict(data)[0].all_spans[0] + + def test_event_promotion_decodes_array_value(self): + """Strands stores messages in span events, and newer GenAI semconv makes + them a complex array. Promotion must decode it instead of dropping it.""" + span = self._load_span_with_event_attribute( + { + "key": "gen_ai.input.messages", + "value": { + "arrayValue": { + "values": [ + { + "kvlistValue": { + "values": [ + {"key": "role", "value": {"stringValue": "user"}}, + {"key": "content", "value": {"stringValue": "Hello"}}, + ] + } + } + ] + } + }, + } + ) + assert span.tags["gen_ai.input.messages"] == [{"role": "user", "content": "Hello"}] + + def test_promoted_array_messages_reach_the_consumer(self): + """Past span.tags: a complex-array gen_ai.input.messages promoted out of + a span event must still yield the user text downstream, which is what a + consumer actually reads.""" + span = self._load_span_with_event_attribute( + { + "key": "gen_ai.input.messages", + "value": { + "arrayValue": { + "values": [ + { + "kvlistValue": { + "values": [ + {"key": "role", "value": {"stringValue": "user"}}, + { + "key": "content", + "value": {"stringValue": "What is the weather?"}, + }, + ] + } + } + ] + } + }, + } + ) + assert extract_user_text_from_attrs(span.tags) == "What is the weather?" + + def test_event_promotion_keeps_string_value(self): + """The pre-existing stringValue path must keep working unchanged.""" + messages_json = '[{"role": "user", "content": "Hello"}]' + span = self._load_span_with_event_attribute( + {"key": "gen_ai.output.messages", "value": {"stringValue": messages_json}} + ) + assert span.tags["gen_ai.output.messages"] == messages_json + + class TestFlatDictAttributes: """Tests for flat dict attribute format (e.g. from simplified producers).""" diff --git a/tests/test_otlp_receiver.py b/tests/test_otlp_receiver.py index 629e9dc..ae8cc56 100644 --- a/tests/test_otlp_receiver.py +++ b/tests/test_otlp_receiver.py @@ -233,6 +233,60 @@ def test_missing_keys_are_none(self): assert meta["session_name"] is None assert meta["service_name"] is None + def test_non_string_session_name_is_ignored(self): + """session_name is used as a dict key in _active_session_for_name. + + The shared AnyValue decoder can now return lists and dicts, which are + unhashable; reading stringValue only keeps them out of the key path. + """ + attrs = [ + { + "key": "agentevals.session_name", + "value": {"arrayValue": {"values": [{"stringValue": "run-42"}]}}, + } + ] + meta = _extract_agentevals_metadata(attrs) + assert meta["session_name"] is None + # Would raise TypeError: unhashable type if a list leaked through. + assert {}.get(meta["session_name"]) is None + + def test_non_string_eval_set_id_is_ignored(self): + """eval_set_id is typed ``str | None`` on the session models.""" + attrs = [ + { + "key": "agentevals.eval_set_id", + "value": {"kvlistValue": {"values": [{"key": "id", "value": {"stringValue": "my-eval"}}]}}, + } + ] + meta = _extract_agentevals_metadata(attrs) + assert meta["eval_set_id"] is None + + def test_non_string_values_are_dropped_from_resource_attrs(self): + """agentevals.session_name is not in SPEC_CONTAINER_ATTRS, so a container + never lands in resource_attrs and can never reach a dict key.""" + attrs = [ + { + "key": "agentevals.session_name", + "value": {"arrayValue": {"values": [{"stringValue": "run-42"}]}}, + }, + _make_otlp_attr("service.name", "test-agent"), + ] + meta = _extract_agentevals_metadata(attrs) + assert "agentevals.session_name" not in meta["resource_attrs"] + assert meta["resource_attrs"]["service.name"] == "test-agent" + + def test_spec_array_attributes_keep_their_container(self): + """Keys in SPEC_CONTAINER_ATTRS keep their decoded container - that is + what #173 fixes.""" + attrs = [ + { + "key": "gen_ai.response.finish_reasons", + "value": {"arrayValue": {"values": [{"stringValue": "stop"}]}}, + } + ] + meta = _extract_agentevals_metadata(attrs) + assert meta["resource_attrs"]["gen_ai.response.finish_reasons"] == ["stop"] + # --------------------------------------------------------------------------- # OTLP log record conversion