Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
19 changes: 18 additions & 1 deletion src/agents/run_internal/run_loop.py
Original file line number Diff line number Diff line change
Expand Up @@ -2645,10 +2645,21 @@ async def rewind_model_request() -> None:
),
)

stream_event_count = 0
last_stream_event_type: str | None = None
stream_response_id: str | None = None
stream_request_id: str | None = None

# Raising out of this loop leaves the model stream suspended at its yield, so close it
# explicitly instead of waiting for garbage collection to finalize the generator chain.
async with aclosing(model_run_context_stream(retry_stream, tool_use_tracker)) as model_events:
async for event in model_events:
stream_event_count += 1
last_stream_event_type = getattr(event, "type", None)
event_response = getattr(event, "response", None)
if isinstance(event_response, Response):
stream_response_id = event_response.id
stream_request_id = getattr(event_response, "_request_id", None)
streamed_result._event_queue.put_nowait(RawResponsesStreamEvent(data=event))

terminal_response: Response | None = None
Expand Down Expand Up @@ -2702,7 +2713,13 @@ async def rewind_model_request() -> None:
streamed_response_output.append(event.item)

if final_response is None:
raise ModelBehaviorError("Model did not produce a final response!")
raise ModelBehaviorError(
"Model did not produce a final response! The stream ended before a "
"response.completed event was received. "
f"event_count={stream_event_count}, last_event_type={last_stream_event_type!r}, "
f"response_id={stream_response_id!r}, request_id={stream_request_id!r}. "
"Share these IDs when reporting this issue; None means no value was observed."
)

record_compaction_model_response(
session, context_wrapper, compaction_input_digests, final_response, reasoning_item_id_policy
Expand Down
132 changes: 132 additions & 0 deletions tests/models/test_responses_stream_diagnostics.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,132 @@
"""Exercise missing-completion diagnostics through the real 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.exceptions import ModelBehaviorError
from agents.models.openai_responses import OpenAIResponsesModel
from agents.retry import ModelRetrySettings
from tests.model_test_helpers import get_response_obj
from tests.test_responses import get_text_message

pytestmark = [pytest.mark.asyncio, pytest.mark.allow_call_model_methods]


def _events(terminal: str | None = None) -> list[dict[str, Any]]:
response = get_response_obj([], response_id="resp_synthetic").model_dump()
response["status"] = "in_progress"
response["instructions"] = "private synthetic instructions"
events: list[dict[str, Any]] = [
{"type": "response.created", "sequence_number": 0, "response": response},
{
"type": "response.output_text.delta",
"sequence_number": 1,
"item_id": "msg_synthetic",
"output_index": 0,
"content_index": 0,
"delta": "private synthetic output",
"logprobs": [],
},
]
if terminal is not None:
final_response = get_response_obj(
[get_text_message("Answer")], response_id="resp_synthetic"
).model_dump()
final_response["status"] = terminal
if terminal == "incomplete":
final_response["incomplete_details"] = {"reason": "max_output_tokens"}
events.append(
{
"type": f"response.{terminal}",
"sequence_number": 2,
"response": final_response,
}
)
return events


def _body(events: list[dict[str, Any]]) -> bytes:
# Mock the HTTP transport, retaining real event parsing and request-ID propagation.
return "".join(
f"event: {event['type']}\ndata: {json.dumps(event)}\n\n" for event in events
).encode()


@pytest.mark.parametrize("empty", [False, True], ids=["partial-output", "empty"])
@pytest.mark.parametrize("request_id", ["req_synthetic", None], ids=["request-id", "no-request-id"])
async def test_missing_completion_reports_observed_metadata(
empty: bool, request_id: str | None
) -> None:
headers = {"content-type": "text/event-stream"}
if request_id is not None:
headers["x-request-id"] = request_id
http_response = httpx2.Response(200, headers=headers, content=_body([] if empty else _events()))
requests: list[httpx2.Request] = []

def respond(request: httpx2.Request) -> httpx2.Response:
requests.append(request)
return http_response

async with AsyncOpenAI(
api_key="synthetic",
http_client=httpx2.AsyncClient(transport=httpx2.MockTransport(respond), trust_env=False),
) as client:
result = Runner.run_streamed(
Agent(
name="Synthetic",
model=OpenAIResponsesModel("synthetic-model", client),
model_settings=ModelSettings(
retry=ModelRetrySettings(max_retries=1, policy=lambda _: True)
),
),
"private synthetic input",
run_config=RunConfig(trace_include_sensitive_data=False),
)
with pytest.raises(ModelBehaviorError) as exc_info:
async for _ in result.stream_events():
pass

message = str(exc_info.value)
assert message.startswith("Model did not produce a final response!")
assert "stream ended before a response.completed event was received" in message
assert f"event_count={0 if empty else 2}" in message
assert f"last_event_type={None if empty else 'response.output_text.delta'!r}" in message
assert f"response_id={None if empty else 'resp_synthetic'!r}" in message
assert f"request_id={None if empty else request_id!r}" in message
assert "private synthetic" not in message
assert len(requests) == 1
assert result.final_output is None
assert http_response.is_closed


@pytest.mark.parametrize("terminal", ["completed", "incomplete", "failed"])
async def test_explicit_terminal_retains_existing_behavior(terminal: str) -> None:
http_response = httpx2.Response(
200, content=_body(_events(terminal)), headers={"content-type": "text/event-stream"}
)
async with AsyncOpenAI(
api_key="synthetic",
http_client=httpx2.AsyncClient(
transport=httpx2.MockTransport(lambda _: http_response), trust_env=False
),
) as client:
result = Runner.run_streamed(
Agent(name="Synthetic", model=OpenAIResponsesModel("synthetic-model", client)),
"Answer",
run_config=RunConfig(tracing_disabled=True),
)
if terminal == "completed":
async for _ in result.stream_events():
pass
assert result.final_output == "Answer"
else:
with pytest.raises(ModelBehaviorError, match=f"terminal event `response.{terminal}`"):
async for _ in result.stream_events():
pass
assert result.final_output is None
assert http_response.is_closed
Loading