From 93233dac21d42d54881a0a5d706c84048cafcca1 Mon Sep 17 00:00:00 2001 From: Justin Beckwith Date: Sun, 27 Sep 2026 18:04:33 -0700 Subject: [PATCH 1/2] fix: reject premature Chat Completions stream EOF --- src/agents/models/chatcmpl_stream_handler.py | 25 +- src/agents/models/openai_chatcompletions.py | 1 + .../models/test_chatcompletions_stream_eof.py | 234 ++++++++++++++++++ .../test_openai_chatcompletions_stream.py | 34 ++- tests/models/test_reasoning_content.py | 3 + 5 files changed, 286 insertions(+), 11 deletions(-) create mode 100644 tests/models/test_chatcompletions_stream_eof.py diff --git a/src/agents/models/chatcmpl_stream_handler.py b/src/agents/models/chatcmpl_stream_handler.py index 289e116b06..797f90fb96 100644 --- a/src/agents/models/chatcmpl_stream_handler.py +++ b/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,15 @@ 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) + 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 diff --git a/src/agents/models/openai_chatcompletions.py b/src/agents/models/openai_chatcompletions.py index 292348a8e0..71990fc7e4 100644 --- a/src/agents/models/openai_chatcompletions.py +++ b/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": diff --git a/tests/models/test_chatcompletions_stream_eof.py b/tests/models/test_chatcompletions_stream_eof.py new file mode 100644 index 0000000000..eda1eb5bad --- /dev/null +++ b/tests/models/test_chatcompletions_stream_eof.py @@ -0,0 +1,234 @@ +"""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 + +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(tracing_disabled=True)) + 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 + + +@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 diff --git a/tests/models/test_openai_chatcompletions_stream.py b/tests/models/test_openai_chatcompletions_stream.py index 7d484981f2..3aa3b2cf77 100644 --- a/tests/models/test_openai_chatcompletions_stream.py +++ b/tests/models/test_openai_chatcompletions_stream.py @@ -280,6 +280,7 @@ async def test_stream_response_yields_events_for_text_content(monkeypatch) -> No async def fake_stream() -> AsyncIterator[ChatCompletionChunk]: for c in (chunk1, chunk2): yield c + yield _chunk_with([Choice(index=0, delta=ChoiceDelta(), finish_reason="stop")]) # Patch _fetch_response to inject our fake stream async def patched_fetch_response(self, *args, **kwargs): @@ -1681,7 +1682,9 @@ async def test_stream_response_warns_and_ignores_server_managed_conversation_sta async def patched_fetch_response(self, *args, **kwargs): nonlocal called called = True - return _empty_response(), _empty_chat_completion_stream() + return _empty_response(), _completion_stream( + _chunk_with([Choice(index=0, delta=ChoiceDelta(), finish_reason="stop")]) + ) monkeypatch.setattr(OpenAIChatCompletionsModel, "_fetch_response", patched_fetch_response) model = OpenAIProvider(use_responses=False).get_model("gpt-4") @@ -1716,7 +1719,9 @@ async def test_stream_response_warns_and_ignores_prompt_by_default( async def patched_fetch_response(self, *args, **kwargs): nonlocal captured_prompt captured_prompt = kwargs.get("prompt") - return _empty_response(), _empty_chat_completion_stream() + return _empty_response(), _completion_stream( + _chunk_with([Choice(index=0, delta=ChoiceDelta(), finish_reason="stop")]) + ) monkeypatch.setattr(OpenAIChatCompletionsModel, "_fetch_response", patched_fetch_response) model = OpenAIProvider(use_responses=False).get_model("gpt-4") @@ -1874,6 +1879,7 @@ async def test_stream_response_includes_logprobs(monkeypatch) -> None: async def fake_stream() -> AsyncIterator[ChatCompletionChunk]: for c in (chunk1, chunk2): yield c + yield _chunk_with([Choice(index=0, delta=ChoiceDelta(), finish_reason="stop")]) async def patched_fetch_response(self, *args, **kwargs): resp = Response( @@ -1957,6 +1963,7 @@ def make_chunk(token: str) -> ChatCompletionChunk: async def fake_stream() -> AsyncIterator[ChatCompletionChunk]: for token in tokens: yield make_chunk(token) + yield _chunk_with([Choice(index=0, delta=ChoiceDelta(), finish_reason="stop")]) async def patched_fetch_response(self, *args, **kwargs): resp = Response( @@ -2029,6 +2036,7 @@ async def test_stream_response_yields_events_for_refusal_content(monkeypatch) -> async def fake_stream() -> AsyncIterator[ChatCompletionChunk]: for c in (chunk1, chunk2): yield c + yield _chunk_with([Choice(index=0, delta=ChoiceDelta(), finish_reason="stop")]) async def patched_fetch_response(self, *args, **kwargs): resp = Response( @@ -2120,6 +2128,7 @@ async def test_stream_response_yields_events_for_tool_call(monkeypatch) -> None: async def fake_stream() -> AsyncIterator[ChatCompletionChunk]: for c in (chunk1, chunk2): yield c + yield _chunk_with([Choice(index=0, delta=ChoiceDelta(), finish_reason="tool_calls")]) async def patched_fetch_response(self, *args, **kwargs): resp = Response( @@ -2208,6 +2217,7 @@ async def test_stream_response_buffers_tool_call_deltas_when_enabled(monkeypatch async def fake_stream() -> AsyncIterator[ChatCompletionChunk]: for chunk in (chunk1, chunk2): yield chunk + yield _chunk_with([Choice(index=0, delta=ChoiceDelta(), finish_reason="tool_calls")]) async def patched_fetch_response(self, *args, **kwargs): return _empty_response(), fake_stream() @@ -2404,6 +2414,7 @@ async def test_stream_response_buffers_tool_call_usage_chunk_without_replay( async def fake_stream() -> AsyncIterator[ChatCompletionChunk]: yield chunk + yield _chunk_with([Choice(index=0, delta=ChoiceDelta(), finish_reason="tool_calls")]) async def patched_fetch_response(self, *args, **kwargs): return _empty_response(), fake_stream() @@ -2483,6 +2494,7 @@ async def test_stream_response_buffers_tool_call_provider_fields(monkeypatch) -> async def fake_stream() -> AsyncIterator[ChatCompletionChunk]: for chunk in (chunk1, chunk2): yield chunk + yield _chunk_with([Choice(index=0, delta=ChoiceDelta(), finish_reason="tool_calls")]) async def patched_fetch_response(self, *args, **kwargs): return _empty_response(), fake_stream() @@ -2636,6 +2648,7 @@ async def test_buffered_tool_calls_do_not_merge_nonzero_choice_tool_call_indexes async def fake_stream() -> AsyncIterator[ChatCompletionChunk]: yield chunk + yield _chunk_with([Choice(index=0, delta=ChoiceDelta(), finish_reason="tool_calls")]) async def patched_fetch_response(self, *args, **kwargs): return _empty_response(), fake_stream() @@ -2872,6 +2885,7 @@ async def test_stream_response_ignores_custom_tool_call_chunks_by_default(monkey async def fake_stream() -> AsyncIterator[ChatCompletionChunk]: for chunk in chunks: yield chunk + yield _chunk_with([Choice(index=0, delta=ChoiceDelta(), finish_reason="tool_calls")]) async def patched_fetch_response(self, *args, **kwargs): return _empty_response(), fake_stream() @@ -2977,6 +2991,7 @@ async def test_stream_response_yields_real_time_function_call_arguments(monkeypa async def fake_stream() -> AsyncIterator[ChatCompletionChunk]: for c in (chunk1, chunk2, chunk3, chunk4): yield c + yield _chunk_with([Choice(index=0, delta=ChoiceDelta(), finish_reason="tool_calls")]) async def patched_fetch_response(self, *args, **kwargs): resp = Response( @@ -3091,6 +3106,7 @@ async def test_fallback_function_calls_have_unique_output_indexes(monkeypatch) - async def fake_stream() -> AsyncIterator[ChatCompletionChunk]: for c in (chunk1, chunk2): yield c + yield _chunk_with([Choice(index=0, delta=ChoiceDelta(), finish_reason="tool_calls")]) async def patched_fetch_response(self, *args, **kwargs): resp = Response( @@ -3192,6 +3208,7 @@ async def test_fallback_function_call_keeps_index_before_streamed_call(monkeypat async def fake_stream() -> AsyncIterator[ChatCompletionChunk]: for c in (chunk1, chunk2, chunk3): yield c + yield _chunk_with([Choice(index=0, delta=ChoiceDelta(), finish_reason="tool_calls")]) async def patched_fetch_response(self, *args, **kwargs): resp = Response( @@ -3286,6 +3303,7 @@ async def test_fallback_function_call_before_text_uses_final_output_index( async def fake_stream() -> AsyncIterator[ChatCompletionChunk]: for chunk in (chunk1, chunk2): yield chunk + yield _chunk_with([Choice(index=0, delta=ChoiceDelta(), finish_reason="tool_calls")]) async def patched_fetch_response(self, *args, **kwargs): response = Response( @@ -3389,6 +3407,7 @@ async def test_streamed_function_call_before_text_keeps_realtime_order( async def fake_stream() -> AsyncIterator[ChatCompletionChunk]: for chunk in (chunk1, chunk2, chunk3): yield chunk + yield _chunk_with([Choice(index=0, delta=ChoiceDelta(), finish_reason="tool_calls")]) async def patched_fetch_response(self, *args, **kwargs): response = Response( @@ -3504,6 +3523,7 @@ async def test_mixed_function_calls_before_text_keep_tracked_order( async def fake_stream() -> AsyncIterator[ChatCompletionChunk]: for chunk in (chunk1, chunk2, chunk3, chunk4): yield chunk + yield _chunk_with([Choice(index=0, delta=ChoiceDelta(), finish_reason="tool_calls")]) async def patched_fetch_response(self, *args, **kwargs): response = Response( @@ -3930,8 +3950,7 @@ async def source() -> AsyncIterator[ChatCompletionChunk]: @pytest.mark.asyncio async def test_buffer_tool_call_stream_does_not_duplicate_tool_calls_finish() -> None: - """finish_reason == "tool_calls" is still emitted only by the synthesized - buffered chunk, so the terminal choice is not forwarded twice.""" + """Only the real terminal choice carries a finish reason after buffering.""" tool_call_delta = ChoiceDeltaToolCall( index=0, id="tool-id", @@ -3956,7 +3975,9 @@ async def source() -> AsyncIterator[ChatCompletionChunk]: if choice.finish_reason == "tool_calls" ] assert len(finish_choices) == 1 - assert finish_choices[0].delta.tool_calls + assert finish_choices[0].delta.tool_calls is None + assert buffered[-1].choices[0].delta.tool_calls == [tool_call_delta] + assert buffered[-1].choices[0].finish_reason is None @pytest.mark.allow_call_model_methods @@ -3989,6 +4010,7 @@ def __init__(self) -> None: def __aiter__(self) -> AsyncIterator[ChatCompletionChunk]: async def gen() -> AsyncIterator[ChatCompletionChunk]: yield chunk + yield _chunk_with([Choice(index=0, delta=ChoiceDelta(), finish_reason="stop")]) return gen() @@ -4042,6 +4064,7 @@ async def test_stream_response_without_http_response_has_no_request_id(monkeypat async def fake_stream() -> AsyncIterator[ChatCompletionChunk]: yield chunk + yield _chunk_with([Choice(index=0, delta=ChoiceDelta(), finish_reason="stop")]) async def patched_fetch_response(self, *args, **kwargs): resp = Response( @@ -4222,6 +4245,7 @@ def _usageless_stream_patch(usage: CompletionUsage | None = None): async def fake_stream() -> AsyncIterator[ChatCompletionChunk]: yield chunk + yield _chunk_with([Choice(index=0, delta=ChoiceDelta(), finish_reason="stop")]) async def patched_fetch_response(self, *args, **kwargs): resp = Response( diff --git a/tests/models/test_reasoning_content.py b/tests/models/test_reasoning_content.py index fc86dafd12..765389162b 100644 --- a/tests/models/test_reasoning_content.py +++ b/tests/models/test_reasoning_content.py @@ -391,6 +391,9 @@ async def create_fake_stream( ) -> AsyncIterator[ChatCompletionChunk]: for chunk in chunks: yield chunk + terminal_chunk = create_chunk({}) + terminal_chunk.choices[0].finish_reason = "stop" + yield terminal_chunk @pytest.mark.allow_call_model_methods From af0ea10395d6961656ea4b796c20e22f5a5121e9 Mon Sep 17 00:00:00 2001 From: Justin Beckwith Date: Sun, 27 Sep 2026 18:15:49 -0700 Subject: [PATCH 2/2] fix: count interrupted Chat Completions requests in traces --- src/agents/models/chatcmpl_stream_handler.py | 2 ++ tests/models/test_chatcompletions_stream_eof.py | 9 ++++++++- 2 files changed, 10 insertions(+), 1 deletion(-) diff --git a/src/agents/models/chatcmpl_stream_handler.py b/src/agents/models/chatcmpl_stream_handler.py index 797f90fb96..b0c041f9cb 100644 --- a/src/agents/models/chatcmpl_stream_handler.py +++ b/src/agents/models/chatcmpl_stream_handler.py @@ -1193,6 +1193,8 @@ async def handle_stream( 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( diff --git a/tests/models/test_chatcompletions_stream_eof.py b/tests/models/test_chatcompletions_stream_eof.py index eda1eb5bad..6758bccfae 100644 --- a/tests/models/test_chatcompletions_stream_eof.py +++ b/tests/models/test_chatcompletions_stream_eof.py @@ -11,6 +11,7 @@ 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] @@ -90,7 +91,9 @@ def lookup() -> str: "gpt-4o-mini", client, buffer_streamed_tool_calls=buffered ), ) - result = Runner.run_streamed(agent, "Answer", run_config=RunConfig(tracing_disabled=True)) + 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(): @@ -102,6 +105,10 @@ def lookup() -> str: 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])