docs(pydantic-ai): document unified sync path; deprecate bespoke tracing handler (docstring)

declan-scale · claude · declan-scale · commit d8116eff9044 · 2026-06-22T09:48:08.000-04:00
- _pydantic_ai_sync.py: add "Recommended: unified surface" section to module
  docstring showing PydanticAITurn + UnifiedEmitter usage with automatic span
  derivation; bare converter docstring/code unchanged.
- _pydantic_ai_tracing.py: deprecation notes (docstring-only) on module,
  AgentexPydanticAITracingHandler, and create_pydantic_ai_tracing_handler;
  no runtime warnings.warn so warnings-as-errors does not break callers;
  NOTE: comment explains the deferral rationale.
- tests/lib/adk/test_pydantic_ai_sync_unified.py: 6 new tests covering the
  unified sync path: passthrough equality + tool/reasoning span derivation
  via _FakeTracing injection, no-trace-id no-op, tracer=False suppression.

Co-Authored-By: Claude Sonnet 4.6 &lt;noreply@anthropic.com&gt;
diff --git a/src/agentex/lib/adk/_modules/_pydantic_ai_sync.py b/src/agentex/lib/adk/_modules/_pydantic_ai_sync.py
@@ -16,6 +16,25 @@ async def handle_message_send(params):
         async with agent.run_stream_events(params.content.content) as stream:
             async for event in convert_pydantic_ai_to_agentex_events(stream):
                 yield event
+
+Recommended: unified surface
+-----------------------------
+For new handlers, prefer ``UnifiedEmitter`` + ``PydanticAITurn`` over the
+bare converter. The unified surface wires tracing automatically when a
+``trace_id`` is provided, so tool and reasoning spans are derived from the
+same event stream with no extra setup:
+
+    from agentex.lib.core.harness import UnifiedEmitter
+    from agentex.lib.adk._modules._pydantic_ai_turn import PydanticAITurn
+
+    emitter = UnifiedEmitter(task_id=task_id, trace_id=trace_id, parent_span_id=parent_span_id)
+    turn = PydanticAITurn(agent.run_stream_events(prompt), model="openai:gpt-4o")
+    async for event in emitter.yield_turn(turn):
+        yield event   # forwarded over the ACP streaming response; spans derived automatically
+
+``convert_pydantic_ai_to_agentex_events`` remains the low-level tap for
+callers that manage their own tracing or need direct access to the raw
+converted stream.
 """
 
 from __future__ import annotations
diff --git a/src/agentex/lib/adk/_modules/_pydantic_ai_tracing.py b/src/agentex/lib/adk/_modules/_pydantic_ai_tracing.py
@@ -1,5 +1,29 @@
 """Tracing handler that records Agentex spans for tool calls in a pydantic-ai agent run.
 
+.. deprecated::
+    ``AgentexPydanticAITracingHandler`` and ``create_pydantic_ai_tracing_handler``
+    are superseded by the unified harness surface (``UnifiedEmitter`` in
+    ``agentex.lib.core.harness``). The unified surface derives tool and
+    reasoning spans directly from the canonical ``StreamTaskMessage*`` stream,
+    so no separate handler is required. Both symbols remain fully importable
+    and functional; they will be removed in a future release. New code should
+    construct a ``UnifiedEmitter`` with a ``trace_id`` instead:
+
+        from agentex.lib.core.harness import UnifiedEmitter
+        from agentex.lib.adk._modules._pydantic_ai_turn import PydanticAITurn
+
+        emitter = UnifiedEmitter(task_id=task_id, trace_id=trace_id, parent_span_id=parent_span_id)
+        turn = PydanticAITurn(agent.run_stream_events(prompt), model="openai:gpt-4o")
+        async for event in emitter.yield_turn(turn):
+            yield event
+
+# NOTE: A runtime ``warnings.warn(..., DeprecationWarning)`` is intentionally
+# omitted here. The repo's pyproject ``filterwarnings = ["error"]`` would turn
+# it into a test/caller failure, and the async helper (``stream_pydantic_ai_events``)
+# still threads this handler through for existing callers that lack a ``trace_id``
+# on the async path. The runtime warning and caller migration are deferred until
+# ``trace_id`` threading lands on the async helper in a future API-versioning change.
+
 Mirrors the LangGraph tracing handler pattern: the caller creates a handler
 bound to a ``trace_id`` and a ``parent_span_id``, then hands it to
 ``stream_pydantic_ai_events(..., tracing_handler=handler)``. The streamer
@@ -63,6 +87,14 @@ def _tool_span_id(trace_id: str, tool_call_id: str) -> str:
 class AgentexPydanticAITracingHandler:
     """Records Agentex tracing spans for tool calls observed in a pydantic-ai event stream.
 
+    .. deprecated::
+        Superseded by ``UnifiedEmitter`` (``agentex.lib.core.harness``), which
+        derives tool and reasoning spans from the canonical ``StreamTaskMessage*``
+        stream automatically when ``trace_id`` is provided. This class remains
+        fully functional but will be removed in a future release. New code should
+        use ``UnifiedEmitter`` with a trace context instead of constructing this
+        handler directly.
+
     Pass an instance to ``stream_pydantic_ai_events(..., tracing_handler=...)``
     or call ``on_tool_start`` / ``on_tool_end`` yourself if you're consuming
     the event stream by hand.
@@ -165,6 +197,13 @@ def create_pydantic_ai_tracing_handler(
 ) -> AgentexPydanticAITracingHandler:
     """Create a tracing handler that records Agentex spans for pydantic-ai tool calls.
 
+    .. deprecated::
+        Superseded by ``UnifiedEmitter`` (``agentex.lib.core.harness``), which
+        derives tool and reasoning spans from the canonical ``StreamTaskMessage*``
+        stream automatically when ``trace_id`` is provided. This function remains
+        fully functional but will be removed in a future release. New code should
+        construct a ``UnifiedEmitter`` with a trace context instead.
+
     Args:
         trace_id: The trace ID. Typically the Agentex task ID.
         parent_span_id: Optional parent span ID to nest tool spans under. If
diff --git a/tests/lib/adk/test_pydantic_ai_sync_unified.py b/tests/lib/adk/test_pydantic_ai_sync_unified.py
@@ -0,0 +1,209 @@
+"""Tests for the unified sync (HTTP ACP) path: PydanticAITurn + UnifiedEmitter.
+
+Exercises the path documented in _pydantic_ai_sync.py under "Recommended: unified surface":
+- events forwarded by yield_turn equal PydanticAITurn(stream).events (passthrough)
+- with a trace context + fake tracing backend, tool spans are derived (start_span / end_span called)
+- with a trace context + fake tracing backend, reasoning spans are derived
+"""
+
+from __future__ import annotations
+
+from typing import Any, AsyncIterator
+
+from pydantic_ai.run import AgentRunResult, AgentRunResultEvent
+from pydantic_ai.usage import RunUsage
+from pydantic_ai.messages import (
+    TextPart,
+    PartEndEvent,
+    ThinkingPart,
+    ToolCallPart,
+    TextPartDelta,
+    PartDeltaEvent,
+    PartStartEvent,
+    ThinkingPartDelta,
+    ToolCallPartDelta,
+)
+
+from agentex.lib.core.harness import UnifiedEmitter
+from agentex.lib.adk._modules._pydantic_ai_turn import PydanticAITurn
+
+
+async def _aiter(events: list[Any]) -> AsyncIterator[Any]:
+    for e in events:
+        yield e
+
+
+async def _collect(stream: AsyncIterator[Any]) -> list[Any]:
+    return [e async for e in stream]
+
+
+class _FakeSpan:
+    def __init__(self, name: str):
+        self.name = name
+        self.output: Any = None
+
+
+class _FakeTracing:
+    def __init__(self) -> None:
+        self.started: list[tuple[str, str | None, Any]] = []
+        self.ended: list[tuple[str, Any]] = []
+
+    async def start_span(self, *, trace_id, name, input=None, parent_id=None, data=None, task_id=None):
+        self.started.append((name, parent_id, input))
+        return _FakeSpan(name)
+
+    async def end_span(self, *, trace_id, span):
+        self.ended.append((span.name, span.output))
+
+
+def _make_result_event(usage: RunUsage | None = None) -> AgentRunResultEvent:
+    result = AgentRunResult(output="done", _output_tool_name=None)
+    if usage is not None:
+        result._state.usage = usage
+    return AgentRunResultEvent(result=result)
+
+
+class TestUnifiedSyncPathPassthrough:
+    """The events forwarded by yield_turn are identical to PydanticAITurn.events."""
+
+    async def test_text_stream_passthrough(self):
+        raw_events = [
+            PartStartEvent(index=0, part=TextPart(content="")),
+            PartDeltaEvent(index=0, delta=TextPartDelta(content_delta="hello")),
+            PartEndEvent(index=0, part=TextPart(content="hello")),
+        ]
+
+        turn_a = PydanticAITurn(_aiter(raw_events), model="openai:gpt-4o")
+        direct = await _collect(turn_a.events)
+
+        turn_b = PydanticAITurn(_aiter(raw_events), model="openai:gpt-4o")
+        emitter = UnifiedEmitter(task_id="t", trace_id=None, parent_span_id=None)
+        via_emitter = await _collect(emitter.yield_turn(turn_b))
+
+        assert len(via_emitter) == len(direct)
+        for a, b in zip(via_emitter, direct):
+            assert type(a) is type(b)
+            assert a.model_dump() == b.model_dump()
+
+    async def test_tool_call_stream_passthrough(self):
+        raw_events = [
+            PartStartEvent(index=0, part=ToolCallPart(tool_name="Bash", args=None, tool_call_id="c1")),
+            PartDeltaEvent(index=0, delta=ToolCallPartDelta(args_delta='{"cmd":"ls"}')),
+            PartEndEvent(
+                index=0,
+                part=ToolCallPart(tool_name="Bash", args='{"cmd":"ls"}', tool_call_id="c1"),
+            ),
+        ]
+
+        turn_a = PydanticAITurn(_aiter(raw_events), model="openai:gpt-4o")
+        direct = await _collect(turn_a.events)
+
+        turn_b = PydanticAITurn(_aiter(raw_events), model="openai:gpt-4o")
+        emitter = UnifiedEmitter(task_id="t", trace_id=None, parent_span_id=None)
+        via_emitter = await _collect(emitter.yield_turn(turn_b))
+
+        assert len(via_emitter) == len(direct)
+        for a, b in zip(via_emitter, direct):
+            assert type(a) is type(b)
+            assert a.model_dump() == b.model_dump()
+
+
+class TestUnifiedSyncPathSpanDerivation:
+    """With trace context + fake tracing, spans are derived from the stream."""
+
+    async def test_tool_span_opened_and_closed(self):
+        """A tool call produces start_span + end_span on the fake tracing backend."""
+        from pydantic_ai.messages import ToolReturnPart, FunctionToolResultEvent
+
+        tool_events = [
+            PartStartEvent(
+                index=0,
+                part=ToolCallPart(tool_name="Bash", args={"cmd": "ls"}, tool_call_id="call_1"),
+            ),
+            PartEndEvent(
+                index=0,
+                part=ToolCallPart(tool_name="Bash", args='{"cmd":"ls"}', tool_call_id="call_1"),
+            ),
+            FunctionToolResultEvent(
+                part=ToolReturnPart(tool_name="Bash", content="files", tool_call_id="call_1"),
+            ),
+        ]
+
+        fake = _FakeTracing()
+        turn = PydanticAITurn(_aiter(tool_events), model="openai:gpt-4o")
+        emitter = UnifiedEmitter(task_id="t", trace_id="tr", parent_span_id="p", tracing=fake)
+
+        events = await _collect(emitter.yield_turn(turn))
+
+        assert len(events) >= 2, "at least Start(tool) + Done + Full(response)"
+        assert len(fake.started) == 1, "one tool span opened"
+        assert len(fake.ended) == 1, "one tool span closed"
+        span_name, parent_id, span_input = fake.started[0]
+        assert span_name == "Bash"
+        assert parent_id == "p"
+        closed_name, closed_output = fake.ended[0]
+        assert closed_name == "Bash"
+
+    async def test_reasoning_span_opened_and_closed(self):
+        """A thinking/reasoning block produces start_span + end_span."""
+        reasoning_events = [
+            PartStartEvent(index=0, part=ThinkingPart(content="")),
+            PartDeltaEvent(index=0, delta=ThinkingPartDelta(content_delta="let me think")),
+            PartEndEvent(index=0, part=ThinkingPart(content="let me think")),
+        ]
+
+        fake = _FakeTracing()
+        turn = PydanticAITurn(_aiter(reasoning_events), model="openai:gpt-4o")
+        emitter = UnifiedEmitter(task_id="t", trace_id="tr", parent_span_id="p", tracing=fake)
+
+        await _collect(emitter.yield_turn(turn))
+
+        assert len(fake.started) == 1, "one reasoning span opened"
+        assert len(fake.ended) == 1, "one reasoning span closed"
+        span_name, parent_id, _ = fake.started[0]
+        assert span_name == "reasoning"
+        assert parent_id == "p"
+
+    async def test_no_trace_id_means_no_spans(self):
+        """When trace_id is None, no spans are derived even with a fake tracing backend."""
+        raw_events = [
+            PartStartEvent(
+                index=0,
+                part=ToolCallPart(tool_name="Bash", args={"cmd": "ls"}, tool_call_id="c2"),
+            ),
+            PartEndEvent(
+                index=0,
+                part=ToolCallPart(tool_name="Bash", args='{"cmd":"ls"}', tool_call_id="c2"),
+            ),
+        ]
+
+        fake = _FakeTracing()
+        turn = PydanticAITurn(_aiter(raw_events), model="openai:gpt-4o")
+        emitter = UnifiedEmitter(task_id="t", trace_id=None, parent_span_id=None, tracing=fake)
+
+        await _collect(emitter.yield_turn(turn))
+
+        assert fake.started == [], "no spans when trace_id is absent"
+        assert fake.ended == []
+
+    async def test_tracer_false_suppresses_spans_even_with_trace_id(self):
+        """tracer=False disables span derivation regardless of trace_id."""
+        raw_events = [
+            PartStartEvent(
+                index=0,
+                part=ToolCallPart(tool_name="Bash", args={"cmd": "ls"}, tool_call_id="c3"),
+            ),
+            PartEndEvent(
+                index=0,
+                part=ToolCallPart(tool_name="Bash", args='{"cmd":"ls"}', tool_call_id="c3"),
+            ),
+        ]
+
+        fake = _FakeTracing()
+        turn = PydanticAITurn(_aiter(raw_events), model="openai:gpt-4o")
+        emitter = UnifiedEmitter(task_id="t", trace_id="tr", parent_span_id="p", tracer=False, tracing=fake)
+
+        await _collect(emitter.yield_turn(turn))
+
+        assert fake.started == []
+        assert fake.ended == []
