fix: reject premature Chat Completions stream EOF

src/agents/models/chatcmpl_stream_handler.py

@@ -383,7 +383,8 @@ def _buffered_tool_calls_chunk(
         choice = Choice(
             index=0,
             delta=ChoiceDelta(tool_calls=tool_call_deltas),
-            finish_reason="tool_calls",
+            # Assembling buffered deltas does not prove the provider finished the choice.
+            finish_reason=None,
         )
         return template_chunk.model_copy(update={"choices": [choice], "usage": None})
 
@@ -445,11 +446,8 @@ async def buffer_tool_call_stream(
 
                 if has_passthrough_output:
                     passthrough_choices.append(choice)
-                elif choice.finish_reason in {"content_filter", "length"}:
-                    # A content-filtered or truncated choice ends the stream with an empty
-                    # delta, so it would otherwise be dropped here and the handler would
-                    # never see the finish_reason it needs to act on.
-                    # Forward a delta-stripped copy so buffering semantics are unchanged.
+                elif choice.finish_reason is not None:
+                    # Preserve real terminal evidence even when all output was buffered.
                     passthrough_choices.append(choice.model_copy(update={"delta": ChoiceDelta()}))
 
             if passthrough_choices or chunk.usage is not None:
@@ -610,6 +608,7 @@ async def handle_stream(
         strict_feature_validation: bool = False,
         preserve_raw_usage: bool = False,
         raise_on_length_truncation: bool = False,
+        require_finish_reason: bool = False,
     ) -> AsyncIterator[TResponseStreamEvent]:
         """
         Handle a streaming chat completion response and yield response events.
@@ -626,6 +625,8 @@ async def handle_stream(
                 This is an internal option enabled only by OpenAIChatCompletionsModel:
                 the shared handler also serves LiteLLM and AnyLLM, whose streaming
                 behavior must remain unchanged.
+            require_finish_reason: Whether choice zero must have a provider finish reason
+                before completion. Enabled for the official OpenAI endpoint only.
         """
         usage: CompletionUsage | None = None
         raw_usage: dict[str, Any] | None = None
@@ -641,6 +642,7 @@ async def handle_stream(
         # behavior error instead of collapsing into an empty turn.
         saw_content_filter = False
         saw_length = False
+        saw_finish_reason = False
         async for chunk in stream:
             if not state.started:
                 state.started = True
@@ -683,6 +685,8 @@ async def handle_stream(
             if choice is None:
                 continue
 
+            if choice.finish_reason is not None:
+                saw_finish_reason = True
             if choice.finish_reason == "content_filter":
                 saw_content_filter = True
             elif choice.finish_reason == "length":
@@ -1186,6 +1190,17 @@ async def handle_stream(
                 sequence_number=sequence_number.get_and_increment(),
             )
 
+        if require_finish_reason and not saw_finish_reason:
+            if usage is not None:
+                response.usage = cls._build_response_usage(usage)
+            else:
+                _mark_request_completed_without_usage(response)
+            if preserve_raw_usage and raw_usage is not None:
+                _attach_raw_usage_snapshot(response, raw_usage)
+            raise ModelBehaviorError(
+                "Chat Completions stream ended before receiving a finish_reason for choice 0."
+            )
+
         # A completion truncated before any visible token (finish_reason ==
         # "length") is a token- or reasoning-budget exhaustion, not a policy
         # refusal. Surface it as a model behavior error rather than manufacturing

src/agents/models/openai_chatcompletions.py

@@ -494,6 +494,7 @@ async def stream_response(
                     model=self.model,
                     strict_feature_validation=self._strict_feature_validation,
                     raise_on_length_truncation=True,
+                    require_finish_reason=ChatCmplHelpers.is_openai(self._client),
                     **raw_usage_options,
                 ):
                     if chunk.type == "response.completed":

tests/models/test_chatcompletions_stream_eof.py

@@ -0,0 +1,241 @@
+"""Exercise terminal detection through the production OpenAI SSE parser and Runner."""
+
+import json
+from typing import Any
+
+import httpx2
+import pytest
+from openai import AsyncOpenAI
+
+from agents import Agent, ModelSettings, RunConfig, Runner
+from agents.decorators import tool
+from agents.exceptions import ModelBehaviorError
+from agents.models.openai_chatcompletions import OpenAIChatCompletionsModel
+from tests.testing_processor import fetch_ordered_spans
+
+pytestmark = [pytest.mark.asyncio, pytest.mark.allow_call_model_methods]
+
+
+def _sse_body(delta: dict[str, Any] | None, finish_reason: str | None) -> bytes:
+    # Mock only the HTTP transport: parsing and adapter finalization stay real.
+    chunks = []
+    if delta is not None:
+        chunks.append({"index": 0, "delta": delta, "finish_reason": None})
+    if finish_reason is not None:
+        chunks.append({"index": 0, "delta": {}, "finish_reason": finish_reason})
+    payloads = [
+        {
+            "id": "chatcmpl-synthetic",
+            "object": "chat.completion.chunk",
+            "created": 0,
+            "model": "gpt-4o-mini",
+            "choices": [choice],
+        }
+        for choice in chunks
+    ]
+    if finish_reason is not None:
+        payloads.append(
+            {
+                "id": "chatcmpl-synthetic",
+                "object": "chat.completion.chunk",
+                "created": 0,
+                "model": "gpt-4o-mini",
+                "choices": [],
+                "usage": {"prompt_tokens": 3, "completion_tokens": 2, "total_tokens": 5},
+            }
+        )
+    body = "".join(f"data: {json.dumps(payload)}\n\n" for payload in payloads)
+    return body.encode()
+
+
+@pytest.mark.parametrize("buffered", [False, True])
+@pytest.mark.parametrize("output", ["text", "tool", "empty"])
+async def test_official_stream_rejects_eof_before_terminal_choice(
+    buffered: bool, output: str
+) -> None:
+    calls = []
+
+    @tool
+    def lookup() -> str:
+        calls.append("lookup")
+        return "result"
+
+    delta = {
+        "text": {"content": "The answer is "},
+        "tool": {
+            "tool_calls": [
+                {
+                    "index": 0,
+                    "id": "call-synthetic",
+                    "type": "function",
+                    "function": {"name": "lookup", "arguments": "{}"},
+                }
+            ]
+        },
+        "empty": None,
+    }[output]
+    http_response = httpx2.Response(
+        200, content=_sse_body(delta, None), headers={"content-type": "text/event-stream"}
+    )
+    async with AsyncOpenAI(
+        api_key="synthetic",
+        base_url="https://api.openai.com/v1",
+        http_client=httpx2.AsyncClient(
+            transport=httpx2.MockTransport(lambda request: http_response), trust_env=False
+        ),
+    ) as client:
+        agent = Agent(
+            name="Synthetic",
+            tools=[lookup],
+            model=OpenAIChatCompletionsModel(
+                "gpt-4o-mini", client, buffer_streamed_tool_calls=buffered
+            ),
+        )
+        result = Runner.run_streamed(
+            agent, "Answer", run_config=RunConfig(trace_include_sensitive_data=False)
+        )
+        raw_types = []
+        with pytest.raises(ModelBehaviorError, match="before receiving a finish_reason"):
+            async for event in result.stream_events():
+                if event.type == "raw_response_event":
+                    raw_types.append(event.data.type)
+
+    assert "response.completed" not in raw_types
+    assert "response.output_item.done" not in raw_types
+    assert result.final_output is None
+    assert calls == []
+    assert http_response.is_closed
+    generation = next(span for span in fetch_ordered_spans() if span.span_data.type == "generation")
+    assert generation.span_data.usage is not None
+    assert generation.span_data.usage["requests"] == 1
+    assert generation.span_data.usage["total_tokens"] == 0
+
+
+@pytest.mark.parametrize("buffered", [False, True])
+@pytest.mark.parametrize("finish_reason", ["stop", "length"])
+async def test_official_stream_accepts_terminal_choice_with_usage_trailer(
+    buffered: bool, finish_reason: str
+) -> None:
+    body = _sse_body({"content": "Answer"}, finish_reason) + b"data: [DONE]\n\n"
+    async with AsyncOpenAI(
+        api_key="synthetic",
+        base_url="https://api.openai.com/v1",
+        http_client=httpx2.AsyncClient(
+            transport=httpx2.MockTransport(
+                lambda request: httpx2.Response(
+                    200, content=body, headers={"content-type": "text/event-stream"}
+                )
+            ),
+            trust_env=False,
+        ),
+    ) as client:
+        result = Runner.run_streamed(
+            Agent(
+                name="Synthetic",
+                model=OpenAIChatCompletionsModel(
+                    "gpt-4o-mini", client, buffer_streamed_tool_calls=buffered
+                ),
+                model_settings=ModelSettings(preserve_raw_usage=True),
+            ),
+            "Answer",
+            run_config=RunConfig(tracing_disabled=True),
+        )
+        events = [event async for event in result.stream_events()]
+
+    assert result.final_output == "Answer"
+    assert (
+        sum(
+            event.type == "raw_response_event" and event.data.type == "response.completed"
+            for event in events
+        )
+        == 1
+    )
+    assert result.context_wrapper.usage.total_tokens == 5
+
+
+@pytest.mark.parametrize("buffered", [False, True])
+async def test_third_party_stream_preserves_eof_completion(buffered: bool) -> None:
+    body = _sse_body({"content": "Answer"}, None)
+    async with AsyncOpenAI(
+        api_key="synthetic",
+        base_url="https://provider.example/v1",
+        http_client=httpx2.AsyncClient(
+            transport=httpx2.MockTransport(
+                lambda request: httpx2.Response(
+                    200, content=body, headers={"content-type": "text/event-stream"}
+                )
+            ),
+            trust_env=False,
+        ),
+    ) as client:
+        result = Runner.run_streamed(
+            Agent(
+                name="Synthetic",
+                model=OpenAIChatCompletionsModel(
+                    "provider-model", client, buffer_streamed_tool_calls=buffered
+                ),
+            ),
+            "Answer",
+            run_config=RunConfig(tracing_disabled=True),
+        )
+        async for _ in result.stream_events():
+            pass
+
+    assert result.final_output == "Answer"
+
+
+@pytest.mark.parametrize("buffered", [False, True])
+async def test_official_stream_executes_tool_after_terminal_choice(buffered: bool) -> None:
+    calls = []
+
+    @tool
+    def lookup() -> str:
+        calls.append("lookup")
+        return "result"
+
+    body = (
+        _sse_body(
+            {
+                "tool_calls": [
+                    {
+                        "index": 0,
+                        "id": "call-synthetic",
+                        "type": "function",
+                        "function": {"name": "lookup", "arguments": "{}"},
+                    }
+                ]
+            },
+            "tool_calls",
+        )
+        + b"data: [DONE]\n\n"
+    )
+    async with AsyncOpenAI(
+        api_key="synthetic",
+        base_url="https://api.openai.com/v1",
+        http_client=httpx2.AsyncClient(
+            transport=httpx2.MockTransport(
+                lambda request: httpx2.Response(
+                    200, content=body, headers={"content-type": "text/event-stream"}
+                )
+            ),
+            trust_env=False,
+        ),
+    ) as client:
+        result = Runner.run_streamed(
+            Agent(
+                name="Synthetic",
+                tools=[lookup],
+                tool_use_behavior="stop_on_first_tool",
+                model=OpenAIChatCompletionsModel(
+                    "gpt-4o-mini", client, buffer_streamed_tool_calls=buffered
+                ),
+            ),
+            "Answer",
+            run_config=RunConfig(tracing_disabled=True),
+        )
+        async for _ in result.stream_events():
+            pass
+
+    assert calls == ["lookup"]
+    assert result.final_output == "result"
+    assert result.context_wrapper.usage.total_tokens == 5