from __future__ import annotations

import logging
from collections.abc import AsyncIterator
from typing import Any, cast

import httpx
import pytest
from openai import APIConnectionError, APIStatusError, AsyncOpenAI, omit
from openai.types.chat.chat_completion import ChatCompletion, Choice, ChoiceLogprobs
from openai.types.chat.chat_completion_chunk import ChatCompletionChunk
from openai.types.chat.chat_completion_message import ChatCompletionMessage
from openai.types.chat.chat_completion_message_custom_tool_call import (
    ChatCompletionMessageCustomToolCall,
    Custom,
)
from openai.types.chat.chat_completion_message_tool_call import (  # type: ignore[attr-defined]
    ChatCompletionMessageFunctionToolCall,
    Function,
)
from openai.types.chat.chat_completion_token_logprob import (
    ChatCompletionTokenLogprob,
    TopLogprob,
)
from openai.types.completion_usage import (
    CompletionUsage,
    PromptTokensDetails,
)
from openai.types.responses import (
    Response,
    ResponseFunctionToolCall,
    ResponseOutputMessage,
    ResponseOutputRefusal,
    ResponseOutputText,
)
from openai.types.shared import Reasoning

from agents import (
    Agent,
    ModelResponse,
    ModelRetryAdviceRequest,
    ModelSettings,
    ModelTracing,
    OpenAIChatCompletionsModel,
    OpenAIProvider,
    Runner,
    __version__,
    generation_span,
)
from agents.exceptions import UserError
from agents.models._retry_runtime import provider_managed_retries_disabled
from agents.models.chatcmpl_helpers import HEADERS_OVERRIDE, ChatCmplHelpers
from agents.models.fake_id import FAKE_RESPONSES_ID


def _minimal_chat_completion(content: str = "ok") -> ChatCompletion:
    return ChatCompletion(
        id="resp-id",
        created=0,
        model="fake",
        object="chat.completion",
        choices=[
            Choice(
                index=0,
                finish_reason="stop",
                message=ChatCompletionMessage(role="assistant", content=content),
            )
        ],
    )


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

        async def create(self, **kwargs: Any) -> Any:
            self.kwargs = kwargs
            return ChatCompletion(
                id="resp-id",
                created=0,
                model="fake",
                object="chat.completion",
                choices=[
                    Choice(
                        index=0,
                        finish_reason="stop",
                        message=ChatCompletionMessage(role="assistant", content="ok"),
                    )
                ],
            )

    class DummyClient:
        def __init__(self, completions: DummyCompletions) -> None:
            self.chat = type("_Chat", (), {"completions": completions})()
            self.base_url = httpx.URL("https://custom.example.test/v1/")

    completions = DummyCompletions()
    model = OpenAIChatCompletionsModel(
        model="gpt-4",
        openai_client=DummyClient(completions),  # type: ignore[arg-type]
    )
    agent = Agent(name="test", model=model, model_settings=model_settings or ModelSettings())

    await Runner.run(agent, "hi")

    return completions.kwargs


@pytest.mark.allow_call_model_methods
@pytest.mark.asyncio
async def test_get_response_with_text_message(monkeypatch) -> None:
    """
    When the model returns a ChatCompletionMessage with plain text content,
    `get_response` should produce a single `ResponseOutputMessage` containing
    a `ResponseOutputText` with that content, and a `Usage` populated from
    the completion's usage.
    """
    msg = ChatCompletionMessage(role="assistant", content="Hello")
    choice = Choice(index=0, finish_reason="stop", message=msg)
    chat = ChatCompletion(
        id="resp-id",
        created=0,
        model="fake",
        object="chat.completion",
        choices=[choice],
        usage=CompletionUsage(
            completion_tokens=5,
            prompt_tokens=7,
            total_tokens=12,
            # completion_tokens_details left blank to test default
            prompt_tokens_details=PromptTokensDetails.model_validate(
                {"cached_tokens": 3, "cache_write_tokens": 4}
            ),
        ),
    )

    async def patched_fetch_response(self, *args, **kwargs):
        return chat

    monkeypatch.setattr(OpenAIChatCompletionsModel, "_fetch_response", patched_fetch_response)
    model = OpenAIProvider(use_responses=False).get_model("gpt-4")
    resp: ModelResponse = await model.get_response(
        system_instructions=None,
        input="",
        model_settings=ModelSettings(),
        tools=[],
        output_schema=None,
        handoffs=[],
        tracing=ModelTracing.DISABLED,
        previous_response_id=None,
        conversation_id=None,
        prompt=None,
    )
    # Should have produced exactly one output message with one text part
    assert isinstance(resp, ModelResponse)
    assert len(resp.output) == 1
    assert isinstance(resp.output[0], ResponseOutputMessage)
    msg_item = resp.output[0]
    assert len(msg_item.content) == 1
    assert isinstance(msg_item.content[0], ResponseOutputText)
    assert msg_item.content[0].text == "Hello"
    # Usage should be preserved from underlying ChatCompletion.usage
    assert resp.usage.input_tokens == 7
    assert resp.usage.output_tokens == 5
    assert resp.usage.total_tokens == 12
    assert resp.usage.input_tokens_details.cached_tokens == 3
    assert getattr(resp.usage.input_tokens_details, "cache_write_tokens", None) == 4
    assert resp.usage.output_tokens_details.reasoning_tokens == 0
    assert resp.response_id is None


@pytest.mark.allow_call_model_methods
@pytest.mark.asyncio
@pytest.mark.parametrize(
    ("previous_response_id", "conversation_id", "expected_param"),
    [
        ("resp_123", None, "previous_response_id"),
        (None, "conv_123", "conversation_id"),
    ],
)
async def test_get_response_warns_and_ignores_server_managed_conversation_state_by_default(
    monkeypatch: pytest.MonkeyPatch,
    caplog: pytest.LogCaptureFixture,
    previous_response_id: str | None,
    conversation_id: str | None,
    expected_param: str,
) -> None:
    called = False

    async def patched_fetch_response(self, *args, **kwargs):
        nonlocal called
        called = True
        return _minimal_chat_completion()

    monkeypatch.setattr(OpenAIChatCompletionsModel, "_fetch_response", patched_fetch_response)
    model = OpenAIProvider(use_responses=False).get_model("gpt-4")
    caplog.set_level(logging.WARNING, logger="openai.agents")

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

    assert expected_param in caplog.text
    assert "Ignoring unsupported server-managed conversation state" in caplog.text
    assert called is True


@pytest.mark.allow_call_model_methods
@pytest.mark.asyncio
async def test_get_response_warns_and_ignores_prompt_by_default(
    monkeypatch: pytest.MonkeyPatch, caplog: pytest.LogCaptureFixture
) -> None:
    captured_prompt: Any = None

    async def patched_fetch_response(self, *args, **kwargs):
        nonlocal captured_prompt
        captured_prompt = kwargs.get("prompt")
        return _minimal_chat_completion()

    monkeypatch.setattr(OpenAIChatCompletionsModel, "_fetch_response", patched_fetch_response)
    model = OpenAIProvider(use_responses=False).get_model("gpt-4")
    caplog.set_level(logging.WARNING, logger="openai.agents")

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

    assert "Reusable prompts are only supported by the Responses API" in caplog.text
    assert "Ignoring `prompt`" in caplog.text
    assert captured_prompt is None


@pytest.mark.allow_call_model_methods
@pytest.mark.asyncio
@pytest.mark.parametrize(
    ("previous_response_id", "conversation_id", "expected_param"),
    [
        ("resp_123", None, "previous_response_id"),
        (None, "conv_123", "conversation_id"),
    ],
)
async def test_get_response_rejects_server_managed_conversation_state_in_strict_mode(
    monkeypatch: pytest.MonkeyPatch,
    previous_response_id: str | None,
    conversation_id: str | None,
    expected_param: str,
) -> None:
    called = False

    async def patched_fetch_response(self, *args, **kwargs):
        nonlocal called
        called = True
        raise AssertionError("_fetch_response should not be called")

    monkeypatch.setattr(OpenAIChatCompletionsModel, "_fetch_response", patched_fetch_response)
    model = OpenAIProvider(
        use_responses=False,
        strict_feature_validation=True,
    ).get_model("gpt-4")

    with pytest.raises(UserError, match="server-managed conversation state") as exc_info:
        await model.get_response(
            system_instructions=None,
            input="",
            model_settings=ModelSettings(),
            tools=[],
            output_schema=None,
            handoffs=[],
            tracing=ModelTracing.DISABLED,
            previous_response_id=previous_response_id,
            conversation_id=conversation_id,
            prompt=None,
        )

    assert expected_param in str(exc_info.value)
    assert called is False


@pytest.mark.allow_call_model_methods
@pytest.mark.asyncio
async def test_get_response_rejects_prompt_in_strict_mode(monkeypatch) -> None:
    async def patched_fetch_response(self, *args, **kwargs):
        raise AssertionError("_fetch_response should not run when prompt is unsupported")

    monkeypatch.setattr(OpenAIChatCompletionsModel, "_fetch_response", patched_fetch_response)
    model = OpenAIProvider(
        use_responses=False,
        strict_feature_validation=True,
    ).get_model("gpt-4")

    with pytest.raises(UserError, match="Reusable prompts"):
        await model.get_response(
            system_instructions=None,
            input="",
            model_settings=ModelSettings(),
            tools=[],
            output_schema=None,
            handoffs=[],
            tracing=ModelTracing.DISABLED,
            previous_response_id=None,
            conversation_id=None,
            prompt=cast(Any, {"id": "pmpt_123"}),
        )


@pytest.mark.allow_call_model_methods
@pytest.mark.asyncio
async def test_get_response_rejects_non_text_tool_output_in_strict_mode() -> None:
    class DummyCompletions:
        async def create(self, **kwargs: Any) -> Any:
            raise AssertionError("chat.completions.create should not run")

    class DummyClient:
        def __init__(self) -> None:
            self.chat = type("_Chat", (), {"completions": DummyCompletions()})()
            self.base_url = httpx.URL("http://fake")

    model = OpenAIChatCompletionsModel(
        model="gpt-4",
        openai_client=DummyClient(),  # type: ignore[arg-type]
        strict_feature_validation=True,
    )

    with pytest.raises(UserError, match="cannot be empty or contain only non-text content"):
        await model.get_response(
            system_instructions=None,
            input=[
                {
                    "type": "function_call_output",
                    "call_id": "call_image",
                    "output": [
                        {
                            "type": "input_image",
                            "image_url": "https://example.com/image.png",
                        }
                    ],
                }
            ],
            model_settings=ModelSettings(),
            tools=[],
            output_schema=None,
            handoffs=[],
            tracing=ModelTracing.DISABLED,
            previous_response_id=None,
            conversation_id=None,
            prompt=None,
        )


@pytest.mark.allow_call_model_methods
@pytest.mark.asyncio
async def test_get_response_warns_and_sends_placeholder_for_non_text_tool_output(
    caplog: pytest.LogCaptureFixture,
) -> None:
    class DummyCompletions:
        def __init__(self) -> None:
            self.kwargs: dict[str, Any] = {}

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

    class DummyClient:
        def __init__(self) -> None:
            self.completions = DummyCompletions()
            self.chat = type("_Chat", (), {"completions": self.completions})()
            self.base_url = httpx.URL("http://fake")

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

    with caplog.at_level(logging.WARNING, logger="openai.agents"):
        await model.get_response(
            system_instructions=None,
            input=[
                {
                    "type": "function_call_output",
                    "call_id": "call_image",
                    "output": [
                        {
                            "type": "input_image",
                            "image_url": "https://example.com/image.png",
                        }
                    ],
                }
            ],
            model_settings=ModelSettings(),
            tools=[],
            output_schema=None,
            handoffs=[],
            tracing=ModelTracing.DISABLED,
            previous_response_id=None,
            conversation_id=None,
            prompt=None,
        )

    assert client.completions.kwargs["messages"] == [
        {
            "role": "tool",
            "tool_call_id": "call_image",
            "content": "[tool output omitted]",
        }
    ]
    assert "Replacing the tool output with a placeholder" in caplog.text


@pytest.mark.allow_call_model_methods
@pytest.mark.asyncio
async def test_get_response_attaches_logprobs(monkeypatch) -> None:
    msg = ChatCompletionMessage(role="assistant", content="Hi!")
    choice = Choice(
        index=0,
        finish_reason="stop",
        message=msg,
        logprobs=ChoiceLogprobs(
            content=[
                ChatCompletionTokenLogprob(
                    token="Hi",
                    logprob=-0.5,
                    bytes=[1],
                    top_logprobs=[TopLogprob(token="Hi", logprob=-0.5, bytes=[1])],
                ),
                ChatCompletionTokenLogprob(
                    token="!",
                    logprob=-0.1,
                    bytes=[2],
                    top_logprobs=[TopLogprob(token="!", logprob=-0.1, bytes=[2])],
                ),
            ]
        ),
    )
    chat = ChatCompletion(
        id="resp-id",
        created=0,
        model="fake",
        object="chat.completion",
        choices=[choice],
        usage=None,
    )

    async def patched_fetch_response(self, *args, **kwargs):
        return chat

    monkeypatch.setattr(OpenAIChatCompletionsModel, "_fetch_response", patched_fetch_response)
    model = OpenAIProvider(use_responses=False).get_model("gpt-4")
    resp: ModelResponse = await model.get_response(
        system_instructions=None,
        input="",
        model_settings=ModelSettings(),
        tools=[],
        output_schema=None,
        handoffs=[],
        tracing=ModelTracing.DISABLED,
        previous_response_id=None,
        conversation_id=None,
        prompt=None,
    )
    assert len(resp.output) == 1
    assert isinstance(resp.output[0], ResponseOutputMessage)
    text_part = resp.output[0].content[0]
    assert isinstance(text_part, ResponseOutputText)
    assert text_part.logprobs is not None
    assert [lp.token for lp in text_part.logprobs] == ["Hi", "!"]


@pytest.mark.allow_call_model_methods
@pytest.mark.asyncio
async def test_get_response_with_refusal(monkeypatch) -> None:
    """
    When the model returns a ChatCompletionMessage with a `refusal` instead
    of normal `content`, `get_response` should produce a single
    `ResponseOutputMessage` containing a `ResponseOutputRefusal` part.
    """
    msg = ChatCompletionMessage(role="assistant", refusal="No thanks")
    choice = Choice(index=0, finish_reason="stop", message=msg)
    chat = ChatCompletion(
        id="resp-id",
        created=0,
        model="fake",
        object="chat.completion",
        choices=[choice],
        usage=None,
    )

    async def patched_fetch_response(self, *args, **kwargs):
        return chat

    monkeypatch.setattr(OpenAIChatCompletionsModel, "_fetch_response", patched_fetch_response)
    model = OpenAIProvider(use_responses=False).get_model("gpt-4")
    resp: ModelResponse = await model.get_response(
        system_instructions=None,
        input="",
        model_settings=ModelSettings(),
        tools=[],
        output_schema=None,
        handoffs=[],
        tracing=ModelTracing.DISABLED,
        previous_response_id=None,
        conversation_id=None,
        prompt=None,
    )
    assert len(resp.output) == 1
    assert isinstance(resp.output[0], ResponseOutputMessage)
    refusal_part = resp.output[0].content[0]
    assert isinstance(refusal_part, ResponseOutputRefusal)
    assert refusal_part.refusal == "No thanks"
    # With no usage from the completion, usage defaults to zeros.
    assert resp.usage.requests == 0
    assert resp.usage.input_tokens == 0
    assert resp.usage.output_tokens == 0
    assert resp.usage.input_tokens_details.cached_tokens == 0
    assert resp.usage.output_tokens_details.reasoning_tokens == 0


@pytest.mark.allow_call_model_methods
@pytest.mark.asyncio
async def test_get_response_with_tool_call(monkeypatch) -> None:
    """
    If the ChatCompletionMessage includes one or more tool_calls, `get_response`
    should append corresponding `ResponseFunctionToolCall` items after the
    assistant message item with matching name/arguments.
    """
    tool_call = ChatCompletionMessageFunctionToolCall(
        id="call-id",
        type="function",
        function=Function(name="do_thing", arguments="{'x':1}"),
    )
    msg = ChatCompletionMessage(role="assistant", content="Hi", tool_calls=[tool_call])
    choice = Choice(index=0, finish_reason="stop", message=msg)
    chat = ChatCompletion(
        id="resp-id",
        created=0,
        model="fake",
        object="chat.completion",
        choices=[choice],
        usage=None,
    )

    async def patched_fetch_response(self, *args, **kwargs):
        return chat

    monkeypatch.setattr(OpenAIChatCompletionsModel, "_fetch_response", patched_fetch_response)
    model = OpenAIProvider(use_responses=False).get_model("gpt-4")
    resp: ModelResponse = await model.get_response(
        system_instructions=None,
        input="",
        model_settings=ModelSettings(),
        tools=[],
        output_schema=None,
        handoffs=[],
        tracing=ModelTracing.DISABLED,
        previous_response_id=None,
        conversation_id=None,
        prompt=None,
    )
    # Expect a message item followed by a function tool call item.
    assert len(resp.output) == 2
    assert isinstance(resp.output[0], ResponseOutputMessage)
    fn_call_item = resp.output[1]
    assert isinstance(fn_call_item, ResponseFunctionToolCall)
    assert fn_call_item.call_id == "call-id"
    assert fn_call_item.name == "do_thing"
    assert fn_call_item.arguments == "{'x':1}"


@pytest.mark.allow_call_model_methods
@pytest.mark.asyncio
async def test_get_response_rejects_custom_tool_call_in_strict_mode(monkeypatch) -> None:
    tool_call = ChatCompletionMessageCustomToolCall(
        id="tool1",
        type="custom",
        custom=Custom(name="raw_tool", input="payload"),
    )
    msg = ChatCompletionMessage(role="assistant", tool_calls=[tool_call])
    chat = ChatCompletion(
        id="resp-id",
        created=0,
        model="fake",
        object="chat.completion",
        choices=[Choice(index=0, finish_reason="tool_calls", message=msg)],
        usage=None,
    )

    async def patched_fetch_response(self, *args, **kwargs):
        return chat

    monkeypatch.setattr(OpenAIChatCompletionsModel, "_fetch_response", patched_fetch_response)
    model = OpenAIProvider(use_responses=False, strict_feature_validation=True).get_model("gpt-4")

    with pytest.raises(UserError, match="Custom tool calls are not supported"):
        await model.get_response(
            system_instructions=None,
            input="",
            model_settings=ModelSettings(),
            tools=[],
            output_schema=None,
            handoffs=[],
            tracing=ModelTracing.DISABLED,
            previous_response_id=None,
            conversation_id=None,
            prompt=None,
        )


def test_get_client_disables_provider_managed_retries_on_runner_retry() -> None:
    class DummyChatCompletionsClient:
        def __init__(self) -> None:
            self.base_url = httpx.URL("https://api.openai.com/v1/")
            self.chat = type("ChatNamespace", (), {"completions": object()})()
            self.with_options_calls: list[dict[str, Any]] = []

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

    client = DummyChatCompletionsClient()
    model = OpenAIChatCompletionsModel(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_get_response_with_no_message(monkeypatch) -> None:
    """If the model returns no message, get_response should return an empty output."""
    msg = ChatCompletionMessage(role="assistant", content="ignored")
    choice = Choice(index=0, finish_reason="content_filter", message=msg)
    choice.message = None  # type: ignore[assignment]
    chat = ChatCompletion(
        id="resp-id",
        created=0,
        model="fake",
        object="chat.completion",
        choices=[choice],
        usage=None,
    )

    async def patched_fetch_response(self, *args, **kwargs):
        return chat

    monkeypatch.setattr(OpenAIChatCompletionsModel, "_fetch_response", patched_fetch_response)
    model = OpenAIProvider(use_responses=False).get_model("gpt-4")
    resp: ModelResponse = await model.get_response(
        system_instructions=None,
        input="",
        model_settings=ModelSettings(),
        tools=[],
        output_schema=None,
        handoffs=[],
        tracing=ModelTracing.DISABLED,
        previous_response_id=None,
        conversation_id=None,
        prompt=None,
    )
    assert resp.output == []


@pytest.mark.asyncio
async def test_fetch_response_non_stream(monkeypatch) -> None:
    """
    Verify that `_fetch_response` builds the correct OpenAI API call when not
    streaming and returns the ChatCompletion object directly. We supply a
    dummy ChatCompletion through a stubbed OpenAI client and inspect the
    captured kwargs.
    """

    # Dummy completions to record kwargs
    class DummyCompletions:
        def __init__(self) -> None:
            self.kwargs: dict[str, Any] = {}

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

    class DummyClient:
        def __init__(self, completions: DummyCompletions) -> None:
            self.chat = type("_Chat", (), {"completions": completions})()
            self.base_url = httpx.URL("http://fake")

    msg = ChatCompletionMessage(role="assistant", content="ignored")
    choice = Choice(index=0, finish_reason="stop", message=msg)
    chat = ChatCompletion(
        id="resp-id",
        created=0,
        model="fake",
        object="chat.completion",
        choices=[choice],
    )
    completions = DummyCompletions()
    dummy_client = DummyClient(completions)
    model = OpenAIChatCompletionsModel(model="gpt-4", openai_client=dummy_client)  # type: ignore
    # Execute the private fetch with a system instruction and simple string input.
    with generation_span(disabled=True) as span:
        result = await model._fetch_response(
            system_instructions="sys",
            input="hi",
            model_settings=ModelSettings(
                reasoning=Reasoning(effort="xhigh"),
                prompt_cache_retention="24h",
                prompt_cache_options={"mode": "explicit", "ttl": "30m"},
            ),
            tools=[],
            output_schema=None,
            handoffs=[],
            span=span,
            tracing=ModelTracing.DISABLED,
            stream=False,
        )
    assert result is chat
    # Ensure expected args were passed through to OpenAI client.
    kwargs = completions.kwargs
    assert kwargs["stream"] is omit
    assert kwargs["store"] is omit
    assert kwargs["model"] == "gpt-4"
    assert kwargs["messages"][0]["role"] == "system"
    assert kwargs["messages"][0]["content"] == "sys"
    assert kwargs["messages"][1]["role"] == "user"
    # Defaults for optional fields become the omit sentinel
    assert kwargs["tools"] is omit
    assert kwargs["tool_choice"] is omit
    assert kwargs["response_format"] is omit
    assert kwargs["stream_options"] is omit
    assert kwargs["reasoning_effort"] == "xhigh"
    assert kwargs["prompt_cache_retention"] == "24h"
    assert kwargs["prompt_cache_options"] == {"mode": "explicit", "ttl": "30m"}


def test_chat_completions_warns_once_for_responses_only_reasoning_settings(
    caplog: pytest.LogCaptureFixture,
) -> None:
    model = OpenAIChatCompletionsModel(
        model="gpt-5.6-sol",
        openai_client=cast(Any, object()),
    )
    model_settings = ModelSettings(
        reasoning=Reasoning(mode="pro", effort="max", context="all_turns")
    )
    caplog.set_level(logging.WARNING, logger="openai.agents")

    model._handle_unsupported_reasoning_settings(model_settings)
    model._handle_unsupported_reasoning_settings(model_settings)

    assert caplog.text.count("Ignoring unsupported reasoning settings") == 1
    assert "reasoning.mode" in caplog.text
    assert "reasoning.context" in caplog.text


def test_chat_completions_rejects_responses_only_reasoning_settings_in_strict_mode() -> None:
    model = OpenAIChatCompletionsModel(
        model="gpt-5.6-sol",
        openai_client=cast(Any, object()),
        strict_feature_validation=True,
    )

    with pytest.raises(UserError, match="reasoning.mode"):
        model._handle_unsupported_reasoning_settings(
            ModelSettings(reasoning=Reasoning(mode="pro", context="all_turns"))
        )


@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_chat_completions_model_with_custom_base_url()
    explicit_kwargs = await _run_chat_completions_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
@pytest.mark.asyncio
async def test_extra_args_prompt_cache_options_allowed_when_direct_field_is_omitted() -> None:
    prompt_cache_options = {"mode": "explicit", "ttl": "30m"}

    kwargs = await _run_chat_completions_model_with_custom_base_url(
        model_settings=ModelSettings(extra_args={"prompt_cache_options": prompt_cache_options})
    )

    assert kwargs["prompt_cache_options"] == prompt_cache_options


@pytest.mark.allow_call_model_methods
@pytest.mark.asyncio
async def test_duplicate_prompt_cache_options_rejected() -> None:
    with pytest.raises(TypeError, match="multiple values.*prompt_cache_options"):
        await _run_chat_completions_model_with_custom_base_url(
            model_settings=ModelSettings(
                prompt_cache_options={"mode": "explicit", "ttl": "30m"},
                extra_args={"prompt_cache_options": {"mode": "implicit"}},
            )
        )


@pytest.mark.allow_call_model_methods
@pytest.mark.asyncio
async def test_get_response_accepts_raw_chat_completions_image_content() -> None:
    """
    Raw Chat Completions content parts should be accepted on the SDK input path
    when using the Chat Completions backend.
    """

    class DummyCompletions:
        def __init__(self) -> None:
            self.kwargs: dict[str, Any] = {}

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

    class DummyClient:
        def __init__(self, completions: DummyCompletions) -> None:
            self.chat = type("_Chat", (), {"completions": completions})()
            self.base_url = httpx.URL("https://api.openai.com/v1/")

    msg = ChatCompletionMessage(role="assistant", content="ok")
    choice = Choice(index=0, finish_reason="stop", message=msg)
    chat = ChatCompletion(
        id="resp-id",
        created=0,
        model="fake",
        object="chat.completion",
        choices=[choice],
        usage=None,
    )
    completions = DummyCompletions()
    dummy_client = DummyClient(completions)
    model = OpenAIChatCompletionsModel(model="gpt-4", openai_client=dummy_client)  # type: ignore[arg-type]

    await model.get_response(
        system_instructions=None,
        input=[
            # Cast the fixture because the raw chat-style alias is intentionally outside the
            # canonical TypedDict shape that mypy expects for ordinary SDK inputs.
            cast(
                Any,
                {
                    "role": "user",
                    "content": [
                        {"type": "text", "text": "What is in this image?"},
                        {
                            "type": "image_url",
                            "image_url": {
                                "url": "data:image/png;base64,AAAA",
                                "detail": "high",
                            },
                        },
                    ],
                },
            )
        ],
        model_settings=ModelSettings(),
        tools=[],
        output_schema=None,
        handoffs=[],
        tracing=ModelTracing.DISABLED,
        previous_response_id=None,
        conversation_id=None,
        prompt=None,
    )

    assert completions.kwargs["messages"] == [
        {
            "role": "user",
            "content": [
                {"type": "text", "text": "What is in this image?"},
                {
                    "type": "image_url",
                    "image_url": {
                        "url": "data:image/png;base64,AAAA",
                        "detail": "high",
                    },
                },
            ],
        }
    ]


@pytest.mark.asyncio
async def test_fetch_response_stream(monkeypatch) -> None:
    """
    When `stream=True`, `_fetch_response` should return a bare `Response`
    object along with the underlying async stream. The OpenAI client call
    should include `stream_options` to request usage-delimited chunks.
    """

    async def event_stream() -> AsyncIterator[ChatCompletionChunk]:
        if False:  # pragma: no cover
            yield  # pragma: no cover

    class DummyCompletions:
        def __init__(self) -> None:
            self.kwargs: dict[str, Any] = {}

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

    class DummyClient:
        def __init__(self, completions: DummyCompletions) -> None:
            self.chat = type("_Chat", (), {"completions": completions})()
            self.base_url = httpx.URL("http://fake")

    completions = DummyCompletions()
    dummy_client = DummyClient(completions)
    model = OpenAIChatCompletionsModel(model="gpt-4", openai_client=dummy_client)  # type: ignore
    with generation_span(disabled=True) as span:
        response, stream = await model._fetch_response(
            system_instructions=None,
            input="hi",
            model_settings=ModelSettings(),
            tools=[],
            output_schema=None,
            handoffs=[],
            span=span,
            tracing=ModelTracing.DISABLED,
            stream=True,
        )
    # Check OpenAI client was called for streaming
    assert completions.kwargs["stream"] is True
    assert completions.kwargs["store"] is omit
    assert completions.kwargs["stream_options"] is omit
    # Response is a proper openai Response
    assert isinstance(response, Response)
    assert response.id == FAKE_RESPONSES_ID
    assert response.model == "gpt-4"
    assert response.object == "response"
    assert response.output == []
    # We returned the async iterator produced by our dummy.
    assert hasattr(stream, "__aiter__")


def test_store_param():
    """Should default to True for OpenAI API calls, and False otherwise."""

    model_settings = ModelSettings()
    client = AsyncOpenAI()
    assert ChatCmplHelpers.get_store_param(client, model_settings) is True, (
        "Should default to True for OpenAI API calls"
    )

    model_settings = ModelSettings(store=False)
    assert ChatCmplHelpers.get_store_param(client, model_settings) is False, (
        "Should respect explicitly set store=False"
    )

    model_settings = ModelSettings(store=True)
    assert ChatCmplHelpers.get_store_param(client, model_settings) is True, (
        "Should respect explicitly set store=True"
    )


def test_clean_gemini_tool_call_id_removes_thought_suffix() -> None:
    assert (
        ChatCmplHelpers.clean_gemini_tool_call_id(
            "call_123__thought__signature",
            model="gemini-2.5-pro",
        )
        == "call_123"
    )


def test_get_retry_advice_uses_openai_headers() -> None:
    request = httpx.Request("POST", "https://api.openai.com/v1/chat/completions")
    response = httpx.Response(
        429,
        request=request,
        headers={
            "x-should-retry": "true",
            "retry-after-ms": "500",
            "x-request-id": "req_123",
        },
        json={"error": {"code": "rate_limit"}},
    )
    error = APIStatusError(
        "rate limited", response=response, body={"error": {"code": "rate_limit"}}
    )
    model = OpenAIChatCompletionsModel(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.5
    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_123"


def test_get_retry_advice_keeps_stateful_transport_failures_ambiguous() -> None:
    model = OpenAIChatCompletionsModel(model="gpt-4", openai_client=cast(Any, object()))
    error = APIConnectionError(
        message="connection error",
        request=httpx.Request("POST", "https://api.openai.com/v1/chat/completions"),
    )

    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


def test_get_retry_advice_marks_stateful_http_failures_replay_safe() -> None:
    request = httpx.Request("POST", "https://api.openai.com/v1/chat/completions")
    response = httpx.Response(
        429,
        request=request,
        json={"error": {"code": "rate_limit"}},
    )
    error = APIStatusError(
        "rate limited", response=response, body={"error": {"code": "rate_limit"}}
    )
    model = OpenAIChatCompletionsModel(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


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 = OpenAIChatCompletionsModel(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}]


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

    class DummyCompletions:
        async def create(self, **kwargs):
            nonlocal called_kwargs
            called_kwargs = kwargs
            msg = ChatCompletionMessage(role="assistant", content="Hello")
            choice = Choice(index=0, finish_reason="stop", message=msg)
            return ChatCompletion(
                id="resp-id",
                created=0,
                model="fake",
                object="chat.completion",
                choices=[choice],
                usage=None,
            )

    class DummyChatClient:
        def __init__(self):
            self.chat = type("_Chat", (), {"completions": DummyCompletions()})()
            self.base_url = "https://api.openai.com"

    model = OpenAIChatCompletionsModel(model="gpt-4", openai_client=DummyChatClient())  # type: ignore

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

    try:
        await model.get_response(
            system_instructions=None,
            input="hi",
            model_settings=ModelSettings(),
            tools=[],
            output_schema=None,
            handoffs=[],
            tracing=ModelTracing.DISABLED,
            previous_response_id=None,
            conversation_id=None,
        )
    finally:
        if token is not None:
            HEADERS_OVERRIDE.reset(token)

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

    client = AsyncOpenAI(base_url="http://www.notopenai.com")
    model_settings = ModelSettings()
    assert ChatCmplHelpers.get_store_param(client, model_settings) is None, (
        "Should default to None for non-OpenAI API calls"
    )

    model_settings = ModelSettings(store=False)
    assert ChatCmplHelpers.get_store_param(client, model_settings) is False, (
        "Should respect explicitly set store=False"
    )

    model_settings = ModelSettings(store=True)
    assert ChatCmplHelpers.get_store_param(client, model_settings) is True, (
        "Should respect explicitly set store=True"
    )
