fix: backfill streamed terminal output (#3000)

Author: ankitphogat

src/agents/run_internal/run_loop.py

@@ -17,7 +17,7 @@
     ResponseFunctionToolCall,
     ResponseOutputItemDoneEvent,
 )
-from openai.types.responses.response_output_item import McpCall, McpListTools
+from openai.types.responses.response_output_item import McpCall, McpListTools, ResponseOutputItem
 from openai.types.responses.response_prompt_param import ResponsePromptParam
 from openai.types.responses.response_reasoning_item import ResponseReasoningItem
 
@@ -1337,6 +1337,7 @@ def _tool_search_fingerprint(raw_item: Any) -> str:
     model_settings = maybe_reset_tool_choice(public_agent, tool_use_tracker, model_settings)
 
     final_response: ModelResponse | None = None
+    streamed_response_output: list[ResponseOutputItem] = []
 
     if server_conversation_tracker is not None:
         items_for_input = (
@@ -1474,14 +1475,21 @@ async def rewind_model_request() -> None:
         streamed_result._event_queue.put_nowait(RawResponsesStreamEvent(data=event))
 
         terminal_response: Response | None = None
+        is_completed_event = False
         if isinstance(event, ResponseCompletedEvent):
+            is_completed_event = True
             terminal_response = event.response
         elif getattr(event, "type", None) in {"response.incomplete", "response.failed"}:
             maybe_response = getattr(event, "response", None)
             if isinstance(maybe_response, Response):
                 terminal_response = maybe_response
 
         if terminal_response is not None:
+            if is_completed_event and not terminal_response.output and streamed_response_output:
+                # Some streaming backends emit output items during item.done events while leaving
+                # the terminal response output empty. Preserve those items so the runner can
+                # resolve the completed step correctly.
+                terminal_response.output = list(streamed_response_output)
             usage = (
                 apply_retry_attempt_usage(
                     Usage(
@@ -1506,6 +1514,7 @@ async def rewind_model_request() -> None:
 
         if isinstance(event, ResponseOutputItemDoneEvent):
             output_item = event.item
+            streamed_response_output.append(output_item)
             output_item_type = getattr(output_item, "type", None)
 
             if output_item_type == "tool_search_call":

tests/test_streamed_terminal_output_backfill.py

@@ -0,0 +1,219 @@
+from __future__ import annotations
+
+import json
+from collections.abc import AsyncIterator
+from typing import Any
+
+import pytest
+from openai.types.responses import (
+    ResponseCompletedEvent,
+    ResponseCreatedEvent,
+    ResponseInProgressEvent,
+    ResponseOutputItemDoneEvent,
+)
+
+from agents import Agent, Runner
+from agents.agent_output import AgentOutputSchemaBase
+from agents.handoffs import Handoff
+from agents.items import TResponseInputItem, TResponseOutputItem, TResponseStreamEvent
+from agents.model_settings import ModelSettings
+from agents.models.interface import ModelTracing
+from agents.tool import Tool, function_tool
+
+from .fake_model import FakeModel, get_response_obj
+from .test_responses import get_final_output_message, get_function_tool_call
+
+
+class TerminalOutputStreamModel(FakeModel):
+    def __init__(self) -> None:
+        super().__init__()
+        self.terminal_turn_outputs: list[list[TResponseOutputItem]] = []
+
+    def add_terminal_turn_outputs(
+        self,
+        outputs: list[list[TResponseOutputItem]],
+    ) -> None:
+        self.terminal_turn_outputs.extend(outputs)
+
+    def get_next_terminal_output(self) -> list[TResponseOutputItem]:
+        if not self.terminal_turn_outputs:
+            return []
+        return self.terminal_turn_outputs.pop(0)
+
+    async def stream_response(
+        self,
+        system_instructions: str | None,
+        input: str | list[TResponseInputItem],
+        model_settings: ModelSettings,
+        tools: list[Tool],
+        output_schema: AgentOutputSchemaBase | None,
+        handoffs: list[Handoff],
+        tracing: ModelTracing,
+        *,
+        previous_response_id: str | None = None,
+        conversation_id: str | None = None,
+        prompt: Any | None = None,
+    ) -> AsyncIterator[TResponseStreamEvent]:
+        turn_args = {
+            "system_instructions": system_instructions,
+            "input": input,
+            "model_settings": model_settings,
+            "tools": tools,
+            "output_schema": output_schema,
+            "previous_response_id": previous_response_id,
+            "conversation_id": conversation_id,
+        }
+
+        if self.first_turn_args is None:
+            self.first_turn_args = turn_args.copy()
+
+        self.last_turn_args = turn_args
+        streamed_output = self.get_next_output()
+        if isinstance(streamed_output, Exception):
+            raise streamed_output
+
+        terminal_response = get_response_obj(
+            self.get_next_terminal_output(),
+            usage=self.hardcoded_usage,
+        )
+        sequence_number = 0
+
+        yield ResponseCreatedEvent(
+            type="response.created",
+            response=terminal_response,
+            sequence_number=sequence_number,
+        )
+        sequence_number += 1
+
+        yield ResponseInProgressEvent(
+            type="response.in_progress",
+            response=terminal_response,
+            sequence_number=sequence_number,
+        )
+        sequence_number += 1
+
+        for output_index, output_item in enumerate(streamed_output):
+            yield ResponseOutputItemDoneEvent(
+                type="response.output_item.done",
+                item=output_item,
+                output_index=output_index,
+                sequence_number=sequence_number,
+            )
+            sequence_number += 1
+
+        yield ResponseCompletedEvent(
+            type="response.completed",
+            response=terminal_response,
+            sequence_number=sequence_number,
+        )
+
+
+@pytest.mark.asyncio
+async def test_streamed_runner_backfills_empty_terminal_output_before_step_resolution() -> None:
+    tool_inputs: list[str] = []
+
+    async def test_tool(a: str) -> str:
+        tool_inputs.append(a)
+        return "tool_result"
+
+    tool = function_tool(test_tool, name_override="foo")
+    model = TerminalOutputStreamModel()
+    agent = Agent(name="test", model=model, tools=[tool])
+
+    model.add_multiple_turn_outputs(
+        [
+            [get_function_tool_call("foo", json.dumps({"a": "b"}), call_id="call-1")],
+            [get_final_output_message("done")],
+        ]
+    )
+    model.add_terminal_turn_outputs(
+        [
+            [],
+            [get_final_output_message("done")],
+        ]
+    )
+
+    result = Runner.run_streamed(agent, input="test")
+    async for _ in result.stream_events():
+        pass
+
+    assert tool_inputs == ["b"]
+    assert [item.type for item in result.raw_responses[0].output] == ["function_call"]
+    assert result.final_output == "done"
+
+
+@pytest.mark.asyncio
+async def test_streamed_runner_preserves_populated_terminal_output() -> None:
+    tool_inputs: list[str] = []
+
+    async def test_tool(a: str) -> str:
+        tool_inputs.append(a)
+        return "tool_result"
+
+    tool = function_tool(test_tool, name_override="foo")
+    model = TerminalOutputStreamModel()
+    agent = Agent(name="test", model=model, tools=[tool])
+
+    model.add_multiple_turn_outputs(
+        [
+            [get_function_tool_call("foo", json.dumps({"a": "b"}), call_id="call-1")],
+        ]
+    )
+    model.add_terminal_turn_outputs(
+        [
+            [get_final_output_message("done")],
+        ]
+    )
+
+    result = Runner.run_streamed(agent, input="test")
+    async for _ in result.stream_events():
+        pass
+
+    assert tool_inputs == []
+    assert [item.type for item in result.raw_responses[0].output] == ["message"]
+    assert result.final_output == "done"
+
+
+@pytest.mark.asyncio
+async def test_streamed_runner_backfills_multiple_tool_calls_in_order() -> None:
+    tool_inputs: list[tuple[str, str]] = []
+
+    async def foo_tool(a: str) -> str:
+        tool_inputs.append(("foo", a))
+        return "foo_result"
+
+    async def bar_tool(b: str) -> str:
+        tool_inputs.append(("bar", b))
+        return "bar_result"
+
+    foo = function_tool(foo_tool, name_override="foo")
+    bar = function_tool(bar_tool, name_override="bar")
+    model = TerminalOutputStreamModel()
+    agent = Agent(name="test", model=model, tools=[foo, bar])
+
+    model.add_multiple_turn_outputs(
+        [
+            [
+                get_function_tool_call("foo", json.dumps({"a": "first"}), call_id="call-1"),
+                get_function_tool_call("bar", json.dumps({"b": "second"}), call_id="call-2"),
+            ],
+            [get_final_output_message("done")],
+        ]
+    )
+    model.add_terminal_turn_outputs(
+        [
+            [],
+            [get_final_output_message("done")],
+        ]
+    )
+
+    result = Runner.run_streamed(agent, input="test")
+    async for _ in result.stream_events():
+        pass
+
+    assert tool_inputs == [("foo", "first"), ("bar", "second")]
+    assert [item.type for item in result.raw_responses[0].output] == [
+        "function_call",
+        "function_call",
+    ]
+    assert result.final_output == "done"