feat(tracing): snapshot effective tools and handoffs per turn

src/agents/run_internal/run_loop.py

@@ -22,6 +22,7 @@
 from openai.types.responses.response_prompt_param import ResponsePromptParam
 
 from .._tool_identity import (
+    get_function_tool_trace_name,
     get_tool_trace_name_for_tool,
     resolve_tool_name_collisions,
 )
@@ -85,10 +86,18 @@
     dispose_resolved_computers,
 )
 from ..tool_guardrails import ToolInputGuardrailResult, ToolOutputGuardrailResult
-from ..tracing import Span, SpanError, agent_span, get_current_trace, task_span, turn_span
+from ..tracing import (
+    Span,
+    SpanError,
+    agent_span,
+    get_current_span,
+    get_current_trace,
+    task_span,
+    turn_span,
+)
 from ..tracing.config import include_task_and_turn_spans
 from ..tracing.model_tracing import get_model_tracing_impl
-from ..tracing.span_data import AgentSpanData, TaskSpanData
+from ..tracing.span_data import AgentSpanData, TaskSpanData, TurnSpanData
 from ..usage import (
     Usage,
     _extract_raw_usage_snapshot,
@@ -319,6 +328,34 @@ def _agent_diagnostic_extra(agent: Agent[Any]) -> dict[str, object]:
     return {"agent_name": agent.name}
 
 
+def _record_effective_model_capabilities(
+    agent_span: Span[AgentSpanData] | None,
+    tools: list[Tool],
+    handoffs: list[Handoff],
+) -> None:
+    """Record the resolved model-visible capability set for the agent and current turn."""
+    agent_handoff_names = [handoff.agent_name for handoff in handoffs]
+    turn_handoff_names = [handoff.tool_name for handoff in handoffs]
+    agent_tool_names = [
+        tool_name for tool in tools if (tool_name := get_tool_trace_name_for_tool(tool)) is not None
+    ]
+    turn_tool_names = [
+        trace_name
+        for tool in tools
+        if (trace_name := get_function_tool_trace_name(tool) or get_tool_trace_name_for_tool(tool))
+        is not None
+    ]
+
+    if agent_span is not None:
+        agent_span.span_data.handoffs = agent_handoff_names
+        agent_span.span_data.tools = agent_tool_names
+
+    current_span = get_current_span()
+    if current_span is not None and isinstance(current_span.span_data, TurnSpanData):
+        current_span.span_data.handoffs = turn_handoff_names
+        current_span.span_data.tools = turn_tool_names
+
+
 async def _should_persist_stream_items(
     *,
     session: Session | None,
@@ -2105,13 +2142,11 @@ async def raise_if_input_guardrail_tripwire_known() -> None:
         handoffs,
         collision_policy=run_config.tool_name_collision_policy,
     )
-    if agent_span is not None:
-        agent_span.span_data.handoffs = [handoff.agent_name for handoff in handoffs]
-        agent_span.span_data.tools = [
-            tool_name
-            for tool in all_tools
-            if (tool_name := get_tool_trace_name_for_tool(tool)) is not None
-        ]
+    _record_effective_model_capabilities(
+        agent_span=agent_span,
+        tools=all_tools,
+        handoffs=handoffs,
+    )
 
     model = get_model(execution_agent, run_config)
     tool_use_tracker.record_model(model)
@@ -2432,13 +2467,11 @@ async def run_single_turn(
         handoffs,
         collision_policy=run_config.tool_name_collision_policy,
     )
-    if agent_span is not None:
-        agent_span.span_data.handoffs = [handoff.agent_name for handoff in handoffs]
-        agent_span.span_data.tools = [
-            tool_name
-            for tool in all_tools
-            if (tool_name := get_tool_trace_name_for_tool(tool)) is not None
-        ]
+    _record_effective_model_capabilities(
+        agent_span=agent_span,
+        tools=all_tools,
+        handoffs=handoffs,
+    )
 
     output_schema = get_output_schema(execution_agent)
     if server_conversation_tracker is not None:

src/agents/tracing/span_data.py

@@ -98,19 +98,23 @@ def export(self) -> dict[str, Any]:
 class TurnSpanData(SpanData):
     """Represents one agent loop turn."""
 
-    __slots__ = ("turn", "agent_name", "usage", "metadata")
+    __slots__ = ("turn", "agent_name", "usage", "metadata", "tools", "handoffs")
 
     def __init__(
         self,
         turn: int,
         agent_name: str,
         usage: dict[str, Any] | None = None,
         metadata: dict[str, Any] | None = None,
+        tools: list[str] | None = None,
+        handoffs: list[str] | None = None,
     ):
         self.turn = turn
         self.agent_name = agent_name
         self.usage = usage
         self.metadata = metadata
+        self.tools = tools
+        self.handoffs = handoffs
 
     @property
     def type(self) -> str:
@@ -122,6 +126,10 @@ def export(self) -> dict[str, Any]:
             "turn": self.turn,
             "agent_name": self.agent_name,
         }
+        if self.tools is not None:
+            data["tools"] = self.tools
+        if self.handoffs is not None:
+            data["handoffs"] = self.handoffs
         if self.usage is not None:
             data["usage"] = self.usage
 

tests/test_agent_tracing.py

@@ -8,6 +8,7 @@
 from openai.types.responses.response_usage import InputTokensDetails
 
 from agents import Agent, RunConfig, Runner, RunState, custom_span, function_tool, trace
+from agents.handoffs import handoff
 from agents.sandbox.runtime import SandboxRuntime
 from agents.testing import ScriptedModel
 from agents.usage import Usage
@@ -130,6 +131,70 @@ def second_lookup() -> str:
     assert agent_spans[0].span_data.handoffs == expected_handoffs
 
 
+@pytest.mark.parametrize("streamed", [False, True])
+@pytest.mark.asyncio
+async def test_turn_span_snapshots_effective_capabilities_per_model_turn(
+    streamed: bool,
+) -> None:
+    phase = {"value": "draft"}
+
+    def draft_enabled(_ctx: object, _agent: object) -> bool:
+        return phase["value"] == "draft"
+
+    def send_enabled(_ctx: object, _agent: object) -> bool:
+        return phase["value"] == "send"
+
+    @function_tool(name_override="draft_invoice", is_enabled=draft_enabled)
+    def draft_invoice() -> str:
+        phase["value"] = "send"
+        return "drafted"
+
+    @function_tool(name_override="send_invoice", is_enabled=send_enabled)
+    def send_invoice() -> str:
+        return "sent"
+
+    model = ScriptedModel(emit_traces=True)
+    model.extend(
+        [
+            [get_function_tool_call("draft_invoice", "{}", call_id="call-1")],
+            [get_text_message("done")],
+        ]
+    )
+    agent = Agent(
+        name="test_agent",
+        model=model,
+        tools=[draft_invoice, send_invoice],
+        handoffs=[
+            handoff(Agent(name="Draft Review"), is_enabled=draft_enabled),
+            handoff(Agent(name="Send Review"), is_enabled=send_enabled),
+        ],
+    )
+
+    if streamed:
+        result = Runner.run_streamed(agent, input="test")
+        async for _ in result.stream_events():
+            pass
+    else:
+        await Runner.run(agent, input="test")
+
+    spans = fetch_ordered_spans()
+    turn_spans = [span for span in spans if span.span_data.type == "turn"]
+    assert len(turn_spans) == 2
+    assert [span.span_data.tools for span in turn_spans] == [
+        ["draft_invoice"],
+        ["send_invoice"],
+    ]
+    assert [span.span_data.handoffs for span in turn_spans] == [
+        ["transfer_to_draft_review"],
+        ["transfer_to_send_review"],
+    ]
+
+    agent_spans = [span for span in spans if span.span_data.type == "agent"]
+    assert len(agent_spans) == 1
+    assert agent_spans[0].span_data.tools == ["send_invoice"]
+    assert agent_spans[0].span_data.handoffs == ["Send Review"]
+
+
 @pytest.mark.asyncio
 async def test_task_and_turn_spans_export_aggregate_usage():
     @function_tool
@@ -205,6 +270,8 @@ def foo_tool() -> str:
                 "sdk_span_type": "turn",
                 "turn": 1,
                 "agent_name": "test_agent",
+                "tools": ["foo_tool"],
+                "handoffs": [],
                 "usage": {
                     "input_tokens": 10,
                     "output_tokens": 3,
@@ -220,6 +287,8 @@ def foo_tool() -> str:
                 "sdk_span_type": "turn",
                 "turn": 2,
                 "agent_name": "test_agent",
+                "tools": ["foo_tool"],
+                "handoffs": [],
                 "usage": {
                     "input_tokens": 10,
                     "output_tokens": 3,

tests/test_turn_capability_identities.py

@@ -0,0 +1,51 @@
+from __future__ import annotations
+
+import pytest
+
+from agents import Agent, Runner, function_tool, tool_namespace
+from agents.handoffs import handoff
+from agents.testing import ScriptedModel
+
+from .test_responses import get_text_message
+from .testing_processor import fetch_ordered_spans
+
+
+@pytest.mark.asyncio
+async def test_turn_span_uses_model_visible_capability_identities() -> None:
+    @function_tool
+    def lookup_invoice() -> str:
+        return "invoice"
+
+    namespaced_tools = tool_namespace(
+        name="billing",
+        description="Billing lookup tools.",
+        tools=[lookup_invoice],
+    )
+    billing_agent = Agent(name="Billing Review")
+    billing_handoff = handoff(
+        billing_agent,
+        tool_name_override="escalate_to_billing",
+    )
+    model = ScriptedModel(emit_traces=True)
+    model.extend([[get_text_message("done")]])
+    agent = Agent(
+        name="test_agent",
+        model=model,
+        tools=namespaced_tools,
+        handoffs=[billing_handoff],
+    )
+
+    await Runner.run(agent, input="test")
+
+    spans = fetch_ordered_spans()
+    turn_spans = [span for span in spans if span.span_data.type == "turn"]
+    assert len(turn_spans) == 1
+    exported = turn_spans[0].export()
+    assert exported is not None
+    assert exported["span_data"]["data"]["tools"] == ["billing.lookup_invoice"]
+    assert exported["span_data"]["data"]["handoffs"] == ["escalate_to_billing"]
+
+    agent_spans = [span for span in spans if span.span_data.type == "agent"]
+    assert len(agent_spans) == 1
+    assert agent_spans[0].span_data.tools == ["lookup_invoice"]
+    assert agent_spans[0].span_data.handoffs == ["Billing Review"]