fix(streaming): ensure processed_response.new_items, including HandoffCallItem, are pushed to event_queue

- Added `event_queue` parameter to `_get_single_step_result_from_response`
- Push `processed_response.new_items` to queue in streaming mode
- Non-streaming path (`_run_single_turn`) remains unchanged
- Fixes loss of HandoffCallItem in streaming scenarios
- Add test_stream_events_main_with_handoff test

Author: Wujiaxuan007

src/agents/run.py

@@ -60,7 +60,7 @@
 from .models.multi_provider import MultiProvider
 from .result import RunResult, RunResultStreaming
 from .run_context import RunContextWrapper, TContext
-from .stream_events import AgentUpdatedStreamEvent, RawResponsesStreamEvent, RunItemStreamEvent
+from .stream_events import AgentUpdatedStreamEvent, RawResponsesStreamEvent, RunItemStreamEvent, StreamEvent
 from .tool import Tool
 from .tracing import Span, SpanError, agent_span, get_current_trace, trace
 from .tracing.span_data import AgentSpanData
@@ -1084,6 +1084,7 @@ async def _run_single_turn_streamed(
             context_wrapper=context_wrapper,
             run_config=run_config,
             tool_use_tracker=tool_use_tracker,
+            event_queue=streamed_result._event_queue,
         )
 
         if emitted_tool_call_ids:
@@ -1196,6 +1197,7 @@ async def _get_single_step_result_from_response(
         context_wrapper: RunContextWrapper[TContext],
         run_config: RunConfig,
         tool_use_tracker: AgentToolUseTracker,
+        event_queue: asyncio.Queue[StreamEvent | QueueCompleteSentinel] | None = None,
     ) -> SingleStepResult:
         processed_response = RunImpl.process_model_response(
             agent=agent,
@@ -1207,6 +1209,9 @@ async def _get_single_step_result_from_response(
 
         tool_use_tracker.add_tool_use(agent, processed_response.tools_used)
 
+        if event_queue is not None and processed_response.new_items:
+            RunImpl.stream_step_items_to_queue(processed_response.new_items, event_queue)
+
         return await RunImpl.execute_tools_and_side_effects(
             agent=agent,
             original_input=original_input,

tests/test_stream_events.py

@@ -3,11 +3,12 @@
 
 import pytest
 
-from agents import Agent, Runner, function_tool
+from agents import Agent, Runner, function_tool, HandoffCallItem, HandoffOutputItem
+from agents.handoffs import handoff
+from agents.extensions.handoff_filters import remove_all_tools
 
 from .fake_model import FakeModel
-from .test_responses import get_function_tool_call, get_text_message
-
+from .test_responses import get_function_tool_call, get_text_message, get_handoff_tool_call
 
 @function_tool
 async def foo() -> str:
@@ -52,3 +53,65 @@ async def test_stream_events_main():
     assert tool_call_start_time > 0, "tool_call_item was not observed"
     assert tool_call_end_time > 0, "tool_call_output_item was not observed"
     assert tool_call_start_time < tool_call_end_time, "Tool call ended before or equals it started?"
+
+@pytest.mark.asyncio
+async def test_stream_events_main_with_handoff():
+    @function_tool
+    async def foo(args: str) -> str:
+        return f"foo_result_{args}"
+
+    english_agent = Agent(
+        name="EnglishAgent",
+        instructions="You only speak English.",
+        model=FakeModel(),
+    )
+
+    model = FakeModel()
+    model.add_multiple_turn_outputs(
+        [
+            [
+                get_text_message("Hello"),
+                get_function_tool_call("foo", '{"args": "arg1"}'),
+                get_handoff_tool_call(english_agent),
+            ],
+            [get_text_message("Done")],
+        ]
+    )
+
+    triage_agent = Agent(
+        name="TriageAgent",
+        instructions="Handoff to the appropriate agent based on the language of the request.",
+        handoffs=[
+            handoff(english_agent, input_filter=remove_all_tools),
+        ],
+        tools=[foo],
+        model=model,
+    )
+
+    result = Runner.run_streamed(
+        triage_agent,
+        input="Start",
+    )
+
+    tool_call_start_time = -1
+    tool_call_end_time = -1
+    handoff_requested_seen = False
+    handoff_occured_seen = False
+    agent_switched_to_english = False
+
+    async for event in result.stream_events():
+        if event.type == "run_item_stream_event":
+            if isinstance(event.item, HandoffCallItem):
+                handoff_requested_seen = True
+            elif isinstance(event.item, HandoffOutputItem):
+                handoff_occured_seen = True
+            elif event.item.type == "tool_call_item":
+                tool_call_start_time = time.time_ns()
+            elif event.item.type == "tool_call_output_item":
+                tool_call_end_time = time.time_ns()
+        elif event.type == "agent_updated_stream_event":
+            if hasattr(event, 'new_agent') and event.new_agent.name == "EnglishAgent":
+                agent_switched_to_english = True
+
+    assert handoff_requested_seen, "handoff_requested event not observed"
+    assert agent_switched_to_english, "Agent did not switch to EnglishAgent"