fix(tracing): record gen_ai.output.messages on the chat span

Per spec (docs/specs/2026-05-18-cubepi-tracing-design.md §10.3/§10.5), the
`chat` span's gen_ai.output.messages should reflect what the provider
actually returned - independent of the turn/agent-level rollup. The code
in _on_provider_response already had a comment saying it would "record ...
the normalized output messages where derivable", but only ever set
cubepi.llm.raw_response; gen_ai.output.messages was never written on this
span at all (only on invoke_agent and cubepi.turn).

Add _derive_output_message_from_body(), mirroring the existing three-way
provider-shape dispatch in _record_chat_response_attrs (Anthropic-shaped,
OpenAI chat.completion-shaped, OpenAI Responses-shaped) to reconstruct the
semconv output-message parts (text/reasoning/tool_call) directly from the
assembled response body - this span only ever sees that raw body via
subscribe_response, not the parsed AssistantMessage the agent loop builds
later. Defensive throughout: an unrecognized shape or empty content just
sets nothing, identical to today's behavior (no regression).

Author: xfgong

cubepi/tracing/recorder.py

@@ -1178,6 +1178,11 @@ def _on_provider_response(
                     # response (JSON dict) for backends that prefer it,
                     # and the normalized output messages where derivable.
                     self._set_content_attr(span, CUBEPI_LLM_RAW_RESPONSE, body)
+                    output_message = _derive_output_message_from_body(body)
+                    if output_message:
+                        self._set_content_attr(
+                            span, GEN_AI_OUTPUT_MESSAGES, output_message
+                        )
             # Cooperative abort: providers may finish the response
             # listener with ``exc is None`` and a body whose finish
             # reason is ``"aborted"`` (faux + the agent's signal path).
@@ -1402,6 +1407,121 @@ def _maybe_record_system_prompt_hash(self, payload: dict, run: _RunState) -> Non
         run.agent_span.set_attribute(CUBEPI_AGENT_SYSTEM_PROMPT_SHA256, digest)
 
 
+def _derive_output_message_from_body(body: dict) -> list[dict[str, Any]] | None:
+    """Reconstruct the semconv output-message shape (§10.5) directly from an
+    assembled provider response body, for the ``chat`` span's own
+    ``gen_ai.output.messages`` (§10.3).
+
+    The `chat` span only ever sees the raw provider-shaped ``body`` (via
+    ``subscribe_response``) — never the parsed cubepi ``AssistantMessage``
+    that ``messages_to_semconv`` normally converts, since that object is
+    built later in the agent loop. Mirrors the same three-way provider-shape
+    dispatch already used by ``_record_chat_response_attrs`` right above.
+
+    Returns ``None`` (never raises) when the shape isn't recognized or
+    carries no content — the caller treats that as "nothing to record",
+    identical to today's behavior.
+    """
+    # Anthropic-shaped: {"content": [...], "stop_reason": ..., "usage": {...}}
+    if (
+        "stop_reason" in body
+        and "usage" in body
+        and isinstance(body.get("content"), list)
+    ):
+        parts: list[dict[str, Any]] = []
+        for block in body["content"]:
+            if not isinstance(block, dict):
+                continue
+            btype = block.get("type")
+            if btype == "text" and block.get("text"):
+                parts.append({"type": "text", "content": block["text"]})
+            elif btype == "thinking" and block.get("thinking"):
+                parts.append({"type": "reasoning", "content": block["thinking"]})
+            elif btype == "tool_use":
+                parts.append(
+                    {
+                        "type": "tool_call",
+                        "id": block.get("id"),
+                        "name": block.get("name"),
+                        "arguments": block.get("input") or {},
+                    }
+                )
+        return [{"role": "assistant", "parts": parts}] if parts else None
+
+    # OpenAI chat.completion-shaped: {"choices": [{"message": {...}}]}
+    if "choices" in body and isinstance(body.get("choices"), list) and body["choices"]:
+        first = body["choices"][0]
+        message = first.get("message") if isinstance(first, dict) else None
+        if not isinstance(message, dict):
+            return None
+        parts = []
+        if message.get("content"):
+            parts.append({"type": "text", "content": message["content"]})
+        for tc in message.get("tool_calls") or []:
+            if not isinstance(tc, dict):
+                continue
+            fn = tc.get("function") or {}
+            args = _try_parse_json(fn.get("arguments"))
+            parts.append(
+                {
+                    "type": "tool_call",
+                    "id": tc.get("id"),
+                    "name": fn.get("name"),
+                    "arguments": args,
+                }
+            )
+        return [{"role": "assistant", "parts": parts}] if parts else None
+
+    # OpenAI Responses-shaped: {"object": "response", "output": [...]}
+    if body.get("object") == "response" or isinstance(body.get("output"), list):
+        output = body.get("output")
+        if not isinstance(output, list):
+            return None
+        parts = []
+        for item in output:
+            if not isinstance(item, dict):
+                continue
+            itype = item.get("type")
+            if itype == "message":
+                for c in item.get("content") or []:
+                    if (
+                        isinstance(c, dict)
+                        and c.get("type") == "output_text"
+                        and c.get("text")
+                    ):
+                        parts.append({"type": "text", "content": c["text"]})
+            elif itype == "reasoning":
+                for s in item.get("summary") or []:
+                    if isinstance(s, dict) and s.get("text"):
+                        parts.append({"type": "reasoning", "content": s["text"]})
+            elif itype == "function_call":
+                args = _try_parse_json(item.get("arguments"))
+                parts.append(
+                    {
+                        "type": "tool_call",
+                        "id": item.get("call_id") or item.get("id"),
+                        "name": item.get("name"),
+                        "arguments": args,
+                    }
+                )
+        return [{"role": "assistant", "parts": parts}] if parts else None
+
+    return None
+
+
+def _try_parse_json(value: Any) -> Any:
+    """Tool call arguments arrive as a JSON-encoded string on the wire;
+    decode to match the object shape ``ToolCall.arguments`` uses elsewhere
+    (§10.5). Falls back to the raw value if it isn't valid JSON rather than
+    dropping it."""
+    if isinstance(value, str):
+        try:
+            return json.loads(value)
+        except (TypeError, ValueError):
+            return value
+    return value or {}
+
+
 def _extract_system_prompt(payload: dict) -> str | None:
     """Find the system prompt text in a provider's wire payload.
 

tests/tracing/test_recorder.py

@@ -7,6 +7,7 @@
 from __future__ import annotations
 
 import asyncio
+import json
 from typing import Any
 
 from opentelemetry.sdk.trace import ReadableSpan
@@ -828,6 +829,124 @@ async def test_max_output_tokens_lands_in_request_max_tokens(self):
         assert attrs.get("gen_ai.request.max_tokens") == 4096
 
 
+class TestChatSpanOutputMessages:
+    """Per spec §10.3/§10.5 (docs/specs/2026-05-18-cubepi-tracing-design.md),
+    the `chat` span's own `gen_ai.output.messages` should reflect what the
+    provider actually returned - independent of the turn/agent-level rollup,
+    which is built later from the agent's own message accumulation and isn't
+    available yet inside the `subscribe_response` callback. Previously only
+    `cubepi.llm.raw_response` was recorded here despite the code comment
+    already saying otherwise.
+    """
+
+    async def _drive(self, body: dict) -> dict[str, Any]:
+        provider = FauxProvider(provider_id="faux")
+        agent = Agent(model=provider.model(MODEL.id), system_prompt="s")
+        exporter = InMemoryExporter()
+        tracer = Tracer(
+            service_name="t", agent_name="a", exporters=[exporter], record_content=True
+        )
+        tracer.attach(agent)
+
+        recorder = _find_attached_recorder(provider)
+        assert recorder is not None
+        from cubepi.agent.types import AgentStartEvent, TurnStartEvent
+
+        await recorder._on_agent_event(AgentStartEvent())
+        await recorder._on_agent_event(TurnStartEvent())
+        recorder._on_provider_request({"messages": []}, MODEL)
+        recorder._on_provider_response(body, MODEL, None)
+        await tracer.shutdown()
+
+        chats = [s for s in exporter.spans if s.name.startswith("chat ")]
+        assert chats, "no chat span captured"
+        return _attrs(chats[-1])
+
+    async def test_openai_chat_completions_text(self):
+        attrs = await self._drive(
+            {
+                "id": "resp1",
+                "model": "faux-1",
+                "choices": [{"message": {"role": "assistant", "content": "hi there"}}],
+            }
+        )
+        raw = attrs.get("gen_ai.output.messages")
+        assert raw, "gen_ai.output.messages missing on chat span"
+        messages = json.loads(raw)
+        assert messages == [
+            {"role": "assistant", "parts": [{"type": "text", "content": "hi there"}]}
+        ]
+
+    async def test_openai_chat_completions_tool_call(self):
+        attrs = await self._drive(
+            {
+                "id": "resp1",
+                "model": "faux-1",
+                "choices": [
+                    {
+                        "message": {
+                            "role": "assistant",
+                            "content": None,
+                            "tool_calls": [
+                                {
+                                    "id": "call_1",
+                                    "function": {
+                                        "name": "search",
+                                        "arguments": '{"query": "weather"}',
+                                    },
+                                }
+                            ],
+                        }
+                    }
+                ],
+            }
+        )
+        messages = json.loads(attrs["gen_ai.output.messages"])
+        assert messages == [
+            {
+                "role": "assistant",
+                "parts": [
+                    {
+                        "type": "tool_call",
+                        "id": "call_1",
+                        "name": "search",
+                        "arguments": {"query": "weather"},
+                    }
+                ],
+            }
+        ]
+
+    async def test_anthropic_shaped_text_and_thinking(self):
+        attrs = await self._drive(
+            {
+                "id": "msg1",
+                "type": "message",
+                "role": "assistant",
+                "model": "faux-1",
+                "content": [
+                    {"type": "thinking", "thinking": "let me think"},
+                    {"type": "text", "text": "the answer"},
+                ],
+                "stop_reason": "end_turn",
+                "usage": {"input_tokens": 1, "output_tokens": 1},
+            }
+        )
+        messages = json.loads(attrs["gen_ai.output.messages"])
+        assert messages == [
+            {
+                "role": "assistant",
+                "parts": [
+                    {"type": "reasoning", "content": "let me think"},
+                    {"type": "text", "content": "the answer"},
+                ],
+            }
+        ]
+
+    async def test_unrecognized_shape_sets_nothing(self):
+        attrs = await self._drive({"model": "faux-1", "weird": True})
+        assert "gen_ai.output.messages" not in attrs
+
+
 class TestDetachFlushGuarantee:
     """``Tracer.attach()``'s ``detach()`` must let callers await the
     flush so buffered spans land before they proceed — previously the