from __future__ import annotations

import asyncio
import json
from types import SimpleNamespace
from typing import Any, cast

import httpx
import pytest
from openai import NOT_GIVEN, APIConnectionError, AsyncOpenAI, RateLimitError, omit
from openai.types.responses import ResponseCompletedEvent, ResponseErrorEvent
from openai.types.responses.response_create_params import ContextManagement, PromptCacheOptions
from openai.types.shared.reasoning import Reasoning

from agents import (
    Agent,
    AsyncComputer,
    Computer,
    ComputerTool,
    ModelSettings,
    ModelTracing,
    Runner,
    ToolSearchTool,
    __version__,
    trace,
)
from agents.exceptions import ModelBehaviorError, UserError
from agents.models._retry_runtime import (
    provider_managed_retries_disabled,
    websocket_pre_event_retries_disabled,
)
from agents.models.openai_responses import (
    _HEADERS_OVERRIDE as RESP_HEADERS,
    ConvertedTools,
    Converter,
    OpenAIResponsesModel,
    OpenAIResponsesWSModel,
    ResponsesWebSocketError,
    _should_retry_pre_event_websocket_disconnect,
)
from agents.retry import ModelRetryAdviceRequest
from agents.usage import Usage
from tests.fake_model import get_response_obj
from tests.testing_processor import fetch_ordered_spans


async def _run_responses_model_with_custom_base_url(
    model_settings: ModelSettings | None = None,
) -> dict[str, Any]:
    class DummyResponses:
        def __init__(self) -> None:
            self.kwargs: dict[str, Any] = {}

        async def create(self, **kwargs: Any) -> Any:
            self.kwargs = kwargs
            return get_response_obj([])

    class DummyResponsesClient:
        def __init__(self, responses: DummyResponses) -> None:
            self.responses = responses
            self.base_url = httpx.URL("https://custom.example.test/v1/")

    responses = DummyResponses()
    model = OpenAIResponsesModel(
        model="gpt-4",
        openai_client=DummyResponsesClient(responses),  # type: ignore[arg-type]
    )
    agent = Agent(name="test", model=model, model_settings=model_settings or ModelSettings())

    await Runner.run(agent, "hi")

    return responses.kwargs


async def _run_responses_model_with_official_client(
    model_settings: ModelSettings | None = None,
) -> list[httpx.Request]:
    requests: list[httpx.Request] = []

    async def handler(request: httpx.Request) -> httpx.Response:
        requests.append(request)
        return httpx.Response(
            200,
            content=get_response_obj([]).model_dump_json(),
            headers={"content-type": "application/json"},
            request=request,
        )

    http_client = httpx.AsyncClient(transport=httpx.MockTransport(handler))
    try:
        client = AsyncOpenAI(
            api_key="test-key",
            base_url="https://example.test/v1",
            http_client=http_client,
        )
        model = OpenAIResponsesModel(model="gpt-4", openai_client=client)

        await model.get_response(
            system_instructions=None,
            input="hi",
            model_settings=model_settings or ModelSettings(),
            tools=[],
            output_schema=None,
            handoffs=[],
            tracing=ModelTracing.DISABLED,
        )
    finally:
        await http_client.aclose()

    return requests


class DummyWSConnection:
    def __init__(self, frames: list[str]):
        self._frames = frames
        self.sent_messages: list[dict[str, Any]] = []
        self.close_calls = 0
        self.close_code: int | None = None

    async def send(self, payload: str) -> None:
        self.sent_messages.append(json.loads(payload))

    async def recv(self) -> str:
        if not self._frames:
            raise RuntimeError("No more websocket frames configured")
        return self._frames.pop(0)

    async def close(self) -> None:
        self.close_calls += 1
        if self.close_code is None:
            self.close_code = 1000


class DummyWSClient:
    def __init__(self):
        self.base_url = httpx.URL("https://api.openai.com/v1/")
        self.websocket_base_url = None
        self.default_query: dict[str, Any] = {}
        self.auth_headers = {"Authorization": "Bearer test-key"}
        self.default_headers = {"User-Agent": "AsyncOpenAI/Python test"}
        self.timeout: Any = None
        self.refresh_calls = 0

    async def _refresh_api_key(self) -> None:
        self.refresh_calls += 1


def _response_event_frame(event_type: str, response_id: str, sequence_number: int) -> str:
    response = get_response_obj([]).model_dump()
    response["id"] = response_id
    return json.dumps(
        {
            "type": event_type,
            "response": response,
            "sequence_number": sequence_number,
        }
    )


def _response_completed_frame(response_id: str, sequence_number: int) -> str:
    return _response_event_frame("response.completed", response_id, sequence_number)


def _response_error_frame(code: str, message: str, sequence_number: int) -> str:
    return json.dumps(
        {
            "type": "response.error",
            "error": {"code": code, "message": message, "param": None},
            "sequence_number": sequence_number,
        }
    )


def _connection_closed_error(message: str) -> Exception:
    class ConnectionClosedError(Exception):
        pass

    ConnectionClosedError.__module__ = "websockets.client"
    return ConnectionClosedError(message)


@pytest.mark.allow_call_model_methods
@pytest.mark.asyncio
@pytest.mark.parametrize("override_ua", [None, "test_user_agent"])
async def test_user_agent_header_responses(override_ua: str | None):
    called_kwargs: dict[str, Any] = {}
    expected_ua = override_ua or f"Agents/Python {__version__}"

    class DummyStream:
        def __aiter__(self):
            async def gen():
                yield ResponseCompletedEvent(
                    type="response.completed",
                    response=get_response_obj([]),
                    sequence_number=0,
                )

            return gen()

    class DummyResponses:
        async def create(self, **kwargs):
            nonlocal called_kwargs
            called_kwargs = kwargs
            return DummyStream()

    class DummyResponsesClient:
        def __init__(self):
            self.responses = DummyResponses()

    model = OpenAIResponsesModel(model="gpt-4", openai_client=DummyResponsesClient())  # type: ignore

    if override_ua is not None:
        token = RESP_HEADERS.set({"User-Agent": override_ua})
    else:
        token = None

    try:
        stream = model.stream_response(
            system_instructions=None,
            input="hi",
            model_settings=ModelSettings(),
            tools=[],
            output_schema=None,
            handoffs=[],
            tracing=ModelTracing.DISABLED,
        )
        async for _ in stream:
            pass
    finally:
        if token is not None:
            RESP_HEADERS.reset(token)

    assert "extra_headers" in called_kwargs
    assert called_kwargs["extra_headers"]["User-Agent"] == expected_ua


@pytest.mark.allow_call_model_methods
@pytest.mark.asyncio
async def test_get_response_exposes_request_id():
    class DummyResponses:
        async def create(self, **kwargs):
            response = get_response_obj([], response_id="resp-request-id")
            response._request_id = "req_nonstream_123"
            return response

    class DummyResponsesClient:
        def __init__(self):
            self.responses = DummyResponses()

    model = OpenAIResponsesModel(model="gpt-4", openai_client=DummyResponsesClient())  # type: ignore[arg-type]

    response = await model.get_response(
        system_instructions=None,
        input="hi",
        model_settings=ModelSettings(),
        tools=[],
        output_schema=None,
        handoffs=[],
        tracing=ModelTracing.DISABLED,
    )

    assert response.response_id == "resp-request-id"
    assert response.request_id == "req_nonstream_123"


@pytest.mark.allow_call_model_methods
@pytest.mark.asyncio
async def test_get_response_span_exports_usage():
    class DummyResponses:
        async def create(self, **kwargs):
            return get_response_obj(
                [],
                response_id="resp-usage",
                usage=Usage(requests=1, input_tokens=10, output_tokens=4, total_tokens=14),
            )

    class DummyResponsesClient:
        def __init__(self):
            self.responses = DummyResponses()

    model = OpenAIResponsesModel(model="gpt-4", openai_client=DummyResponsesClient())  # type: ignore[arg-type]

    with trace("test"):
        await model.get_response(
            system_instructions=None,
            input="hi",
            model_settings=ModelSettings(),
            tools=[],
            output_schema=None,
            handoffs=[],
            tracing=ModelTracing.ENABLED,
        )

    response_spans = [
        span.export() for span in fetch_ordered_spans() if span.span_data.type == "response"
    ]
    assert len(response_spans) == 1
    assert response_spans[0]
    assert response_spans[0]["span_data"] == {
        "type": "response",
        "response_id": "resp-usage",
        "usage": {
            "requests": 1,
            "input_tokens": 10,
            "output_tokens": 4,
            "total_tokens": 14,
            "input_tokens_details": {"cached_tokens": 0, "cache_write_tokens": 0},
            "output_tokens_details": {"reasoning_tokens": 0},
        },
    }


def test_get_client_disables_provider_managed_retries_on_runner_retry() -> None:
    class DummyResponsesClient:
        def __init__(self) -> None:
            self.responses = SimpleNamespace()
            self.with_options_calls: list[dict[str, Any]] = []

        def with_options(self, **kwargs):
            self.with_options_calls.append(kwargs)
            return self

    client = DummyResponsesClient()
    model = OpenAIResponsesModel(model="gpt-4", openai_client=client)  # type: ignore[arg-type]

    assert cast(object, model._get_client()) is client
    with provider_managed_retries_disabled(True):
        assert cast(object, model._get_client()) is client

    assert client.with_options_calls == [{"max_retries": 0}]


@pytest.mark.allow_call_model_methods
@pytest.mark.asyncio
async def test_fetch_response_stream_attaches_request_id_to_terminal_response():
    class DummyHTTPStream:
        def __init__(self):
            self._yielded = False

        def __aiter__(self):
            return self

        async def __anext__(self):
            if self._yielded:
                raise StopAsyncIteration
            self._yielded = True
            return ResponseCompletedEvent(
                type="response.completed",
                response=get_response_obj([], response_id="resp-stream-request-id"),
                sequence_number=0,
            )

    inner_stream = DummyHTTPStream()

    class DummyAPIResponse:
        def __init__(self):
            self.request_id = "req_stream_123"
            self.close_calls = 0
            self.parse_calls = 0

        async def parse(self):
            self.parse_calls += 1
            return inner_stream

        async def close(self) -> None:
            self.close_calls += 1

    api_response = DummyAPIResponse()
    aexit_calls: list[tuple[Any, Any, Any]] = []

    class DummyStreamingContextManager:
        async def __aenter__(self):
            return api_response

        async def __aexit__(self, exc_type, exc, tb):
            aexit_calls.append((exc_type, exc, tb))
            await api_response.close()
            return False

    class DummyResponses:
        def __init__(self):
            self.with_streaming_response = SimpleNamespace(create=self.create_streaming)

        def create_streaming(self, **kwargs):
            return DummyStreamingContextManager()

    class DummyResponsesClient:
        def __init__(self):
            self.responses = DummyResponses()

    model = OpenAIResponsesModel(model="gpt-4", openai_client=DummyResponsesClient())  # type: ignore[arg-type]

    stream = await model._fetch_response(
        system_instructions=None,
        input="hi",
        model_settings=ModelSettings(),
        tools=[],
        output_schema=None,
        handoffs=[],
        previous_response_id=None,
        conversation_id=None,
        stream=True,
    )

    stream_agen = cast(Any, stream)
    event = await stream_agen.__anext__()

    assert getattr(stream, "request_id", None) == "req_stream_123"
    assert getattr(event.response, "_request_id", None) == "req_stream_123"

    with pytest.raises(StopAsyncIteration):
        await stream_agen.__anext__()

    assert api_response.parse_calls == 1
    assert api_response.close_calls == 1
    assert aexit_calls == [(None, None, None)]


@pytest.mark.allow_call_model_methods
@pytest.mark.asyncio
async def test_fetch_response_stream_parse_failure_exits_streaming_context():
    parse_error = RuntimeError("parse failed")
    aexit_calls: list[tuple[Any, Any, Any]] = []

    class DummyAPIResponse:
        request_id = "req_stream_123"

        async def parse(self):
            raise parse_error

    api_response = DummyAPIResponse()

    class DummyStreamingContextManager:
        async def __aenter__(self):
            return api_response

        async def __aexit__(self, exc_type, exc, tb):
            aexit_calls.append((exc_type, exc, tb))
            return False

    class DummyResponses:
        def __init__(self):
            self.with_streaming_response = SimpleNamespace(create=self.create_streaming)

        def create_streaming(self, **kwargs):
            return DummyStreamingContextManager()

    class DummyResponsesClient:
        def __init__(self):
            self.responses = DummyResponses()

    model = OpenAIResponsesModel(model="gpt-4", openai_client=DummyResponsesClient())  # type: ignore[arg-type]

    with pytest.raises(RuntimeError, match="parse failed"):
        await model._fetch_response(
            system_instructions=None,
            input="hi",
            model_settings=ModelSettings(),
            tools=[],
            output_schema=None,
            handoffs=[],
            previous_response_id=None,
            conversation_id=None,
            stream=True,
        )

    assert len(aexit_calls) == 1
    exc_type, exc, tb = aexit_calls[0]
    assert exc_type is RuntimeError
    assert exc is parse_error
    assert tb is not None


@pytest.mark.allow_call_model_methods
@pytest.mark.asyncio
async def test_fetch_response_stream_without_request_id_still_returns_events():
    class DummyHTTPStream:
        def __init__(self):
            self._yielded = False

        def __aiter__(self):
            return self

        async def __anext__(self):
            if self._yielded:
                raise StopAsyncIteration
            self._yielded = True
            return ResponseCompletedEvent(
                type="response.completed",
                response=get_response_obj([], response_id="resp-stream-request-id"),
                sequence_number=0,
            )

    inner_stream = DummyHTTPStream()
    aexit_calls: list[tuple[Any, Any, Any]] = []

    class DummyAPIResponse:
        def __init__(self):
            self.close_calls = 0
            self.parse_calls = 0

        async def parse(self):
            self.parse_calls += 1
            return inner_stream

        async def close(self) -> None:
            self.close_calls += 1

    api_response = DummyAPIResponse()

    class DummyStreamingContextManager:
        async def __aenter__(self):
            return api_response

        async def __aexit__(self, exc_type, exc, tb):
            aexit_calls.append((exc_type, exc, tb))
            await api_response.close()
            return False

    class DummyResponses:
        def __init__(self):
            self.with_streaming_response = SimpleNamespace(create=self.create_streaming)

        def create_streaming(self, **kwargs):
            return DummyStreamingContextManager()

    class DummyResponsesClient:
        def __init__(self):
            self.responses = DummyResponses()

    model = OpenAIResponsesModel(model="gpt-4", openai_client=DummyResponsesClient())  # type: ignore[arg-type]

    stream = await model._fetch_response(
        system_instructions=None,
        input="hi",
        model_settings=ModelSettings(),
        tools=[],
        output_schema=None,
        handoffs=[],
        previous_response_id=None,
        conversation_id=None,
        stream=True,
    )

    stream_agen = cast(Any, stream)
    event = await stream_agen.__anext__()

    assert getattr(stream, "request_id", None) is None
    assert getattr(event.response, "_request_id", None) is None

    with pytest.raises(StopAsyncIteration):
        await stream_agen.__anext__()

    assert api_response.parse_calls == 1
    assert api_response.close_calls == 1
    assert aexit_calls == [(None, None, None)]


@pytest.mark.allow_call_model_methods
@pytest.mark.asyncio
async def test_stream_response_ignores_streaming_context_exit_failure_after_terminal_event():
    class DummyHTTPStream:
        def __init__(self):
            self._yielded = False

        def __aiter__(self):
            return self

        async def __anext__(self):
            if self._yielded:
                raise StopAsyncIteration
            self._yielded = True
            return ResponseCompletedEvent(
                type="response.completed",
                response=get_response_obj([], response_id="resp-stream-request-id"),
                sequence_number=0,
            )

    inner_stream = DummyHTTPStream()
    aexit_calls: list[tuple[Any, Any, Any]] = []

    class DummyAPIResponse:
        request_id = "req_stream_123"

        async def parse(self):
            return inner_stream

    api_response = DummyAPIResponse()

    class DummyStreamingContextManager:
        async def __aenter__(self):
            return api_response

        async def __aexit__(self, exc_type, exc, tb):
            aexit_calls.append((exc_type, exc, tb))
            raise RuntimeError("stream context exit failed")

    class DummyResponses:
        def __init__(self):
            self.with_streaming_response = SimpleNamespace(create=self.create_streaming)

        def create_streaming(self, **kwargs):
            return DummyStreamingContextManager()

    class DummyResponsesClient:
        def __init__(self):
            self.responses = DummyResponses()

    model = OpenAIResponsesModel(model="gpt-4", openai_client=DummyResponsesClient())  # type: ignore[arg-type]

    events: list[ResponseCompletedEvent] = []
    async for event in model.stream_response(
        system_instructions=None,
        input="hi",
        model_settings=ModelSettings(),
        tools=[],
        output_schema=None,
        handoffs=[],
        tracing=ModelTracing.DISABLED,
    ):
        assert isinstance(event, ResponseCompletedEvent)
        events.append(event)

    assert len(events) == 1
    assert aexit_calls == [(None, None, None)]


@pytest.mark.allow_call_model_methods
@pytest.mark.asyncio
async def test_stream_response_close_closes_inner_http_stream_with_async_close(monkeypatch):
    client = DummyWSClient()
    model = OpenAIResponsesModel(model="gpt-4", openai_client=client)  # type: ignore[arg-type]

    class DummyHTTPStream:
        def __init__(self):
            self._yielded = False
            self.close_calls = 0

        def __aiter__(self):
            return self

        async def __anext__(self):
            if self._yielded:
                raise StopAsyncIteration
            self._yielded = True
            return ResponseCompletedEvent(
                type="response.completed",
                response=get_response_obj([]),
                sequence_number=0,
            )

        async def close(self) -> None:
            self.close_calls += 1

    inner_stream = DummyHTTPStream()

    async def fake_fetch_response(*args: Any, **kwargs: Any) -> DummyHTTPStream:
        return inner_stream

    monkeypatch.setattr(model, "_fetch_response", fake_fetch_response)

    stream = model.stream_response(
        system_instructions=None,
        input="hi",
        model_settings=ModelSettings(),
        tools=[],
        output_schema=None,
        handoffs=[],
        tracing=ModelTracing.DISABLED,
    )
    stream_agen = cast(Any, stream)

    event = await stream_agen.__anext__()
    assert event.type == "response.completed"

    await stream_agen.aclose()

    assert inner_stream.close_calls == 1


@pytest.mark.allow_call_model_methods
@pytest.mark.asyncio
async def test_stream_response_normal_exhaustion_closes_inner_http_stream(monkeypatch):
    client = DummyWSClient()
    model = OpenAIResponsesModel(model="gpt-4", openai_client=client)  # type: ignore[arg-type]

    class DummyHTTPStream:
        def __init__(self):
            self._yielded = False
            self.close_calls = 0

        def __aiter__(self):
            return self

        async def __anext__(self):
            if self._yielded:
                raise StopAsyncIteration
            self._yielded = True
            return ResponseCompletedEvent(
                type="response.completed",
                response=get_response_obj([]),
                sequence_number=0,
            )

        async def close(self) -> None:
            self.close_calls += 1

    inner_stream = DummyHTTPStream()

    async def fake_fetch_response(*args: Any, **kwargs: Any) -> DummyHTTPStream:
        return inner_stream

    monkeypatch.setattr(model, "_fetch_response", fake_fetch_response)

    events: list[ResponseCompletedEvent] = []
    async for event in model.stream_response(
        system_instructions=None,
        input="hi",
        model_settings=ModelSettings(),
        tools=[],
        output_schema=None,
        handoffs=[],
        tracing=ModelTracing.DISABLED,
    ):
        assert isinstance(event, ResponseCompletedEvent)
        events.append(event)

    assert len(events) == 1
    assert inner_stream.close_calls == 1


@pytest.mark.allow_call_model_methods
@pytest.mark.asyncio
async def test_stream_response_ignores_inner_close_failure_after_terminal_event(monkeypatch):
    client = DummyWSClient()
    model = OpenAIResponsesModel(model="gpt-4", openai_client=client)  # type: ignore[arg-type]

    class DummyHTTPStream:
        def __init__(self):
            self._yielded = False
            self.close_calls = 0

        def __aiter__(self):
            return self

        async def __anext__(self):
            if self._yielded:
                raise StopAsyncIteration
            self._yielded = True
            return ResponseCompletedEvent(
                type="response.completed",
                response=get_response_obj([]),
                sequence_number=0,
            )

        async def close(self) -> None:
            self.close_calls += 1
            raise RuntimeError("stream close failed")

    inner_stream = DummyHTTPStream()

    async def fake_fetch_response(*args: Any, **kwargs: Any) -> DummyHTTPStream:
        return inner_stream

    monkeypatch.setattr(model, "_fetch_response", fake_fetch_response)

    events: list[ResponseCompletedEvent] = []
    async for event in model.stream_response(
        system_instructions=None,
        input="hi",
        model_settings=ModelSettings(),
        tools=[],
        output_schema=None,
        handoffs=[],
        tracing=ModelTracing.DISABLED,
    ):
        assert isinstance(event, ResponseCompletedEvent)
        events.append(event)

    assert len(events) == 1
    assert inner_stream.close_calls == 1


@pytest.mark.allow_call_model_methods
@pytest.mark.asyncio
async def test_stream_response_cancellation_does_not_block_on_inner_stream_close(monkeypatch):
    client = DummyWSClient()
    model = OpenAIResponsesModel(model="gpt-4", openai_client=client)  # type: ignore[arg-type]

    class BlockingHTTPStream:
        def __init__(self):
            self.next_started = asyncio.Event()
            self.close_started = asyncio.Event()
            self.close_release = asyncio.Event()
            self.close_calls = 0

        def __aiter__(self):
            return self

        async def __anext__(self):
            self.next_started.set()
            await asyncio.Event().wait()
            raise StopAsyncIteration

        async def aclose(self) -> None:
            self.close_calls += 1
            self.close_started.set()
            await self.close_release.wait()

    inner_stream = BlockingHTTPStream()

    async def fake_fetch_response(*args: Any, **kwargs: Any) -> BlockingHTTPStream:
        return inner_stream

    monkeypatch.setattr(model, "_fetch_response", fake_fetch_response)

    stream = model.stream_response(
        system_instructions=None,
        input="hi",
        model_settings=ModelSettings(),
        tools=[],
        output_schema=None,
        handoffs=[],
        tracing=ModelTracing.DISABLED,
    )
    stream_agen = cast(Any, stream)
    next_task = asyncio.create_task(stream_agen.__anext__())

    await asyncio.wait_for(inner_stream.next_started.wait(), timeout=1.0)
    next_task.cancel()

    try:
        with pytest.raises(asyncio.CancelledError):
            await asyncio.wait_for(next_task, timeout=0.5)
        await asyncio.wait_for(inner_stream.close_started.wait(), timeout=1.0)
        assert inner_stream.close_calls == 1
    finally:
        inner_stream.close_release.set()
        await asyncio.sleep(0)


@pytest.mark.allow_call_model_methods
def test_build_response_create_kwargs_rejects_duplicate_extra_args_keys():
    client = DummyWSClient()
    model = OpenAIResponsesModel(model="gpt-4", openai_client=client)  # type: ignore[arg-type]

    with pytest.raises(TypeError, match="multiple values.*stream"):
        model._build_response_create_kwargs(
            system_instructions=None,
            input="hi",
            model_settings=ModelSettings(extra_args={"stream": False}),
            tools=[],
            output_schema=None,
            handoffs=[],
            previous_response_id=None,
            conversation_id=None,
            stream=True,
            prompt=None,
        )


@pytest.mark.allow_call_model_methods
def test_build_response_create_kwargs_includes_extra_args_prompt_cache_key():
    client = DummyWSClient()
    model = OpenAIResponsesModel(model="gpt-4", openai_client=client)  # type: ignore[arg-type]

    kwargs = model._build_response_create_kwargs(
        system_instructions=None,
        input="hi",
        model_settings=ModelSettings(extra_args={"prompt_cache_key": "cache-key"}),
        tools=[],
        output_schema=None,
        handoffs=[],
        previous_response_id=None,
        conversation_id=None,
        stream=False,
        prompt=None,
    )

    assert kwargs["prompt_cache_key"] == "cache-key"


@pytest.mark.allow_call_model_methods
def test_build_response_create_kwargs_includes_context_management():
    client = DummyWSClient()
    model = OpenAIResponsesModel(model="gpt-4", openai_client=client)  # type: ignore[arg-type]
    context_management: list[ContextManagement] = [
        {"type": "compaction", "compact_threshold": 200000}
    ]

    kwargs = model._build_response_create_kwargs(
        system_instructions=None,
        input="hi",
        model_settings=ModelSettings(context_management=context_management),
        tools=[],
        output_schema=None,
        handoffs=[],
        previous_response_id=None,
        conversation_id=None,
        stream=False,
        prompt=None,
    )

    assert kwargs["context_management"] == context_management


@pytest.mark.allow_call_model_methods
def test_build_response_create_kwargs_includes_gpt_5_6_request_controls():
    client = DummyWSClient()
    model = OpenAIResponsesModel(model="gpt-5.6-sol", openai_client=client)  # type: ignore[arg-type]
    reasoning = Reasoning(mode="pro", effort="max", context="all_turns")
    prompt_cache_options: PromptCacheOptions = {"mode": "explicit", "ttl": "30m"}

    kwargs = model._build_response_create_kwargs(
        system_instructions=None,
        input="hi",
        model_settings=ModelSettings(
            reasoning=reasoning,
            prompt_cache_retention="24h",
            prompt_cache_options=prompt_cache_options,
        ),
        tools=[],
        output_schema=None,
        handoffs=[],
        previous_response_id="resp-previous",
        conversation_id=None,
        stream=False,
        prompt=None,
    )

    assert kwargs["reasoning"] is reasoning
    assert kwargs["prompt_cache_retention"] == "24h"
    assert kwargs["prompt_cache_options"] == prompt_cache_options
    assert kwargs["previous_response_id"] == "resp-previous"


@pytest.mark.allow_call_model_methods
def test_build_response_create_kwargs_rejects_duplicate_prompt_cache_options_extra_args():
    client = DummyWSClient()
    model = OpenAIResponsesModel(model="gpt-5.6-sol", openai_client=client)  # type: ignore[arg-type]

    with pytest.raises(TypeError, match="multiple values.*prompt_cache_options"):
        model._build_response_create_kwargs(
            system_instructions=None,
            input="hi",
            model_settings=ModelSettings(
                prompt_cache_options={"mode": "explicit", "ttl": "30m"},
                extra_args={"prompt_cache_options": {"mode": "implicit"}},
            ),
            tools=[],
            output_schema=None,
            handoffs=[],
            previous_response_id=None,
            conversation_id=None,
            stream=False,
            prompt=None,
        )


@pytest.mark.allow_call_model_methods
def test_build_response_create_kwargs_allows_prompt_cache_options_extra_args_when_direct_omitted():
    client = DummyWSClient()
    model = OpenAIResponsesModel(model="gpt-5.6-sol", openai_client=client)  # type: ignore[arg-type]
    prompt_cache_options = {"mode": "explicit", "ttl": "30m"}

    kwargs = model._build_response_create_kwargs(
        system_instructions=None,
        input="hi",
        model_settings=ModelSettings(extra_args={"prompt_cache_options": prompt_cache_options}),
        tools=[],
        output_schema=None,
        handoffs=[],
        previous_response_id=None,
        conversation_id=None,
        stream=False,
        prompt=None,
    )

    assert kwargs["prompt_cache_options"] == prompt_cache_options


@pytest.mark.allow_call_model_methods
def test_build_response_create_kwargs_allows_extra_arg_when_explicit_arg_is_omitted():
    client = DummyWSClient()
    model = OpenAIResponsesModel(model="gpt-4", openai_client=client)  # type: ignore[arg-type]
    context_management: list[ContextManagement] = [
        {"type": "compaction", "compact_threshold": 200000}
    ]

    kwargs = model._build_response_create_kwargs(
        system_instructions=None,
        input="hi",
        model_settings=ModelSettings(extra_args={"context_management": context_management}),
        tools=[],
        output_schema=None,
        handoffs=[],
        previous_response_id=None,
        conversation_id=None,
        stream=False,
        prompt=None,
    )

    assert kwargs["context_management"] == context_management


@pytest.mark.allow_call_model_methods
def test_build_response_create_kwargs_rejects_duplicate_context_management_extra_args():
    client = DummyWSClient()
    model = OpenAIResponsesModel(model="gpt-4", openai_client=client)  # type: ignore[arg-type]

    with pytest.raises(TypeError, match="multiple values.*context_management"):
        model._build_response_create_kwargs(
            system_instructions=None,
            input="hi",
            model_settings=ModelSettings(
                context_management=[{"type": "compaction", "compact_threshold": 200000}],
                extra_args={"context_management": [{"type": "compaction"}]},
            ),
            tools=[],
            output_schema=None,
            handoffs=[],
            previous_response_id=None,
            conversation_id=None,
            stream=False,
            prompt=None,
        )


@pytest.mark.allow_call_model_methods
def test_build_response_create_kwargs_keeps_unset_transport_extra_kwargs_as_none():
    client = DummyWSClient()
    model = OpenAIResponsesModel(model="gpt-4", openai_client=client)  # type: ignore[arg-type]

    kwargs = model._build_response_create_kwargs(
        system_instructions=None,
        input="hi",
        model_settings=ModelSettings(),
        tools=[],
        output_schema=None,
        handoffs=[],
        previous_response_id=None,
        conversation_id=None,
        stream=False,
        prompt=None,
    )

    assert kwargs["extra_query"] is None
    assert kwargs["extra_body"] is None


@pytest.mark.allow_call_model_methods
@pytest.mark.asyncio
async def test_get_response_with_official_client_accepts_unset_transport_extra_kwargs() -> None:
    requests = await _run_responses_model_with_official_client()

    assert len(requests) == 1
    assert requests[0].url == "https://example.test/v1/responses"


@pytest.mark.allow_call_model_methods
@pytest.mark.asyncio
async def test_get_response_with_official_client_applies_transport_extra_kwargs() -> None:
    requests = await _run_responses_model_with_official_client(
        ModelSettings(
            extra_query={"api-version": "2026-01-01-preview"},
            extra_body={"extra_transport_field": "enabled"},
        )
    )

    assert len(requests) == 1
    assert requests[0].url == ("https://example.test/v1/responses?api-version=2026-01-01-preview")
    assert json.loads(requests[0].content)["extra_transport_field"] == "enabled"


@pytest.mark.allow_call_model_methods
@pytest.mark.asyncio
async def test_custom_base_url_prompt_cache_key_uses_model_settings_only() -> None:
    default_kwargs = await _run_responses_model_with_custom_base_url()
    explicit_kwargs = await _run_responses_model_with_custom_base_url(
        model_settings=ModelSettings(extra_args={"prompt_cache_key": "cache-key"})
    )

    assert "prompt_cache_key" not in default_kwargs
    assert explicit_kwargs["prompt_cache_key"] == "cache-key"


@pytest.mark.allow_call_model_methods
def test_build_response_create_kwargs_preserves_unknown_response_include_values():
    client = DummyWSClient()
    model = OpenAIResponsesModel(model="gpt-4", openai_client=client)  # type: ignore[arg-type]

    kwargs = model._build_response_create_kwargs(
        system_instructions=None,
        input="hi",
        model_settings=ModelSettings(response_include=["response.future_flag"]),
        tools=[],
        output_schema=None,
        handoffs=[],
        previous_response_id=None,
        conversation_id=None,
        stream=False,
        prompt=None,
    )

    assert kwargs["include"] == ["response.future_flag"]


@pytest.mark.allow_call_model_methods
def test_build_response_create_kwargs_preserves_unknown_tool_types(monkeypatch) -> None:
    client = DummyWSClient()
    model = OpenAIResponsesModel(model="gpt-4", openai_client=client)  # type: ignore[arg-type]

    future_tool = cast(Any, {"type": "future_beta_tool", "label": "preview"})

    monkeypatch.setattr(
        Converter,
        "convert_tools",
        classmethod(
            lambda cls, tools, handoffs, **kwargs: ConvertedTools(tools=[future_tool], includes=[])
        ),
    )

    kwargs = model._build_response_create_kwargs(
        system_instructions=None,
        input="hi",
        model_settings=ModelSettings(),
        tools=[],
        output_schema=None,
        handoffs=[],
        previous_response_id=None,
        conversation_id=None,
        stream=False,
        prompt=None,
    )

    assert kwargs["tools"] == [future_tool]


@pytest.mark.allow_call_model_methods
@pytest.mark.asyncio
async def test_prompt_id_omits_model_parameter():
    called_kwargs: dict[str, Any] = {}

    class DummyResponses:
        async def create(self, **kwargs):
            nonlocal called_kwargs
            called_kwargs = kwargs
            return get_response_obj([])

    class DummyResponsesClient:
        def __init__(self):
            self.responses = DummyResponses()

    model = OpenAIResponsesModel(
        model="gpt-4",
        openai_client=DummyResponsesClient(),  # type: ignore[arg-type]
        model_is_explicit=False,
    )

    await model.get_response(
        system_instructions=None,
        input="hi",
        model_settings=ModelSettings(),
        tools=[],
        output_schema=None,
        handoffs=[],
        tracing=ModelTracing.DISABLED,
        prompt={"id": "pmpt_123"},
    )

    assert called_kwargs["prompt"] == {"id": "pmpt_123"}
    assert called_kwargs["model"] is omit


@pytest.mark.allow_call_model_methods
@pytest.mark.asyncio
async def test_prompt_id_omits_tools_parameter_when_no_tools_configured():
    called_kwargs: dict[str, Any] = {}

    class DummyResponses:
        async def create(self, **kwargs):
            nonlocal called_kwargs
            called_kwargs = kwargs
            return get_response_obj([])

    class DummyResponsesClient:
        def __init__(self):
            self.responses = DummyResponses()

    model = OpenAIResponsesModel(
        model="gpt-4",
        openai_client=DummyResponsesClient(),  # type: ignore[arg-type]
        model_is_explicit=False,
    )

    await model.get_response(
        system_instructions=None,
        input="hi",
        model_settings=ModelSettings(),
        tools=[],
        output_schema=None,
        handoffs=[],
        tracing=ModelTracing.DISABLED,
        prompt={"id": "pmpt_123"},
    )

    assert called_kwargs["tools"] is omit


@pytest.mark.allow_call_model_methods
@pytest.mark.asyncio
async def test_prompt_id_omits_tool_choice_when_no_tools_configured():
    called_kwargs: dict[str, Any] = {}

    class DummyResponses:
        async def create(self, **kwargs):
            nonlocal called_kwargs
            called_kwargs = kwargs
            return get_response_obj([])

    class DummyResponsesClient:
        def __init__(self):
            self.responses = DummyResponses()

    model = OpenAIResponsesModel(
        model="gpt-4",
        openai_client=DummyResponsesClient(),  # type: ignore[arg-type]
        model_is_explicit=False,
    )

    await model.get_response(
        system_instructions=None,
        input="hi",
        model_settings=ModelSettings(tool_choice="web_search_preview"),
        tools=[],
        output_schema=None,
        handoffs=[],
        tracing=ModelTracing.DISABLED,
        prompt={"id": "pmpt_123"},
    )

    assert called_kwargs["tools"] is omit
    assert called_kwargs["tool_choice"] is omit


@pytest.mark.allow_call_model_methods
@pytest.mark.asyncio
@pytest.mark.parametrize("tool_choice", ["none", "required"])
async def test_prompt_id_keeps_literal_tool_choice_without_local_tools(tool_choice: str):
    called_kwargs: dict[str, Any] = {}

    class DummyResponses:
        async def create(self, **kwargs):
            nonlocal called_kwargs
            called_kwargs = kwargs
            return get_response_obj([])

    class DummyResponsesClient:
        def __init__(self):
            self.responses = DummyResponses()

    model = OpenAIResponsesModel(
        model="gpt-4",
        openai_client=DummyResponsesClient(),  # type: ignore[arg-type]
        model_is_explicit=False,
    )

    await model.get_response(
        system_instructions=None,
        input="hi",
        model_settings=ModelSettings(tool_choice=tool_choice),
        tools=[],
        output_schema=None,
        handoffs=[],
        tracing=ModelTracing.DISABLED,
        prompt={"id": "pmpt_123"},
    )

    assert called_kwargs["tools"] is omit
    assert called_kwargs["tool_choice"] == tool_choice


@pytest.mark.allow_call_model_methods
@pytest.mark.asyncio
async def test_prompt_id_keeps_explicit_tool_search_without_local_surface() -> None:
    called_kwargs: dict[str, Any] = {}

    class DummyResponses:
        async def create(self, **kwargs):
            nonlocal called_kwargs
            called_kwargs = kwargs
            return get_response_obj([])

    class DummyResponsesClient:
        def __init__(self):
            self.responses = DummyResponses()

    model = OpenAIResponsesModel(
        model="gpt-4",
        openai_client=DummyResponsesClient(),  # type: ignore[arg-type]
        model_is_explicit=False,
    )

    await model.get_response(
        system_instructions=None,
        input="hi",
        model_settings=ModelSettings(),
        tools=[ToolSearchTool()],
        output_schema=None,
        handoffs=[],
        tracing=ModelTracing.DISABLED,
        prompt={"id": "pmpt_123"},
    )

    assert called_kwargs["prompt"] == {"id": "pmpt_123"}
    assert called_kwargs["tools"] == [{"type": "tool_search"}]


@pytest.mark.allow_call_model_methods
@pytest.mark.asyncio
async def test_ga_computer_tool_does_not_require_preview_metadata() -> None:
    called_kwargs: dict[str, Any] = {}

    class DummyComputer(AsyncComputer):
        async def screenshot(self) -> str:
            return "screenshot"

        async def click(self, x: int, y: int, button: str) -> None:
            pass

        async def double_click(self, x: int, y: int) -> None:
            pass

        async def drag(self, path: list[tuple[int, int]]) -> None:
            pass

        async def keypress(self, keys: list[str]) -> None:
            pass

        async def move(self, x: int, y: int) -> None:
            pass

        async def scroll(self, x: int, y: int, scroll_x: int, scroll_y: int) -> None:
            pass

        async def type(self, text: str) -> None:
            pass

        async def wait(self) -> None:
            pass

    class DummyResponses:
        async def create(self, **kwargs):
            nonlocal called_kwargs
            called_kwargs = kwargs
            return get_response_obj([])

    class DummyResponsesClient:
        def __init__(self):
            self.responses = DummyResponses()

    model = OpenAIResponsesModel(
        model="gpt-5.4",
        openai_client=DummyResponsesClient(),  # type: ignore[arg-type]
        model_is_explicit=True,
    )

    await model.get_response(
        system_instructions=None,
        input="hi",
        model_settings=ModelSettings(),
        tools=[ComputerTool(computer=DummyComputer())],
        output_schema=None,
        handoffs=[],
        tracing=ModelTracing.DISABLED,
        prompt=None,
    )

    assert called_kwargs["tools"] == [{"type": "computer"}]


@pytest.mark.allow_call_model_methods
@pytest.mark.asyncio
async def test_prompt_id_uses_preview_computer_payload_when_prompt_owns_model() -> None:
    called_kwargs: dict[str, Any] = {}

    class DummyComputer(Computer):
        @property
        def environment(self) -> str:  # type: ignore[override]
            return "mac"

        @property
        def dimensions(self) -> tuple[int, int]:
            return (800, 600)

        def screenshot(self) -> str:
            return "screenshot"

        def click(self, x: int, y: int, button: str) -> None:
            pass

        def double_click(self, x: int, y: int) -> None:
            pass

        def drag(self, path: list[tuple[int, int]]) -> None:
            pass

        def keypress(self, keys: list[str]) -> None:
            pass

        def move(self, x: int, y: int) -> None:
            pass

        def scroll(self, x: int, y: int, scroll_x: int, scroll_y: int) -> None:
            pass

        def type(self, text: str) -> None:
            pass

        def wait(self) -> None:
            pass

    class DummyResponses:
        async def create(self, **kwargs):
            nonlocal called_kwargs
            called_kwargs = kwargs
            return get_response_obj([])

    class DummyResponsesClient:
        def __init__(self):
            self.responses = DummyResponses()

    model = OpenAIResponsesModel(
        model="gpt-5.4",
        openai_client=DummyResponsesClient(),  # type: ignore[arg-type]
        model_is_explicit=False,
    )

    await model.get_response(
        system_instructions=None,
        input="hi",
        model_settings=ModelSettings(),
        tools=[ComputerTool(computer=DummyComputer())],
        output_schema=None,
        handoffs=[],
        tracing=ModelTracing.DISABLED,
        prompt={"id": "pmpt_123"},
    )

    assert called_kwargs["model"] is omit
    assert called_kwargs["tool_choice"] is omit
    assert called_kwargs["tools"] == [
        {
            "type": "computer_use_preview",
            "environment": "mac",
            "display_width": 800,
            "display_height": 600,
        }
    ]


@pytest.mark.allow_call_model_methods
@pytest.mark.asyncio
async def test_prompt_id_computer_without_preview_metadata_raises_clear_error() -> None:
    called_kwargs: dict[str, Any] = {}

    class DummyComputer(Computer):
        def screenshot(self) -> str:
            return "screenshot"

        def click(self, x: int, y: int, button: str) -> None:
            pass

        def double_click(self, x: int, y: int) -> None:
            pass

        def drag(self, path: list[tuple[int, int]]) -> None:
            pass

        def keypress(self, keys: list[str]) -> None:
            pass

        def move(self, x: int, y: int) -> None:
            pass

        def scroll(self, x: int, y: int, scroll_x: int, scroll_y: int) -> None:
            pass

        def type(self, text: str) -> None:
            pass

        def wait(self) -> None:
            pass

    class DummyResponses:
        async def create(self, **kwargs):
            nonlocal called_kwargs
            called_kwargs = kwargs
            return get_response_obj([])

    class DummyResponsesClient:
        def __init__(self):
            self.responses = DummyResponses()

    model = OpenAIResponsesModel(
        model="gpt-5.4",
        openai_client=DummyResponsesClient(),  # type: ignore[arg-type]
        model_is_explicit=False,
    )

    with pytest.raises(
        UserError,
        match="Preview computer tool payloads require `environment` and `dimensions`",
    ):
        await model.get_response(
            system_instructions=None,
            input="hi",
            model_settings=ModelSettings(),
            tools=[ComputerTool(computer=DummyComputer())],
            output_schema=None,
            handoffs=[],
            tracing=ModelTracing.DISABLED,
            prompt={"id": "pmpt_123"},
        )

    assert called_kwargs == {}


@pytest.mark.allow_call_model_methods
@pytest.mark.asyncio
async def test_prompt_id_unresolved_computer_uses_preview_payload_shape() -> None:
    called_kwargs: dict[str, Any] = {}

    class DummyComputer(Computer):
        @property
        def environment(self) -> str:  # type: ignore[override]
            return "mac"

        @property
        def dimensions(self) -> tuple[int, int]:
            return (800, 600)

        def screenshot(self) -> str:
            return "screenshot"

        def click(self, x: int, y: int, button: str) -> None:
            pass

        def double_click(self, x: int, y: int) -> None:
            pass

        def drag(self, path: list[tuple[int, int]]) -> None:
            pass

        def keypress(self, keys: list[str]) -> None:
            pass

        def move(self, x: int, y: int) -> None:
            pass

        def scroll(self, x: int, y: int, scroll_x: int, scroll_y: int) -> None:
            pass

        def type(self, text: str) -> None:
            pass

        def wait(self) -> None:
            pass

    class DummyResponses:
        async def create(self, **kwargs):
            nonlocal called_kwargs
            called_kwargs = kwargs
            return get_response_obj([])

    class DummyResponsesClient:
        def __init__(self):
            self.responses = DummyResponses()

    model = OpenAIResponsesModel(
        model="gpt-5.4",
        openai_client=DummyResponsesClient(),  # type: ignore[arg-type]
        model_is_explicit=False,
    )

    with pytest.raises(UserError, match="Computer tool is not initialized for serialization"):
        await model.get_response(
            system_instructions=None,
            input="hi",
            model_settings=ModelSettings(),
            tools=[ComputerTool(computer=lambda **_: DummyComputer())],
            output_schema=None,
            handoffs=[],
            tracing=ModelTracing.DISABLED,
            prompt={"id": "pmpt_123"},
        )

    assert called_kwargs == {}


@pytest.mark.allow_call_model_methods
@pytest.mark.asyncio
@pytest.mark.parametrize("tool_choice", ["computer", "computer_use"])
async def test_prompt_id_explicit_ga_computer_tool_choice_uses_ga_selector_and_tool(
    tool_choice: str,
) -> None:
    called_kwargs: dict[str, Any] = {}

    class DummyComputer(Computer):
        @property
        def environment(self) -> str:  # type: ignore[override]
            return "mac"

        @property
        def dimensions(self) -> tuple[int, int]:
            return (800, 600)

        def screenshot(self) -> str:
            return "screenshot"

        def click(self, x: int, y: int, button: str) -> None:
            pass

        def double_click(self, x: int, y: int) -> None:
            pass

        def drag(self, path: list[tuple[int, int]]) -> None:
            pass

        def keypress(self, keys: list[str]) -> None:
            pass

        def move(self, x: int, y: int) -> None:
            pass

        def scroll(self, x: int, y: int, scroll_x: int, scroll_y: int) -> None:
            pass

        def type(self, text: str) -> None:
            pass

        def wait(self) -> None:
            pass

    class DummyResponses:
        async def create(self, **kwargs):
            nonlocal called_kwargs
            called_kwargs = kwargs
            return get_response_obj([])

    class DummyResponsesClient:
        def __init__(self):
            self.responses = DummyResponses()

    model = OpenAIResponsesModel(
        model="gpt-5.4",
        openai_client=DummyResponsesClient(),  # type: ignore[arg-type]
        model_is_explicit=False,
    )

    await model.get_response(
        system_instructions=None,
        input="hi",
        model_settings=ModelSettings(tool_choice=tool_choice),
        tools=[ComputerTool(computer=DummyComputer())],
        output_schema=None,
        handoffs=[],
        tracing=ModelTracing.DISABLED,
        prompt={"id": "pmpt_123"},
    )

    assert called_kwargs["model"] is omit
    assert called_kwargs["tool_choice"] == {"type": "computer"}
    assert called_kwargs["tools"] == [{"type": "computer"}]


@pytest.mark.allow_call_model_methods
@pytest.mark.asyncio
@pytest.mark.parametrize("tool_choice", ["computer", "computer_use"])
async def test_preview_model_forced_computer_tool_choice_uses_preview_selector(
    tool_choice: str,
) -> None:
    called_kwargs: dict[str, Any] = {}

    class DummyComputer(Computer):
        @property
        def environment(self) -> str:  # type: ignore[override]
            return "mac"

        @property
        def dimensions(self) -> tuple[int, int]:
            return (800, 600)

        def screenshot(self) -> str:
            return "screenshot"

        def click(self, x: int, y: int, button: str) -> None:
            pass

        def double_click(self, x: int, y: int) -> None:
            pass

        def drag(self, path: list[tuple[int, int]]) -> None:
            pass

        def keypress(self, keys: list[str]) -> None:
            pass

        def move(self, x: int, y: int) -> None:
            pass

        def scroll(self, x: int, y: int, scroll_x: int, scroll_y: int) -> None:
            pass

        def type(self, text: str) -> None:
            pass

        def wait(self) -> None:
            pass

    class DummyResponses:
        async def create(self, **kwargs):
            nonlocal called_kwargs
            called_kwargs = kwargs
            return get_response_obj([])

    class DummyResponsesClient:
        def __init__(self):
            self.responses = DummyResponses()

    model = OpenAIResponsesModel(
        model="computer-use-preview",
        openai_client=DummyResponsesClient(),  # type: ignore[arg-type]
    )

    await model.get_response(
        system_instructions=None,
        input="hi",
        model_settings=ModelSettings(tool_choice=tool_choice),
        tools=[ComputerTool(computer=DummyComputer())],
        output_schema=None,
        handoffs=[],
        tracing=ModelTracing.DISABLED,
    )

    assert called_kwargs["model"] == "computer-use-preview"
    assert called_kwargs["tool_choice"] == {"type": "computer_use_preview"}
    assert called_kwargs["tools"] == [
        {
            "type": "computer_use_preview",
            "environment": "mac",
            "display_width": 800,
            "display_height": 600,
        }
    ]


@pytest.mark.allow_call_model_methods
@pytest.mark.asyncio
async def test_websocket_model_reuses_connection_and_sends_response_create_frames(monkeypatch):
    client = DummyWSClient()
    ws = DummyWSConnection(
        [
            _response_completed_frame("resp-1", 1),
            _response_completed_frame("resp-2", 2),
        ]
    )
    model = OpenAIResponsesWSModel(model="gpt-4", openai_client=client)  # type: ignore[arg-type]

    opened: list[tuple[str, dict[str, str]]] = []

    async def fake_open(
        ws_url: str, headers: dict[str, str], *, connect_timeout: float | None = None
    ) -> DummyWSConnection:
        opened.append((ws_url, headers))
        return ws

    monkeypatch.setattr(model, "_open_websocket_connection", fake_open)

    first = await model.get_response(
        system_instructions=None,
        input="hi",
        model_settings=ModelSettings(
            reasoning=Reasoning(mode="pro", effort="max", context="all_turns"),
            prompt_cache_options={"mode": "explicit", "ttl": "30m"},
        ),
        tools=[],
        output_schema=None,
        handoffs=[],
        tracing=ModelTracing.DISABLED,
    )
    second = await model.get_response(
        system_instructions=None,
        input="next",
        model_settings=ModelSettings(),
        tools=[],
        output_schema=None,
        handoffs=[],
        tracing=ModelTracing.DISABLED,
        previous_response_id="resp-1",
    )

    assert first.response_id == "resp-1"
    assert second.response_id == "resp-2"
    assert client.refresh_calls == 2
    assert len(opened) == 1
    assert ws.sent_messages[0]["type"] == "response.create"
    assert ws.sent_messages[0]["stream"] is True
    assert ws.sent_messages[0]["reasoning"] == {
        "context": "all_turns",
        "effort": "max",
        "mode": "pro",
    }
    assert ws.sent_messages[0]["prompt_cache_options"] == {
        "mode": "explicit",
        "ttl": "30m",
    }
    assert ws.sent_messages[1]["type"] == "response.create"
    assert ws.sent_messages[1]["stream"] is True
    assert ws.sent_messages[1]["previous_response_id"] == "resp-1"


@pytest.mark.asyncio
async def test_websocket_model_passes_keepalive_options_to_connect(monkeypatch):
    import websockets.asyncio.client as websockets_client

    client = DummyWSClient()
    model = OpenAIResponsesWSModel(
        model="gpt-4",
        openai_client=client,  # type: ignore[arg-type]
        websocket_options={"ping_interval": 45.0, "ping_timeout": None},
    )
    ws = DummyWSConnection([])
    captured_kwargs: dict[str, Any] = {}

    async def fake_connect(ws_url: str, **kwargs: Any) -> DummyWSConnection:
        captured_kwargs["ws_url"] = ws_url
        captured_kwargs.update(kwargs)
        return ws

    monkeypatch.setattr(websockets_client, "connect", fake_connect)

    opened = await model._open_websocket_connection(
        "wss://example.test/v1/responses",
        {"Authorization": "Bearer test-key"},
        connect_timeout=10.0,
    )

    assert opened is ws
    assert captured_kwargs["ws_url"] == "wss://example.test/v1/responses"
    assert captured_kwargs["additional_headers"] == {"Authorization": "Bearer test-key"}
    assert captured_kwargs["open_timeout"] == 10.0
    assert captured_kwargs["ping_interval"] == 45.0
    assert captured_kwargs["ping_timeout"] is None


@pytest.mark.asyncio
async def test_websocket_model_passes_max_size_to_connect(monkeypatch):
    import websockets.asyncio.client as websockets_client

    client = DummyWSClient()
    model = OpenAIResponsesWSModel(
        model="gpt-4",
        openai_client=client,  # type: ignore[arg-type]
        websocket_options={"max_size": 8 * 1024 * 1024},
    )
    ws = DummyWSConnection([])
    captured_kwargs: dict[str, Any] = {}

    async def fake_connect(ws_url: str, **kwargs: Any) -> DummyWSConnection:
        captured_kwargs["ws_url"] = ws_url
        captured_kwargs.update(kwargs)
        return ws

    monkeypatch.setattr(websockets_client, "connect", fake_connect)

    opened = await model._open_websocket_connection(
        "wss://example.test/v1/responses",
        {"Authorization": "Bearer test-key"},
        connect_timeout=10.0,
    )

    assert opened is ws
    assert captured_kwargs["max_size"] == 8 * 1024 * 1024


@pytest.mark.allow_call_model_methods
def test_websocket_model_reconnects_when_reused_from_different_event_loop(monkeypatch):
    client = DummyWSClient()
    ws1 = DummyWSConnection([_response_completed_frame("resp-1", 1)])
    ws2 = DummyWSConnection([_response_completed_frame("resp-2", 2)])
    model = OpenAIResponsesWSModel(model="gpt-4", openai_client=client)  # type: ignore[arg-type]

    opened: list[tuple[str, dict[str, str]]] = []
    ws_connections = [ws1, ws2]

    async def fake_open(
        ws_url: str, headers: dict[str, str], *, connect_timeout: float | None = None
    ) -> DummyWSConnection:
        opened.append((ws_url, headers))
        return ws_connections.pop(0)

    monkeypatch.setattr(model, "_open_websocket_connection", fake_open)

    async def get_response(input_text: str, previous_response_id: str | None = None):
        return await model.get_response(
            system_instructions=None,
            input=input_text,
            model_settings=ModelSettings(),
            tools=[],
            output_schema=None,
            handoffs=[],
            tracing=ModelTracing.DISABLED,
            previous_response_id=previous_response_id,
        )

    loop1 = asyncio.new_event_loop()
    loop2 = asyncio.new_event_loop()
    try:
        first = loop1.run_until_complete(get_response("hi"))
        second = loop2.run_until_complete(get_response("next", previous_response_id="resp-1"))
    finally:
        loop1.close()
        loop2.close()
        asyncio.set_event_loop(None)

    assert first.response_id == "resp-1"
    assert second.response_id == "resp-2"
    assert len(opened) == 2
    assert ws1.close_calls == 1
    assert ws2.close_calls == 0


@pytest.mark.allow_call_model_methods
def test_websocket_model_init_lazily_creates_request_lock(monkeypatch):
    client = DummyWSClient()

    def fail_lock(*args, **kwargs):
        raise RuntimeError("asyncio.Lock() should not be called in __init__")

    monkeypatch.setattr("agents.models.openai_responses.asyncio.Lock", fail_lock)

    model = OpenAIResponsesWSModel(model="gpt-4", openai_client=client)  # type: ignore[arg-type]

    assert model._ws_request_lock is None


@pytest.mark.allow_call_model_methods
@pytest.mark.asyncio
async def test_websocket_model_stream_response_yields_typed_events(monkeypatch):
    client = DummyWSClient()
    ws = DummyWSConnection([_response_completed_frame("resp-stream", 1)])
    model = OpenAIResponsesWSModel(model="gpt-4", openai_client=client)  # type: ignore[arg-type]

    async def fake_open(
        ws_url: str, headers: dict[str, str], *, connect_timeout: float | None = None
    ) -> DummyWSConnection:
        return ws

    monkeypatch.setattr(model, "_open_websocket_connection", fake_open)

    events = []
    async for event in model.stream_response(
        system_instructions=None,
        input="hi",
        model_settings=ModelSettings(),
        tools=[],
        output_schema=None,
        handoffs=[],
        tracing=ModelTracing.DISABLED,
    ):
        events.append(event)

    assert len(events) == 1
    assert isinstance(events[0], ResponseCompletedEvent)
    assert events[0].response.id == "resp-stream"


@pytest.mark.allow_call_model_methods
@pytest.mark.asyncio
@pytest.mark.parametrize("terminal_event_type", ["response.incomplete", "response.failed"])
async def test_websocket_model_get_response_rejects_failed_terminal_response_payload_events(
    monkeypatch, terminal_event_type: str
):
    client = DummyWSClient()
    ws = DummyWSConnection([_response_event_frame(terminal_event_type, "resp-terminal", 1)])
    model = OpenAIResponsesWSModel(model="gpt-4", openai_client=client)  # type: ignore[arg-type]

    async def fake_open(
        ws_url: str, headers: dict[str, str], *, connect_timeout: float | None = None
    ) -> DummyWSConnection:
        return ws

    monkeypatch.setattr(model, "_open_websocket_connection", fake_open)

    with pytest.raises(ModelBehaviorError, match=terminal_event_type):
        await model.get_response(
            system_instructions=None,
            input="hi",
            model_settings=ModelSettings(),
            tools=[],
            output_schema=None,
            handoffs=[],
            tracing=ModelTracing.DISABLED,
        )


@pytest.mark.allow_call_model_methods
@pytest.mark.asyncio
@pytest.mark.parametrize("terminal_event_type", ["response.incomplete", "response.failed"])
async def test_websocket_model_stream_response_rejects_failed_terminal_response_payload_events(
    monkeypatch, terminal_event_type: str
):
    client = DummyWSClient()
    ws = DummyWSConnection([_response_event_frame(terminal_event_type, "resp-terminal", 1)])
    model = OpenAIResponsesWSModel(model="gpt-4", openai_client=client)  # type: ignore[arg-type]

    async def fake_open(
        ws_url: str, headers: dict[str, str], *, connect_timeout: float | None = None
    ) -> DummyWSConnection:
        return ws

    monkeypatch.setattr(model, "_open_websocket_connection", fake_open)

    events = []
    with pytest.raises(ModelBehaviorError, match=terminal_event_type):
        async for event in model.stream_response(
            system_instructions=None,
            input="hi",
            model_settings=ModelSettings(),
            tools=[],
            output_schema=None,
            handoffs=[],
            tracing=ModelTracing.DISABLED,
        ):
            events.append(event)

    assert len(events) == 1
    assert events[0].type == terminal_event_type
    assert cast(Any, events[0]).response.id == "resp-terminal"


@pytest.mark.allow_call_model_methods
@pytest.mark.asyncio
async def test_stream_response_rejects_response_error_terminal_event(monkeypatch):
    model = OpenAIResponsesModel(model="gpt-4", openai_client=object())  # type: ignore[arg-type]

    async def dummy_fetch_response(
        system_instructions,
        input,
        model_settings,
        tools,
        output_schema,
        handoffs,
        previous_response_id,
        conversation_id,
        stream,
        prompt,
    ):
        class DummyStream:
            async def __aiter__(self):
                yield ResponseErrorEvent(
                    type="error",
                    code="invalid_request_error",
                    message="bad request",
                    param=None,
                    sequence_number=0,
                )

        return DummyStream()

    monkeypatch.setattr(model, "_fetch_response", dummy_fetch_response)

    events = []
    with pytest.raises(ModelBehaviorError, match="invalid_request_error"):
        async for event in model.stream_response(
            system_instructions=None,
            input="hi",
            model_settings=ModelSettings(),
            tools=[],
            output_schema=None,
            handoffs=[],
            tracing=ModelTracing.DISABLED,
        ):
            events.append(event)

    assert len(events) == 1
    assert events[0].type == "error"
    assert events[0].code == "invalid_request_error"
    assert events[0].message == "bad request"


@pytest.mark.allow_call_model_methods
@pytest.mark.asyncio
async def test_websocket_model_get_response_surfaces_response_error_event(monkeypatch):
    client = DummyWSClient()
    ws = DummyWSConnection([_response_error_frame("invalid_request_error", "bad request", 1)])
    model = OpenAIResponsesWSModel(model="gpt-4", openai_client=client)  # type: ignore[arg-type]

    async def fake_open(
        ws_url: str, headers: dict[str, str], *, connect_timeout: float | None = None
    ) -> DummyWSConnection:
        return ws

    monkeypatch.setattr(model, "_open_websocket_connection", fake_open)

    with pytest.raises(ResponsesWebSocketError, match="response\\.error") as exc_info:
        await model.get_response(
            system_instructions=None,
            input="hi",
            model_settings=ModelSettings(),
            tools=[],
            output_schema=None,
            handoffs=[],
            tracing=ModelTracing.DISABLED,
        )

    assert "invalid_request_error" in str(exc_info.value)
    assert "bad request" in str(exc_info.value)
    assert exc_info.value.event_type == "response.error"
    assert exc_info.value.code == "invalid_request_error"
    assert exc_info.value.error_message == "bad request"


@pytest.mark.allow_call_model_methods
@pytest.mark.asyncio
async def test_websocket_model_stream_response_raises_on_response_error_event(monkeypatch):
    client = DummyWSClient()
    ws = DummyWSConnection([_response_error_frame("invalid_request_error", "bad request", 1)])
    model = OpenAIResponsesWSModel(model="gpt-4", openai_client=client)  # type: ignore[arg-type]

    async def fake_open(
        ws_url: str, headers: dict[str, str], *, connect_timeout: float | None = None
    ) -> DummyWSConnection:
        return ws

    monkeypatch.setattr(model, "_open_websocket_connection", fake_open)

    with pytest.raises(ResponsesWebSocketError, match="response\\.error") as exc_info:
        async for _event in model.stream_response(
            system_instructions=None,
            input="hi",
            model_settings=ModelSettings(),
            tools=[],
            output_schema=None,
            handoffs=[],
            tracing=ModelTracing.DISABLED,
        ):
            pass

    assert "invalid_request_error" in str(exc_info.value)
    assert "bad request" in str(exc_info.value)


@pytest.mark.allow_call_model_methods
@pytest.mark.asyncio
async def test_websocket_model_stream_break_drops_persistent_connection(monkeypatch):
    client = DummyWSClient()
    ws = DummyWSConnection(
        [
            _response_event_frame("response.created", "resp-created", 1),
            _response_completed_frame("resp-complete", 2),
        ]
    )
    model = OpenAIResponsesWSModel(model="gpt-4", openai_client=client)  # type: ignore[arg-type]

    async def fake_open(
        ws_url: str, headers: dict[str, str], *, connect_timeout: float | None = None
    ) -> DummyWSConnection:
        return ws

    monkeypatch.setattr(model, "_open_websocket_connection", fake_open)

    stream = await model._fetch_response(
        system_instructions=None,
        input="hi",
        model_settings=ModelSettings(),
        tools=[],
        output_schema=None,
        handoffs=[],
        previous_response_id=None,
        conversation_id=None,
        stream=True,
    )

    stream_agen = cast(Any, stream)
    event = await stream_agen.__anext__()
    assert event.type == "response.created"
    await stream_agen.aclose()

    assert ws.close_calls == 0
    assert model._ws_connection is None


@pytest.mark.allow_call_model_methods
@pytest.mark.asyncio
async def test_websocket_model_stream_close_after_terminal_event_preserves_persistent_connection(
    monkeypatch,
):
    client = DummyWSClient()
    ws = DummyWSConnection(
        [
            _response_completed_frame("resp-complete-1", 1),
            _response_completed_frame("resp-complete-2", 2),
        ]
    )
    model = OpenAIResponsesWSModel(model="gpt-4", openai_client=client)  # type: ignore[arg-type]

    opened: list[DummyWSConnection] = []

    async def fake_open(
        ws_url: str, headers: dict[str, str], *, connect_timeout: float | None = None
    ) -> DummyWSConnection:
        opened.append(ws)
        return ws

    monkeypatch.setattr(model, "_open_websocket_connection", fake_open)

    stream = await model._fetch_response(
        system_instructions=None,
        input="hi",
        model_settings=ModelSettings(),
        tools=[],
        output_schema=None,
        handoffs=[],
        previous_response_id=None,
        conversation_id=None,
        stream=True,
    )

    stream_agen = cast(Any, stream)
    event = await stream_agen.__anext__()
    assert event.type == "response.completed"
    await stream_agen.aclose()

    assert ws.close_calls == 0
    assert model._ws_connection is ws
    assert model._ws_request_lock is not None
    assert model._ws_request_lock.locked() is False

    second = await model.get_response(
        system_instructions=None,
        input="next",
        model_settings=ModelSettings(),
        tools=[],
        output_schema=None,
        handoffs=[],
        tracing=ModelTracing.DISABLED,
    )

    assert second.response_id == "resp-complete-2"
    assert len(opened) == 1


@pytest.mark.allow_call_model_methods
@pytest.mark.asyncio
async def test_websocket_model_stream_response_terminal_close_keeps_connection(
    monkeypatch,
):
    client = DummyWSClient()
    ws = DummyWSConnection(
        [
            _response_completed_frame("resp-complete-1", 1),
            _response_completed_frame("resp-complete-2", 2),
        ]
    )
    model = OpenAIResponsesWSModel(model="gpt-4", openai_client=client)  # type: ignore[arg-type]

    opened: list[DummyWSConnection] = []

    async def fake_open(
        ws_url: str, headers: dict[str, str], *, connect_timeout: float | None = None
    ) -> DummyWSConnection:
        opened.append(ws)
        return ws

    monkeypatch.setattr(model, "_open_websocket_connection", fake_open)

    stream = model.stream_response(
        system_instructions=None,
        input="hi",
        model_settings=ModelSettings(),
        tools=[],
        output_schema=None,
        handoffs=[],
        tracing=ModelTracing.DISABLED,
    )

    stream_agen = cast(Any, stream)
    event = await stream_agen.__anext__()
    assert event.type == "response.completed"
    await stream_agen.aclose()

    assert ws.close_calls == 0
    assert model._ws_connection is ws

    second = await model.get_response(
        system_instructions=None,
        input="next",
        model_settings=ModelSettings(),
        tools=[],
        output_schema=None,
        handoffs=[],
        tracing=ModelTracing.DISABLED,
    )

    assert second.response_id == "resp-complete-2"
    assert len(opened) == 1


@pytest.mark.allow_call_model_methods
@pytest.mark.asyncio
async def test_websocket_model_stream_response_close_releases_inner_iterator(monkeypatch):
    client = DummyWSClient()
    ws = DummyWSConnection(
        [
            _response_event_frame("response.created", "resp-created", 1),
            _response_completed_frame("resp-complete", 2),
        ]
    )
    model = OpenAIResponsesWSModel(model="gpt-4", openai_client=client)  # type: ignore[arg-type]

    async def fake_open(
        ws_url: str, headers: dict[str, str], *, connect_timeout: float | None = None
    ) -> DummyWSConnection:
        return ws

    monkeypatch.setattr(model, "_open_websocket_connection", fake_open)

    stream = model.stream_response(
        system_instructions=None,
        input="hi",
        model_settings=ModelSettings(),
        tools=[],
        output_schema=None,
        handoffs=[],
        tracing=ModelTracing.DISABLED,
    )

    stream_agen = cast(Any, stream)
    event = await stream_agen.__anext__()
    assert event.type == "response.created"
    await stream_agen.aclose()

    assert ws.close_calls == 0
    assert model._ws_connection is None
    assert model._ws_request_lock is not None
    assert model._ws_request_lock.locked() is False


@pytest.mark.allow_call_model_methods
@pytest.mark.asyncio
async def test_websocket_model_stream_response_non_terminal_close_does_not_await_close_handshake(
    monkeypatch,
):
    class BlockingCloseWSConnection(DummyWSConnection):
        def __init__(self):
            super().__init__(
                [
                    _response_event_frame("response.created", "resp-created", 1),
                    _response_completed_frame("resp-complete", 2),
                ]
            )
            self.close_started = asyncio.Event()
            self.close_release = asyncio.Event()

            class DummyTransport:
                def __init__(inner_self, outer: BlockingCloseWSConnection):
                    inner_self.outer = outer
                    inner_self.abort_calls = 0

                def abort(inner_self) -> None:
                    inner_self.abort_calls += 1

            self.transport = DummyTransport(self)

        async def close(self) -> None:
            self.close_calls += 1
            self.close_started.set()
            await self.close_release.wait()

    client = DummyWSClient()
    ws = BlockingCloseWSConnection()
    model = OpenAIResponsesWSModel(model="gpt-4", openai_client=client)  # type: ignore[arg-type]

    async def fake_open(
        ws_url: str, headers: dict[str, str], *, connect_timeout: float | None = None
    ) -> DummyWSConnection:
        return ws

    monkeypatch.setattr(model, "_open_websocket_connection", fake_open)

    stream = model.stream_response(
        system_instructions=None,
        input="hi",
        model_settings=ModelSettings(),
        tools=[],
        output_schema=None,
        handoffs=[],
        tracing=ModelTracing.DISABLED,
    )

    stream_agen = cast(Any, stream)
    event = await stream_agen.__anext__()
    assert event.type == "response.created"

    try:
        await asyncio.wait_for(stream_agen.aclose(), timeout=0.5)
        assert ws.transport.abort_calls == 1
        assert ws.close_calls == 0
        assert model._ws_connection is None
        assert model._ws_request_lock is not None
        assert model._ws_request_lock.locked() is False
    finally:
        ws.close_release.set()
        await asyncio.sleep(0)


@pytest.mark.allow_call_model_methods
@pytest.mark.asyncio
async def test_websocket_model_cancellation_drops_persistent_connection(monkeypatch):
    class CancelOnRecvWSConnection(DummyWSConnection):
        async def recv(self) -> str:
            raise asyncio.CancelledError()

    client = DummyWSClient()
    ws = CancelOnRecvWSConnection([])
    model = OpenAIResponsesWSModel(model="gpt-4", openai_client=client)  # type: ignore[arg-type]

    async def fake_open(
        ws_url: str, headers: dict[str, str], *, connect_timeout: float | None = None
    ) -> DummyWSConnection:
        return ws

    monkeypatch.setattr(model, "_open_websocket_connection", fake_open)

    with pytest.raises(asyncio.CancelledError):
        await model.get_response(
            system_instructions=None,
            input="hi",
            model_settings=ModelSettings(),
            tools=[],
            output_schema=None,
            handoffs=[],
            tracing=ModelTracing.DISABLED,
        )

    assert ws.close_calls == 0
    assert model._ws_connection is None


@pytest.mark.allow_call_model_methods
@pytest.mark.asyncio
async def test_websocket_model_cancellation_does_not_await_close_handshake(monkeypatch):
    class BlockingCloseCancelOnRecvWSConnection(DummyWSConnection):
        def __init__(self):
            super().__init__([])
            self.recv_started = asyncio.Event()
            self.close_started = asyncio.Event()
            self.close_release = asyncio.Event()

            class DummyTransport:
                def __init__(inner_self, outer: BlockingCloseCancelOnRecvWSConnection):
                    inner_self.outer = outer
                    inner_self.abort_calls = 0

                def abort(inner_self) -> None:
                    inner_self.abort_calls += 1

            self.transport = DummyTransport(self)

        async def recv(self) -> str:
            self.recv_started.set()
            await asyncio.Event().wait()
            raise RuntimeError("unreachable")

        async def close(self) -> None:
            self.close_calls += 1
            self.close_started.set()
            await self.close_release.wait()

    client = DummyWSClient()
    ws = BlockingCloseCancelOnRecvWSConnection()
    model = OpenAIResponsesWSModel(model="gpt-4", openai_client=client)  # type: ignore[arg-type]

    async def fake_open(
        ws_url: str, headers: dict[str, str], *, connect_timeout: float | None = None
    ) -> DummyWSConnection:
        return ws

    monkeypatch.setattr(model, "_open_websocket_connection", fake_open)

    request_task = asyncio.create_task(
        model.get_response(
            system_instructions=None,
            input="hi",
            model_settings=ModelSettings(),
            tools=[],
            output_schema=None,
            handoffs=[],
            tracing=ModelTracing.DISABLED,
        )
    )

    await asyncio.wait_for(ws.recv_started.wait(), timeout=1.0)
    request_task.cancel()

    try:
        with pytest.raises(asyncio.CancelledError):
            await asyncio.wait_for(request_task, timeout=0.5)
        assert ws.transport.abort_calls == 1
        assert ws.close_calls == 0
        assert model._ws_connection is None
    finally:
        ws.close_release.set()
        await asyncio.sleep(0)


@pytest.mark.allow_call_model_methods
@pytest.mark.asyncio
async def test_websocket_model_preserves_pre_event_usererror(monkeypatch):
    client = DummyWSClient()
    model = OpenAIResponsesWSModel(model="gpt-4", openai_client=client)  # type: ignore[arg-type]

    async def fake_open(
        ws_url: str, headers: dict[str, str], *, connect_timeout: float | None = None
    ) -> DummyWSConnection:
        raise UserError("websockets dependency missing")

    monkeypatch.setattr(model, "_open_websocket_connection", fake_open)

    with pytest.raises(UserError, match="websockets dependency missing"):
        await model.get_response(
            system_instructions=None,
            input="hi",
            model_settings=ModelSettings(),
            tools=[],
            output_schema=None,
            handoffs=[],
            tracing=ModelTracing.DISABLED,
        )


@pytest.mark.allow_call_model_methods
@pytest.mark.asyncio
async def test_websocket_model_preserves_pre_event_server_error_frame_message(monkeypatch):
    client = DummyWSClient()
    ws = DummyWSConnection(
        [
            json.dumps(
                {
                    "type": "error",
                    "error": {"message": "bad auth", "type": "invalid_request_error"},
                }
            )
        ]
    )
    model = OpenAIResponsesWSModel(model="gpt-4", openai_client=client)  # type: ignore[arg-type]

    async def fake_open(
        ws_url: str, headers: dict[str, str], *, connect_timeout: float | None = None
    ) -> DummyWSConnection:
        return ws

    monkeypatch.setattr(model, "_open_websocket_connection", fake_open)

    with pytest.raises(ResponsesWebSocketError, match="Responses websocket error:") as exc_info:
        await model.get_response(
            system_instructions=None,
            input="hi",
            model_settings=ModelSettings(),
            tools=[],
            output_schema=None,
            handoffs=[],
            tracing=ModelTracing.DISABLED,
        )

    assert "feature may not be enabled" not in str(exc_info.value)
    assert "invalid_request_error" in str(exc_info.value)
    assert exc_info.value.event_type == "error"
    assert exc_info.value.error_type == "invalid_request_error"


@pytest.mark.allow_call_model_methods
@pytest.mark.asyncio
async def test_websocket_model_reconnects_if_cached_connection_is_closed(monkeypatch):
    client = DummyWSClient()
    ws1 = DummyWSConnection([_response_completed_frame("resp-1", 1)])
    ws2 = DummyWSConnection([_response_completed_frame("resp-2", 2)])
    model = OpenAIResponsesWSModel(model="gpt-4", openai_client=client)  # type: ignore[arg-type]

    opened: list[DummyWSConnection] = []
    queue = [ws1, ws2]

    async def fake_open(
        ws_url: str, headers: dict[str, str], *, connect_timeout: float | None = None
    ) -> DummyWSConnection:
        next_ws = queue.pop(0)
        opened.append(next_ws)
        return next_ws

    monkeypatch.setattr(model, "_open_websocket_connection", fake_open)

    first = await model.get_response(
        system_instructions=None,
        input="hi",
        model_settings=ModelSettings(),
        tools=[],
        output_schema=None,
        handoffs=[],
        tracing=ModelTracing.DISABLED,
    )
    assert first.response_id == "resp-1"
    assert len(opened) == 1

    # Simulate an idle timeout/server-side close on the cached websocket connection.
    ws1.close_code = 1001

    second = await model.get_response(
        system_instructions=None,
        input="next",
        model_settings=ModelSettings(),
        tools=[],
        output_schema=None,
        handoffs=[],
        tracing=ModelTracing.DISABLED,
    )

    assert second.response_id == "resp-2"
    assert len(opened) == 2
    assert ws1.close_calls == 1
    assert model._ws_connection is ws2


@pytest.mark.allow_call_model_methods
@pytest.mark.asyncio
async def test_websocket_model_does_not_retry_if_send_raises_after_writing_on_reused_connection(
    monkeypatch,
):
    client = DummyWSClient()

    class ConnectionClosedError(Exception):
        pass

    ConnectionClosedError.__module__ = "websockets.client"

    class DropAfterSendWriteOnReuseWSConnection(DummyWSConnection):
        def __init__(self, frames: list[str]):
            super().__init__(frames)
            self.send_calls = 0

        async def send(self, payload: str) -> None:
            self.send_calls += 1
            if self.send_calls > 1:
                await super().send(payload)
                raise ConnectionClosedError("peer closed during send after request write")
            await super().send(payload)

    ws1 = DropAfterSendWriteOnReuseWSConnection([_response_completed_frame("resp-1", 1)])
    model = OpenAIResponsesWSModel(model="gpt-4", openai_client=client)  # type: ignore[arg-type]

    open_calls = 0

    async def fake_open(
        ws_url: str, headers: dict[str, str], *, connect_timeout: float | None = None
    ) -> DummyWSConnection:
        nonlocal open_calls
        open_calls += 1
        if open_calls > 1:
            raise AssertionError("Unexpected websocket retry after send started")
        return ws1

    monkeypatch.setattr(model, "_open_websocket_connection", fake_open)

    first = await model.get_response(
        system_instructions=None,
        input="hi",
        model_settings=ModelSettings(),
        tools=[],
        output_schema=None,
        handoffs=[],
        tracing=ModelTracing.DISABLED,
    )
    with pytest.raises(RuntimeError, match="before any response events were received"):
        await model.get_response(
            system_instructions=None,
            input="next",
            model_settings=ModelSettings(),
            tools=[],
            output_schema=None,
            handoffs=[],
            tracing=ModelTracing.DISABLED,
        )

    assert first.response_id == "resp-1"
    assert open_calls == 1
    assert ws1.send_calls == 2
    assert len(ws1.sent_messages) == 2
    assert ws1.close_calls == 1
    assert model._ws_connection is None


@pytest.mark.allow_call_model_methods
@pytest.mark.asyncio
async def test_websocket_model_does_not_retry_after_pre_event_disconnect_once_request_sent(
    monkeypatch,
):
    client = DummyWSClient()

    class ConnectionClosedError(Exception):
        pass

    ConnectionClosedError.__module__ = "websockets.client"

    class DisconnectAfterSendWSConnection(DummyWSConnection):
        def __init__(self):
            super().__init__([])
            self.send_calls = 0
            self.recv_calls = 0

        async def send(self, payload: str) -> None:
            self.send_calls += 1
            await super().send(payload)

        async def recv(self) -> str:
            self.recv_calls += 1
            raise ConnectionClosedError("peer closed after request send")

    ws = DisconnectAfterSendWSConnection()
    model = OpenAIResponsesWSModel(model="gpt-4", openai_client=client)  # type: ignore[arg-type]
    open_calls = 0

    async def fake_open(
        ws_url: str, headers: dict[str, str], *, connect_timeout: float | None = None
    ) -> DisconnectAfterSendWSConnection:
        nonlocal open_calls
        open_calls += 1
        if open_calls > 1:
            raise AssertionError("Unexpected websocket retry after request frame was sent")
        return ws

    monkeypatch.setattr(model, "_open_websocket_connection", fake_open)

    with pytest.raises(RuntimeError, match="before any response events were received"):
        await model.get_response(
            system_instructions=None,
            input="hi",
            model_settings=ModelSettings(),
            tools=[],
            output_schema=None,
            handoffs=[],
            tracing=ModelTracing.DISABLED,
        )

    assert open_calls == 1
    assert ws.send_calls == 1
    assert ws.recv_calls == 1
    assert ws.close_calls == 1
    assert model._ws_connection is None


@pytest.mark.allow_call_model_methods
@pytest.mark.asyncio
async def test_websocket_model_does_not_retry_after_client_initiated_close(monkeypatch):
    client = DummyWSClient()

    class ConnectionClosedError(Exception):
        pass

    ConnectionClosedError.__module__ = "websockets.client"

    class AbortableRecvWSConnection(DummyWSConnection):
        def __init__(self):
            super().__init__([])
            self.send_calls = 0
            self.recv_started = asyncio.Event()
            self.abort_event = asyncio.Event()

            class DummyTransport:
                def __init__(inner_self, outer: AbortableRecvWSConnection):
                    inner_self.outer = outer
                    inner_self.abort_calls = 0

                def abort(inner_self) -> None:
                    inner_self.abort_calls += 1
                    inner_self.outer.abort_event.set()

            self.transport = DummyTransport(self)

        async def send(self, payload: str) -> None:
            self.send_calls += 1
            await super().send(payload)

        async def recv(self) -> str:
            self.recv_started.set()
            await self.abort_event.wait()
            raise ConnectionClosedError("client closed websocket")

    ws = AbortableRecvWSConnection()
    model = OpenAIResponsesWSModel(model="gpt-4", openai_client=client)  # type: ignore[arg-type]
    open_calls = 0

    async def fake_open(
        ws_url: str, headers: dict[str, str], *, connect_timeout: float | None = None
    ) -> AbortableRecvWSConnection:
        nonlocal open_calls
        open_calls += 1
        if open_calls > 1:
            raise AssertionError("Unexpected websocket reconnect after client close")
        return ws

    monkeypatch.setattr(model, "_open_websocket_connection", fake_open)

    request_task = asyncio.create_task(
        model.get_response(
            system_instructions=None,
            input="hi",
            model_settings=ModelSettings(),
            tools=[],
            output_schema=None,
            handoffs=[],
            tracing=ModelTracing.DISABLED,
        )
    )

    await asyncio.wait_for(ws.recv_started.wait(), timeout=1.0)
    await asyncio.wait_for(model.close(), timeout=1.0)

    with pytest.raises(ConnectionClosedError, match="client closed websocket"):
        await asyncio.wait_for(request_task, timeout=1.0)

    assert open_calls == 1
    assert ws.send_calls == 1
    assert ws.transport.abort_calls == 1
    assert model._ws_connection is None


@pytest.mark.allow_call_model_methods
def test_websocket_model_prepare_websocket_url_preserves_non_tls_scheme_mapping():
    client = DummyWSClient()
    client.base_url = httpx.URL("http://127.0.0.1:8080/v1/")
    model = OpenAIResponsesWSModel(model="gpt-4", openai_client=client)  # type: ignore[arg-type]

    ws_url = model._prepare_websocket_url(extra_query=None)

    assert ws_url == "ws://127.0.0.1:8080/v1/responses"


@pytest.mark.allow_call_model_methods
def test_websocket_model_prepare_websocket_url_appends_path_with_existing_query():
    client = DummyWSClient()
    client.websocket_base_url = "wss://proxy.example.test/v1?token=abc"
    model = OpenAIResponsesWSModel(model="gpt-4", openai_client=client)  # type: ignore[arg-type]

    ws_url = model._prepare_websocket_url(extra_query={"route": "team-a"})
    parsed = httpx.URL(ws_url)

    assert parsed.path == "/v1/responses"
    assert dict(parsed.params) == {"token": "abc", "route": "team-a"}


@pytest.mark.allow_call_model_methods
@pytest.mark.parametrize(
    ("configured_ws_base_url", "expected_scheme"),
    [
        ("http://proxy.example.test/v1?token=abc", "ws"),
        ("https://proxy.example.test/v1?token=abc", "wss"),
    ],
)
def test_websocket_model_prepare_websocket_url_normalizes_explicit_http_schemes(
    configured_ws_base_url: str, expected_scheme: str
):
    client = DummyWSClient()
    client.websocket_base_url = configured_ws_base_url
    model = OpenAIResponsesWSModel(model="gpt-4", openai_client=client)  # type: ignore[arg-type]

    ws_url = model._prepare_websocket_url(extra_query={"route": "team-a"})
    parsed = httpx.URL(ws_url)

    assert parsed.scheme == expected_scheme
    assert parsed.path == "/v1/responses"
    assert dict(parsed.params) == {"token": "abc", "route": "team-a"}


@pytest.mark.allow_call_model_methods
@pytest.mark.parametrize("extra_query", [omit, NOT_GIVEN])
def test_websocket_model_prepare_websocket_url_treats_top_level_omit_sentinels_as_absent(
    extra_query,
):
    client = DummyWSClient()
    client.websocket_base_url = "wss://proxy.example.test/v1?token=abc"
    model = OpenAIResponsesWSModel(model="gpt-4", openai_client=client)  # type: ignore[arg-type]

    ws_url = model._prepare_websocket_url(extra_query=extra_query)
    parsed = httpx.URL(ws_url)

    assert parsed.path == "/v1/responses"
    assert dict(parsed.params) == {"token": "abc"}


@pytest.mark.allow_call_model_methods
def test_websocket_model_prepare_websocket_url_skips_not_given_query_values():
    client = DummyWSClient()
    client.websocket_base_url = "wss://proxy.example.test/v1?token=abc"
    client.default_query = {"api-version": NOT_GIVEN, "route": "team-a"}
    model = OpenAIResponsesWSModel(model="gpt-4", openai_client=client)  # type: ignore[arg-type]

    ws_url = model._prepare_websocket_url(extra_query={"tenant": NOT_GIVEN, "region": "us"})
    parsed = httpx.URL(ws_url)

    assert parsed.path == "/v1/responses"
    assert dict(parsed.params) == {"token": "abc", "route": "team-a", "region": "us"}


@pytest.mark.allow_call_model_methods
@pytest.mark.asyncio
async def test_websocket_model_prepare_websocket_request_filters_omit_from_extra_body():
    client = DummyWSClient()
    model = OpenAIResponsesWSModel(model="gpt-4", openai_client=client)  # type: ignore[arg-type]

    frame, _ws_url, _headers = await model._prepare_websocket_request(
        {
            "model": "gpt-4",
            "input": "hi",
            "stream": True,
            "extra_body": {"keep": "value", "drop": omit},
        }
    )

    assert frame["type"] == "response.create"
    assert frame["keep"] == "value"
    assert "drop" not in frame


@pytest.mark.allow_call_model_methods
@pytest.mark.asyncio
@pytest.mark.parametrize("extra_body", [omit, NOT_GIVEN])
async def test_websocket_model_prepare_websocket_request_ignores_top_level_extra_body_sentinels(
    extra_body,
):
    client = DummyWSClient()
    model = OpenAIResponsesWSModel(model="gpt-4", openai_client=client)  # type: ignore[arg-type]

    frame, _ws_url, _headers = await model._prepare_websocket_request(
        {
            "model": "gpt-4",
            "input": "hi",
            "stream": True,
            "extra_body": extra_body,
        }
    )

    assert frame["type"] == "response.create"
    assert frame["stream"] is True
    assert frame["model"] == "gpt-4"
    assert frame["input"] == "hi"


@pytest.mark.allow_call_model_methods
@pytest.mark.asyncio
async def test_websocket_model_prepare_websocket_request_preserves_envelope_fields():
    client = DummyWSClient()
    model = OpenAIResponsesWSModel(model="gpt-4", openai_client=client)  # type: ignore[arg-type]

    frame, _ws_url, _headers = await model._prepare_websocket_request(
        {
            "model": "gpt-4",
            "input": "hi",
            "stream": True,
            "extra_body": {
                "type": "not-response-create",
                "stream": False,
                "custom": "value",
            },
        }
    )

    assert frame["type"] == "response.create"
    assert frame["stream"] is True
    assert frame["custom"] == "value"


@pytest.mark.allow_call_model_methods
@pytest.mark.asyncio
async def test_websocket_model_prepare_websocket_request_strips_client_timeout_kwarg():
    client = DummyWSClient()
    model = OpenAIResponsesWSModel(model="gpt-4", openai_client=client)  # type: ignore[arg-type]

    frame, _ws_url, _headers = await model._prepare_websocket_request(
        {
            "model": "gpt-4",
            "input": "hi",
            "stream": True,
            "timeout": 30.0,
            "metadata": {"request_id": "123"},
        }
    )

    assert frame["type"] == "response.create"
    assert frame["metadata"] == {"request_id": "123"}
    assert "timeout" not in frame


@pytest.mark.allow_call_model_methods
@pytest.mark.asyncio
async def test_websocket_model_prepare_websocket_request_skips_not_given_values():
    client = DummyWSClient()
    model = OpenAIResponsesWSModel(model="gpt-4", openai_client=client)  # type: ignore[arg-type]

    frame, _ws_url, _headers = await model._prepare_websocket_request(
        {
            "model": "gpt-4",
            "input": "hi",
            "stream": True,
            "user": NOT_GIVEN,
            "stream_options": NOT_GIVEN,
            "extra_body": {
                "metadata": {"request_id": "123"},
                "optional_field": NOT_GIVEN,
            },
        }
    )

    assert frame["type"] == "response.create"
    assert frame["stream"] is True
    assert frame["metadata"] == {"request_id": "123"}
    assert "user" not in frame
    assert "stream_options" not in frame
    assert "optional_field" not in frame
    json.dumps(frame)


@pytest.mark.allow_call_model_methods
@pytest.mark.asyncio
async def test_websocket_model_get_response_applies_timeout_to_recv(monkeypatch):
    client = DummyWSClient()
    model = OpenAIResponsesWSModel(model="gpt-4", openai_client=client)  # type: ignore[arg-type]

    class SlowRecvWSConnection(DummyWSConnection):
        async def recv(self) -> str:
            await asyncio.sleep(0.2)
            return await super().recv()

    ws = SlowRecvWSConnection([_response_completed_frame("resp-timeout", 1)])

    async def fake_open(
        ws_url: str, headers: dict[str, str], *, connect_timeout: float | None = None
    ) -> DummyWSConnection:
        return ws

    monkeypatch.setattr(model, "_open_websocket_connection", fake_open)

    with pytest.raises(TimeoutError, match="Responses websocket receive timed out"):
        await model.get_response(
            system_instructions=None,
            input="hi",
            model_settings=ModelSettings(extra_args={"timeout": 0.01}),
            tools=[],
            output_schema=None,
            handoffs=[],
            tracing=ModelTracing.DISABLED,
        )

    assert ws.close_calls == 1


@pytest.mark.allow_call_model_methods
@pytest.mark.asyncio
async def test_websocket_model_get_response_marks_partial_receive_timeout_unsafe_to_replay(
    monkeypatch,
):
    client = DummyWSClient()
    model = OpenAIResponsesWSModel(model="gpt-4", openai_client=client)  # type: ignore[arg-type]

    class PartialThenSlowRecvWSConnection(DummyWSConnection):
        def __init__(self) -> None:
            super().__init__([_response_event_frame("response.created", "resp-partial", 1)])
            self.recv_calls = 0

        async def recv(self) -> str:
            self.recv_calls += 1
            if self.recv_calls == 1:
                return await super().recv()
            await asyncio.sleep(0.2)
            return await super().recv()

    ws = PartialThenSlowRecvWSConnection()

    async def fake_open(
        ws_url: str, headers: dict[str, str], *, connect_timeout: float | None = None
    ) -> DummyWSConnection:
        return ws

    monkeypatch.setattr(model, "_open_websocket_connection", fake_open)

    with pytest.raises(TimeoutError, match="Responses websocket receive timed out") as exc_info:
        await model.get_response(
            system_instructions=None,
            input="hi",
            model_settings=ModelSettings(extra_args={"timeout": 0.01}),
            tools=[],
            output_schema=None,
            handoffs=[],
            tracing=ModelTracing.DISABLED,
        )

    error = exc_info.value
    assert getattr(error, "_openai_agents_ws_replay_safety", None) == "unsafe"

    advice = model.get_retry_advice(
        ModelRetryAdviceRequest(
            error=error,
            attempt=1,
            stream=False,
        )
    )

    assert advice is not None
    assert advice.suggested is False
    assert advice.replay_safety == "unsafe"
    assert ws.close_calls == 1


@pytest.mark.allow_call_model_methods
@pytest.mark.asyncio
async def test_websocket_model_get_response_applies_timeout_while_waiting_for_request_lock(
    monkeypatch,
):
    client = DummyWSClient()
    model = OpenAIResponsesWSModel(model="gpt-4", openai_client=client)  # type: ignore[arg-type]
    recv_started = asyncio.Event()
    release_first_request = asyncio.Event()

    class BlockingRecvWSConnection(DummyWSConnection):
        async def recv(self) -> str:
            recv_started.set()
            await release_first_request.wait()
            return await super().recv()

    ws = BlockingRecvWSConnection(
        [
            _response_completed_frame("resp-lock-1", 1),
            _response_completed_frame("resp-lock-2", 2),
        ]
    )

    async def fake_open(
        ws_url: str, headers: dict[str, str], *, connect_timeout: float | None = None
    ) -> DummyWSConnection:
        return ws

    monkeypatch.setattr(model, "_open_websocket_connection", fake_open)

    first_task = asyncio.create_task(
        model.get_response(
            system_instructions=None,
            input="first",
            model_settings=ModelSettings(),
            tools=[],
            output_schema=None,
            handoffs=[],
            tracing=ModelTracing.DISABLED,
        )
    )

    await asyncio.wait_for(recv_started.wait(), timeout=1.0)

    with pytest.raises(TimeoutError, match="request lock wait timed out"):
        await model.get_response(
            system_instructions=None,
            input="second",
            model_settings=ModelSettings(extra_args={"timeout": 0.01}),
            tools=[],
            output_schema=None,
            handoffs=[],
            tracing=ModelTracing.DISABLED,
        )

    release_first_request.set()
    first_response = await first_task

    assert first_response.response_id == "resp-lock-1"
    assert len(ws.sent_messages) == 1


@pytest.mark.allow_call_model_methods
@pytest.mark.asyncio
async def test_websocket_model_get_response_allows_zero_pool_timeout_when_lock_uncontended(
    monkeypatch,
):
    client = DummyWSClient()
    client.timeout = httpx.Timeout(connect=1.0, read=1.0, write=1.0, pool=0.0)
    model = OpenAIResponsesWSModel(model="gpt-4", openai_client=client)  # type: ignore[arg-type]
    ws = DummyWSConnection([_response_completed_frame("resp-zero-pool", 1)])

    async def fake_open(
        ws_url: str, headers: dict[str, str], *, connect_timeout: float | None = None
    ) -> DummyWSConnection:
        return ws

    monkeypatch.setattr(model, "_open_websocket_connection", fake_open)

    response = await model.get_response(
        system_instructions=None,
        input="hi",
        model_settings=ModelSettings(),
        tools=[],
        output_schema=None,
        handoffs=[],
        tracing=ModelTracing.DISABLED,
    )

    assert response.response_id == "resp-zero-pool"
    assert len(ws.sent_messages) == 1


@pytest.mark.allow_call_model_methods
@pytest.mark.asyncio
async def test_websocket_model_get_response_allows_zero_timeout_when_ws_ops_are_immediate(
    monkeypatch,
):
    client = DummyWSClient()
    model = OpenAIResponsesWSModel(model="gpt-4", openai_client=client)  # type: ignore[arg-type]
    ws = DummyWSConnection([_response_completed_frame("resp-zero-timeout", 1)])

    async def fake_open(
        ws_url: str, headers: dict[str, str], *, connect_timeout: float | None = None
    ) -> DummyWSConnection:
        return ws

    monkeypatch.setattr(model, "_open_websocket_connection", fake_open)

    response = await model.get_response(
        system_instructions=None,
        input="hi",
        model_settings=ModelSettings(extra_args={"timeout": 0}),
        tools=[],
        output_schema=None,
        handoffs=[],
        tracing=ModelTracing.DISABLED,
    )

    assert response.response_id == "resp-zero-timeout"
    assert len(ws.sent_messages) == 1


@pytest.mark.allow_call_model_methods
@pytest.mark.asyncio
async def test_websocket_model_get_response_uses_client_default_timeout_when_no_override(
    monkeypatch,
):
    client = DummyWSClient()
    client.timeout = httpx.Timeout(connect=1.0, read=0.01, write=1.0, pool=1.0)
    model = OpenAIResponsesWSModel(model="gpt-4", openai_client=client)  # type: ignore[arg-type]

    class SlowRecvWSConnection(DummyWSConnection):
        async def recv(self) -> str:
            await asyncio.sleep(0.2)
            return await super().recv()

    ws = SlowRecvWSConnection([_response_completed_frame("resp-timeout-default", 1)])

    async def fake_open(
        ws_url: str, headers: dict[str, str], *, connect_timeout: float | None = None
    ) -> DummyWSConnection:
        return ws

    monkeypatch.setattr(model, "_open_websocket_connection", fake_open)

    with pytest.raises(TimeoutError, match="Responses websocket receive timed out"):
        await model.get_response(
            system_instructions=None,
            input="hi",
            model_settings=ModelSettings(),
            tools=[],
            output_schema=None,
            handoffs=[],
            tracing=ModelTracing.DISABLED,
        )

    assert ws.close_calls == 1


@pytest.mark.allow_call_model_methods
@pytest.mark.asyncio
async def test_websocket_model_get_response_uses_client_default_timeout_when_override_is_not_given(
    monkeypatch,
):
    client = DummyWSClient()
    client.timeout = httpx.Timeout(connect=1.0, read=0.01, write=1.0, pool=1.0)
    model = OpenAIResponsesWSModel(model="gpt-4", openai_client=client)  # type: ignore[arg-type]

    class SlowRecvWSConnection(DummyWSConnection):
        async def recv(self) -> str:
            await asyncio.sleep(0.2)
            return await super().recv()

    ws = SlowRecvWSConnection([_response_completed_frame("resp-timeout-not-given", 1)])

    async def fake_open(
        ws_url: str, headers: dict[str, str], *, connect_timeout: float | None = None
    ) -> DummyWSConnection:
        return ws

    monkeypatch.setattr(model, "_open_websocket_connection", fake_open)

    with pytest.raises(TimeoutError, match="Responses websocket receive timed out"):
        await model.get_response(
            system_instructions=None,
            input="hi",
            model_settings=ModelSettings(extra_args={"timeout": NOT_GIVEN}),
            tools=[],
            output_schema=None,
            handoffs=[],
            tracing=ModelTracing.DISABLED,
        )

    assert ws.close_calls == 1


@pytest.mark.allow_call_model_methods
@pytest.mark.asyncio
async def test_websocket_model_prepare_websocket_request_includes_client_auth_headers():
    client = DummyWSClient()
    model = OpenAIResponsesWSModel(model="gpt-4", openai_client=client)  # type: ignore[arg-type]

    _frame, _ws_url, headers = await model._prepare_websocket_request(
        {
            "model": "gpt-4",
            "input": "hi",
            "stream": True,
        }
    )

    assert headers["Authorization"] == "Bearer test-key"


@pytest.mark.allow_call_model_methods
@pytest.mark.asyncio
async def test_websocket_model_default_headers_override_auth_case_insensitively():
    client = DummyWSClient()
    client.default_headers["authorization"] = "Bearer override-key"
    model = OpenAIResponsesWSModel(model="gpt-4", openai_client=client)  # type: ignore[arg-type]

    _frame, _ws_url, headers = await model._prepare_websocket_request(
        {
            "model": "gpt-4",
            "input": "hi",
            "stream": True,
        }
    )

    assert headers["authorization"] == "Bearer override-key"
    assert "Authorization" not in headers


@pytest.mark.allow_call_model_methods
@pytest.mark.asyncio
async def test_websocket_model_prepare_websocket_request_omit_removes_inherited_header():
    client = DummyWSClient()
    model = OpenAIResponsesWSModel(model="gpt-4", openai_client=client)  # type: ignore[arg-type]

    _frame, _ws_url, headers = await model._prepare_websocket_request(
        {
            "model": "gpt-4",
            "input": "hi",
            "stream": True,
            "extra_headers": {"User-Agent": omit},
        }
    )

    assert "Authorization" in headers
    assert "User-Agent" not in headers


@pytest.mark.allow_call_model_methods
@pytest.mark.asyncio
async def test_websocket_model_prepare_websocket_request_replaces_header_case_insensitively():
    client = DummyWSClient()
    model = OpenAIResponsesWSModel(model="gpt-4", openai_client=client)  # type: ignore[arg-type]

    _frame, _ws_url, headers = await model._prepare_websocket_request(
        {
            "model": "gpt-4",
            "input": "hi",
            "stream": True,
            "extra_headers": {
                "authorization": "Bearer override-key",
                "user-agent": "Custom UA",
            },
        }
    )

    assert headers["authorization"] == "Bearer override-key"
    assert headers["user-agent"] == "Custom UA"
    assert "Authorization" not in headers
    assert "User-Agent" not in headers


@pytest.mark.allow_call_model_methods
@pytest.mark.asyncio
async def test_websocket_model_prepare_websocket_request_skips_not_given_header_values():
    client = DummyWSClient()
    model = OpenAIResponsesWSModel(model="gpt-4", openai_client=client)  # type: ignore[arg-type]

    _frame, _ws_url, headers = await model._prepare_websocket_request(
        {
            "model": "gpt-4",
            "input": "hi",
            "stream": True,
            "extra_headers": {
                "Authorization": NOT_GIVEN,
                "X-Optional": NOT_GIVEN,
            },
        }
    )

    assert headers["Authorization"] == "Bearer test-key"
    assert "X-Optional" not in headers
    assert "NOT_GIVEN" not in headers.values()


@pytest.mark.allow_call_model_methods
def test_websocket_model_prepare_websocket_url_includes_client_default_query():
    client = DummyWSClient()
    client.websocket_base_url = "wss://proxy.example.test/v1?token=abc"
    client.default_query = {"api-version": "2025-01-01-preview", "omit_me": omit}
    model = OpenAIResponsesWSModel(model="gpt-4", openai_client=client)  # type: ignore[arg-type]

    ws_url = model._prepare_websocket_url(
        extra_query={"route": "team-a", "api-version": "2026-01-01-preview"}
    )
    parsed = httpx.URL(ws_url)

    assert parsed.path == "/v1/responses"
    assert dict(parsed.params) == {
        "token": "abc",
        "api-version": "2026-01-01-preview",
        "route": "team-a",
    }


@pytest.mark.allow_call_model_methods
def test_websocket_model_prepare_websocket_url_omit_removes_inherited_query_params():
    client = DummyWSClient()
    client.websocket_base_url = "wss://proxy.example.test/v1?token=abc"
    client.default_query = {"route": "team-a", "region": "us"}
    model = OpenAIResponsesWSModel(model="gpt-4", openai_client=client)  # type: ignore[arg-type]

    ws_url = model._prepare_websocket_url(extra_query={"token": omit, "route": omit, "keep": "1"})
    parsed = httpx.URL(ws_url)

    assert parsed.path == "/v1/responses"
    assert dict(parsed.params) == {"region": "us", "keep": "1"}


@pytest.mark.allow_call_model_methods
@pytest.mark.asyncio
async def test_websocket_model_close_closes_persistent_connection(monkeypatch):
    client = DummyWSClient()
    ws = DummyWSConnection([_response_completed_frame("resp-close", 1)])
    model = OpenAIResponsesWSModel(model="gpt-4", openai_client=client)  # type: ignore[arg-type]

    async def fake_open(
        ws_url: str, headers: dict[str, str], *, connect_timeout: float | None = None
    ) -> DummyWSConnection:
        return ws

    monkeypatch.setattr(model, "_open_websocket_connection", fake_open)

    await model.get_response(
        system_instructions=None,
        input="hi",
        model_settings=ModelSettings(),
        tools=[],
        output_schema=None,
        handoffs=[],
        tracing=ModelTracing.DISABLED,
    )

    assert ws.close_calls == 0
    await model.close()
    assert ws.close_calls == 1


@pytest.mark.allow_call_model_methods
@pytest.mark.asyncio
async def test_websocket_model_close_falls_back_to_transport_abort_on_close_error():
    client = DummyWSClient()
    model = OpenAIResponsesWSModel(model="gpt-4", openai_client=client)  # type: ignore[arg-type]

    class DummyTransport:
        def __init__(self):
            self.abort_calls = 0

        def abort(self):
            self.abort_calls += 1

    class FailingWSConnection:
        def __init__(self):
            self.transport = DummyTransport()

        async def close(self):
            raise RuntimeError("attached to a different loop")

    ws = FailingWSConnection()
    model._ws_connection = ws
    model._ws_connection_identity = ("wss://example.test", (("authorization", "x"),))

    await model.close()

    assert ws.transport.abort_calls == 1
    assert model._ws_connection is None
    assert model._ws_connection_identity is None


@pytest.mark.allow_call_model_methods
@pytest.mark.asyncio
async def test_websocket_model_close_does_not_wait_for_held_request_lock():
    client = DummyWSClient()
    model = OpenAIResponsesWSModel(model="gpt-4", openai_client=client)  # type: ignore[arg-type]
    request_lock = model._get_ws_request_lock()
    await request_lock.acquire()

    class DummyTransport:
        def __init__(self):
            self.abort_calls = 0

        def abort(self):
            self.abort_calls += 1

    class HangingCloseWSConnection:
        def __init__(self):
            self.transport = DummyTransport()
            self.close_calls = 0

        async def close(self) -> None:
            self.close_calls += 1
            await asyncio.sleep(3600)

    ws = HangingCloseWSConnection()
    model._ws_connection = ws
    model._ws_connection_identity = ("wss://example.test", (("authorization", "x"),))

    try:
        await asyncio.wait_for(model.close(), timeout=0.1)
    finally:
        if request_lock.locked():
            request_lock.release()

    assert ws.transport.abort_calls == 1
    assert ws.close_calls == 0
    assert model._ws_connection is None
    assert model._ws_connection_identity is None


@pytest.mark.allow_call_model_methods
@pytest.mark.asyncio
async def test_websocket_model_open_websocket_connection_disables_message_size_limit(monkeypatch):
    client = DummyWSClient()
    model = OpenAIResponsesWSModel(model="gpt-4", openai_client=client)  # type: ignore[arg-type]
    captured: dict[str, Any] = {}
    sentinel = object()

    async def fake_connect(*args: Any, **kwargs: Any) -> object:
        captured["args"] = args
        captured["kwargs"] = kwargs
        return sentinel

    monkeypatch.setattr("websockets.asyncio.client.connect", fake_connect)

    result = await model._open_websocket_connection(
        "wss://proxy.example.test/v1/responses",
        {"Authorization": "Bearer test-key"},
        connect_timeout=None,
    )

    assert result is sentinel
    assert captured["args"] == ("wss://proxy.example.test/v1/responses",)
    assert captured["kwargs"]["user_agent_header"] is None
    assert captured["kwargs"]["additional_headers"] == {"Authorization": "Bearer test-key"}
    assert captured["kwargs"]["max_size"] is None
    assert captured["kwargs"]["open_timeout"] is None


@pytest.mark.allow_call_model_methods
@pytest.mark.asyncio
async def test_websocket_model_open_websocket_connection_honors_connect_timeout(monkeypatch):
    client = DummyWSClient()
    model = OpenAIResponsesWSModel(model="gpt-4", openai_client=client)  # type: ignore[arg-type]
    captured: dict[str, Any] = {}
    sentinel = object()

    async def fake_connect(*args: Any, **kwargs: Any) -> object:
        captured["args"] = args
        captured["kwargs"] = kwargs
        return sentinel

    monkeypatch.setattr("websockets.asyncio.client.connect", fake_connect)

    result = await model._open_websocket_connection(
        "wss://proxy.example.test/v1/responses",
        {"Authorization": "Bearer test-key"},
        connect_timeout=42.0,
    )

    assert result is sentinel
    assert captured["kwargs"]["open_timeout"] == 42.0


@pytest.mark.allow_call_model_methods
def test_get_retry_advice_uses_openai_headers() -> None:
    request = httpx.Request("POST", "https://api.openai.com/v1/responses")
    response = httpx.Response(
        429,
        request=request,
        headers={
            "x-should-retry": "true",
            "retry-after-ms": "250",
            "x-request-id": "req_456",
        },
        json={"error": {"code": "rate_limit"}},
    )
    error = RateLimitError(
        "rate limited", response=response, body={"error": {"code": "rate_limit"}}
    )
    model = OpenAIResponsesModel(model="gpt-4", openai_client=cast(Any, object()))

    advice = model.get_retry_advice(
        ModelRetryAdviceRequest(
            error=error,
            attempt=1,
            stream=False,
        )
    )

    assert advice is not None
    assert advice.suggested is True
    assert advice.retry_after == 0.25
    assert advice.replay_safety == "safe"
    assert advice.normalized is not None
    assert advice.normalized.error_code == "rate_limit"
    assert advice.normalized.status_code == 429
    assert advice.normalized.request_id == "req_456"


@pytest.mark.allow_call_model_methods
def test_get_retry_advice_keeps_stateful_transport_failures_ambiguous() -> None:
    model = OpenAIResponsesModel(model="gpt-4", openai_client=cast(Any, object()))
    error = APIConnectionError(
        message="connection error",
        request=httpx.Request("POST", "https://api.openai.com/v1/responses"),
    )

    advice = model.get_retry_advice(
        ModelRetryAdviceRequest(
            error=error,
            attempt=1,
            stream=False,
            previous_response_id="resp_prev",
        )
    )

    assert advice is not None
    assert advice.suggested is True
    assert advice.replay_safety is None
    assert advice.normalized is not None
    assert advice.normalized.is_network_error is True


@pytest.mark.allow_call_model_methods
def test_get_retry_advice_marks_stateful_http_failures_replay_safe() -> None:
    request = httpx.Request("POST", "https://api.openai.com/v1/responses")
    response = httpx.Response(
        429,
        request=request,
        json={"error": {"code": "rate_limit"}},
    )
    error = RateLimitError(
        "rate limited", response=response, body={"error": {"code": "rate_limit"}}
    )
    model = OpenAIResponsesModel(model="gpt-4", openai_client=cast(Any, object()))

    advice = model.get_retry_advice(
        ModelRetryAdviceRequest(
            error=error,
            attempt=1,
            stream=False,
            previous_response_id="resp_prev",
        )
    )

    assert advice is not None
    assert advice.suggested is True
    assert advice.replay_safety == "safe"
    assert advice.normalized is not None
    assert advice.normalized.status_code == 429


@pytest.mark.allow_call_model_methods
def test_get_retry_advice_keeps_stateless_transport_failures_retryable() -> None:
    model = OpenAIResponsesModel(model="gpt-4", openai_client=cast(Any, object()))
    error = APIConnectionError(
        message="connection error",
        request=httpx.Request("POST", "https://api.openai.com/v1/responses"),
    )

    advice = model.get_retry_advice(
        ModelRetryAdviceRequest(
            error=error,
            attempt=1,
            stream=False,
        )
    )

    assert advice is not None
    assert advice.suggested is True
    assert advice.replay_safety is None
    assert advice.normalized is not None
    assert advice.normalized.is_network_error is True


@pytest.mark.allow_call_model_methods
def test_websocket_get_retry_advice_marks_ambiguous_replay_unsafe() -> None:
    model = OpenAIResponsesWSModel(model="gpt-4", openai_client=cast(Any, DummyWSClient()))
    error = RuntimeError("Responses websocket connection closed before a terminal response event.")
    error.__cause__ = _connection_closed_error("peer closed after request send")

    advice = model.get_retry_advice(
        ModelRetryAdviceRequest(
            error=error,
            attempt=1,
            stream=True,
            previous_response_id="resp_prev",
        )
    )

    assert advice is not None
    assert advice.suggested is False
    assert advice.replay_safety == "unsafe"


@pytest.mark.allow_call_model_methods
def test_websocket_get_retry_advice_allows_stateless_ambiguous_disconnect_retry() -> None:
    model = OpenAIResponsesWSModel(model="gpt-4", openai_client=cast(Any, DummyWSClient()))
    error = RuntimeError("Responses websocket connection closed before a terminal response event.")
    error.__cause__ = _connection_closed_error("peer closed after request send")

    advice = model.get_retry_advice(
        ModelRetryAdviceRequest(
            error=error,
            attempt=1,
            stream=True,
        )
    )

    assert advice is not None
    assert advice.suggested is True
    assert advice.replay_safety is None


@pytest.mark.allow_call_model_methods
def test_websocket_get_retry_advice_keeps_wrapped_pre_send_disconnect_safe() -> None:
    model = OpenAIResponsesWSModel(model="gpt-4", openai_client=cast(Any, DummyWSClient()))
    error = RuntimeError(
        "Responses websocket connection closed before any response events were received."
    )
    setattr(error, "_openai_agents_ws_replay_safety", "safe")  # noqa: B010
    error.__cause__ = _connection_closed_error("peer closed before request send")

    advice = model.get_retry_advice(
        ModelRetryAdviceRequest(
            error=error,
            attempt=1,
            stream=True,
            previous_response_id="resp_prev",
        )
    )

    assert advice is not None
    assert advice.suggested is True
    assert advice.replay_safety == "safe"


@pytest.mark.allow_call_model_methods
def test_websocket_get_retry_advice_allows_stateless_wrapped_post_send_disconnect_retry() -> None:
    model = OpenAIResponsesWSModel(model="gpt-4", openai_client=cast(Any, DummyWSClient()))
    error = RuntimeError(
        "Responses websocket connection closed before any response events were received."
    )
    setattr(error, "_openai_agents_ws_replay_safety", "unsafe")  # noqa: B010
    error.__cause__ = _connection_closed_error("peer closed after request send")

    advice = model.get_retry_advice(
        ModelRetryAdviceRequest(
            error=error,
            attempt=1,
            stream=True,
        )
    )

    assert advice is not None
    assert advice.suggested is True
    assert advice.replay_safety is None


@pytest.mark.allow_call_model_methods
def test_websocket_get_retry_advice_allows_stateless_nonstream_post_send_retry() -> None:
    model = OpenAIResponsesWSModel(model="gpt-4", openai_client=cast(Any, DummyWSClient()))
    error = RuntimeError(
        "Responses websocket connection closed before any response events were received."
    )
    setattr(error, "_openai_agents_ws_replay_safety", "unsafe")  # noqa: B010
    error.__cause__ = _connection_closed_error("peer closed after request send")

    advice = model.get_retry_advice(
        ModelRetryAdviceRequest(
            error=error,
            attempt=1,
            stream=False,
        )
    )

    assert advice is not None
    assert advice.suggested is True
    assert advice.replay_safety is None


@pytest.mark.allow_call_model_methods
def test_websocket_get_retry_advice_marks_wrapped_post_send_disconnect_unsafe() -> None:
    model = OpenAIResponsesWSModel(model="gpt-4", openai_client=cast(Any, DummyWSClient()))
    error = RuntimeError(
        "Responses websocket connection closed before any response events were received."
    )
    setattr(error, "_openai_agents_ws_replay_safety", "unsafe")  # noqa: B010
    error.__cause__ = _connection_closed_error("peer closed after request send")

    advice = model.get_retry_advice(
        ModelRetryAdviceRequest(
            error=error,
            attempt=1,
            stream=True,
            previous_response_id="resp_prev",
        )
    )

    assert advice is not None
    assert advice.suggested is False
    assert advice.replay_safety == "unsafe"


@pytest.mark.allow_call_model_methods
def test_websocket_get_retry_advice_marks_partial_nonstream_failure_unsafe() -> None:
    model = OpenAIResponsesWSModel(model="gpt-4", openai_client=cast(Any, DummyWSClient()))
    error = TimeoutError("Responses websocket receive timed out after 5.0 seconds.")
    setattr(error, "_openai_agents_ws_replay_safety", "unsafe")  # noqa: B010
    setattr(error, "_openai_agents_ws_response_started", True)  # noqa: B010

    advice = model.get_retry_advice(
        ModelRetryAdviceRequest(
            error=error,
            attempt=1,
            stream=False,
        )
    )

    assert advice is not None
    assert advice.suggested is False
    assert advice.replay_safety == "unsafe"


@pytest.mark.allow_call_model_methods
def test_websocket_get_retry_advice_marks_connect_timeout_replay_safe() -> None:
    model = OpenAIResponsesWSModel(model="gpt-4", openai_client=cast(Any, DummyWSClient()))
    error = TimeoutError("Responses websocket connect timed out after 5.0 seconds.")

    advice = model.get_retry_advice(
        ModelRetryAdviceRequest(
            error=error,
            attempt=1,
            stream=True,
            previous_response_id="resp_prev",
        )
    )

    assert advice is not None
    assert advice.suggested is True
    assert advice.replay_safety == "safe"


@pytest.mark.allow_call_model_methods
def test_websocket_get_retry_advice_marks_request_lock_timeout_replay_safe() -> None:
    model = OpenAIResponsesWSModel(model="gpt-4", openai_client=cast(Any, DummyWSClient()))
    error = TimeoutError("Responses websocket request lock wait timed out after 5.0 seconds.")

    advice = model.get_retry_advice(
        ModelRetryAdviceRequest(
            error=error,
            attempt=1,
            stream=False,
            previous_response_id="resp_prev",
        )
    )

    assert advice is not None
    assert advice.suggested is True
    assert advice.replay_safety == "safe"


@pytest.mark.allow_call_model_methods
def test_websocket_get_retry_advice_marks_stateful_receive_timeout_unsafe() -> None:
    model = OpenAIResponsesWSModel(model="gpt-4", openai_client=cast(Any, DummyWSClient()))
    error = TimeoutError("Responses websocket receive timed out after 5.0 seconds.")

    advice = model.get_retry_advice(
        ModelRetryAdviceRequest(
            error=error,
            attempt=1,
            stream=True,
            previous_response_id="resp_prev",
        )
    )

    assert advice is not None
    assert advice.suggested is False
    assert advice.replay_safety == "unsafe"


@pytest.mark.allow_call_model_methods
def test_websocket_get_retry_advice_allows_stateless_receive_timeout_retry() -> None:
    model = OpenAIResponsesWSModel(model="gpt-4", openai_client=cast(Any, DummyWSClient()))
    error = TimeoutError("Responses websocket receive timed out after 5.0 seconds.")

    advice = model.get_retry_advice(
        ModelRetryAdviceRequest(
            error=error,
            attempt=1,
            stream=True,
        )
    )

    assert advice is not None
    assert advice.suggested is True
    assert advice.replay_safety is None


def test_get_client_disables_provider_managed_retries_when_requested() -> None:
    class DummyClient:
        def __init__(self):
            self.calls: list[dict[str, int]] = []

        def with_options(self, **kwargs):
            self.calls.append(kwargs)
            return "retry-client"

    client = DummyClient()
    model = OpenAIResponsesModel(model="gpt-4", openai_client=cast(Any, client))

    assert cast(object, model._get_client()) is client

    with provider_managed_retries_disabled(True):
        assert cast(object, model._get_client()) == "retry-client"

    assert client.calls == [{"max_retries": 0}]


def test_websocket_pre_event_disconnect_retry_respects_websocket_retry_disable() -> None:
    assert _should_retry_pre_event_websocket_disconnect() is True

    with websocket_pre_event_retries_disabled(True):
        assert _should_retry_pre_event_websocket_disconnect() is False
