from __future__ import annotations

import asyncio
import base64
import inspect
import json
import math
import os
import time
from collections.abc import Callable, Mapping
from dataclasses import dataclass
from typing import Annotated, Any, Literal, TypeAlias, cast

import pydantic
import websockets
from openai.types.realtime import realtime_audio_config as _rt_audio_config
from openai.types.realtime.conversation_item import (
    ConversationItem,
    ConversationItem as OpenAIConversationItem,
)
from openai.types.realtime.conversation_item_create_event import (
    ConversationItemCreateEvent as OpenAIConversationItemCreateEvent,
)
from openai.types.realtime.conversation_item_retrieve_event import (
    ConversationItemRetrieveEvent as OpenAIConversationItemRetrieveEvent,
)
from openai.types.realtime.conversation_item_truncate_event import (
    ConversationItemTruncateEvent as OpenAIConversationItemTruncateEvent,
)
from openai.types.realtime.input_audio_buffer_append_event import (
    InputAudioBufferAppendEvent as OpenAIInputAudioBufferAppendEvent,
)
from openai.types.realtime.input_audio_buffer_commit_event import (
    InputAudioBufferCommitEvent as OpenAIInputAudioBufferCommitEvent,
)
from openai.types.realtime.realtime_audio_formats import (
    AudioPCM,
    AudioPCMA,
    AudioPCMU,
)
from openai.types.realtime.realtime_client_event import (
    RealtimeClientEvent as OpenAIRealtimeClientEvent,
)
from openai.types.realtime.realtime_conversation_item_assistant_message import (
    RealtimeConversationItemAssistantMessage,
)
from openai.types.realtime.realtime_conversation_item_function_call_output import (
    RealtimeConversationItemFunctionCallOutput,
)
from openai.types.realtime.realtime_conversation_item_system_message import (
    RealtimeConversationItemSystemMessage,
)
from openai.types.realtime.realtime_conversation_item_user_message import (
    Content,
    RealtimeConversationItemUserMessage,
)
from openai.types.realtime.realtime_function_tool import (
    RealtimeFunctionTool as OpenAISessionFunction,
)
from openai.types.realtime.realtime_response_usage import RealtimeResponseUsage
from openai.types.realtime.realtime_server_event import (
    RealtimeServerEvent as OpenAIRealtimeServerEvent,
)
from openai.types.realtime.realtime_session_create_request import (
    RealtimeSessionCreateRequest as OpenAISessionCreateRequest,
)
from openai.types.realtime.realtime_tracing_config import (
    TracingConfiguration as OpenAITracingConfiguration,
)
from openai.types.realtime.realtime_transcription_session_create_request import (
    RealtimeTranscriptionSessionCreateRequest as OpenAIRealtimeTranscriptionSessionCreateRequest,
)
from openai.types.realtime.response_audio_delta_event import ResponseAudioDeltaEvent
from openai.types.realtime.response_cancel_event import (
    ResponseCancelEvent as OpenAIResponseCancelEvent,
)
from openai.types.realtime.response_create_event import (
    ResponseCreateEvent as OpenAIResponseCreateEvent,
)
from openai.types.realtime.session_update_event import (
    SessionUpdateEvent as OpenAISessionUpdateEvent,
)
from openai.types.responses.response_prompt import ResponsePrompt
from pydantic import Field, TypeAdapter
from typing_extensions import NotRequired, TypedDict, assert_never
from websockets.asyncio.client import ClientConnection

from agents.handoffs import Handoff
from agents.prompts import Prompt
from agents.realtime._default_tracker import ModelAudioTracker
from agents.realtime.audio_formats import to_realtime_audio_format
from agents.tool import (
    FunctionTool,
    Tool,
    ensure_function_tool_supports_responses_only_features,
    ensure_tool_choice_supports_backend,
)
from agents.util._types import MaybeAwaitable

from .. import _debug
from ..exceptions import UserError
from ..logger import logger
from ..run_context import RunContextWrapper, TContext
from ..usage import Usage
from ..version import __version__
from ._tool_filtering import filter_enabled_tools, filter_statically_enabled_tools
from ._tool_validation import validate_realtime_tool_names
from .agent import RealtimeAgent
from .config import (
    RealtimeModelTracingConfig,
    RealtimeRunConfig,
    RealtimeSessionModelSettings,
)
from .handoffs import collect_enabled_handoffs, filter_enabled_handoffs
from .items import RealtimeMessageItem, RealtimeToolCallItem
from .model import (
    RealtimeModel,
    RealtimeModelConfig,
    RealtimeModelListener,
    RealtimePlaybackState,
    RealtimePlaybackTracker,
)
from .model_events import (
    RealtimeModelAudioDoneEvent,
    RealtimeModelAudioEvent,
    RealtimeModelAudioInterruptedEvent,
    RealtimeModelCachedTokensDetails,
    RealtimeModelErrorEvent,
    RealtimeModelEvent,
    RealtimeModelExceptionEvent,
    RealtimeModelInputAudioTimeoutTriggeredEvent,
    RealtimeModelInputAudioTranscriptionCompletedEvent,
    RealtimeModelInputTokensDetails,
    RealtimeModelItemDeletedEvent,
    RealtimeModelItemUpdatedEvent,
    RealtimeModelOutputTokensDetails,
    RealtimeModelRawServerEvent,
    RealtimeModelToolCallEvent,
    RealtimeModelTranscriptDeltaEvent,
    RealtimeModelTurnEndedEvent,
    RealtimeModelTurnStartedEvent,
    RealtimeModelUsageEvent,
)
from .model_inputs import (
    RealtimeModelSendAudio,
    RealtimeModelSendEvent,
    RealtimeModelSendInterrupt,
    RealtimeModelSendRawMessage,
    RealtimeModelSendSessionUpdate,
    RealtimeModelSendToolOutput,
    RealtimeModelSendUserInput,
)

FormatInput: TypeAlias = str | AudioPCM | AudioPCMU | AudioPCMA | Mapping[str, Any] | None


# Avoid direct imports of non-exported names by referencing via module
OpenAIRealtimeAudioConfig = _rt_audio_config.RealtimeAudioConfig
OpenAIRealtimeAudioInput = _rt_audio_config.RealtimeAudioConfigInput  # type: ignore[attr-defined]
OpenAIRealtimeAudioOutput = _rt_audio_config.RealtimeAudioConfigOutput  # type: ignore[attr-defined]


_USER_AGENT = f"Agents/Python {__version__}"
DEFAULT_REALTIME_MODEL = "gpt-realtime-2.1"

DEFAULT_MODEL_SETTINGS: RealtimeSessionModelSettings = {
    "voice": "ash",
    "modalities": ["audio"],
    "input_audio_format": "pcm16",
    "output_audio_format": "pcm16",
    "input_audio_transcription": {
        "model": "gpt-4o-mini-transcribe",
    },
    "turn_detection": {"type": "semantic_vad", "interrupt_response": True},
}


async def get_api_key(key: str | Callable[[], MaybeAwaitable[str]] | None) -> str | None:
    if isinstance(key, str):
        return key
    elif callable(key):
        result = key()
        if inspect.isawaitable(result):
            return await result
        return result

    return os.getenv("OPENAI_API_KEY")


AllRealtimeServerEvents = Annotated[
    OpenAIRealtimeServerEvent,
    Field(discriminator="type"),
]

ServerEventTypeAdapter: TypeAdapter[AllRealtimeServerEvents] | None = None


def _server_event_validation_summary(error: BaseException) -> str:
    if isinstance(error, pydantic.ValidationError):
        return f"{error.error_count()} validation error(s)"

    return error.__class__.__name__


def _server_event_identity(event: Any) -> tuple[Any, Any]:
    if not isinstance(event, dict):
        return "unknown", None

    return event.get("type", "unknown"), event.get("event_id")


def _log_server_event_validation_failure(event: Any, error: BaseException) -> str:
    event_type, event_id = _server_event_identity(event)

    if _debug.DONT_LOG_MODEL_DATA:
        logger.error(
            "Failed to validate server event type=%s event_id=%s: %s",
            event_type,
            event_id,
            _server_event_validation_summary(error),
        )
    else:
        logger.error("Failed to validate server event: %s", event, exc_info=True)

    return str(event_type)


@dataclass(frozen=True)
class _PendingResponseCreate:
    event_id: str
    request_version: int
    target_version: int
    is_manual: bool


class _ResponseCreateSequencer:
    """Tracks local response sequencing around response.create and response.cancel."""

    def __init__(self) -> None:
        self._ongoing_response = False
        self._response_control: Literal["free", "create_requested", "cancel_requested"] = "free"
        self._response_create_request_version = 0
        self._response_create_event_counter = 0
        self._pending_request_versions: set[int] = set()
        self._manual_response_create_versions: set[int] = set()
        self._pending_response_create: _PendingResponseCreate | None = None
        self._condition = asyncio.Condition()

    @property
    def ongoing_response(self) -> bool:
        return self._ongoing_response

    @property
    def response_control(self) -> Literal["free", "create_requested", "cancel_requested"]:
        return self._response_control

    @property
    def pending_response_create_event_id(self) -> str | None:
        return self._pending_response_create.event_id if self._pending_response_create else None

    def _next_pending_request_version(self) -> int | None:
        return min(self._pending_request_versions) if self._pending_request_versions else None

    def _auto_response_create_target_version(self, request_version: int) -> int:
        next_manual_version = min(
            (
                version
                for version in self._manual_response_create_versions
                if version >= request_version
            ),
            default=None,
        )
        if next_manual_version is None:
            eligible_versions = self._pending_request_versions
        else:
            eligible_versions = {
                version
                for version in self._pending_request_versions
                if version < next_manual_version
            }
        return max(eligible_versions)

    def set_ongoing_response_for_test(self, value: bool) -> None:
        self._ongoing_response = value

    async def set_response_control(
        self, control: Literal["free", "create_requested", "cancel_requested"]
    ) -> None:
        async with self._condition:
            self._response_control = control
            self._condition.notify_all()

    async def mark_response_created(self) -> None:
        async with self._condition:
            self._ongoing_response = True
            self._pending_response_create = None
            self._response_control = "free"
            self._condition.notify_all()

    async def mark_response_done(self) -> None:
        async with self._condition:
            self._ongoing_response = False
            self._pending_response_create = None
            self._response_control = "free"
            self._condition.notify_all()

    async def release_waiters(self) -> None:
        async with self._condition:
            self._ongoing_response = False
            self._pending_response_create = None
            self._pending_request_versions.clear()
            self._manual_response_create_versions.clear()
            self._response_create_request_version = 0
            self._response_create_event_counter = 0
            self._response_control = "free"
            self._condition.notify_all()

    async def reserve_response_create_request(self, *, manual: bool = False) -> int:
        async with self._condition:
            self._response_create_request_version += 1
            request_version = self._response_create_request_version
            self._pending_request_versions.add(request_version)
            if manual:
                self._manual_response_create_versions.add(request_version)
            self._condition.notify_all()
            return request_version

    async def clear_pending_response_create(self, event_id: str | None = None) -> bool:
        async with self._condition:
            if (
                self._response_control != "create_requested"
                or self._pending_response_create is None
            ):
                return False
            if event_id is not None and self._pending_response_create.event_id != event_id:
                return False
            # The caller only uses the no-event-id path for response.create-like
            # server errors, so clearing here won't release unrelated requests.
            self._pending_request_versions.discard(self._pending_response_create.request_version)
            if self._pending_response_create.is_manual:
                self._manual_response_create_versions.discard(
                    self._pending_response_create.request_version
                )
            self._pending_response_create = None
            self._response_control = "free"
            self._condition.notify_all()
            return True

    async def wait_for_response_create_slot(
        self, request_version: int, *, manual: bool = False, event_id: str | None = None
    ) -> _PendingResponseCreate | None:
        while True:
            async with self._condition:
                await self._condition.wait_for(
                    lambda: request_version not in self._pending_request_versions
                    or (
                        not self._ongoing_response
                        and self._response_control == "free"
                        and self._next_pending_request_version() == request_version
                    )
                )
                if request_version not in self._pending_request_versions:
                    return None

                self._response_control = "create_requested"
                resolved_event_id = event_id
                if resolved_event_id is None:
                    self._response_create_event_counter += 1
                    resolved_event_id = (
                        f"agents_py_response_create_{self._response_create_event_counter}"
                    )
                target_version = (
                    request_version
                    if manual
                    else self._auto_response_create_target_version(request_version)
                )
                pending = _PendingResponseCreate(
                    event_id=resolved_event_id,
                    request_version=request_version,
                    target_version=target_version,
                    is_manual=manual,
                )
                self._pending_response_create = pending
                return pending

    async def mark_response_create_sent(self, pending: _PendingResponseCreate) -> None:
        async with self._condition:
            covered_versions = {
                version
                for version in self._pending_request_versions
                if version <= pending.target_version
            }
            self._pending_request_versions.difference_update(covered_versions)
            self._manual_response_create_versions.difference_update(covered_versions)
            self._condition.notify_all()

    async def begin_cancel_response(self) -> bool:
        async with self._condition:
            if not self._ongoing_response or self._response_control == "cancel_requested":
                return False
            self._response_control = "cancel_requested"
            return True


def get_server_event_type_adapter() -> TypeAdapter[AllRealtimeServerEvents]:
    global ServerEventTypeAdapter
    if not ServerEventTypeAdapter:
        ServerEventTypeAdapter = TypeAdapter(AllRealtimeServerEvents)
    return ServerEventTypeAdapter


_SERVER_EVENT_TYPES_WITH_CUSTOM_VOICE = frozenset(
    {
        "session.created",
        "session.updated",
        "response.created",
        "response.done",
    }
)


def _should_normalize_custom_voice_for_server_event(event: Any) -> bool:
    return isinstance(event, dict) and event.get("type") in _SERVER_EVENT_TYPES_WITH_CUSTOM_VOICE


def _normalize_custom_voice_for_server_event_validation(value: Any) -> Any:
    # TODO: Remove this once generated Realtime server event models accept custom voice objects.
    if isinstance(value, list):
        return [_normalize_custom_voice_for_server_event_validation(item) for item in value]

    if not isinstance(value, dict):
        return value

    normalized: dict[str, Any] = {}
    for key, item in value.items():
        if key == "voice" and isinstance(item, Mapping):
            voice_id = item.get("id")
            if isinstance(voice_id, str):
                normalized[key] = voice_id
                continue
        normalized[key] = _normalize_custom_voice_for_server_event_validation(item)
    return normalized


async def _collect_enabled_handoffs(
    agent: RealtimeAgent[Any], context_wrapper: RunContextWrapper[Any]
) -> list[Handoff[Any, RealtimeAgent[Any]]]:
    return await collect_enabled_handoffs(agent, context_wrapper)


async def _build_model_settings_from_agent(
    *,
    agent: RealtimeAgent[Any],
    context_wrapper: RunContextWrapper[Any],
    base_settings: RealtimeSessionModelSettings,
    starting_settings: RealtimeSessionModelSettings | None,
    run_config: RealtimeRunConfig | None,
) -> RealtimeSessionModelSettings:
    updated_settings = base_settings.copy()

    if agent.prompt is not None:
        updated_settings["prompt"] = agent.prompt

    instructions, tools, handoffs = await asyncio.gather(
        agent.get_system_prompt(context_wrapper),
        agent.get_all_tools(context_wrapper),
        _collect_enabled_handoffs(agent, context_wrapper),
    )
    updated_settings["instructions"] = instructions or ""
    updated_settings["tools"] = tools or []
    updated_settings["handoffs"] = handoffs or []

    if starting_settings:
        updated_settings.update(starting_settings)
        if "tools" in starting_settings:
            updated_settings["tools"] = await filter_enabled_tools(
                updated_settings.get("tools") or [],
                context_wrapper,
                agent,
            )
        if "handoffs" in starting_settings:
            updated_settings["handoffs"] = await filter_enabled_handoffs(
                updated_settings.get("handoffs") or [],
                context_wrapper,
                agent,
            )

    if run_config and run_config.get("tracing_disabled", False):
        updated_settings["tracing"] = None

    return updated_settings


class TransportConfig(TypedDict):
    """Low-level network transport configuration."""

    ping_interval: NotRequired[float | None]
    """Time in seconds between keepalive pings sent by the client.
    Default is usually 20.0. Set to None to disable."""

    ping_timeout: NotRequired[float | None]
    """Time in seconds to wait for a pong response before disconnecting.
    Set to None to disable ping timeout and keep an open connection (ignore network lag)."""

    handshake_timeout: NotRequired[float]
    """Time in seconds to wait for the connection handshake to complete."""

    max_size: NotRequired[int | None]
    """Maximum size in bytes of an incoming websocket message.
    Defaults to None (no limit). Set an explicit byte limit to bound memory usage for
    long-lived connections behind proxies or in memory-constrained containers."""


class OpenAIRealtimeWebSocketModel(RealtimeModel):
    """A model that uses OpenAI's WebSocket API."""

    def __init__(self, *, transport_config: TransportConfig | None = None) -> None:
        self.model = DEFAULT_REALTIME_MODEL
        self._websocket: ClientConnection | None = None
        self._websocket_task: asyncio.Task[None] | None = None
        self._response_create_tasks: set[asyncio.Task[None]] = set()
        self._listeners: list[RealtimeModelListener] = []
        self._current_item_id: str | None = None
        self._audio_state_tracker: ModelAudioTracker = ModelAudioTracker()
        self._response_create_sequencer = _ResponseCreateSequencer()
        self._tracing_config: RealtimeModelTracingConfig | Literal["auto"] | None = None
        self._playback_tracker: RealtimePlaybackTracker | None = None
        self._created_session: OpenAISessionCreateRequest | None = None
        self._server_event_type_adapter = get_server_event_type_adapter()
        self._call_id: str | None = None
        self._transport_config: TransportConfig | None = transport_config

    @property
    def _ongoing_response(self) -> bool:
        return self._response_create_sequencer.ongoing_response

    @_ongoing_response.setter
    def _ongoing_response(self, value: bool) -> None:
        self._response_create_sequencer.set_ongoing_response_for_test(value)

    @property
    def _response_control(self) -> Literal["free", "create_requested", "cancel_requested"]:
        return self._response_create_sequencer.response_control

    @property
    def _pending_response_create_event_id(self) -> str | None:
        return self._response_create_sequencer.pending_response_create_event_id

    async def connect(self, options: RealtimeModelConfig) -> None:
        """Establish a connection to the model and keep it alive."""
        assert self._websocket is None, "Already connected"
        assert self._websocket_task is None, "Already connected"

        model_settings: RealtimeSessionModelSettings = options.get("initial_model_settings", {})

        self._playback_tracker = options.get("playback_tracker", None)

        call_id = options.get("call_id")
        model_name = model_settings.get("model_name")
        if call_id and model_name:
            error_message = (
                "Cannot specify both `call_id` and `model_name` "
                "when attaching to an existing realtime call."
            )
            raise UserError(error_message)

        if model_name:
            self.model = model_name

        self._call_id = call_id
        api_key = await get_api_key(options.get("api_key"))

        if "tracing" in model_settings:
            self._tracing_config = model_settings["tracing"]
        else:
            self._tracing_config = "auto"

        if call_id:
            url = options.get("url", f"wss://api.openai.com/v1/realtime?call_id={call_id}")
        else:
            url = options.get("url", f"wss://api.openai.com/v1/realtime?model={self.model}")

        headers: dict[str, str] = {}
        if options.get("headers") is not None:
            # For customizing request headers
            headers.update(options["headers"])
        else:
            # OpenAI's Realtime API
            if not api_key:
                raise UserError("API key is required but was not provided.")

            headers.update({"Authorization": f"Bearer {api_key}"})

        self._websocket = await self._create_websocket_connection(
            url=url,
            headers=headers,
            transport_config=self._transport_config,
        )
        self._websocket_task = asyncio.create_task(self._listen_for_messages())
        await self._update_session_config(model_settings)

    async def _create_websocket_connection(
        self,
        url: str,
        headers: dict[str, str],
        transport_config: TransportConfig | None = None,
    ) -> ClientConnection:
        """Create a WebSocket connection with the given configuration.

        Args:
            url: The WebSocket URL to connect to.
            headers: HTTP headers to include in the connection request.
            transport_config: Optional low-level transport configuration.

        Returns:
            A connected WebSocket client connection.
        """
        connect_kwargs: dict[str, Any] = {
            "user_agent_header": _USER_AGENT,
            "additional_headers": headers,
            "max_size": None,  # Allow any size of message
        }

        if transport_config:
            if "ping_interval" in transport_config:
                connect_kwargs["ping_interval"] = transport_config["ping_interval"]
            if "ping_timeout" in transport_config:
                connect_kwargs["ping_timeout"] = transport_config["ping_timeout"]
            if "handshake_timeout" in transport_config:
                connect_kwargs["open_timeout"] = transport_config["handshake_timeout"]
            if "max_size" in transport_config:
                connect_kwargs["max_size"] = transport_config["max_size"]

        return await websockets.connect(url, **connect_kwargs)

    async def _send_tracing_config(
        self, tracing_config: RealtimeModelTracingConfig | Literal["auto"] | None
    ) -> None:
        """Update tracing configuration via session.update event."""
        if tracing_config is not None:
            converted_tracing_config = _ConversionHelper.convert_tracing_config(tracing_config)
            await self._send_raw_message(
                OpenAISessionUpdateEvent(
                    session=OpenAISessionCreateRequest(
                        model=self.model,
                        type="realtime",
                        tracing=converted_tracing_config,
                    ),
                    type="session.update",
                )
            )

    def add_listener(self, listener: RealtimeModelListener) -> None:
        """Add a listener to the model."""
        if listener not in self._listeners:
            self._listeners.append(listener)

    def remove_listener(self, listener: RealtimeModelListener) -> None:
        """Remove a listener from the model."""
        if listener in self._listeners:
            self._listeners.remove(listener)

    async def _emit_event(self, event: RealtimeModelEvent) -> None:
        """Emit an event to the listeners."""
        # Copy list to avoid modification during iteration
        for listener in list(self._listeners):
            await listener.on_event(event)

    async def _listen_for_messages(self):
        assert self._websocket is not None, "Not connected"

        try:
            async for message in self._websocket:
                try:
                    parsed = json.loads(message)
                    await self._handle_ws_event(parsed)
                except json.JSONDecodeError as e:
                    await self._emit_event(
                        RealtimeModelExceptionEvent(
                            exception=e, context="Failed to parse WebSocket message as JSON"
                        )
                    )
                except Exception as e:
                    await self._emit_event(
                        RealtimeModelExceptionEvent(
                            exception=e, context="Error handling WebSocket event"
                        )
                    )

        except websockets.exceptions.ConnectionClosedOK:
            # Normal connection closure - no exception event needed
            logger.debug("WebSocket connection closed normally")
        except websockets.exceptions.ConnectionClosed as e:
            await self._emit_event(
                RealtimeModelExceptionEvent(
                    exception=e, context="WebSocket connection closed unexpectedly"
                )
            )
        except Exception as e:
            await self._emit_event(
                RealtimeModelExceptionEvent(
                    exception=e, context="WebSocket error in message listener"
                )
            )
        finally:
            await self._cancel_response_create_tasks()
            await self._release_response_waiters()

    async def send_event(self, event: RealtimeModelSendEvent) -> None:
        """Send an event to the model."""
        if isinstance(event, RealtimeModelSendRawMessage):
            converted = _ConversionHelper.try_convert_raw_message(event)
            if converted is not None:
                if converted.type == "response.create":
                    request_version = await self._reserve_response_create_request(manual=True)
                    self._start_response_create(
                        request_version,
                        response_create=converted,
                        manual=True,
                    )
                else:
                    await self._send_raw_message(converted)
            else:
                logger.error("Failed to convert raw message: %s", event)
        elif isinstance(event, RealtimeModelSendUserInput):
            await self._send_user_input(event)
        elif isinstance(event, RealtimeModelSendAudio):
            await self._send_audio(event)
        elif isinstance(event, RealtimeModelSendToolOutput):
            await self._send_tool_output(event)
        elif isinstance(event, RealtimeModelSendInterrupt):
            await self._send_interrupt(event)
        elif isinstance(event, RealtimeModelSendSessionUpdate):
            await self._send_session_update(event)
        else:
            assert_never(event)
            raise ValueError(f"Unknown event type: {type(event)}")

    async def _send_raw_message(self, event: OpenAIRealtimeClientEvent) -> None:
        """Send a raw message to the model."""
        assert self._websocket is not None, "Not connected"
        payload = event.model_dump_json(exclude_unset=True)
        await self._websocket.send(payload)

    async def _set_response_control(
        self, control: Literal["free", "create_requested", "cancel_requested"]
    ) -> None:
        await self._response_create_sequencer.set_response_control(control)

    async def _mark_response_created(self) -> None:
        await self._response_create_sequencer.mark_response_created()

    async def _mark_response_done(self) -> None:
        await self._response_create_sequencer.mark_response_done()

    async def _release_response_waiters(self) -> None:
        # Connection teardown means no response.done will arrive, so local
        # response sequencing must be released explicitly.
        await self._response_create_sequencer.release_waiters()

    async def _reserve_response_create_request(self, *, manual: bool = False) -> int:
        return await self._response_create_sequencer.reserve_response_create_request(manual=manual)

    async def _clear_pending_response_create(self, event_id: str | None = None) -> bool:
        return await self._response_create_sequencer.clear_pending_response_create(event_id)

    async def _send_response_create_when_idle(
        self,
        request_version: int,
        *,
        response_create: OpenAIResponseCreateEvent | None = None,
        manual: bool = False,
    ) -> None:
        pending = await self._response_create_sequencer.wait_for_response_create_slot(
            request_version,
            manual=manual,
            event_id=response_create.event_id if response_create is not None else None,
        )
        if pending is None:
            return

        try:
            response_create_event = (
                response_create.model_copy(update={"event_id": pending.event_id})
                if response_create is not None
                else OpenAIResponseCreateEvent(type="response.create", event_id=pending.event_id)
            )
            await self._send_raw_message(response_create_event)
        except BaseException:
            await self._clear_pending_response_create(pending.event_id)
            raise

        await self._response_create_sequencer.mark_response_create_sent(pending)

    async def _send_response_create_in_background(
        self,
        request_version: int,
        *,
        response_create: OpenAIResponseCreateEvent | None = None,
        manual: bool = False,
    ) -> None:
        try:
            await self._send_response_create_when_idle(
                request_version,
                response_create=response_create,
                manual=manual,
            )
        except asyncio.CancelledError:
            logger.debug("Deferred response.create task was cancelled")
        except AssertionError as exc:
            if str(exc) != "Not connected":
                await self._emit_event(
                    RealtimeModelExceptionEvent(
                        exception=exc, context="Error sending deferred response.create"
                    )
                )
        except websockets.exceptions.ConnectionClosed:
            logger.debug("Skipping deferred response.create because the websocket is closed")
        except Exception as exc:
            await self._emit_event(
                RealtimeModelExceptionEvent(
                    exception=exc, context="Error sending deferred response.create"
                )
            )

    def _start_response_create(
        self,
        request_version: int,
        *,
        response_create: OpenAIResponseCreateEvent | None = None,
        manual: bool = False,
    ) -> None:
        task = asyncio.create_task(
            self._send_response_create_in_background(
                request_version,
                response_create=response_create,
                manual=manual,
            )
        )
        self._response_create_tasks.add(task)
        task.add_done_callback(self._response_create_tasks.discard)

    async def _cancel_response_create_tasks(self) -> None:
        if not self._response_create_tasks:
            return

        current_task = asyncio.current_task()
        tasks_to_await = []
        for task in list(self._response_create_tasks):
            task.cancel()
            if task is not current_task:
                tasks_to_await.append(task)

        if tasks_to_await:
            await asyncio.gather(*tasks_to_await, return_exceptions=True)

    async def _send_user_input(self, event: RealtimeModelSendUserInput) -> None:
        converted = _ConversionHelper.convert_user_input_to_item_create(event)
        await self._send_raw_message(converted)
        request_version = await self._reserve_response_create_request()
        self._start_response_create(request_version)

    async def _send_audio(self, event: RealtimeModelSendAudio) -> None:
        converted = _ConversionHelper.convert_audio_to_input_audio_buffer_append(event)
        await self._send_raw_message(converted)
        if event.commit:
            await self._send_raw_message(
                OpenAIInputAudioBufferCommitEvent(type="input_audio_buffer.commit")
            )

    async def _send_tool_output(self, event: RealtimeModelSendToolOutput) -> None:
        converted = _ConversionHelper.convert_tool_output(event)
        await self._send_raw_message(converted)

        tool_item = RealtimeToolCallItem(
            item_id=event.tool_call.id or "",
            previous_item_id=event.tool_call.previous_item_id,
            call_id=event.tool_call.call_id,
            type="function_call",
            status="completed",
            arguments=event.tool_call.arguments,
            name=event.tool_call.name,
            output=event.output,
        )
        await self._emit_event(RealtimeModelItemUpdatedEvent(item=tool_item))

        if event.start_response:
            request_version = await self._reserve_response_create_request()
            self._start_response_create(request_version)

    def _get_playback_state(self) -> RealtimePlaybackState:
        if self._playback_tracker:
            return self._playback_tracker.get_state()

        if last_audio_item_id := self._audio_state_tracker.get_last_audio_item():
            item_id, item_content_index = last_audio_item_id
            audio_state = self._audio_state_tracker.get_state(item_id, item_content_index)
            if audio_state:
                elapsed_ms = (time.monotonic() - audio_state.initial_received_time) * 1000
                return {
                    "current_item_id": item_id,
                    "current_item_content_index": item_content_index,
                    "elapsed_ms": elapsed_ms,
                }

        return {
            "current_item_id": None,
            "current_item_content_index": None,
            "elapsed_ms": None,
        }

    def _get_audio_limits(self, item_id: str, item_content_index: int) -> tuple[float, int] | None:
        audio_state = self._audio_state_tracker.get_state(item_id, item_content_index)
        if audio_state is None:
            return None
        max_audio_ms = int(math.ceil(audio_state.audio_length_ms))
        return audio_state.audio_length_ms, max_audio_ms

    async def _send_interrupt(self, event: RealtimeModelSendInterrupt) -> None:
        playback_state = self._get_playback_state()
        current_item_id = playback_state.get("current_item_id")
        current_item_content_index = playback_state.get("current_item_content_index")
        elapsed_ms = playback_state.get("elapsed_ms")

        if current_item_id is None or elapsed_ms is None:
            logger.debug(
                "Skipping interrupt. Item id: %s, elapsed ms: %s, content index: %s",
                current_item_id,
                elapsed_ms,
                current_item_content_index,
            )
        else:
            current_item_content_index = current_item_content_index or 0
            if elapsed_ms > 0:
                await self._emit_event(
                    RealtimeModelAudioInterruptedEvent(
                        item_id=current_item_id,
                        content_index=current_item_content_index,
                    )
                )
                max_audio_ms: int | None = None
                audio_limits = self._get_audio_limits(current_item_id, current_item_content_index)
                if audio_limits is not None:
                    _, max_audio_ms = audio_limits
                truncated_ms = max(int(elapsed_ms), 0)
                if self._ongoing_response or max_audio_ms is None or truncated_ms < max_audio_ms:
                    converted = _ConversionHelper.convert_interrupt(
                        current_item_id,
                        current_item_content_index,
                        truncated_ms,
                    )
                    await self._send_raw_message(converted)
            else:
                logger.debug(
                    "Didn't interrupt bc elapsed ms is < 0. Item id: %s, "
                    "elapsed ms: %s, content index: %s",
                    current_item_id,
                    elapsed_ms,
                    current_item_content_index,
                )

        session = self._created_session
        automatic_response_cancellation_enabled = (
            session
            and session.audio is not None
            and session.audio.input is not None
            and session.audio.input.turn_detection is not None
            and session.audio.input.turn_detection.interrupt_response is True
        )
        should_cancel_response = event.force_response_cancel or (
            not automatic_response_cancellation_enabled
        )
        if should_cancel_response:
            await self._cancel_response()

        if current_item_id is not None and elapsed_ms is not None:
            self._audio_state_tracker.on_interrupted()
            if self._playback_tracker:
                self._playback_tracker.on_interrupted()

    async def _send_session_update(self, event: RealtimeModelSendSessionUpdate) -> None:
        """Send a session update to the model."""
        await self._update_session_config(event.session_settings)

    async def _handle_audio_delta(self, parsed: ResponseAudioDeltaEvent) -> None:
        """Handle audio delta events and update audio tracking state."""
        self._current_item_id = parsed.item_id

        audio_bytes = base64.b64decode(parsed.delta)

        self._audio_state_tracker.on_audio_delta(parsed.item_id, parsed.content_index, audio_bytes)

        await self._emit_event(
            RealtimeModelAudioEvent(
                data=audio_bytes,
                response_id=parsed.response_id,
                item_id=parsed.item_id,
                content_index=parsed.content_index,
            )
        )

    async def _handle_output_item(self, item: ConversationItem) -> None:
        """Handle response output item events (function calls and messages)."""
        if item.type == "function_call" and item.status == "completed":
            tool_call = RealtimeToolCallItem(
                item_id=item.id or "",
                previous_item_id=None,
                call_id=item.call_id,
                type="function_call",
                # We use the same item for tool call and output, so it will be completed by the
                # output being added
                status="in_progress",
                arguments=item.arguments or "",
                name=item.name or "",
                output=None,
            )
            await self._emit_event(RealtimeModelItemUpdatedEvent(item=tool_call))
            await self._emit_event(
                RealtimeModelToolCallEvent(
                    call_id=item.call_id or "",
                    name=item.name or "",
                    arguments=item.arguments or "",
                    id=item.id or "",
                )
            )
        elif item.type == "message":
            # Handle message items from output_item events (no previous_item_id)
            message_item: RealtimeMessageItem = TypeAdapter(RealtimeMessageItem).validate_python(
                {
                    "item_id": item.id or "",
                    "type": item.type,
                    "role": item.role,
                    "content": (
                        [content.model_dump() for content in item.content] if item.content else []
                    ),
                    "status": "in_progress",
                }
            )
            await self._emit_event(RealtimeModelItemUpdatedEvent(item=message_item))

    async def _handle_conversation_item(
        self, item: ConversationItem, previous_item_id: str | None
    ) -> None:
        """Handle conversation item creation/retrieval events."""
        message_item = _ConversionHelper.conversation_item_to_realtime_message_item(
            item, previous_item_id
        )
        await self._emit_event(RealtimeModelItemUpdatedEvent(item=message_item))

    async def close(self) -> None:
        """Close the session."""
        await self._cancel_response_create_tasks()
        if self._websocket:
            await self._websocket.close()
            self._websocket = None
        if self._websocket_task:
            self._websocket_task.cancel()
            try:
                await self._websocket_task
            except asyncio.CancelledError:
                pass
            self._websocket_task = None
        else:
            await self._release_response_waiters()

    async def _cancel_response(self) -> None:
        if not await self._response_create_sequencer.begin_cancel_response():
            return

        try:
            await self._send_raw_message(OpenAIResponseCancelEvent(type="response.cancel"))
        except Exception:
            await self._set_response_control("free")
            raise

    def _error_matches_pending_response_create(self, error: Any) -> bool:
        if error.event_id is not None:
            return True

        code = getattr(error, "code", None)
        message = (getattr(error, "message", None) or "").lower()
        return code == "bad_response_create" or "response.create" in message

    async def _handle_ws_event(self, event: dict[str, Any]):
        await self._emit_event(RealtimeModelRawServerEvent(data=event))
        # The public interface defined on this Agents SDK side (e.g., RealtimeMessageItem)
        # must be the same even after the GA migration, so this part does the conversion
        if isinstance(event, dict) and event.get("type") in (
            "response.output_item.added",
            "response.output_item.done",
        ):
            item = event.get("item")
            if isinstance(item, dict) and item.get("type") == "message":
                raw_content = item.get("content") or []
                converted_content: list[dict[str, Any]] = []
                for part in raw_content:
                    if not isinstance(part, dict):
                        continue
                    if part.get("type") in ("audio", "output_audio"):
                        converted_content.append(
                            {
                                "type": "audio",
                                "audio": part.get("audio"),
                                "transcript": part.get("transcript"),
                            }
                        )
                    elif part.get("type") in ("text", "output_text"):
                        converted_content.append({"type": "text", "text": part.get("text")})
                status = item.get("status")
                if status not in ("in_progress", "completed", "incomplete"):
                    is_done = event.get("type") == "response.output_item.done"
                    status = "completed" if is_done else "in_progress"
                # Explicitly type the adapter for mypy
                type_adapter: TypeAdapter[RealtimeMessageItem] = TypeAdapter(RealtimeMessageItem)
                message_item: RealtimeMessageItem = type_adapter.validate_python(
                    {
                        "item_id": item.get("id", ""),
                        "type": "message",
                        "role": item.get("role", "assistant"),
                        "content": converted_content,
                        "status": status,
                    }
                )
                await self._emit_event(RealtimeModelItemUpdatedEvent(item=message_item))
                return

        try:
            if "previous_item_id" in event and event["previous_item_id"] is None:
                event["previous_item_id"] = ""  # TODO (rm) remove
            validation_event = (
                _normalize_custom_voice_for_server_event_validation(event)
                if _should_normalize_custom_voice_for_server_event(event)
                else event
            )
            parsed: AllRealtimeServerEvents = self._server_event_type_adapter.validate_python(
                validation_event
            )
        except pydantic.ValidationError as e:
            _log_server_event_validation_failure(event, e)
            await self._emit_event(RealtimeModelErrorEvent(error=e))
            return
        except Exception as e:
            event_type = _log_server_event_validation_failure(event, e)
            exception_event = RealtimeModelExceptionEvent(
                exception=e,
                context=f"Failed to validate server event: {event_type}",
            )
            await self._emit_event(exception_event)
            return

        if parsed.type == "response.output_audio.delta":
            await self._handle_audio_delta(parsed)
        elif parsed.type == "response.output_audio.done":
            audio_done_event = RealtimeModelAudioDoneEvent(
                item_id=parsed.item_id,
                content_index=parsed.content_index,
            )
            await self._emit_event(audio_done_event)
        elif parsed.type == "input_audio_buffer.speech_started":
            # On VAD speech start, immediately stop local playback so the user can
            # barge‑in without overlapping assistant audio.
            last_audio = self._audio_state_tracker.get_last_audio_item()
            if last_audio is not None:
                item_id, content_index = last_audio
                playback_state = self._get_playback_state()
                playback_item_id = playback_state.get("current_item_id")
                playback_content_index = playback_state.get("current_item_content_index") or 0
                playback_elapsed_ms = playback_state.get("elapsed_ms")
                await self._emit_event(
                    RealtimeModelAudioInterruptedEvent(item_id=item_id, content_index=content_index)
                )

                elapsed_override = getattr(parsed, "audio_end_ms", None)
                if elapsed_override is None or elapsed_override <= 0:
                    effective_elapsed_ms = playback_elapsed_ms
                else:
                    effective_elapsed_ms = float(elapsed_override)

                if playback_item_id and effective_elapsed_ms is not None:
                    max_audio_ms: int | None = None
                    audio_limits = self._get_audio_limits(playback_item_id, playback_content_index)
                    if audio_limits is not None:
                        _, max_audio_ms = audio_limits
                    truncated_ms = max(int(round(effective_elapsed_ms)), 0)
                    if (
                        max_audio_ms is not None
                        and truncated_ms >= max_audio_ms
                        and not self._ongoing_response
                    ):
                        logger.debug(
                            "Skipping truncate because playback appears complete. Item id: %s, "
                            "elapsed ms: %s, content index: %s, audio length ms: %s",
                            playback_item_id,
                            effective_elapsed_ms,
                            playback_content_index,
                            max_audio_ms,
                        )
                    else:
                        if max_audio_ms is not None:
                            truncated_ms = min(truncated_ms, max_audio_ms)
                        await self._send_raw_message(
                            _ConversionHelper.convert_interrupt(
                                playback_item_id,
                                playback_content_index,
                                truncated_ms,
                            )
                        )

                # Reset trackers so subsequent playback state queries don't
                # reference audio that has been interrupted client‑side.
                self._audio_state_tracker.on_interrupted()
                if self._playback_tracker:
                    self._playback_tracker.on_interrupted()

                # If server isn't configured to auto‑interrupt/cancel, cancel the
                # response to prevent further audio.
                session = self._created_session
                automatic_response_cancellation_enabled = (
                    session
                    and session.audio is not None
                    and session.audio.input is not None
                    and session.audio.input.turn_detection is not None
                    and session.audio.input.turn_detection.interrupt_response is True
                )
                if not automatic_response_cancellation_enabled:
                    await self._cancel_response()
        elif parsed.type == "response.created":
            await self._mark_response_created()
            await self._emit_event(RealtimeModelTurnStartedEvent())
        elif parsed.type == "response.done":
            await self._mark_response_done()
            if parsed.response.usage is not None:
                await self._emit_event(
                    _ConversionHelper.convert_response_usage(parsed.response.usage)
                )
            await self._emit_event(RealtimeModelTurnEndedEvent())
        elif parsed.type == "session.created":
            await self._send_tracing_config(self._tracing_config)
            self._update_created_session(parsed.session)
        elif parsed.type == "session.updated":
            self._update_created_session(parsed.session)
        elif parsed.type == "error":
            if (
                not self._ongoing_response
                and self._response_control == "create_requested"
                and self._error_matches_pending_response_create(parsed.error)
            ):
                await self._clear_pending_response_create(parsed.error.event_id)
            await self._emit_event(RealtimeModelErrorEvent(error=parsed.error))
        elif parsed.type == "conversation.item.deleted":
            await self._emit_event(RealtimeModelItemDeletedEvent(item_id=parsed.item_id))
        elif (
            parsed.type == "conversation.item.added"
            or parsed.type == "conversation.item.created"
            or parsed.type == "conversation.item.retrieved"
        ):
            previous_item_id = (
                parsed.previous_item_id if parsed.type == "conversation.item.created" else None
            )
            if parsed.item.type == "message":
                await self._handle_conversation_item(parsed.item, previous_item_id)
        elif (
            parsed.type == "conversation.item.input_audio_transcription.completed"
            or parsed.type == "conversation.item.truncated"
        ):
            if self._current_item_id:
                await self._send_raw_message(
                    OpenAIConversationItemRetrieveEvent(
                        type="conversation.item.retrieve",
                        item_id=self._current_item_id,
                    )
                )
            if parsed.type == "conversation.item.input_audio_transcription.completed":
                await self._emit_event(
                    RealtimeModelInputAudioTranscriptionCompletedEvent(
                        item_id=parsed.item_id, transcript=parsed.transcript
                    )
                )
        elif parsed.type == "response.output_audio_transcript.delta":
            await self._emit_event(
                RealtimeModelTranscriptDeltaEvent(
                    item_id=parsed.item_id, delta=parsed.delta, response_id=parsed.response_id
                )
            )
        elif (
            parsed.type == "conversation.item.input_audio_transcription.delta"
            or parsed.type == "response.output_text.delta"
            or parsed.type == "response.function_call_arguments.delta"
        ):
            # No support for partials yet
            pass
        elif (
            parsed.type == "response.output_item.added"
            or parsed.type == "response.output_item.done"
        ):
            await self._handle_output_item(parsed.item)
        elif parsed.type == "input_audio_buffer.timeout_triggered":
            await self._emit_event(
                RealtimeModelInputAudioTimeoutTriggeredEvent(
                    item_id=parsed.item_id,
                    audio_start_ms=parsed.audio_start_ms,
                    audio_end_ms=parsed.audio_end_ms,
                )
            )

    def _update_created_session(
        self,
        session: OpenAISessionCreateRequest
        | OpenAIRealtimeTranscriptionSessionCreateRequest
        | Mapping[str, object]
        | pydantic.BaseModel,
    ) -> None:
        # Only store/playback-format information for realtime sessions (not transcription-only)
        normalized_session = self._normalize_session_payload(session)
        if not normalized_session:
            return

        self._created_session = normalized_session
        normalized_format = self._extract_audio_format(normalized_session)
        if normalized_format is None:
            return

        self._audio_state_tracker.set_audio_format(normalized_format)
        if self._playback_tracker:
            self._playback_tracker.set_audio_format(normalized_format)

    @staticmethod
    def _normalize_session_payload(
        session: OpenAISessionCreateRequest
        | OpenAIRealtimeTranscriptionSessionCreateRequest
        | Mapping[str, object]
        | pydantic.BaseModel,
    ) -> OpenAISessionCreateRequest | None:
        if isinstance(session, OpenAISessionCreateRequest):
            return session

        if isinstance(session, OpenAIRealtimeTranscriptionSessionCreateRequest):
            return None

        session_payload: Mapping[str, object]
        if isinstance(session, pydantic.BaseModel):
            session_payload = cast(Mapping[str, object], session.model_dump())
        elif isinstance(session, Mapping):
            session_payload = session
        else:
            return None

        if OpenAIRealtimeWebSocketModel._is_transcription_session(session_payload):
            return None

        try:
            return OpenAISessionCreateRequest.model_validate(session_payload)
        except pydantic.ValidationError:
            return None

    @staticmethod
    def _is_transcription_session(payload: Mapping[str, object]) -> bool:
        try:
            OpenAIRealtimeTranscriptionSessionCreateRequest.model_validate(payload)
        except pydantic.ValidationError:
            return False
        else:
            return True

    @staticmethod
    def _extract_audio_format(session: OpenAISessionCreateRequest) -> str | None:
        audio = session.audio
        if not audio or not audio.output or not audio.output.format:
            return None

        return OpenAIRealtimeWebSocketModel._normalize_audio_format(audio.output.format)

    @staticmethod
    def _normalize_audio_format(fmt: object) -> str:
        if isinstance(fmt, AudioPCM):
            return "pcm16"
        if isinstance(fmt, AudioPCMU):
            return "g711_ulaw"
        if isinstance(fmt, AudioPCMA):
            return "g711_alaw"

        fmt_type = OpenAIRealtimeWebSocketModel._read_format_type(fmt)
        if isinstance(fmt_type, str) and fmt_type:
            return fmt_type

        return str(fmt)

    @staticmethod
    def _read_format_type(fmt: object) -> str | None:
        if isinstance(fmt, str):
            return fmt

        if isinstance(fmt, Mapping):
            type_value = fmt.get("type")
            return type_value if isinstance(type_value, str) else None

        if isinstance(fmt, pydantic.BaseModel):
            type_value = fmt.model_dump().get("type")
            return type_value if isinstance(type_value, str) else None

        try:
            type_value = fmt.type  # type: ignore[attr-defined]
        except AttributeError:
            return None

        return type_value if isinstance(type_value, str) else None

    @staticmethod
    def _normalize_turn_detection_config(config: object) -> object:
        """Normalize camelCase turn detection keys to snake_case for API compatibility."""
        if not isinstance(config, Mapping):
            return config

        normalized = dict(config)
        key_map = {
            "createResponse": "create_response",
            "interruptResponse": "interrupt_response",
            "prefixPaddingMs": "prefix_padding_ms",
            "silenceDurationMs": "silence_duration_ms",
            "idleTimeoutMs": "idle_timeout_ms",
            "modelVersion": "model_version",
        }
        for camel_key, snake_key in key_map.items():
            if camel_key in normalized and snake_key not in normalized:
                normalized[snake_key] = normalized[camel_key]
            normalized.pop(camel_key, None)

        return normalized

    async def _update_session_config(self, model_settings: RealtimeSessionModelSettings) -> None:
        session_config = self._get_session_config(model_settings)
        await self._send_raw_message(
            OpenAISessionUpdateEvent(session=session_config, type="session.update")
        )

    def _get_session_config(
        self, model_settings: RealtimeSessionModelSettings
    ) -> OpenAISessionCreateRequest:
        """Get the session config."""
        audio_input_args: dict[str, Any] = {}
        audio_output_args: dict[str, Any] = {}

        audio_config = model_settings.get("audio")
        audio_config_mapping = audio_config if isinstance(audio_config, Mapping) else None
        # ``audio.input``/``audio.output`` may be omitted or explicitly None; coerce
        # both to an empty mapping so callers can opt out of one channel without
        # tripping the membership checks below.
        input_audio_config: Mapping[str, Any] = (
            cast(Mapping[str, Any], audio_config_mapping.get("input") or {})
            if audio_config_mapping
            else {}
        )
        output_audio_config: Mapping[str, Any] = (
            cast(Mapping[str, Any], audio_config_mapping.get("output") or {})
            if audio_config_mapping
            else {}
        )

        input_format_source: FormatInput = (
            input_audio_config.get("format") if input_audio_config else None
        )
        if input_format_source is None:
            if self._call_id:
                input_format_source = model_settings.get("input_audio_format")
            else:
                input_format_source = model_settings.get(
                    "input_audio_format", DEFAULT_MODEL_SETTINGS.get("input_audio_format")
                )
        input_format = to_realtime_audio_format(input_format_source)
        if input_format is not None:
            audio_input_args["format"] = input_format

        if "noise_reduction" in input_audio_config:
            audio_input_args["noise_reduction"] = input_audio_config.get("noise_reduction")
        elif "input_audio_noise_reduction" in model_settings:
            audio_input_args["noise_reduction"] = model_settings.get("input_audio_noise_reduction")

        if "transcription" in input_audio_config:
            audio_input_args["transcription"] = input_audio_config.get("transcription")
        elif "input_audio_transcription" in model_settings:
            audio_input_args["transcription"] = model_settings.get("input_audio_transcription")
        else:
            audio_input_args["transcription"] = DEFAULT_MODEL_SETTINGS.get(
                "input_audio_transcription"
            )

        if "turn_detection" in input_audio_config:
            audio_input_args["turn_detection"] = self._normalize_turn_detection_config(
                input_audio_config.get("turn_detection")
            )
        elif "turn_detection" in model_settings:
            audio_input_args["turn_detection"] = self._normalize_turn_detection_config(
                model_settings.get("turn_detection")
            )
        else:
            audio_input_args["turn_detection"] = DEFAULT_MODEL_SETTINGS.get("turn_detection")

        requested_voice = output_audio_config.get("voice") if output_audio_config else None
        audio_output_args["voice"] = requested_voice or model_settings.get(
            "voice", DEFAULT_MODEL_SETTINGS.get("voice")
        )

        output_format_source: FormatInput = (
            output_audio_config.get("format") if output_audio_config else None
        )
        if output_format_source is None:
            if self._call_id:
                output_format_source = model_settings.get("output_audio_format")
            else:
                output_format_source = model_settings.get(
                    "output_audio_format", DEFAULT_MODEL_SETTINGS.get("output_audio_format")
                )
        output_format = to_realtime_audio_format(output_format_source)
        if output_format is not None:
            audio_output_args["format"] = output_format

        if "speed" in output_audio_config:
            audio_output_args["speed"] = output_audio_config.get("speed")
        elif "speed" in model_settings:
            audio_output_args["speed"] = model_settings.get("speed")

        output_modalities = (
            model_settings.get("output_modalities")
            or model_settings.get("modalities")
            or DEFAULT_MODEL_SETTINGS.get("modalities")
        )

        session_create_args: dict[str, Any] = {
            "type": "realtime",
            "model": (model_settings.get("model_name") or self.model) or DEFAULT_REALTIME_MODEL,
            "output_modalities": output_modalities,
            "audio": OpenAIRealtimeAudioConfig(
                input=OpenAIRealtimeAudioInput(**audio_input_args),
                output=OpenAIRealtimeAudioOutput(**audio_output_args),
            ),
            "tools": self._tools_to_session_tools(
                tools=model_settings.get("tools", []),
                handoffs=model_settings.get("handoffs", []),
            ),
        }
        if model_settings.get("parallel_tool_calls") is not None:
            session_create_args["parallel_tool_calls"] = model_settings["parallel_tool_calls"]
        if model_settings.get("reasoning") is not None:
            session_create_args["reasoning"] = model_settings["reasoning"]

        # Construct full session object. `type` will be excluded at serialization time for updates.
        session_create_request = OpenAISessionCreateRequest(**session_create_args)

        if "instructions" in model_settings:
            session_create_request.instructions = model_settings.get("instructions")

        if "prompt" in model_settings:
            _passed_prompt: Prompt = model_settings["prompt"]
            variables: dict[str, Any] | None = _passed_prompt.get("variables")
            session_create_request.prompt = ResponsePrompt(
                id=_passed_prompt["id"],
                variables=variables,
                version=_passed_prompt.get("version"),
            )

        if "max_output_tokens" in model_settings:
            session_create_request.max_output_tokens = cast(
                Any, model_settings.get("max_output_tokens")
            )

        if "tool_choice" in model_settings:
            tool_choice = model_settings.get("tool_choice")
            ensure_tool_choice_supports_backend(
                tool_choice,
                backend_name="OpenAI Responses models",
            )
            session_create_request.tool_choice = cast(Any, tool_choice)

        return session_create_request

    def _tools_to_session_tools(
        self, tools: list[Tool], handoffs: list[Handoff]
    ) -> list[OpenAISessionFunction]:
        converted_tools: list[OpenAISessionFunction] = []
        enabled_tools = filter_statically_enabled_tools(tools)
        enabled_handoffs = [handoff for handoff in handoffs if handoff.is_enabled is not False]
        for tool in enabled_tools:
            if not isinstance(tool, FunctionTool):
                raise UserError(f"Tool {tool.name} is unsupported. Must be a function tool.")
            ensure_function_tool_supports_responses_only_features(
                tool,
                backend_name="Realtime models",
            )
            converted_tools.append(
                OpenAISessionFunction(
                    name=tool.name,
                    description=tool.description,
                    parameters=tool.params_json_schema,
                    type="function",
                )
            )

        validate_realtime_tool_names(enabled_tools, enabled_handoffs)

        for handoff in enabled_handoffs:
            converted_tools.append(
                OpenAISessionFunction(
                    name=handoff.tool_name,
                    description=handoff.tool_description,
                    parameters=handoff.input_json_schema,
                    type="function",
                )
            )

        return converted_tools


class OpenAIRealtimeSIPModel(OpenAIRealtimeWebSocketModel):
    """Realtime model that attaches to SIP-originated calls using a call ID."""

    @staticmethod
    async def build_initial_session_payload(
        agent: RealtimeAgent[Any],
        *,
        context: TContext | None = None,
        model_config: RealtimeModelConfig | None = None,
        run_config: RealtimeRunConfig | None = None,
        overrides: RealtimeSessionModelSettings | None = None,
    ) -> OpenAISessionCreateRequest:
        """Build a session payload that mirrors what a RealtimeSession would send on connect.

        This helper can be used to accept SIP-originated calls by forwarding the returned payload to
        the Realtime Calls API without duplicating session setup logic.
        """
        run_config_settings = (run_config or {}).get("model_settings") or {}
        initial_model_settings = (model_config or {}).get("initial_model_settings") or {}
        base_settings: RealtimeSessionModelSettings = {
            **run_config_settings,
            **initial_model_settings,
        }

        context_wrapper = RunContextWrapper(context)
        merged_settings = await _build_model_settings_from_agent(
            agent=agent,
            context_wrapper=context_wrapper,
            base_settings=base_settings,
            starting_settings=initial_model_settings,
            run_config=run_config,
        )

        if overrides:
            merged_settings.update(overrides)
            if "tools" in overrides:
                merged_settings["tools"] = await filter_enabled_tools(
                    merged_settings.get("tools") or [],
                    context_wrapper,
                    agent,
                )
            if "handoffs" in overrides:
                merged_settings["handoffs"] = await filter_enabled_handoffs(
                    merged_settings.get("handoffs") or [],
                    context_wrapper,
                    agent,
                )

        model = OpenAIRealtimeWebSocketModel()
        return model._get_session_config(merged_settings)

    async def connect(self, options: RealtimeModelConfig) -> None:
        call_id = options.get("call_id")
        if not call_id:
            raise UserError("OpenAIRealtimeSIPModel requires `call_id` in the model configuration.")

        sip_options = options.copy()
        await super().connect(sip_options)


class _ConversionHelper:
    @classmethod
    def convert_response_usage(cls, usage: RealtimeResponseUsage) -> RealtimeModelUsageEvent:
        input_tokens = usage.input_tokens or 0
        output_tokens = usage.output_tokens or 0
        total_tokens = (
            usage.total_tokens if usage.total_tokens is not None else input_tokens + output_tokens
        )

        input_details = usage.input_token_details
        output_details = usage.output_token_details

        aggregate_usage = Usage(
            requests=1,
            input_tokens=input_tokens,
            output_tokens=output_tokens,
            total_tokens=total_tokens,
        )
        if input_details is not None:
            aggregate_usage.input_tokens_details.cached_tokens = input_details.cached_tokens or 0

        cached_details = input_details.cached_tokens_details if input_details is not None else None
        return RealtimeModelUsageEvent(
            usage=aggregate_usage,
            input_tokens_details=(
                RealtimeModelInputTokensDetails(
                    text_tokens=input_details.text_tokens,
                    audio_tokens=input_details.audio_tokens,
                    image_tokens=input_details.image_tokens,
                    cached_tokens=input_details.cached_tokens,
                    cached_tokens_details=(
                        RealtimeModelCachedTokensDetails(
                            text_tokens=cached_details.text_tokens,
                            audio_tokens=cached_details.audio_tokens,
                            image_tokens=cached_details.image_tokens,
                        )
                        if cached_details is not None
                        else None
                    ),
                )
                if input_details is not None
                else None
            ),
            output_tokens_details=(
                RealtimeModelOutputTokensDetails(
                    text_tokens=output_details.text_tokens,
                    audio_tokens=output_details.audio_tokens,
                )
                if output_details is not None
                else None
            ),
        )

    @classmethod
    def conversation_item_to_realtime_message_item(
        cls, item: ConversationItem, previous_item_id: str | None
    ) -> RealtimeMessageItem:
        if not isinstance(
            item,
            RealtimeConversationItemUserMessage
            | RealtimeConversationItemAssistantMessage
            | RealtimeConversationItemSystemMessage,
        ):
            raise ValueError("Unsupported conversation item type for message conversion.")
        content: list[dict[str, Any]] = []
        for each in item.content:
            c = each.model_dump()
            if each.type == "output_text":
                # For backward-compatibility of assistant message items
                c["type"] = "text"
            elif each.type == "output_audio":
                # For backward-compatibility of assistant message items
                c["type"] = "audio"
            content.append(c)
        return TypeAdapter(RealtimeMessageItem).validate_python(
            {
                "item_id": item.id or "",
                "previous_item_id": previous_item_id,
                "type": item.type,
                "role": item.role,
                "content": content,
                "status": "in_progress",
            },
        )

    @classmethod
    def try_convert_raw_message(
        cls, message: RealtimeModelSendRawMessage
    ) -> OpenAIRealtimeClientEvent | None:
        try:
            data = {}
            data["type"] = message.message["type"]
            data.update(message.message.get("other_data", {}))
            return TypeAdapter(OpenAIRealtimeClientEvent).validate_python(data)
        except Exception:
            return None

    @classmethod
    def convert_tracing_config(
        cls, tracing_config: RealtimeModelTracingConfig | Literal["auto"] | None
    ) -> OpenAITracingConfiguration | Literal["auto"] | None:
        if tracing_config is None:
            return None
        elif tracing_config == "auto":
            return "auto"
        return OpenAITracingConfiguration(
            group_id=tracing_config.get("group_id"),
            metadata=tracing_config.get("metadata"),
            workflow_name=tracing_config.get("workflow_name"),
        )

    @classmethod
    def convert_user_input_to_conversation_item(
        cls, event: RealtimeModelSendUserInput
    ) -> OpenAIConversationItem:
        user_input = event.user_input

        if isinstance(user_input, dict):
            content: list[Content] = []
            for item in user_input.get("content", []):
                try:
                    if not isinstance(item, dict):
                        continue
                    t = item.get("type")
                    if t == "input_text":
                        _txt = item.get("text")
                        # Skip parts with missing/non-string text rather than
                        # forwarding text=None, which produces an invalid item
                        # the realtime API will reject.
                        if not isinstance(_txt, str):
                            continue
                        content.append(Content(type="input_text", text=_txt))
                    elif t == "input_image":
                        iu = item.get("image_url")
                        if isinstance(iu, str) and iu:
                            d = item.get("detail")
                            detail_val = cast(
                                Literal["auto", "low", "high"] | None,
                                d if isinstance(d, str) and d in ("auto", "low", "high") else None,
                            )
                            if detail_val is None:
                                content.append(
                                    Content(
                                        type="input_image",
                                        image_url=iu,
                                    )
                                )
                            else:
                                content.append(
                                    Content(
                                        type="input_image",
                                        image_url=iu,
                                        detail=detail_val,
                                    )
                                )
                    # ignore unknown types for forward-compat
                except Exception:
                    # best-effort; skip malformed parts
                    continue
            return RealtimeConversationItemUserMessage(
                type="message",
                role="user",
                content=content,
            )
        else:
            return RealtimeConversationItemUserMessage(
                type="message",
                role="user",
                content=[Content(type="input_text", text=user_input)],
            )

    @classmethod
    def convert_user_input_to_item_create(
        cls, event: RealtimeModelSendUserInput
    ) -> OpenAIRealtimeClientEvent:
        return OpenAIConversationItemCreateEvent(
            type="conversation.item.create",
            item=cls.convert_user_input_to_conversation_item(event),
        )

    @classmethod
    def convert_audio_to_input_audio_buffer_append(
        cls, event: RealtimeModelSendAudio
    ) -> OpenAIRealtimeClientEvent:
        base64_audio = base64.b64encode(event.audio).decode("utf-8")
        return OpenAIInputAudioBufferAppendEvent(
            type="input_audio_buffer.append",
            audio=base64_audio,
        )

    @classmethod
    def convert_tool_output(cls, event: RealtimeModelSendToolOutput) -> OpenAIRealtimeClientEvent:
        return OpenAIConversationItemCreateEvent(
            type="conversation.item.create",
            item=RealtimeConversationItemFunctionCallOutput(
                type="function_call_output",
                output=event.output,
                call_id=event.tool_call.call_id,
            ),
        )

    @classmethod
    def convert_interrupt(
        cls,
        current_item_id: str,
        current_audio_content_index: int,
        elapsed_time_ms: int,
    ) -> OpenAIRealtimeClientEvent:
        return OpenAIConversationItemTruncateEvent(
            type="conversation.item.truncate",
            item_id=current_item_id,
            content_index=current_audio_content_index,
            audio_end_ms=elapsed_time_ms,
        )
