from collections.abc import Mapping, Sequence
from types import MappingProxyType
from typing import TYPE_CHECKING, Any, Final, Protocol, cast, get_type_hints

import httpx
from openai.types.responses import ResponseReasoningItem
from pydantic import BaseModel, ValidationError
from typing_extensions import ReadOnly, TypedDict

import litellm
from litellm._logging import verbose_logger
from litellm.litellm_core_utils.core_helpers import process_response_headers
from litellm.litellm_core_utils.llm_response_utils.convert_dict_to_response import (
    _safe_convert_created_field,
)
from litellm.litellm_core_utils.url_utils import encode_url_path_segment
from litellm.llms.base_llm.responses.transformation import BaseResponsesAPIConfig
from litellm.responses.litellm_completion_transformation.custom_tools import TOOL_CALL_ITEM_ID_PREFIX_BY_TYPE
from litellm.secret_managers.main import get_secret_str
from litellm.types.llms.openai import *
from litellm.types.responses.main import *
from litellm.types.router import GenericLiteLLMParams
from litellm.types.utils import LlmProviders

from ..common_utils import OpenAIError
from ..workload_identity import get_workload_identity_bearer_token, resolve_openai_workload_identity_config

OPENAI_RESPONSES_API_MIN_MAX_OUTPUT_TOKENS: Final = 16

if TYPE_CHECKING:
    from litellm.litellm_core_utils.litellm_logging import Logging as _LiteLLMLoggingObj

    LiteLLMLoggingObj = _LiteLLMLoggingObj
else:
    LiteLLMLoggingObj = Any

_NO_TOOL_UPDATE: Final[Mapping[str, object]] = MappingProxyType({})
_MODEL_FAMILIES_REJECTING_TOP_LEVEL_SCHEMA_COMBINATORS: Final = ("gpt-4", "gpt-3.5", "chatgpt-4o", "o1", "o3", "o4")
_PROVIDERS_WITH_COMBINATOR_REJECTING_VALIDATOR: Final = frozenset({LlmProviders.AZURE, LlmProviders.OPENAI})
_PROVIDERS_VALIDATING_TOOL_CALL_ITEM_IDS: Final = frozenset({LlmProviders.AZURE, LlmProviders.OPENAI})


class _DeleteResponseBody(TypedDict):
    """Decoded body of the Responses API delete call."""

    id: ReadOnly[str | None]
    object: ReadOnly[str | None]
    deleted: ReadOnly[bool | None]


class _DeleteResponse(Protocol):
    """The delete call's HTTP response, read for the decoded body it carries."""

    def json(self) -> _DeleteResponseBody: ...


class _JsonObjectResponse(Protocol):
    """A Responses API HTTP response, read for the JSON object it decodes to."""

    def json(self) -> dict[str, object]: ...


def _delete_response_body(response: _DeleteResponse) -> _DeleteResponseBody:
    """Decode a delete response body into the id, object and deleted fields it carries."""
    return response.json()


def _json_object_body(response: _JsonObjectResponse) -> dict[str, object]:
    """Decode a Responses API response body into its JSON object form."""
    return response.json()


class OpenAIResponsesAPIConfig(BaseResponsesAPIConfig):
    @property
    def custom_llm_provider(self) -> LlmProviders:
        return LlmProviders.OPENAI

    def supports_native_file_search(self) -> bool:
        return True

    @staticmethod
    def _is_gpt_5_model(model: str) -> bool:
        """Return True only for actual OpenAI GPT-5 models.

        Excludes pass-through models from other providers that happen to
        reference gpt-5 in their name (e.g. perplexity/openai/gpt-5.2).
        """
        parts: Final = model.split("/")
        if len(parts) > 1 and parts[0] not in ("openai",):
            return False
        return "gpt-5" in model and "gpt-5-chat" not in model

    @staticmethod
    def _supports_reasoning_effort_none(model: str) -> bool:
        """Return True if the model supports reasoning.effort='none'."""
        from litellm.utils import _supports_factory

        return _supports_factory(
            model=model,
            custom_llm_provider=None,
            key="supports_none_reasoning_effort",
        )

    @staticmethod
    def _effort_resolves_to_none(model: str, effort: str | None) -> bool:
        """Whether this request's reasoning effort ends up as "none", the one condition
        under which a non-default temperature is accepted.

        Delegates to the chat-completions gpt-5 config so both surfaces answer from one
        rule: the Responses API reaches the same models over a different wire, and a second
        copy of the rule here is what let this surface keep forwarding temperature after the
        chat surface stopped.
        """
        from litellm.llms.openai.chat.gpt_5_transformation import OpenAIGPT5Config

        return OpenAIGPT5Config.effort_resolves_to_none(model, effort)

    @staticmethod
    def _enforce_min_max_output_tokens(max_output_tokens: "int | None") -> "int | None":
        """Raise sub-minimum max_output_tokens up to the OpenAI Responses API minimum.

        OpenAI's Responses API rejects max_output_tokens below 16 for every model
        (not gpt-5 specific), so a client like Claude Code that sends a max_tokens=1
        warmup probe on model switch would otherwise 400. Values that are None or
        already at/above the minimum are returned unchanged.
        """
        if isinstance(max_output_tokens, int) and max_output_tokens < OPENAI_RESPONSES_API_MIN_MAX_OUTPUT_TOKENS:
            return OPENAI_RESPONSES_API_MIN_MAX_OUTPUT_TOKENS
        return max_output_tokens

    def get_supported_openai_params(self, model: str) -> list:
        """
        All OpenAI Responses API params are supported
        """
        supported_params: Final = get_type_hints(ResponsesAPIRequestParams).keys()
        return list(
            set(
                [
                    "input",
                    "model",
                    "extra_headers",
                    "extra_query",
                    "extra_body",
                    "timeout",
                ]
                + list(supported_params)
            )
        )

    def map_openai_params(
        self,
        response_api_optional_params: ResponsesAPIOptionalRequestParams,
        model: str,
        drop_params: bool,
    ) -> dict:
        """No mapping applied since inputs are in OpenAI spec already.

        GPT-5 models have restrictions on temperature (only temperature=1
        is accepted unless reasoning_effort='none' on models that support it).
        Apply the same validation used by the chat completions path.
        """
        params: Final = dict(response_api_optional_params)

        if "max_output_tokens" in params:
            params["max_output_tokens"] = self._enforce_min_max_output_tokens(params.get("max_output_tokens"))

        if self._is_gpt_5_model(model=model):
            temperature: Final = params.get("temperature")
            if temperature is not None and temperature != 1:
                reasoning: Final = params.get("reasoning") or {}
                effort: Final = reasoning.get("effort") if isinstance(reasoning, dict) else None
                supports_none: Final = self._supports_reasoning_effort_none(model=model)
                if supports_none and self._effort_resolves_to_none(model, effort):
                    pass  # flexible temperature allowed
                elif drop_params or litellm.drop_params:
                    params.pop("temperature", None)
                else:
                    raise litellm.UnsupportedParamsError(
                        message=(
                            f"{model} doesn't support temperature={temperature} while reasoning is "
                            "active. Only temperature=1 is supported unless reasoning.effort resolves "
                            "to 'none', either set explicitly on the request or declared as the "
                            "model's default_reasoning_effort. "
                            "To drop unsupported params set `litellm.drop_params = True`"
                        ),
                        status_code=400,
                    )

        return params

    def transform_responses_api_request(
        self,
        model: str,
        input: str | ResponseInputParam,
        response_api_optional_request_params: dict,
        litellm_params: GenericLiteLLMParams,
        headers: dict,
    ) -> dict:
        """Strip Anthropic-only `cache_control` markers before sending to OpenAI.

        OpenAI's Responses API rejects unknown fields on input content blocks
        with HTTP 400 ("Unknown parameter: 'input[0].content[0].cache_control'").
        Chat Completions strips these in
        `remove_cache_control_flag_from_messages_and_tools`; mirror that here.
        """

        input = self._validate_input_param(input)
        tools = response_api_optional_request_params.get("tools")
        input, tools = self.remove_cache_control_flag_from_input_and_tools(model=model, input=input, tools=tools)
        sanitized_tools: Final = self._flatten_tool_schema_combinators_for_openai(
            model=model, tools=tools, litellm_params=litellm_params
        )
        if sanitized_tools is not None:
            response_api_optional_request_params["tools"] = sanitized_tools
        replay_safe_input: Final = self._drop_foreign_tool_call_item_ids(input)
        final_request_params: Final = dict(
            ResponsesAPIRequestParams(model=model, input=replay_safe_input, **response_api_optional_request_params)
        )

        return final_request_params

    def remove_cache_control_flag_from_input_and_tools(
        self,
        model: str,  # allows overrides to selectively run this
        input: str | ResponseInputParam,
        tools: list[ALL_RESPONSES_API_TOOL_PARAMS] | None = None,
    ) -> tuple[
        str | ResponseInputParam,
        list[ALL_RESPONSES_API_TOOL_PARAMS] | None,
    ]:
        """Sibling of `remove_cache_control_flag_from_messages_and_tools` on
        the chat path. Strips Anthropic-only `cache_control` markers from
        Responses API input content blocks and tools.

        `filter_value_from_dict` mutates each dict in place, so the same
        objects are returned.
        """
        from litellm.litellm_core_utils.prompt_templates.common_utils import (
            filter_value_from_dict,
        )

        if isinstance(input, list):
            for item in input:
                if isinstance(item, dict):
                    filter_value_from_dict(cast(dict, item), "cache_control")

        if tools is not None:
            for tool in tools:
                if isinstance(tool, dict):
                    filter_value_from_dict(cast(dict, tool), "cache_control")

        return input, tools

    def _drop_foreign_tool_call_item_ids(self, input: str | ResponseInputParam) -> str | ResponseInputParam:
        if self.custom_llm_provider not in _PROVIDERS_VALIDATING_TOOL_CALL_ITEM_IDS or not isinstance(input, list):
            return input
        sanitized_items: Final = [self._without_foreign_tool_call_item_id(item) for item in input]
        return cast("ResponseInputParam", sanitized_items)  # cast-ok: items keep their shape, minus a rejected id

    @staticmethod
    def _without_foreign_tool_call_item_id(item: object) -> object:
        if not isinstance(item, dict):
            return item
        item_type: Final = item.get("type")
        item_id: Final = item.get("id")
        genuine_prefix: Final = TOOL_CALL_ITEM_ID_PREFIX_BY_TYPE.get(item_type) if isinstance(item_type, str) else None
        if genuine_prefix is None or not isinstance(item_id, str) or item_id.startswith(genuine_prefix):
            return item
        return {key: value for key, value in item.items() if key != "id"}  # mutable-ok: outgoing JSON request item

    def _flatten_tool_schema_combinators_for_openai(
        self,
        model: str,
        tools: list[ALL_RESPONSES_API_TOOL_PARAMS] | None,  # mutable-ok: request tools are a JSON list
        litellm_params: GenericLiteLLMParams,
    ) -> list[ALL_RESPONSES_API_TOOL_PARAMS] | None:  # mutable-ok: request tools are a JSON list
        """Flatten top-level schema combinators only where OpenAI's validator rejects them.

        OpenAI-compatible backends reusing this config (and the ChatGPT backend
        Codex talks to natively) accept them, and so do GPT-5 and later models,
        which also call tools better with the union intact. Codex wraps MCP tools
        inside namespace entries, so nested ``tools`` arrays are walked too.
        Azure OpenAI shares the validator but names deployments arbitrarily, so
        the router's declared ``model_info.base_model`` wins over the deployment
        name and an unrecognized name without one is left untouched.
        """
        if tools is None or self.custom_llm_provider not in _PROVIDERS_WITH_COMBINATOR_REJECTING_VALIDATOR:
            return tools
        gate_model: Final = self._combinator_gate_model(model=model, litellm_params=litellm_params)
        if not self._rejects_top_level_schema_combinators(gate_model):
            return tools
        flattened: Final = [  # mutable-ok: request tools are a JSON list
            self._flattened_tool_or_passthrough(tool) for tool in tools
        ]
        return cast("list[ALL_RESPONSES_API_TOOL_PARAMS]", flattened)  # cast-ok: dict spread keeps each tool's shape

    @staticmethod
    def _flattened_tool_or_passthrough(tool: object) -> object:
        return OpenAIResponsesAPIConfig._flattened_tool_entry(tool) if isinstance(tool, dict) else tool

    @staticmethod
    def _rejects_top_level_schema_combinators(model: str) -> bool:
        bare_model: Final = model.split("/")[-1]
        base_model: Final = bare_model.split(":")[1] if bare_model.startswith("ft:") else bare_model
        return base_model.startswith(_MODEL_FAMILIES_REJECTING_TOP_LEVEL_SCHEMA_COMBINATORS)

    @staticmethod
    def _combinator_gate_model(model: str, litellm_params: GenericLiteLLMParams) -> str:
        model_info: Final[object] = getattr(litellm_params, "model_info", None)
        base_model: Final[object] = model_info.get("base_model") if isinstance(model_info, dict) else None
        return base_model if isinstance(base_model, str) and base_model else model

    @staticmethod
    def _flattened_tool_entry(
        entry: Mapping[str, object],
    ) -> dict[str, object]:  # mutable-ok: request tools are JSON dicts
        from litellm.litellm_core_utils.prompt_templates.common_utils import (
            flatten_top_level_schema_combinators,
        )

        parameters: Final = entry.get("parameters")
        nested_tools: Final = entry.get("tools")
        parameters_update: Final = (
            MappingProxyType({"parameters": flatten_top_level_schema_combinators(parameters)})
            if isinstance(parameters, dict)
            else _NO_TOOL_UPDATE
        )
        tools_update: Final = (
            MappingProxyType({"tools": OpenAIResponsesAPIConfig._flattened_nested_tools(nested_tools)})
            if isinstance(nested_tools, list)
            else _NO_TOOL_UPDATE
        )
        return {**entry, **parameters_update, **tools_update}  # mutable-ok: request tools are JSON dicts

    @staticmethod
    def _flattened_nested_tools(
        nested_tools: Sequence[object],
    ) -> list[object]:  # mutable-ok: namespace tools are a JSON list
        return [  # mutable-ok: namespace tools are a JSON list
            OpenAIResponsesAPIConfig._flattened_tool_entry(item) if isinstance(item, dict) else item
            for item in nested_tools
        ]

    def _validate_input_param(self, input: str | ResponseInputParam) -> str | ResponseInputParam:
        """
        Ensure all input fields if pydantic are converted to dict

        OpenAI API Fails when we try to JSON dumps specific input pydantic fields.
        This function ensures all input fields are converted to dict.
        """
        if isinstance(input, list):
            validated_input: Final = []
            for item in input:
                # if it's pydantic, convert to dict
                if isinstance(item, BaseModel):
                    validated_input.append(item.model_dump(exclude_none=True))
                elif isinstance(item, dict):
                    # Handle reasoning items specifically to filter out status=None
                    if item.get("type") == "reasoning":
                        verbose_logger.debug("Handling reasoning item: %s", item)
                        # Type assertion since we know it's a dict at this point
                        dict_item = cast(dict[str, Any], item)
                        filtered_item = self._handle_reasoning_item(dict_item)
                    else:
                        # For other dict items, just pass through
                        filtered_item = cast(dict[str, Any], item)
                    validated_input.append(filtered_item)
                else:
                    validated_input.append(item)
            return validated_input
        # Input is expected to be either str or List, no single BaseModel expected
        return input

    def _handle_reasoning_item(self, item: dict[str, Any]) -> dict[str, Any]:
        """
        Handle reasoning items specifically to filter out status=None using OpenAI's model.
        Issue: https://github.com/BerriAI/litellm/issues/13484
        OpenAI API does not accept ReasoningItem(status=None), so we need to:
        1. Check if the item is a reasoning type
        2. Create a ResponseReasoningItem object with the item data
        3. Convert it back to dict with exclude_none=True to filter None values
        """
        if item.get("type") == "reasoning":
            try:
                # Ensure required fields are present for ResponseReasoningItem
                item_data: Final = dict(item)
                if "summary" not in item_data:
                    item_data["summary"] = (
                        item_data.get("reasoning_content", "")[:100] + "..."
                        if len(item_data.get("reasoning_content", "")) > 100
                        else item_data.get("reasoning_content", "")
                    )

                # Create ResponseReasoningItem object from the item data
                reasoning_item: Final = ResponseReasoningItem(**item_data)

                # Convert back to dict with exclude_none=True to exclude None fields
                dict_reasoning_item: Final = reasoning_item.model_dump(exclude_none=True)

                return dict_reasoning_item
            except Exception as e:
                verbose_logger.debug("Failed to create ResponseReasoningItem, falling back to manual filtering: %s", e)
                # Fallback: manually filter out known None fields
                filtered_item: Final = {
                    k: v
                    for k, v in item.items()
                    if v is not None or k not in {"status", "content", "encrypted_content"}
                }
                return filtered_item
        return item

    def transform_response_api_response(
        self,
        model: str,
        raw_response: httpx.Response,
        logging_obj: LiteLLMLoggingObj,
    ) -> ResponsesAPIResponse:
        """No transform applied since outputs are in OpenAI spec already"""
        try:
            logging_obj.post_call(
                original_response=raw_response.text,
                additional_args={"complete_input_dict": {}},
            )
            raw_response_json: Final = raw_response.json()
            raw_response_json["created_at"] = _safe_convert_created_field(raw_response_json["created_at"])
        except Exception:
            raise OpenAIError(message=raw_response.text, status_code=raw_response.status_code)
        raw_response_headers: Final = dict(raw_response.headers)
        processed_headers: Final = process_response_headers(raw_response_headers)
        try:
            response = ResponsesAPIResponse.model_validate(raw_response_json)
        except Exception:
            verbose_logger.debug(
                "Error constructing ResponsesAPIResponse: %s, using model_construct", raw_response_json
            )
            response = ResponsesAPIResponse.model_construct(**raw_response_json)

        # Store processed headers in additional_headers so they get returned to the client
        response._hidden_params["additional_headers"] = processed_headers
        response._hidden_params["headers"] = raw_response_headers
        return response

    def validate_environment(self, headers: dict, model: str, litellm_params: GenericLiteLLMParams | None) -> dict:
        litellm_params = litellm_params or GenericLiteLLMParams()
        api_key = litellm_params.api_key or litellm.api_key or litellm.openai_key or get_secret_str("OPENAI_API_KEY")
        headers.setdefault("Content-Type", "application/json")
        workload_identity_config: Final = (
            resolve_openai_workload_identity_config(api_key=api_key, api_base=litellm_params.api_base)
            if self.custom_llm_provider is LlmProviders.OPENAI
            else None
        )
        if workload_identity_config is not None:
            headers["Authorization"] = f"Bearer {get_workload_identity_bearer_token(workload_identity_config)}"
            return headers
        headers["Authorization"] = f"Bearer {api_key}"
        return headers

    def get_complete_url(
        self,
        api_base: str | None,
        litellm_params: dict,
    ) -> str:
        """
        Get the endpoint for OpenAI responses API
        """
        api_base = (
            api_base
            or litellm.api_base
            or get_secret_str("OPENAI_BASE_URL")
            or get_secret_str("OPENAI_API_BASE")
            or "https://api.openai.com/v1"
        )

        # Remove trailing slashes
        api_base = api_base.rstrip("/")

        return f"{api_base}/responses"

    def transform_streaming_response(
        self,
        model: str,
        parsed_chunk: dict,
        logging_obj: LiteLLMLoggingObj,
    ) -> ResponsesAPIStreamingResponse:
        """
        Transform a parsed streaming response chunk into a ResponsesAPIStreamingResponse
        """
        # Convert the dictionary to a properly typed ResponsesAPIStreamingResponse
        verbose_logger.debug("Raw OpenAI Chunk=%s", parsed_chunk)
        event_type: Final = str(parsed_chunk.get("type"))
        event_pydantic_model: Final = OpenAIResponsesAPIConfig.get_event_model_class(event_type=event_type)
        # Some OpenAI-compatible providers send error.code: null; coalesce so validation succeeds.
        try:
            error_obj: Final = parsed_chunk.get("error")
            if isinstance(error_obj, dict) and error_obj.get("code") is None:
                parsed_chunk = dict(parsed_chunk)
                parsed_chunk["error"] = dict(error_obj)
                parsed_chunk["error"]["code"] = "unknown_error"
        except Exception:
            verbose_logger.debug("Failed to coalesce error.code in parsed_chunk")

        try:
            return event_pydantic_model(**parsed_chunk)
        except ValidationError:
            verbose_logger.debug(
                "Pydantic validation failed for %s with chunk %s, falling back to model_construct",
                event_pydantic_model.__name__,
                parsed_chunk,
            )
            return event_pydantic_model.model_construct(**parsed_chunk)

    @staticmethod
    def parse_terminal_response_from_stream_chunks(all_chunks: list[str]) -> ResponsesAPIResponse | None:
        for chunk_str in reversed(all_chunks):
            for event_model in (ResponseCompletedEvent, ResponseIncompleteEvent, ResponseFailedEvent):
                try:
                    return event_model.model_validate_json(chunk_str.removeprefix("data: ")).response
                except ValueError:
                    continue
        return None

    @staticmethod
    def get_event_model_class(event_type: str) -> type[BaseLiteLLMOpenAIResponseObject]:
        """
        Returns the appropriate event model class based on the event type.

        Args:
            event_type (str): The type of event from the response chunk

        Returns:
            Any: The corresponding event model class

        Raises:
            ValueError: If the event type is unknown
        """
        event_models: Final = {
            ResponsesAPIStreamEvents.RESPONSE_CREATED: ResponseCreatedEvent,
            ResponsesAPIStreamEvents.RESPONSE_IN_PROGRESS: ResponseInProgressEvent,
            ResponsesAPIStreamEvents.RESPONSE_COMPLETED: ResponseCompletedEvent,
            ResponsesAPIStreamEvents.RESPONSE_FAILED: ResponseFailedEvent,
            ResponsesAPIStreamEvents.RESPONSE_INCOMPLETE: ResponseIncompleteEvent,
            ResponsesAPIStreamEvents.OUTPUT_ITEM_ADDED: OutputItemAddedEvent,
            ResponsesAPIStreamEvents.OUTPUT_ITEM_DONE: OutputItemDoneEvent,
            ResponsesAPIStreamEvents.CONTENT_PART_ADDED: ContentPartAddedEvent,
            ResponsesAPIStreamEvents.CONTENT_PART_DONE: ContentPartDoneEvent,
            ResponsesAPIStreamEvents.OUTPUT_TEXT_DELTA: OutputTextDeltaEvent,
            ResponsesAPIStreamEvents.OUTPUT_TEXT_ANNOTATION_ADDED: OutputTextAnnotationAddedEvent,
            ResponsesAPIStreamEvents.OUTPUT_TEXT_DONE: OutputTextDoneEvent,
            ResponsesAPIStreamEvents.REFUSAL_DELTA: RefusalDeltaEvent,
            ResponsesAPIStreamEvents.REFUSAL_DONE: RefusalDoneEvent,
            ResponsesAPIStreamEvents.FUNCTION_CALL_ARGUMENTS_DELTA: FunctionCallArgumentsDeltaEvent,
            ResponsesAPIStreamEvents.FUNCTION_CALL_ARGUMENTS_DONE: FunctionCallArgumentsDoneEvent,
            ResponsesAPIStreamEvents.FILE_SEARCH_CALL_IN_PROGRESS: FileSearchCallInProgressEvent,
            ResponsesAPIStreamEvents.FILE_SEARCH_CALL_SEARCHING: FileSearchCallSearchingEvent,
            ResponsesAPIStreamEvents.FILE_SEARCH_CALL_COMPLETED: FileSearchCallCompletedEvent,
            ResponsesAPIStreamEvents.WEB_SEARCH_CALL_IN_PROGRESS: WebSearchCallInProgressEvent,
            ResponsesAPIStreamEvents.WEB_SEARCH_CALL_SEARCHING: WebSearchCallSearchingEvent,
            ResponsesAPIStreamEvents.WEB_SEARCH_CALL_COMPLETED: WebSearchCallCompletedEvent,
            ResponsesAPIStreamEvents.MCP_LIST_TOOLS_IN_PROGRESS: MCPListToolsInProgressEvent,
            ResponsesAPIStreamEvents.MCP_LIST_TOOLS_COMPLETED: MCPListToolsCompletedEvent,
            ResponsesAPIStreamEvents.MCP_LIST_TOOLS_FAILED: MCPListToolsFailedEvent,
            ResponsesAPIStreamEvents.MCP_CALL_IN_PROGRESS: MCPCallInProgressEvent,
            ResponsesAPIStreamEvents.MCP_CALL_ARGUMENTS_DELTA: MCPCallArgumentsDeltaEvent,
            ResponsesAPIStreamEvents.MCP_CALL_ARGUMENTS_DONE: MCPCallArgumentsDoneEvent,
            ResponsesAPIStreamEvents.MCP_CALL_COMPLETED: MCPCallCompletedEvent,
            ResponsesAPIStreamEvents.MCP_CALL_FAILED: MCPCallFailedEvent,
            ResponsesAPIStreamEvents.IMAGE_GENERATION_PARTIAL_IMAGE: ImageGenerationPartialImageEvent,
            ResponsesAPIStreamEvents.ERROR: ErrorEvent,
            # Shell tool events: passthrough as GenericEvent so payload is preserved
            ResponsesAPIStreamEvents.SHELL_CALL_IN_PROGRESS: GenericEvent,
            ResponsesAPIStreamEvents.SHELL_CALL_COMPLETED: GenericEvent,
            ResponsesAPIStreamEvents.SHELL_CALL_OUTPUT: GenericEvent,
        }

        model_class: Final = event_models.get(cast(ResponsesAPIStreamEvents, event_type))
        if not model_class:
            return GenericEvent

        return model_class

    def should_fake_stream(
        self,
        model: str | None,
        stream: bool | None,
        custom_llm_provider: str | None = None,
    ) -> bool:
        if stream is not True:
            return False
        if model is not None:
            try:
                if (
                    litellm.utils.supports_native_streaming(
                        model=model,
                        custom_llm_provider=custom_llm_provider,
                    )
                    is False
                ):
                    return True
            except Exception as e:
                verbose_logger.debug("Error getting model info in OpenAIResponsesAPIConfig: %s", e)
        return False

    def supports_native_websocket(self) -> bool:
        """OpenAI supports native WebSocket for Responses API"""
        return True

    #########################################################
    ########## DELETE RESPONSE API TRANSFORMATION ##############
    #########################################################
    def transform_delete_response_api_request(
        self,
        response_id: str,
        api_base: str,
        litellm_params: GenericLiteLLMParams,
        headers: dict,
    ) -> tuple[str, dict]:
        """
        Transform the delete response API request into a URL and data

        OpenAI API expects the following request
        - DELETE /v1/responses/{response_id}
        """
        encoded_response_id: Final = encode_url_path_segment(response_id, field_name="response_id")
        url: Final = f"{api_base}/{encoded_response_id}"
        data: Final[dict] = {}
        return url, data

    def transform_delete_response_api_response(
        self,
        raw_response: httpx.Response,
        logging_obj: LiteLLMLoggingObj,
    ) -> DeleteResponseResult:
        """
        Transform the delete response API response into a DeleteResponseResult
        """
        try:
            raw_response_json: Final = _delete_response_body(raw_response)
        except Exception:
            raise OpenAIError(message=raw_response.text, status_code=raw_response.status_code)
        return DeleteResponseResult(**raw_response_json)

    #########################################################
    ########## GET RESPONSE API TRANSFORMATION ###############
    #########################################################
    def transform_get_response_api_request(
        self,
        response_id: str,
        api_base: str,
        litellm_params: GenericLiteLLMParams,
        headers: dict,
    ) -> tuple[str, dict]:
        """
        Transform the get response API request into a URL and data

        OpenAI API expects the following request
        - GET /v1/responses/{response_id}
        """
        encoded_response_id: Final = encode_url_path_segment(response_id, field_name="response_id")
        url: Final = f"{api_base}/{encoded_response_id}"
        data: Final[dict] = {}
        return url, data

    def transform_get_response_api_response(
        self,
        raw_response: httpx.Response,
        logging_obj: LiteLLMLoggingObj,
    ) -> ResponsesAPIResponse:
        """
        Transform the get response API response into a ResponsesAPIResponse
        """
        try:
            raw_response_json: Final = _json_object_body(raw_response)
        except Exception:
            raise OpenAIError(message=raw_response.text, status_code=raw_response.status_code)
        raw_response_headers: Final = dict(raw_response.headers)
        processed_headers: Final = process_response_headers(raw_response_headers)
        response: Final = ResponsesAPIResponse.model_validate(raw_response_json)
        response._hidden_params["additional_headers"] = processed_headers
        response._hidden_params["headers"] = raw_response_headers

        return response

    #########################################################
    ########## LIST INPUT ITEMS TRANSFORMATION #############
    #########################################################
    def transform_list_input_items_request(
        self,
        response_id: str,
        api_base: str,
        litellm_params: GenericLiteLLMParams,
        headers: dict,
        after: str | None = None,
        before: str | None = None,
        include: list[str] | None = None,
        limit: int = 20,
        order: Literal["asc", "desc"] = "desc",
    ) -> tuple[str, dict]:
        encoded_response_id: Final = encode_url_path_segment(response_id, field_name="response_id")
        url: Final = f"{api_base}/{encoded_response_id}/input_items"
        params: Final[dict[str, object]] = {}
        if after is not None:
            params["after"] = after
        if before is not None:
            params["before"] = before
        if include:
            params["include"] = ",".join(include)
        if limit is not None:
            params["limit"] = limit
        if order is not None:
            params["order"] = order
        return url, params

    def transform_list_input_items_response(
        self,
        raw_response: httpx.Response,
        logging_obj: LiteLLMLoggingObj,
    ) -> dict:
        try:
            return _json_object_body(raw_response)
        except Exception:
            raise OpenAIError(message=raw_response.text, status_code=raw_response.status_code)

    #########################################################
    ########## CANCEL RESPONSE API TRANSFORMATION ##########
    #########################################################
    def transform_cancel_response_api_request(
        self,
        response_id: str,
        api_base: str,
        litellm_params: GenericLiteLLMParams,
        headers: dict,
    ) -> tuple[str, dict]:
        """
        Transform the cancel response API request into a URL and data

        OpenAI API expects the following request
        - POST /v1/responses/{response_id}/cancel
        """
        encoded_response_id: Final = encode_url_path_segment(response_id, field_name="response_id")
        url: Final = f"{api_base}/{encoded_response_id}/cancel"
        data: Final[dict] = {}
        return url, data

    def transform_cancel_response_api_response(
        self,
        raw_response: httpx.Response,
        logging_obj: LiteLLMLoggingObj,
    ) -> ResponsesAPIResponse:
        """
        Transform the cancel response API response into a ResponsesAPIResponse
        """
        try:
            raw_response_json: Final = _json_object_body(raw_response)
        except Exception:
            raise OpenAIError(message=raw_response.text, status_code=raw_response.status_code)
        raw_response_headers: Final = dict(raw_response.headers)
        processed_headers: Final = process_response_headers(raw_response_headers)

        response: Final = ResponsesAPIResponse.model_validate(raw_response_json)
        response._hidden_params["additional_headers"] = processed_headers
        response._hidden_params["headers"] = raw_response_headers

        return response

    #########################################################
    ########## COMPACT RESPONSE API TRANSFORMATION ##########
    #########################################################
    def transform_compact_response_api_request(
        self,
        model: str,
        input: str | ResponseInputParam,
        response_api_optional_request_params: dict,
        api_base: str,
        litellm_params: GenericLiteLLMParams,
        headers: dict,
    ) -> tuple[str, dict]:
        """
        Transform the compact response API request into a URL and data

        OpenAI API expects the following request
        - POST /v1/responses/compact
        """
        # Preserve query params (e.g., api-version) while appending /compact.
        parsed_url: Final = httpx.URL(api_base)
        compact_path: Final = parsed_url.path.rstrip("/") + "/compact"
        url: Final = str(parsed_url.copy_with(path=compact_path))

        input = self._validate_input_param(input)
        tools = response_api_optional_request_params.get("tools")
        input, tools = self.remove_cache_control_flag_from_input_and_tools(model=model, input=input, tools=tools)
        sanitized_tools: Final = self._flatten_tool_schema_combinators_for_openai(
            model=model, tools=tools, litellm_params=litellm_params
        )
        if sanitized_tools is not None:
            response_api_optional_request_params["tools"] = sanitized_tools
        replay_safe_input: Final = self._drop_foreign_tool_call_item_ids(input)
        data: Final = dict(
            ResponsesAPIRequestParams(model=model, input=replay_safe_input, **response_api_optional_request_params)
        )

        return url, data

    def transform_compact_response_api_response(
        self,
        raw_response: httpx.Response,
        logging_obj: LiteLLMLoggingObj,
    ) -> ResponsesAPIResponse:
        """
        Transform the compact response API response into a ResponsesAPIResponse
        """
        try:
            logging_obj.post_call(
                original_response=raw_response.text,
                additional_args={"complete_input_dict": {}},
            )
            raw_response_json: Final = raw_response.json()
            raw_response_json["created_at"] = _safe_convert_created_field(raw_response_json["created_at"])
        except Exception:
            raise OpenAIError(message=raw_response.text, status_code=raw_response.status_code)
        raw_response_headers: Final = dict(raw_response.headers)
        processed_headers: Final = process_response_headers(raw_response_headers)

        try:
            response = ResponsesAPIResponse.model_validate(raw_response_json)
        except Exception:
            verbose_logger.debug(
                "Error constructing ResponsesAPIResponse: %s, using model_construct", raw_response_json
            )
            response = ResponsesAPIResponse.model_construct(**raw_response_json)

        response._hidden_params["additional_headers"] = processed_headers
        response._hidden_params["headers"] = raw_response_headers

        return response
