567-labs--instructor/tests/coverage/test_openai_handlers_coverage.py

from __future__ import annotations

import json
from collections.abc import Iterable
from types import SimpleNamespace
from typing import Any, Union, cast

import pytest
from openai.types.chat import ChatCompletionChunk
from openai.types.responses import (
    ResponseFunctionCallArgumentsDeltaEvent,
    ResponseFunctionToolCall,
    ResponseOutputMessage,
    ResponseOutputText,
)
from pydantic import BaseModel

from instructor.v2.core.errors import (
    ConfigurationError,
    IncompleteOutputException,
    ResponseParsingError,
)
from instructor.v2.core.mode import Mode
from instructor.v2.dsl.iterable import IterableModel
from instructor.v2.dsl.parallel import ParallelBase
from instructor.v2.dsl.partial import Partial
from instructor.v2.dsl.simple_type import ModelAdapter
from instructor.v2.providers.openai import handlers
from instructor.v2.providers.openai.handlers import (
    OpenAIHandlerBase,
    OpenAIJSONHandler,
    OpenAIJSONSchemaHandler,
    OpenAIMDJSONHandler,
    OpenAIParallelToolsHandler,
    OpenAIResponsesToolsHandler,
    OpenAIToolsHandler,
    reask_default,
    reask_responses_tools,
)
from tests.coverage._openai import chat_chunk, chat_completion, tool_call
from tests.coverage._streams import async_items


class User(BaseModel):
    name: str


class Search(BaseModel):
    query: str


def base_handler(mode: Mode) -> OpenAIHandlerBase:
    handler = OpenAIToolsHandler()
    handler.mode = mode
    return handler


def response_deltas(
    item_id: str, *deltas: str
) -> list[ResponseFunctionCallArgumentsDeltaEvent]:
    return [
        ResponseFunctionCallArgumentsDeltaEvent(
            delta=delta,
            item_id=item_id,
            output_index=0,
            sequence_number=index,
            type="response.function_call_arguments.delta",
        )
        for index, delta in enumerate(deltas, start=1)
    ]


def test_responses_tool_filter_accepts_legacy_items_and_formats_missing_details() -> (
    None
):
    legacy_call = SimpleNamespace(arguments='{"name":"Ada"}')
    ignored_message = SimpleNamespace(type="message", content="hello")

    assert handlers._filter_responses_tool_calls([ignored_message, legacy_call]) == [
        legacy_call
    ]
    assert handlers._format_responses_tool_call_details(legacy_call) == ""


def test_reask_default_keeps_assistant_message_before_correction() -> None:
    response = chat_completion(content='{"name": 2}')

    result = reask_default(
        {"messages": [{"role": "user", "content": "extract a user"}]},
        response,
        ValueError("name must be a string"),
    )

    assert result["messages"][1]["role"] == "assistant"
    assert result["messages"][1]["content"] == '{"name": 2}'
    assert result["messages"][2] == {
        "role": "user",
        "content": (
            "Recall the function correctly, fix the errors, exceptions found\n"
            "name must be a string"
        ),
    }


def test_tools_reask_preserves_assistant_calls_and_adds_one_tool_error_per_call() -> (
    None
):
    response = chat_completion(
        tool_calls=[
            tool_call("User", '{"name":3}', "call_user"),
            tool_call("Search", '{"query":4}', "call_search"),
        ]
    )

    result = OpenAIToolsHandler().handle_reask(
        {"messages": [{"role": "user", "content": "extract both values"}]},
        response,
        ValueError("invalid fields"),
    )

    assert result["messages"] == [
        {"role": "user", "content": "extract both values"},
        {
            "role": "assistant",
            "content": "",
            "tool_calls": [
                tool_call("User", '{"name":3}', "call_user").model_dump(),
                tool_call("Search", '{"query":4}', "call_search").model_dump(),
            ],
        },
        {
            "role": "tool",
            "tool_call_id": "call_user",
            "name": "User",
            "content": (
                "Validation Error found:\ninvalid fields\n"
                "Recall the function correctly, fix the errors"
            ),
        },
        {
            "role": "tool",
            "tool_call_id": "call_search",
            "name": "Search",
            "content": (
                "Validation Error found:\ninvalid fields\n"
                "Recall the function correctly, fix the errors"
            ),
        },
    ]


def test_responses_reask_uses_legacy_arguments_without_inventing_call_details() -> None:
    response = SimpleNamespace(output=[SimpleNamespace(arguments='{"name": 3}')])

    result = reask_responses_tools(
        {"messages": []}, response, ValueError("name must be a string")
    )

    assert result["messages"] == [
        {
            "role": "user",
            "content": (
                "Validation Error found:\nname must be a string\n"
                'Recall the function correctly, fix the errors with {"name": 3}'
            ),
        }
    ]


def test_streaming_flags_ignore_none_non_classes_and_non_streaming_models() -> None:
    handler = OpenAIToolsHandler()
    iterable_user = IterableModel(User)

    handler.mark_streaming_model(None, True)
    handler.mark_streaming_model(iterable_user, False)
    handler.mark_streaming_model(User, True)

    assert handler._consume_streaming_flag(None) is False
    assert handler._consume_streaming_flag(ParallelBase(User)) is False
    assert handler._consume_streaming_flag(iterable_user) is False


def test_responses_tools_replaces_non_dict_text_format_and_preserves_options() -> None:
    original_text = {"format": "json_object", "verbosity": "low"}

    _, kwargs = OpenAIResponsesToolsHandler().prepare_request(
        User, {"text": original_text}
    )

    assert kwargs["text"] is not original_text
    assert kwargs["text"]["verbosity"] == "low"
    assert kwargs["text"]["format"] == {
        "type": "json_schema",
        "name": "User",
        "strict": True,
        "schema": kwargs["tools"][0]["parameters"],
    }


@pytest.mark.parametrize(
    ("mode", "valid_chunk", "expected"),
    [
        (
            Mode.FUNCTIONS,
            chat_chunk(
                {"function_call": {"name": "User", "arguments": '{"name":"Ada"}'}}
            ),
            ['{"name":"Ada"}'],
        ),
        (Mode.JSON, chat_chunk({"content": '{"name":"Ada"}'}), ['{"name":"Ada"}']),
        (
            Mode.JSON_SCHEMA,
            chat_chunk({"content": '{"name":"Ada"}'}),
            ['{"name":"Ada"}'],
        ),
        (
            Mode.TOOLS,
            chat_chunk(
                {
                    "tool_calls": [
                        {
                            "index": 0,
                            "id": "call_1",
                            "type": "function",
                            "function": {"name": "User", "arguments": '{"name":"Ada"}'},
                        }
                    ]
                }
            ),
            ['{"name":"Ada"}'],
        ),
        (
            Mode.MD_JSON,
            chat_chunk(
                {"content": 'Here is the result:\n```json\n{"name":"Ada"}\n```'}
            ),
            ['{"name":"Ada"}\n'],
        ),
    ],
)
def test_sync_stream_extractor_handles_chat_delta_modes_and_bad_chunks(
    mode: Mode,
    valid_chunk: ChatCompletionChunk,
    expected: list[str],
    monkeypatch: pytest.MonkeyPatch,
) -> None:
    monkeypatch.setattr(
        Mode, "warn_mode_functions_deprecation", staticmethod(lambda: None)
    )
    chunks = [
        SimpleNamespace(choices=[]),
        SimpleNamespace(choices=[SimpleNamespace(delta=SimpleNamespace())]),
        valid_chunk,
    ]

    result = list(base_handler(mode).extract_streaming_json(chunks))
    if mode is Mode.MD_JSON:
        assert json.loads("".join(result)) == {"name": "Ada"}
    else:
        assert result == expected


def test_sync_responses_stream_extractor_only_yields_argument_delta_events() -> None:
    event = ResponseFunctionCallArgumentsDeltaEvent(
        delta='{"name":"Ada"}',
        item_id="fc_1",
        output_index=0,
        sequence_number=1,
        type="response.function_call_arguments.delta",
    )

    assert list(
        base_handler(Mode.RESPONSES_TOOLS).extract_streaming_json(
            [SimpleNamespace(type="response.output_text.delta"), event]
        )
    ) == ['{"name":"Ada"}']


@pytest.mark.parametrize(
    ("mode", "delta"),
    [
        (Mode.FUNCTIONS, {"function_call": {"name": "User", "arguments": ""}}),
        (Mode.JSON, {"content": ""}),
        (Mode.TOOLS, {"tool_calls": []}),
        (
            Mode.TOOLS,
            {
                "tool_calls": [
                    {
                        "index": 0,
                        "id": "call_1",
                        "type": "function",
                        "function": {"name": "User", "arguments": None},
                    }
                ]
            },
        ),
        (Mode.COHERE_JSON_SCHEMA, {"content": '{"name":"ignored"}'}),
    ],
)
def test_sync_stream_extractor_ignores_empty_and_unsupported_deltas(
    mode: Mode, delta: dict[str, Any], monkeypatch: pytest.MonkeyPatch
) -> None:
    monkeypatch.setattr(
        Mode, "warn_mode_functions_deprecation", staticmethod(lambda: None)
    )

    assert list(base_handler(mode).extract_streaming_json([chat_chunk(delta)])) == []


@pytest.mark.asyncio
@pytest.mark.parametrize(
    ("mode", "valid_chunk", "expected"),
    [
        (
            Mode.FUNCTIONS,
            chat_chunk(
                {"function_call": {"name": "User", "arguments": '{"name":"Ada"}'}}
            ),
            ['{"name":"Ada"}'],
        ),
        (Mode.JSON, chat_chunk({"content": '{"name":"Ada"}'}), ['{"name":"Ada"}']),
        (
            Mode.TOOLS,
            chat_chunk(
                {
                    "tool_calls": [
                        {
                            "index": 0,
                            "id": "call_1",
                            "type": "function",
                            "function": {"name": "User", "arguments": '{"name":"Ada"}'},
                        }
                    ]
                }
            ),
            ['{"name":"Ada"}'],
        ),
        (
            Mode.MD_JSON,
            chat_chunk({"content": '```json\n{"name":"Ada"}\n```'}),
            ['{"name":"Ada"}\n'],
        ),
    ],
)
async def test_async_stream_extractor_handles_chat_delta_modes_and_bad_chunks(
    mode: Mode,
    valid_chunk: ChatCompletionChunk,
    expected: list[str],
    monkeypatch: pytest.MonkeyPatch,
) -> None:
    monkeypatch.setattr(
        Mode, "warn_mode_functions_deprecation", staticmethod(lambda: None)
    )
    chunks = [
        SimpleNamespace(choices=[]),
        SimpleNamespace(choices=[SimpleNamespace(delta=SimpleNamespace())]),
        valid_chunk,
    ]

    result = [
        part
        async for part in base_handler(mode).extract_streaming_json_async(
            async_items(chunks)
        )
    ]
    if mode is Mode.MD_JSON:
        assert json.loads("".join(result)) == {"name": "Ada"}
    else:
        assert result == expected


@pytest.mark.asyncio
async def test_async_responses_stream_extractor_only_yields_argument_delta_events() -> (
    None
):
    event = ResponseFunctionCallArgumentsDeltaEvent(
        delta='{"name":"Ada"}',
        item_id="fc_1",
        output_index=0,
        sequence_number=1,
        type="response.function_call_arguments.delta",
    )

    assert [
        part
        async for part in base_handler(
            Mode.RESPONSES_TOOLS
        ).extract_streaming_json_async(
            async_items([SimpleNamespace(type="response.output_text.delta"), event])
        )
    ] == ['{"name":"Ada"}']


@pytest.mark.asyncio
@pytest.mark.parametrize(
    ("mode", "delta"),
    [
        (Mode.FUNCTIONS, {"function_call": {"name": "User", "arguments": ""}}),
        (Mode.JSON, {"content": ""}),
        (Mode.TOOLS, {"tool_calls": []}),
        (
            Mode.TOOLS,
            {
                "tool_calls": [
                    {
                        "index": 0,
                        "id": "call_1",
                        "type": "function",
                        "function": {"name": "User", "arguments": None},
                    }
                ]
            },
        ),
        (Mode.COHERE_JSON_SCHEMA, {"content": '{"name":"ignored"}'}),
    ],
)
async def test_async_stream_extractor_ignores_empty_and_unsupported_deltas(
    mode: Mode, delta: dict[str, Any], monkeypatch: pytest.MonkeyPatch
) -> None:
    monkeypatch.setattr(
        Mode, "warn_mode_functions_deprecation", staticmethod(lambda: None)
    )

    assert [
        part
        async for part in base_handler(mode).extract_streaming_json_async(
            async_items([chat_chunk(delta)])
        )
    ] == []


def test_parse_streaming_response_forwards_context_strict_and_mode() -> None:
    received: dict[str, Any] = {}

    class StreamingUser(BaseModel):
        @classmethod
        def from_streaming_response(
            cls,
            response: list[Any],
            *,
            stream_extractor: Any,
            mode: Mode,
            **kwargs: Any,
        ) -> Iterable[User]:
            received.update(
                response=response, extractor=stream_extractor, mode=mode, **kwargs
            )
            return iter([User(name="Ada")])

    response = [chat_chunk({"content": '{"name":"Ada"}'})]
    result = base_handler(Mode.JSON)._parse_streaming_response(
        StreamingUser,
        response,
        validation_context={"source": "chat"},
        strict=True,
    )

    assert result == [User(name="Ada")]
    assert received["response"] is response
    assert received["mode"] is Mode.JSON
    assert received["context"] == {"source": "chat"}
    assert received["strict"] is True
    assert callable(received["extractor"])


def test_parse_streaming_response_falls_back_when_signature_is_unavailable(
    monkeypatch: pytest.MonkeyPatch,
) -> None:
    class StreamingUser(BaseModel):
        @classmethod
        def from_streaming_response(
            cls, _response: list[Any], *, stream_extractor: Any, **kwargs: Any
        ) -> Iterable[User]:
            assert "mode" not in kwargs
            assert callable(stream_extractor)
            return iter([User(name="Ada")])

    original_signature = handlers.inspect.signature

    def unavailable_signature(value: Any) -> Any:
        if getattr(value, "__name__", "") == "from_streaming_response":
            raise ValueError("extension method has no signature")
        return original_signature(value)

    monkeypatch.setattr(handlers.inspect, "signature", unavailable_signature)

    assert base_handler(Mode.JSON)._parse_streaming_response(
        StreamingUser, [], validation_context=None, strict=None
    ) == [User(name="Ada")]


@pytest.mark.asyncio
async def test_parse_streaming_response_returns_async_iterable_results() -> None:
    iterable_user = IterableModel(User)
    chunks = async_items([chat_chunk({"content": '{"tasks":[{"name":"Ada"}]}'})])

    result = base_handler(Mode.JSON)._parse_streaming_response(
        iterable_user, chunks, validation_context=None, strict=None
    )

    assert [item async for item in result] == [User(name="Ada")]


def test_finalize_parsed_result_handles_parallel_iterable_adapter_and_base_model() -> (
    None
):
    handler = OpenAIToolsHandler()
    response = chat_completion(content="unused")
    iterable_user = IterableModel(User)
    adapter = cast(type[BaseModel], ModelAdapter[str])
    parallel = ParallelBase(User, Search)

    assert handler._finalize_parsed_result(
        iterable_user,
        response,
        iterable_user(tasks=[User(name="Ada")]),
    ) == [User(name="Ada")]
    assert handler._finalize_parsed_result(parallel, response, ("kept",)) == ("kept",)
    assert (
        handler._finalize_parsed_result(
            adapter, response, adapter.model_validate({"content": "ok"})
        )
        == "ok"
    )
    parsed_user = User(name="Ada")
    assert handler._finalize_parsed_result(User, response, parsed_user) is parsed_user
    assert parsed_user._raw_response is response
    assert handler._finalize_parsed_result(User, response, "already parsed") == (
        "already parsed"
    )


def test_extract_tool_call_json_supports_legacy_and_serializable_arguments() -> None:
    handler = OpenAIToolsHandler()
    legacy = chat_completion(function_call=("User", '{"name":"Ada"}'))
    dict_response = SimpleNamespace(
        choices=[
            SimpleNamespace(
                message=SimpleNamespace(
                    refusal=None,
                    tool_calls=[
                        SimpleNamespace(
                            function=SimpleNamespace(arguments={"name": "Ada"})
                        )
                    ],
                )
            )
        ]
    )
    list_response = SimpleNamespace(
        choices=[
            SimpleNamespace(
                message=SimpleNamespace(
                    refusal=None,
                    tool_calls=[
                        SimpleNamespace(function=SimpleNamespace(arguments=["Ada"]))
                    ],
                )
            )
        ]
    )

    assert handler._extract_tool_call_json(legacy) == '{"name":"Ada"}'
    assert json.loads(handler._extract_tool_call_json(dict_response)) == {"name": "Ada"}
    assert json.loads(handler._extract_tool_call_json(list_response)) == ["Ada"]


@pytest.mark.parametrize(
    ("response", "error", "message"),
    [
        (
            chat_completion(refusal="policy"),
            AssertionError,
            "Unable to generate a response due to policy",
        ),
        (
            SimpleNamespace(
                choices=[
                    SimpleNamespace(
                        message=SimpleNamespace(
                            refusal=None,
                            tool_calls=[
                                SimpleNamespace(
                                    function=SimpleNamespace(arguments=None)
                                )
                            ],
                        )
                    )
                ]
            ),
            ResponseParsingError,
            "Tool call arguments missing in response",
        ),
        (
            SimpleNamespace(
                choices=[
                    SimpleNamespace(
                        message=SimpleNamespace(
                            refusal=None,
                            tool_calls=[
                                SimpleNamespace(
                                    function=SimpleNamespace(arguments=object())
                                )
                            ],
                        )
                    )
                ]
            ),
            ResponseParsingError,
            "Tool call arguments must be JSON-serializable",
        ),
        (
            chat_completion(content="no tool call", tool_calls=[]),
            ResponseParsingError,
            "No tool calls or function call found in response",
        ),
    ],
)
def test_extract_tool_call_json_reports_refusals_and_malformed_responses(
    response: Any, error: type[Exception], message: str
) -> None:
    with pytest.raises(error, match=message):
        OpenAIToolsHandler()._extract_tool_call_json(response)


def test_tools_prepare_and_parse_parallel_calls_and_ignore_unknown_tool() -> None:
    handler = OpenAIToolsHandler()
    response_model = Iterable[Union[User, Search]]

    prepared_model, kwargs = handler.prepare_request(response_model, {"messages": []})
    response = chat_completion(
        tool_calls=[
            tool_call("User", '{"name":"Ada"}', "call_user"),
            tool_call("Search", '{"query":"python"}', "call_search"),
            tool_call("Unrelated", "{}", "call_unknown"),
        ]
    )

    assert prepared_model is response_model
    assert kwargs["tool_choice"] == "auto"
    assert {tool["function"]["name"] for tool in kwargs["tools"]} == {"User", "Search"}
    assert list(
        handler.parse_response(
            response,
            prepared_model,
            validation_context={"source": "parallel"},
            strict=True,
        )
    ) == [User(name="Ada"), Search(query="python")]


def test_tools_prepare_strict_schema_and_parse_incomplete_output() -> None:
    handler = OpenAIToolsHandler()

    prepared_model, kwargs = handler.prepare_request(User, {"strict": True})

    assert prepared_model.__name__ == "User"
    assert prepared_model.model_fields["name"].annotation is str
    assert kwargs["tools"][0]["function"]["strict"] is True
    assert "strict" not in kwargs
    with pytest.raises(IncompleteOutputException):
        handler.parse_response(
            chat_completion(
                tool_calls=[tool_call("User", '{"name":"Ada"}')], finish_reason="length"
            ),
            User,
        )


def test_tools_handler_reports_an_empty_choices_response() -> None:
    response = chat_completion(
        tool_calls=[tool_call("User", '{"name":"Ada"}')]
    ).model_copy(update={"choices": []})

    with pytest.raises(ResponseParsingError, match="No choices in OpenAI response") as (
        error
    ):
        OpenAIToolsHandler().parse_response(response, User)

    assert error.value.mode == Mode.TOOLS.value
    assert error.value.raw_response is response


@pytest.mark.parametrize(
    "handler",
    [OpenAIJSONSchemaHandler(), OpenAIJSONHandler(), OpenAIMDJSONHandler()],
)
def test_json_handlers_reject_incomplete_output(handler: OpenAIHandlerBase) -> None:
    with pytest.raises(IncompleteOutputException):
        handler.parse_response(
            chat_completion(content="{", finish_reason="length"), User
        )


@pytest.mark.parametrize(
    "handler",
    [OpenAIJSONSchemaHandler(), OpenAIJSONHandler(), OpenAIMDJSONHandler()],
)
def test_json_handlers_report_an_empty_choices_response(
    handler: OpenAIHandlerBase,
) -> None:
    response = chat_completion(content='{"name":"Ada"}').model_copy(
        update={"choices": []}
    )

    with pytest.raises(ResponseParsingError, match="No choices in OpenAI response") as (
        error
    ):
        handler.parse_response(response, User)

    assert error.value.mode == handler.mode.value
    assert error.value.raw_response is response


def test_json_schema_stream_parses_iterable_and_partial_models() -> None:
    handler = OpenAIJSONSchemaHandler()
    iterable_user = IterableModel(User)
    partial_user = Partial[User]

    iterable_result = handler.parse_response(
        [chat_chunk({"content": '{"tasks":[{"name":"Ada"}]}'})],
        iterable_user,
        stream=True,
    )
    partial_result = handler.parse_response(
        [chat_chunk({"content": '{"name":"Ada"}'})], partial_user, stream=True
    )

    assert list(iterable_result) == [User(name="Ada")]
    assert partial_result[-1] == User(name="Ada")


@pytest.mark.parametrize("handler", [OpenAIJSONHandler(), OpenAIMDJSONHandler()])
def test_json_prompt_handlers_extend_list_system_content_and_create_system_message(
    handler: OpenAIHandlerBase,
) -> None:
    list_messages = [
        {"role": "system", "content": [{"type": "text", "text": "Extract users."}]}
    ]

    _, with_list = handler.prepare_request(User, {"messages": list_messages})
    _, without_messages = handler.prepare_request(User, {"messages": []})

    assert "Extract users." in with_list["messages"][0]["content"][0]["text"]
    assert "json_schema" in with_list["messages"][0]["content"][0]["text"]
    assert without_messages["messages"][0]["role"] == "system"
    assert "User" in without_messages["messages"][0]["content"]
    if isinstance(handler, OpenAIMDJSONHandler):
        assert "```json codeblock" in with_list["messages"][-1]["content"][0]["text"]


@pytest.mark.parametrize(
    ("handler", "content"),
    [
        (OpenAIJSONHandler(), '{"tasks":[{"name":"Ada"}]}'),
        (OpenAIMDJSONHandler(), '```json\n{"tasks":[{"name":"Ada"}]}\n```'),
    ],
)
def test_json_prompt_handlers_consume_registered_streaming_model(
    handler: OpenAIHandlerBase, content: str
) -> None:
    iterable_user = IterableModel(User)
    _, kwargs = handler.prepare_request(iterable_user, {"stream": True})

    parsed = handler.parse_response([chat_chunk({"content": content})], iterable_user)

    assert kwargs["stream"] is True
    assert list(parsed) == [User(name="Ada")]
    assert handler._consume_streaming_flag(iterable_user) is False


def test_parallel_tools_prepare_handles_none_streaming_and_model_union() -> None:
    handler = OpenAIParallelToolsHandler()
    response_model = cast(type[BaseModel], Iterable[Union[User, Search]])

    assert handler.prepare_request(None, {"messages": []}) == (None, {"messages": []})
    with pytest.raises(ConfigurationError, match="stream=True is not supported"):
        handler.prepare_request(response_model, {"stream": True})
    prepared, kwargs = handler.prepare_request(response_model, {"messages": []})

    assert isinstance(prepared, ParallelBase)
    assert prepared.registry == {"User": User, "Search": Search}
    assert kwargs["tool_choice"] == "auto"
    assert {tool["function"]["name"] for tool in kwargs["tools"]} == {"User", "Search"}


def test_parallel_tools_parse_valid_calls_and_report_empty_or_incomplete_output() -> (
    None
):
    handler = OpenAIParallelToolsHandler()
    response_model = cast(type[BaseModel], Iterable[Union[User, Search]])
    valid = chat_completion(
        tool_calls=[
            tool_call("User", '{"name":"Ada"}', "call_user"),
            tool_call("Search", '{"query":"python"}', "call_search"),
            tool_call("Unrelated", "{}", "call_unknown"),
        ]
    )

    assert list(
        handler.parse_response(
            valid,
            response_model,
            validation_context={"source": "parallel"},
            strict=True,
        )
    ) == [User(name="Ada"), Search(query="python")]
    with pytest.raises(ResponseParsingError, match="No tool calls in response"):
        handler.parse_response(chat_completion(tool_calls=[]), response_model)
    empty_response = chat_completion(tool_calls=[]).model_copy(update={"choices": []})
    with pytest.raises(ResponseParsingError, match="No choices in OpenAI response") as (
        error
    ):
        handler.parse_response(empty_response, response_model)
    assert error.value.mode == Mode.PARALLEL_TOOLS.value
    assert error.value.raw_response is empty_response
    with pytest.raises(IncompleteOutputException):
        handler.parse_response(
            chat_completion(tool_calls=[], finish_reason="length"), response_model
        )


def test_responses_tools_converts_max_tokens_and_falls_back_to_chat_tool_call() -> None:
    handler = OpenAIResponsesToolsHandler()

    prepared, kwargs = handler.prepare_request(User, {"max_tokens": 64})
    assert prepared is not None
    parsed = handler.parse_response(
        chat_completion(tool_calls=[tool_call("User", '{"name":"Ada"}')]),
        prepared,
        validation_context={"source": "fallback"},
        strict=True,
    )

    assert prepared.__name__ == "User"
    assert prepared.model_fields["name"].annotation is str
    assert kwargs["max_output_tokens"] == 64
    assert "max_tokens" not in kwargs
    assert parsed.model_dump() == {"name": "Ada"}
    assert parsed._raw_response.choices[0].message.tool_calls[0].function.name == "User"


def test_responses_tools_falls_back_when_output_is_empty_or_has_no_tool_call() -> None:
    handler = OpenAIResponsesToolsHandler()
    message = ResponseOutputMessage(
        id="msg_1",
        content=[
            ResponseOutputText(
                annotations=[],
                text="No structured output was emitted.",
                type="output_text",
            )
        ],
        role="assistant",
        status="completed",
        type="message",
    )

    for output in ([], [message]):
        response = chat_completion(
            tool_calls=[tool_call("User", '{"name":"Ada"}')]
        ).model_copy(update={"output": output})

        parsed = handler.parse_response(response, User)

        assert parsed == User(name="Ada")
        assert parsed._raw_response is response


def test_responses_tools_skips_an_empty_call_and_parses_the_next_call(
    caplog: pytest.LogCaptureFixture,
) -> None:
    empty = ResponseFunctionToolCall(
        arguments="",
        call_id="call_empty",
        name="User",
        type="function_call",
    )
    valid = ResponseFunctionToolCall(
        arguments='{"name":"Ada"}',
        call_id="call_valid",
        name="User",
        type="function_call",
    )
    response = SimpleNamespace(output=[empty, valid])

    with caplog.at_level("WARNING", logger="instructor"):
        parsed = OpenAIResponsesToolsHandler().parse_response(response, User)

    assert parsed == User(name="Ada")
    assert parsed._raw_response is response
    assert len(caplog.records) == 1
    assert "tool 'User' returned empty arguments" in caplog.records[0].message


def test_responses_tools_streams_iterable_and_partial_models() -> None:
    iterable_user = IterableModel(User)
    partial_user = Partial[User]

    iterable_events = response_deltas("fc_iterable", '{"tasks":[', '{"name":"Ada"}]}')
    partial_events = response_deltas("fc_partial", '{"name":', '"Ada"}')

    handler = OpenAIResponsesToolsHandler()
    iterable_result = handler.parse_response(
        iterable_events, iterable_user, stream=True
    )
    partial_result = handler.parse_response(partial_events, partial_user, stream=True)

    assert list(iterable_result) == [User(name="Ada")]
    assert partial_result[-1] == User(name="Ada")


@pytest.mark.asyncio
async def test_responses_tools_streams_async_iterable_and_partial_models() -> None:
    iterable_user = IterableModel(User)
    partial_user = Partial[User]

    iterable_events = response_deltas("fc_iterable", '{"tasks":[', '{"name":"Ada"}]}')
    partial_events = response_deltas("fc_partial", '{"name":', '"Ada"}')

    handler = OpenAIResponsesToolsHandler()
    iterable_result = handler.parse_response(
        async_items(iterable_events), iterable_user, stream=True
    )
    partial_result = handler.parse_response(
        async_items(partial_events), partial_user, stream=True
    )

    assert [item async for item in iterable_result] == [User(name="Ada")]
    assert [item async for item in partial_result][-1] == User(name="Ada")