[agents] Reject unpaired function outputs in Chat Completions

src/agents/models/chatcmpl_converter.py

@@ -818,6 +818,13 @@ def ensure_assistant_message() -> ChatCompletionAssistantMessageParam:
                 asst["tool_calls"] = tool_calls
             # 5) function call output => tool message
             elif func_output := cls.maybe_function_tool_call_output(item):
+                call_id = func_output.get("call_id")
+                if call_id is None:
+                    raise UserError(
+                        "Unpaired function outputs are supported by Responses but cannot be "
+                        "converted to Chat Completions tool messages. "
+                        "Use a Responses model to preserve this input."
+                    )
                 flush_assistant_message()
                 output_content = cast(
                     str | Iterable[ResponseInputContentWithAudioParam], func_output["output"]
@@ -849,7 +856,7 @@ def ensure_assistant_message() -> ChatCompletionAssistantMessageParam:
                             tool_result_content = _OMITTED_TOOL_OUTPUT_PLACEHOLDER
                 msg: ChatCompletionToolMessageParam = {
                     "role": "tool",
-                    "tool_call_id": func_output["call_id"],
+                    "tool_call_id": call_id,
                     "content": tool_result_content,  # type: ignore[typeddict-item]
                 }
                 result.append(msg)

tests/models/test_openai_chatcompletions.py

@@ -743,6 +743,58 @@ async def patched_fetch_response(self, *args, **kwargs):
         )
 
 
+@pytest.mark.allow_call_model_methods
+@pytest.mark.asyncio
+@pytest.mark.parametrize("call_id_fields", [{}, {"call_id": None}], ids=["omitted", "null"])
+@pytest.mark.parametrize("stream", [False, True], ids=["non_streaming", "streaming"])
+@pytest.mark.parametrize("strict_feature_validation", [False, True], ids=["default", "strict"])
+async def test_unpaired_function_output_rejected_before_chat_request(
+    call_id_fields: dict[str, None], stream: bool, strict_feature_validation: bool
+) -> None:
+    """Unpaired Responses context cannot become a Chat Completions tool message."""
+    requests: list[httpx2.Request] = []
+
+    async def handler(request: httpx2.Request) -> httpx2.Response:
+        requests.append(request)
+        raise AssertionError("Unpaired outputs must not reach Chat Completions")
+
+    async with httpx2.AsyncClient(transport=httpx2.MockTransport(handler)) as http_client:
+        model = OpenAIChatCompletionsModel(
+            model="gpt-4",
+            openai_client=AsyncOpenAI(api_key="test-key", http_client=http_client),
+            strict_feature_validation=strict_feature_validation,
+        )
+        request_kwargs: dict[str, Any] = {
+            "system_instructions": None,
+            "input": [
+                {
+                    "type": "function_call_output",
+                    "name": "notifications",
+                    "namespace": "slack",
+                    "output": "Alice mentioned you in #deployments.",
+                    **call_id_fields,
+                }
+            ],
+            "model_settings": ModelSettings(),
+            "tools": [],
+            "output_schema": None,
+            "handoffs": [],
+            "tracing": ModelTracing.DISABLED,
+        }
+
+        with pytest.raises(
+            UserError,
+            match="Unpaired function outputs.*Chat Completions.*Use a Responses model",
+        ):
+            if stream:
+                async for _ in model.stream_response(**request_kwargs):
+                    pass
+            else:
+                await model.get_response(**request_kwargs)
+
+    assert requests == []
+
+
 @pytest.mark.allow_call_model_methods
 @pytest.mark.asyncio
 async def test_get_response_rejects_non_text_tool_output_in_strict_mode() -> None:

tests/models/test_openai_responses.py

@@ -117,6 +117,64 @@ async def handler(request: httpx2.Request) -> httpx2.Response:
     return requests
 
 
+@pytest.mark.allow_call_model_methods
+@pytest.mark.asyncio
+@pytest.mark.parametrize("call_id_fields", [{}, {"call_id": None}], ids=["omitted", "null"])
+@pytest.mark.parametrize("stream", [False, True], ids=["non_streaming", "streaming"])
+async def test_unpaired_function_output_preserved_in_responses_request(
+    call_id_fields: dict[str, None], stream: bool
+) -> None:
+    """Responses keeps external-context output and its source without inventing a call ID."""
+    request_bodies: list[dict[str, Any]] = []
+
+    async def handler(request: httpx2.Request) -> httpx2.Response:
+        request_bodies.append(json.loads(request.content))
+        if stream:
+            event = _response_completed_frame("resp-id", sequence_number=0)
+            return httpx2.Response(
+                200,
+                content=f"event: response.completed\ndata: {event}\n\n",
+                headers={"content-type": "text/event-stream"},
+            )
+        return httpx2.Response(
+            200,
+            content=get_response_obj([]).model_dump_json(),
+            headers={"content-type": "application/json"},
+        )
+
+    expected_input = {
+        "type": "function_call_output",
+        "name": "notifications",
+        "namespace": "slack",
+        "output": [
+            {"type": "input_text", "text": "Alice mentioned you in #deployments."},
+            {"type": "input_image", "image_url": "https://example.com/image.png"},
+        ],
+        **call_id_fields,
+    }
+    async with httpx2.AsyncClient(transport=httpx2.MockTransport(handler)) as http_client:
+        model = OpenAIResponsesModel(
+            model="gpt-4",
+            openai_client=AsyncOpenAI(api_key="test-key", http_client=http_client),
+        )
+        request_kwargs: dict[str, Any] = {
+            "system_instructions": None,
+            "input": [dict(expected_input)],
+            "model_settings": ModelSettings(),
+            "tools": [],
+            "output_schema": None,
+            "handoffs": [],
+            "tracing": ModelTracing.DISABLED,
+        }
+        if stream:
+            async for _ in model.stream_response(**request_kwargs):
+                pass
+        else:
+            await model.get_response(**request_kwargs)
+
+    assert [body["input"] for body in request_bodies] == [[expected_input]]
+
+
 class DummyWSConnection:
     def __init__(self, frames: list[str]):
         self._frames = frames