Fix agent span message aggregation

Author: nagkumar91

instrumentation-genai/opentelemetry-instrumentation-openai-agents-v2/src/opentelemetry/instrumentation/openai_agents/genai_semantic_processor.py

@@ -687,9 +687,12 @@ def _normalize_messages_to_role_parts(
                 )
                 parts.append(p)
 
-            normalized.append(
-                {"role": role, "parts": parts or self._redacted_text_parts()}
-            )
+            if parts:
+                normalized.append({"role": role, "parts": parts})
+            elif not self.include_sensitive_data:
+                normalized.append(
+                    {"role": role, "parts": self._redacted_text_parts()}
+                )
 
         return normalized
 
@@ -946,11 +949,17 @@ def _update_agent_aggregate(
             },
         )
         if payload.input_messages:
-            entry["input_messages"].extend(payload.input_messages)
+            entry["input_messages"] = self._merge_content_sequence(
+                entry["input_messages"], payload.input_messages
+            )
         if payload.output_messages:
-            entry["output_messages"].extend(payload.output_messages)
+            entry["output_messages"] = self._merge_content_sequence(
+                entry["output_messages"], payload.output_messages
+            )
         if payload.system_instructions:
-            entry["system_instructions"].extend(payload.system_instructions)
+            entry["system_instructions"] = self._merge_content_sequence(
+                entry["system_instructions"], payload.system_instructions
+            )
 
     def _infer_output_type(self, span_data: Any) -> str:
         """Infer gen_ai.output.type for multiple span kinds."""
@@ -1512,6 +1521,85 @@ def _get_attributes_from_generation_span_data(
             normalize_output_type(self._infer_output_type(span_data)),
         )
 
+    def _merge_content_sequence(
+        self,
+        existing: list[Any],
+        incoming: Sequence[Any],
+    ) -> list[Any]:
+        """Merge normalized message/content lists without duplicating snapshots."""
+        if not incoming:
+            return existing
+
+        incoming_list = [self._clone_message(item) for item in incoming]
+
+        if self.include_sensitive_data:
+            filtered = [
+                msg
+                for msg in incoming_list
+                if not self._is_placeholder_message(msg)
+            ]
+            if filtered:
+                incoming_list = filtered
+
+        if not existing:
+            return incoming_list
+
+        result = [self._clone_message(item) for item in existing]
+
+        for idx, new_msg in enumerate(incoming_list):
+            if idx < len(result):
+                if (
+                    self.include_sensitive_data
+                    and self._is_placeholder_message(new_msg)
+                    and not self._is_placeholder_message(result[idx])
+                ):
+                    continue
+                if result[idx] != new_msg:
+                    result[idx] = self._clone_message(new_msg)
+            else:
+                if (
+                    self.include_sensitive_data
+                    and self._is_placeholder_message(new_msg)
+                ):
+                    if (
+                        any(
+                            not self._is_placeholder_message(existing_msg)
+                            for existing_msg in result
+                        )
+                        or new_msg in result
+                    ):
+                        continue
+                result.append(self._clone_message(new_msg))
+
+        return result
+
+    def _clone_message(self, message: Any) -> Any:
+        if isinstance(message, dict):
+            return {
+                key: self._clone_message(value)
+                if isinstance(value, (dict, list))
+                else value
+                for key, value in message.items()
+            }
+        if isinstance(message, list):
+            return [self._clone_message(item) for item in message]
+        return message
+
+    def _is_placeholder_message(self, message: Any) -> bool:
+        if not isinstance(message, dict):
+            return False
+        parts = message.get("parts")
+        if not isinstance(parts, list) or not parts:
+            return False
+        for part in parts:
+            if (
+                not isinstance(part, dict)
+                or part.get("type") != "text"
+                or part.get("content") != "readacted"
+            ):
+                return False
+        return True
+
     def _get_attributes_from_agent_span_data(
         self,
         span_data: AgentSpanData,

instrumentation-genai/opentelemetry-instrumentation-openai-agents-v2/tests/test_tracer.py

@@ -3,6 +3,8 @@
 import json
 import sys
 from pathlib import Path
+from types import SimpleNamespace
+from typing import Any
 
 TESTS_ROOT = Path(__file__).resolve().parent
 stub_path = TESTS_ROOT / "stubs"
@@ -25,6 +27,10 @@
 from opentelemetry.instrumentation.openai_agents import (  # noqa: E402
     OpenAIAgentsInstrumentor,
 )
+from opentelemetry.instrumentation.openai_agents.genai_semantic_processor import (  # noqa: E402
+    ContentPayload,
+    GenAISemanticProcessor,
+)
 from opentelemetry.sdk.trace import TracerProvider  # noqa: E402
 
 try:
@@ -202,6 +208,160 @@ def test_agent_create_span_records_attributes():
         exporter.clear()
 
 
+def _placeholder_message() -> dict[str, Any]:
+    return {
+        "role": "user",
+        "parts": [{"type": "text", "content": "readacted"}],
+    }
+
+
+def test_normalize_messages_skips_empty_when_sensitive_enabled():
+    processor = GenAISemanticProcessor(metrics_enabled=False)
+    normalized = processor._normalize_messages_to_role_parts(
+        [{"role": "user", "content": None}]
+    )
+    assert normalized == []
+
+
+def test_normalize_messages_emits_placeholder_when_sensitive_disabled():
+    processor = GenAISemanticProcessor(
+        include_sensitive_data=False, metrics_enabled=False
+    )
+    normalized = processor._normalize_messages_to_role_parts(
+        [{"role": "user", "content": None}]
+    )
+    assert normalized == [_placeholder_message()]
+
+
+def test_agent_content_aggregation_skips_duplicate_snapshots():
+    processor = GenAISemanticProcessor(metrics_enabled=False)
+    agent_id = "agent-span"
+    processor._agent_content[agent_id] = {
+        "input_messages": [],
+        "output_messages": [],
+        "system_instructions": [],
+    }
+
+    payload = ContentPayload(
+        input_messages=[
+            {"role": "user", "parts": [{"type": "text", "content": "hello"}]},
+            {
+                "role": "user",
+                "parts": [{"type": "text", "content": "readacted"}],
+            },
+        ]
+    )
+
+    processor._update_agent_aggregate(
+        SimpleNamespace(span_id="child-1", parent_id=agent_id, span_data=None),
+        payload,
+    )
+    processor._update_agent_aggregate(
+        SimpleNamespace(span_id="child-2", parent_id=agent_id, span_data=None),
+        payload,
+    )
+
+    aggregated = processor._agent_content[agent_id]["input_messages"]
+    assert aggregated == [
+        {"role": "user", "parts": [{"type": "text", "content": "hello"}]}
+    ]
+    # ensure data copied rather than reused to prevent accidental mutation
+    assert aggregated is not payload.input_messages
+
+
+def test_agent_content_aggregation_filters_placeholder_append_when_sensitive():
+    processor = GenAISemanticProcessor(metrics_enabled=False)
+    agent_id = "agent-span"
+    processor._agent_content[agent_id] = {
+        "input_messages": [],
+        "output_messages": [],
+        "system_instructions": [],
+    }
+
+    initial_payload = ContentPayload(
+        input_messages=[
+            {"role": "user", "parts": [{"type": "text", "content": "hello"}]}
+        ]
+    )
+    processor._update_agent_aggregate(
+        SimpleNamespace(span_id="child-1", parent_id=agent_id, span_data=None),
+        initial_payload,
+    )
+
+    placeholder_payload = ContentPayload(
+        input_messages=[_placeholder_message()]
+    )
+    processor._update_agent_aggregate(
+        SimpleNamespace(span_id="child-2", parent_id=agent_id, span_data=None),
+        placeholder_payload,
+    )
+
+    aggregated = processor._agent_content[agent_id]["input_messages"]
+    assert aggregated == [
+        {"role": "user", "parts": [{"type": "text", "content": "hello"}]}
+    ]
+
+
+def test_agent_content_aggregation_retains_placeholder_when_sensitive_disabled():
+    processor = GenAISemanticProcessor(
+        include_sensitive_data=False, metrics_enabled=False
+    )
+    agent_id = "agent-span"
+    processor._agent_content[agent_id] = {
+        "input_messages": [],
+        "output_messages": [],
+        "system_instructions": [],
+    }
+
+    placeholder_payload = ContentPayload(
+        input_messages=[_placeholder_message()]
+    )
+    processor._update_agent_aggregate(
+        SimpleNamespace(span_id="child-1", parent_id=agent_id, span_data=None),
+        placeholder_payload,
+    )
+
+    aggregated = processor._agent_content[agent_id]["input_messages"]
+    assert aggregated == [_placeholder_message()]
+
+
+def test_agent_content_aggregation_appends_new_messages_once():
+    processor = GenAISemanticProcessor(metrics_enabled=False)
+    agent_id = "agent-span"
+    processor._agent_content[agent_id] = {
+        "input_messages": [],
+        "output_messages": [],
+        "system_instructions": [],
+    }
+
+    initial_payload = ContentPayload(
+        input_messages=[
+            {"role": "user", "parts": [{"type": "text", "content": "hello"}]}
+        ]
+    )
+    processor._update_agent_aggregate(
+        SimpleNamespace(span_id="child-1", parent_id=agent_id, span_data=None),
+        initial_payload,
+    )
+
+    extended_messages = [
+        {"role": "user", "parts": [{"type": "text", "content": "hello"}]},
+        {
+            "role": "assistant",
+            "parts": [{"type": "text", "content": "hi there"}],
+        },
+    ]
+    extended_payload = ContentPayload(input_messages=extended_messages)
+    processor._update_agent_aggregate(
+        SimpleNamespace(span_id="child-2", parent_id=agent_id, span_data=None),
+        extended_payload,
+    )
+
+    aggregated = processor._agent_content[agent_id]["input_messages"]
+    assert aggregated == extended_messages
+    assert extended_payload.input_messages == extended_messages
+
+
 def test_agent_span_collects_child_messages():
     instrumentor, exporter = _instrument_with_provider()