backend/services/openai_realtime/bridge.py

"""Translate OpenAI Realtime events to the project's neutral pipeline messages."""

from __future__ import annotations

import asyncio
import base64
import binascii
import json
from typing import Any
from uuid import uuid4

from pipecat.audio.utils import create_stream_resampler
from pipecat.frames.frames import (
    BotStartedSpeakingFrame,
    BotStoppedSpeakingFrame,
    InputAudioRawFrame,
    InputTransportMessageFrame,
    LLMContextFrame,
    OutputAudioRawFrame,
    OutputTransportMessageFrame,
    OutputTransportMessageUrgentFrame,
    StartFrame,
    UserStartedSpeakingFrame,
    UserStoppedSpeakingFrame,
    VADParamsUpdateFrame,
)
from pipecat.processors.frame_processor import FrameDirection, FrameProcessor

from services.input_assets import store_input_image
from services.openai_realtime.events import (
    SUPPORTED_CAPABILITIES,
    RealtimeEventError,
    error_event,
    normalize_turn_detection,
    require_event,
)
from services.openai_realtime.schemas import SERVER_EVENT_TYPES
from services.openai_realtime.session import OpenAIRealtimeSession
from services.pipecat.turn_config import create_vad_params
from services.realtime.protocol import PipelineProtocolRuntime


MAX_BUFFERED_AUDIO_BYTES = 24000 * 2 * 120


def _event_id() -> str:
    return f"event_{uuid4().hex}"


def _server_event(event_type: str, **payload: Any) -> dict[str, Any]:
    event = {"type": event_type, "event_id": _event_id(), **payload}
    if event_type.startswith("x.interactive_media."):
        event["version"] = "1"
    return event


def _decode_base64(value: object, *, param: str) -> bytes:
    if not isinstance(value, str) or not value:
        raise RealtimeEventError(f"{param} must be non-empty Base64", param=param)
    try:
        return base64.b64decode(value, validate=True)
    except (ValueError, binascii.Error) as exc:
        raise RealtimeEventError(f"{param} is not valid Base64", param=param) from exc


def _decode_data_url(value: object) -> bytes:
    if not isinstance(value, str) or not value.startswith("data:image/"):
        raise RealtimeEventError(
            "input_image only supports image Data URLs",
            code="unsupported_content_type",
            param="item.content",
        )
    header, separator, encoded = value.partition(",")
    if not separator or ";base64" not in header:
        raise RealtimeEventError(
            "input_image Data URL must use Base64 encoding",
            param="item.content",
        )
    return _decode_base64(encoded, param="item.content")


class OpenAIRealtimeInputProcessor(FrameProcessor):
    def __init__(self, bridge: "OpenAIRealtimeBridge") -> None:
        super().__init__()
        self._bridge = bridge
        self._started = False

    async def process_frame(self, frame, direction: FrameDirection):
        await super().process_frame(frame, direction)
        if isinstance(frame, StartFrame) and not self._started:
            self._started = True
            # Initialize every downstream processor/serializer before the first
            # business event is sent to the client.
            await self.push_frame(frame, direction)
            await self._bridge.emit(
                _server_event(
                    "session.created",
                    session=self._bridge.session.public_value(),
                )
            )
            return

        if isinstance(frame, InputAudioRawFrame):
            if (
                self._bridge.session.external_turn_control
                and frame.transport_source != "openai-committed"
            ):
                try:
                    self._bridge.buffer_audio(
                        frame.audio,
                        sample_rate=frame.sample_rate,
                        num_channels=frame.num_channels,
                    )
                except RealtimeEventError as exc:
                    await self._bridge.emit(error_event(exc))
                return
            await self.push_frame(frame, direction)
            return

        if not isinstance(frame, InputTransportMessageFrame):
            await self.push_frame(frame, direction)
            return
        if isinstance(frame.message, dict) and frame.message.pop(
            "_pipeline_internal", False
        ):
            await self.push_frame(frame, direction)
            return
        if not self._bridge.client_ready_sent:
            self._bridge.client_ready_sent = True
            await self.push_frame(
                InputTransportMessageFrame(
                    message={"type": "client-ready"}
                ),
                direction,
            )
        client_event: dict[str, Any] | None = None
        try:
            client_event = require_event(frame.message)
            await self._bridge.handle_client_event(client_event)
        except RealtimeEventError as exc:
            if exc.event_id is None and client_event is not None:
                exc.event_id = str(client_event.get("event_id") or "") or None
            await self._bridge.emit(error_event(exc))
        except Exception as exc:  # noqa: BLE001 - event errors must stay connection-local
            client_event_id = (
                str(frame.message.get("event_id") or "") or None
                if isinstance(frame.message, dict)
                else None
            )
            await self._bridge.emit(
                error_event(
                    RealtimeEventError(
                        f"Event processing failed: {exc}",
                        code="event_processing_error",
                        event_id=client_event_id,
                    )
                )
            )


class OpenAIRealtimeOutputProcessor(FrameProcessor):
    def __init__(self, bridge: "OpenAIRealtimeBridge") -> None:
        super().__init__()
        self._bridge = bridge
        self._audio_resampler = create_stream_resampler()

    async def process_frame(self, frame, direction: FrameDirection):
        await super().process_frame(frame, direction)
        if isinstance(frame, BotStartedSpeakingFrame):
            if not self._bridge.session.active_response_id:
                for event in self._bridge._start_assistant_output():
                    await self.push_frame(
                        OutputTransportMessageUrgentFrame(message=event)
                    )
            await self.push_frame(frame, direction)
            return
        if isinstance(frame, BotStoppedSpeakingFrame):
            if self._bridge.session.active_response_id:
                for event in self._bridge._end_assistant_output(False):
                    await self.push_frame(
                        OutputTransportMessageUrgentFrame(message=event)
                    )
            await self.push_frame(frame, direction)
            return
        if isinstance(frame, UserStartedSpeakingFrame):
            item_id = f"item_{uuid4().hex}"
            self._bridge.session.active_input_item_id = item_id
            await self.push_frame(
                OutputTransportMessageUrgentFrame(
                    message=_server_event(
                        "input_audio_buffer.speech_started",
                        audio_start_ms=0,
                        item_id=item_id,
                    )
                )
            )
            await self.push_frame(frame, direction)
            return
        if isinstance(frame, UserStoppedSpeakingFrame):
            item_id = (
                self._bridge.session.active_input_item_id
                or f"item_{uuid4().hex}"
            )
            self._bridge.session.active_input_item_id = None
            await self.push_frame(
                OutputTransportMessageUrgentFrame(
                    message=_server_event(
                        "input_audio_buffer.speech_stopped",
                        audio_end_ms=0,
                        item_id=item_id,
                    )
                )
            )
            await self.push_frame(frame, direction)
            return
        if isinstance(frame, OutputAudioRawFrame):
            if not self._bridge.session.output_is_audio:
                return
            if self._bridge.channel == "websocket":
                if not self._bridge.session.active_response_id:
                    for event in self._bridge._start_assistant_output():
                        await self.push_frame(
                            OutputTransportMessageUrgentFrame(message=event)
                        )
                response_id, item_id = self._bridge.session.begin_response()
                audio = frame.audio
                if frame.sample_rate != 24000:
                    audio = await self._audio_resampler.resample(
                        audio,
                        frame.sample_rate,
                        24000,
                    )
                await self.push_frame(
                    OutputTransportMessageUrgentFrame(
                        message=_server_event(
                            "response.output_audio.delta",
                            response_id=response_id,
                            item_id=item_id,
                            output_index=0,
                            content_index=0,
                            delta=base64.b64encode(audio).decode("ascii"),
                        )
                    )
                )
                return
            await self.push_frame(frame, direction)
            return

        if not isinstance(
            frame,
            (OutputTransportMessageFrame, OutputTransportMessageUrgentFrame),
        ):
            await self.push_frame(frame, direction)
            return
        message = frame.message
        if not isinstance(message, dict):
            return
        translated = self._bridge.translate_server_message(message)
        for event in translated:
            await self.push_frame(
                OutputTransportMessageUrgentFrame(message=event),
                direction,
            )


class OpenAIResponseGateProcessor(FrameProcessor):
    """Hold automatic inference when create_response is disabled or PTT is active."""

    def __init__(self, bridge: "OpenAIRealtimeBridge") -> None:
        super().__init__()
        self._bridge = bridge
        self._held: list[tuple[Any, FrameDirection]] = []
        self._allow_next = False

    async def process_frame(self, frame, direction: FrameDirection):
        await super().process_frame(frame, direction)
        if direction != FrameDirection.DOWNSTREAM or not isinstance(
            frame, LLMContextFrame
        ):
            await self.push_frame(frame, direction)
            return
        turn = self._bridge.session.turn_detection
        auto_response = bool(turn and turn.get("create_response", True))
        if auto_response or self._allow_next:
            self._allow_next = False
            await self.push_frame(frame, direction)
            return
        self._held.append((frame, direction))

    async def allow_one_response(self) -> None:
        if self._held:
            held, self._held = self._held, []
            for frame, direction in held:
                await self.push_frame(frame, direction)
            return
        self._allow_next = True


class OpenAIRealtimeBridge:
    """One adapter instance is owned by exactly one public Realtime session."""

    def __init__(self, session: OpenAIRealtimeSession, *, channel: str) -> None:
        self.session = session
        self.channel = channel
        self._runtime: PipelineProtocolRuntime | None = None
        self.client_ready_sent = False
        self._input = OpenAIRealtimeInputProcessor(self)
        self._inference = OpenAIResponseGateProcessor(self)
        self._output = OpenAIRealtimeOutputProcessor(self)

    def input_processors(self) -> list[FrameProcessor]:
        return [self._input]

    def output_processors(self) -> list[FrameProcessor]:
        return [self._output]

    def inference_processors(self) -> list[FrameProcessor]:
        return [self._inference]

    async def bind(self, runtime: PipelineProtocolRuntime) -> None:
        self._runtime = runtime
        await runtime.set_external_turn_control(self.session.external_turn_control)
        turn = self.session.turn_detection
        if turn:
            await runtime.set_response_interruption(
                bool(turn.get("interrupt_response", True))
            )

    @property
    def runtime(self) -> PipelineProtocolRuntime:
        if self._runtime is None:
            raise RuntimeError("OpenAI Realtime bridge is not bound to a pipeline")
        return self._runtime

    async def emit(self, event: dict[str, Any]) -> None:
        await self._input.push_frame(
            OutputTransportMessageUrgentFrame(message=event)
        )

    async def queue_internal_message(self, message: dict[str, Any]) -> None:
        await self.runtime.queue_frame(
            InputTransportMessageFrame(
                message={**message, "_pipeline_internal": True}
            )
        )

    def buffer_audio(
        self,
        audio: bytes,
        *,
        sample_rate: int,
        num_channels: int,
    ) -> None:
        if self.session.buffered_audio_bytes + len(audio) > MAX_BUFFERED_AUDIO_BYTES:
            self.clear_buffered_audio()
            raise RealtimeEventError(
                "input_audio_buffer exceeds the 120 second MVP limit",
                code="input_audio_buffer_too_large",
            )
        self.session.audio_chunks.append((audio, sample_rate, num_channels))
        self.session.buffered_audio_bytes += len(audio)

    def clear_buffered_audio(self) -> None:
        self.session.audio_chunks.clear()
        self.session.buffered_audio_bytes = 0
        self.session.audio_committed = False

    async def handle_client_event(self, event: dict[str, Any]) -> None:
        event_type = str(event["type"])
        if event_type == "session.update":
            await self._update_session(event)
        elif event_type == "conversation.item.create":
            await self._create_item(event)
        elif event_type == "conversation.item.truncate":
            await self.runtime.cancel_response()
            await self.emit(
                _server_event(
                    "conversation.item.truncated",
                    item_id=event.get("item_id"),
                    content_index=int(event.get("content_index") or 0),
                    audio_end_ms=int(event.get("audio_end_ms") or 0),
                )
            )
        elif event_type == "input_audio_buffer.append":
            await self._append_audio(event)
        elif event_type == "input_audio_buffer.commit":
            await self._commit_audio()
        elif event_type == "input_audio_buffer.clear":
            self.clear_buffered_audio()
            await self.runtime.clear_audio()
            await self.emit(_server_event("input_audio_buffer.cleared"))
        elif event_type == "output_audio_buffer.clear":
            await self.runtime.cancel_response()
            await self._finish_response(status="cancelled")
            await self.emit(_server_event("output_audio_buffer.cleared"))
        elif event_type == "response.create":
            await self._create_response(event)
        elif event_type == "response.cancel":
            await self.runtime.cancel_response()
            await self._finish_response(status="cancelled")
        elif event_type == "x.interactive_media.capabilities.update":
            await self._update_capabilities(event)
        elif event_type == "x.interactive_media.session.variables.update":
            await self._update_variables(event)

    async def _update_session(self, event: dict[str, Any]) -> None:
        update = event.get("session")
        if not isinstance(update, dict):
            raise RealtimeEventError("session.update requires session", param="session")
        locked = {"model", "instructions", "voice", "tools", "tool_choice"}
        changed_locked = sorted(locked.intersection(update))
        audio = update.get("audio")
        if isinstance(audio, dict):
            output = audio.get("output")
            if isinstance(output, dict) and "voice" in output:
                changed_locked.append("audio.output.voice")
        if changed_locked:
            raise RealtimeEventError(
                f"Assistant-owned session fields cannot be changed: {', '.join(changed_locked)}",
                code="immutable_session_field",
                param=changed_locked[0],
                event_id=str(event.get("event_id") or "") or None,
            )

        modalities = update.get("output_modalities")
        if modalities is not None:
            if modalities not in (["audio"], ["text"]):
                raise RealtimeEventError(
                    'output_modalities must be ["audio"] or ["text"]',
                    param="session.output_modalities",
                )
            self.session.output_modalities = list(modalities)

        marker = object()
        turn_detection: object = marker
        if isinstance(audio, dict) and isinstance(audio.get("input"), dict):
            turn_detection = audio["input"].get("turn_detection", marker)
        if turn_detection is marker:
            turn_detection = update.get("turn_detection", marker)
        if turn_detection is not marker:
            self.session.turn_detection = self._validate_turn_detection(turn_detection)
            await self.runtime.set_external_turn_control(
                self.session.external_turn_control
            )
            if self.session.turn_detection:
                await self.runtime.set_response_interruption(
                    bool(self.session.turn_detection.get("interrupt_response", True))
                )
            if self.session.turn_detection:
                turn = self.session.turn_detection
                config = {
                    "vad": {
                        "confidence": turn["threshold"],
                        "start_secs": max(0.05, turn["prefix_padding_ms"] / 1000),
                        "stop_secs": 0.2,
                    },
                    "turn_detection": {
                        "strategy": "silence",
                        "silence_timeout_secs": turn["silence_duration_ms"] / 1000,
                    },
                }
                await self.runtime.queue_frame(
                    VADParamsUpdateFrame(params=create_vad_params(config))
                )
        await self.emit(
            _server_event("session.updated", session=self.session.public_value())
        )

    def _validate_turn_detection(self, value: object) -> dict[str, Any] | None:
        return normalize_turn_detection(value)

    async def _create_item(self, event: dict[str, Any]) -> None:
        item = event.get("item")
        if not isinstance(item, dict):
            raise RealtimeEventError("conversation.item.create requires item", param="item")
        item_type = item.get("type")
        if item_type == "function_call_output":
            call_id = str(item.get("call_id") or "")
            if not call_id:
                raise RealtimeEventError("function_call_output requires call_id", param="item.call_id")
            raw_output = item.get("output")
            try:
                data = json.loads(raw_output) if isinstance(raw_output, str) else raw_output
            except json.JSONDecodeError:
                data = raw_output
            await self.queue_internal_message(
                {
                    "type": "client-tool-result",
                    "tool_call_id": call_id,
                    "status": "ok",
                    "data": data,
                }
            )
            await self._emit_item_ack(item)
            return
        if item_type != "message" or item.get("role") != "user":
            raise RealtimeEventError(
                "Only user messages and function_call_output items are accepted",
                code="unsupported_item_type",
                param="item.type",
            )

        wire_parts: list[dict[str, Any]] = []
        for part in item.get("content") or []:
            if not isinstance(part, dict):
                continue
            if part.get("type") == "input_text":
                text = str(part.get("text") or "").strip()
                if text:
                    wire_parts.append({"type": "input_text", "text": text})
            elif part.get("type") == "input_image":
                if self.session.config.runtimeMode == "realtime":
                    raise RealtimeEventError(
                        "input_image is not supported by the assistant's realtime runtime",
                        code="unsupported_content_type",
                    )
                if not self.session.vision_enabled:
                    raise RealtimeEventError(
                        "This assistant has not enabled image input",
                        code="unsupported_content_type",
                    )
                data = _decode_data_url(part.get("image_url"))
                stored = await asyncio.to_thread(store_input_image, data)
                wire_parts.append(
                    {
                        "type": "input_image",
                        "source": {
                            "type": "uploaded_asset",
                            "asset_token": stored.token,
                        },
                    }
                )
            else:
                raise RealtimeEventError(
                    f"Unsupported content part: {part.get('type')}",
                    code="unsupported_content_type",
                    param="item.content",
                )
        if not wire_parts:
            raise RealtimeEventError("User item has no supported content", param="item.content")
        item_id = str(item.get("id") or f"item_{uuid4().hex}")
        public_item = {**item, "id": item_id, "status": "completed"}
        self.session.pending_item = {
            "type": "user-input",
            "schema_version": 1,
            "input_id": item_id,
            "parts": wire_parts,
            "options": {
                "run_immediately": self.session.config.runtimeMode != "realtime",
                "interrupt": True,
            },
        }
        await self._emit_item_ack(public_item)

    async def _emit_item_ack(self, item: dict[str, Any]) -> None:
        await self.emit(
            _server_event(
                "conversation.item.added",
                previous_item_id=None,
                item=item,
            )
        )
        await self.emit(_server_event("conversation.item.done", item=item))

    async def _append_audio(self, event: dict[str, Any]) -> None:
        audio = _decode_base64(event.get("audio"), param="audio")
        if len(audio) % 2:
            raise RealtimeEventError(
                "WebSocket audio must contain complete PCM16 samples",
                code="invalid_audio_format",
                param="audio",
            )
        if self.channel != "websocket":
            raise RealtimeEventError(
                "input_audio_buffer.append is only used by WebSocket audio",
                code="invalid_event",
            )
        if self.session.external_turn_control:
            self.buffer_audio(audio, sample_rate=24000, num_channels=1)
            return
        await self.runtime.queue_frame(
            InputAudioRawFrame(audio=audio, sample_rate=24000, num_channels=1)
        )

    async def _commit_audio(self) -> None:
        if not self.session.audio_chunks and self.session.external_turn_control:
            raise RealtimeEventError(
                "input_audio_buffer is empty",
                code="input_audio_buffer_commit_empty",
            )
        self.session.audio_committed = True
        item_id = f"item_{uuid4().hex}"
        await self.emit(
            _server_event("input_audio_buffer.committed", item_id=item_id)
        )

    async def _create_response(self, event: dict[str, Any]) -> None:
        response = event.get("response")
        if isinstance(response, dict):
            forbidden = {"instructions", "voice", "tools", "tool_choice", "model"}
            changed = forbidden.intersection(response)
            if changed:
                raise RealtimeEventError(
                    "Per-response assistant configuration is locked",
                    code="immutable_session_field",
                    param=f"response.{sorted(changed)[0]}",
                )
        if (
            self.session.external_turn_control
            and self.session.audio_chunks
            and not self.session.audio_committed
        ):
            raise RealtimeEventError(
                "Commit input_audio_buffer before response.create",
                code="input_audio_buffer_not_committed",
            )
        await self._inference.allow_one_response()
        submitted_item = bool(self.session.pending_item)
        if self.session.pending_item:
            pending = self.session.pending_item
            self.session.pending_item = None
            await self.queue_internal_message(pending)
        if self.session.external_turn_control and self.session.audio_chunks:
            await self.runtime.queue_frame(UserStartedSpeakingFrame())
            for audio, sample_rate, num_channels in self.session.audio_chunks:
                await self.runtime.queue_frame(
                    InputAudioRawFrame(
                        audio=audio,
                        sample_rate=sample_rate,
                        num_channels=num_channels,
                        transport_source="openai-committed",
                    )
                )
            await self.runtime.queue_frame(UserStoppedSpeakingFrame())
            await self.runtime.commit_audio()
            self.clear_buffered_audio()
        if not submitted_item or self.session.config.runtimeMode == "realtime":
            await self.runtime.request_response()

    async def _finish_response(self, *, status: str) -> None:
        response_id, _item_id, text = self.session.finish_response()
        if not response_id:
            return
        await self.emit(
            _server_event(
                "response.done",
                response={
                    "id": response_id,
                    "object": "realtime.response",
                    "status": status,
                    "output": [],
                    "output_text": text,
                },
            )
        )

    async def _update_capabilities(self, event: dict[str, Any]) -> None:
        requested = event.get("capabilities") or []
        if not isinstance(requested, list) or not all(isinstance(x, str) for x in requested):
            raise RealtimeEventError("capabilities must be a string array", param="capabilities")
        enabled = set(requested).intersection(SUPPORTED_CAPABILITIES)
        if "video_track" in enabled and not self.session.vision_enabled:
            enabled.remove("video_track")
        self.session.capabilities = enabled
        await self.emit(
            _server_event(
                "x.interactive_media.capabilities.updated",
                capabilities=sorted(enabled),
                rejected=sorted(set(requested) - enabled),
            )
        )

    async def _update_variables(self, event: dict[str, Any]) -> None:
        if "dynamic_variables" not in self.session.capabilities:
            raise RealtimeEventError(
                "dynamic_variables capability was not negotiated",
                code="extension_not_negotiated",
            )
        variables = event.get("variables")
        if not isinstance(variables, dict) or not variables:
            raise RealtimeEventError("variables must be a non-empty object", param="variables")
        await self.queue_internal_message(
            {
                "type": "session-update",
                "schema_version": 1,
                "update_id": str(event.get("event_id") or f"update_{uuid4().hex}"),
                "dynamic_variables": variables,
            }
        )

    def translate_server_message(self, message: dict[str, Any]) -> list[dict[str, Any]]:
        message_type = str(message.get("type") or "")
        if message_type in SERVER_EVENT_TYPES:
            event = dict(message)
            event.setdefault("event_id", _event_id())
            if message_type.startswith("x.interactive_media."):
                event["version"] = "1"
            return [event]
        if "." in message_type:
            # Dotted wire events outside the documented subset must not leak
            # through from a provider-specific adapter.
            return []
        if message_type == "transcript":
            return self._translate_transcript(message)
        if message_type == "assistant-text-start":
            return self._start_assistant_output()
        if message_type == "assistant-text-delta":
            return self._assistant_delta(str(message.get("delta") or ""))
        if message_type == "assistant-text-end":
            return self._end_assistant_output(bool(message.get("interrupted")))
        if message_type == "client-tool-call":
            return self._client_tool_call(message)
        if message_type == "user-input-result" and message.get("status") == "error":
            return [
                error_event(
                    RealtimeEventError(
                        str(message.get("message") or "User input failed"),
                        code="input_error",
                    )
                )
            ]
        extension = self._extension_event(message)
        return [extension] if extension else []

    def _translate_transcript(self, message: dict[str, Any]) -> list[dict[str, Any]]:
        role = message.get("role")
        text = str(message.get("content") or "")
        item_id = f"item_{uuid4().hex}"
        if role == "user":
            return [
                _server_event(
                    "conversation.item.input_audio_transcription.completed",
                    item_id=item_id,
                    content_index=0,
                    transcript=text,
                )
            ]
        response_id, output_item_id = self.session.begin_response()
        item = {
            "id": output_item_id,
            "type": "message",
            "role": "assistant",
            "status": "completed",
            "content": [{"type": "output_text", "text": text}],
        }
        self.session.response_text += text
        response = {
            "id": response_id,
            "object": "realtime.response",
            "status": "completed",
            "output": [item],
        }
        self.session.finish_response()
        return [
            _server_event("response.created", response={**response, "status": "in_progress", "output": []}),
            _server_event("response.output_item.added", response_id=response_id, output_index=0, item=item),
            _server_event("response.output_item.done", response_id=response_id, output_index=0, item=item),
            _server_event("response.done", response=response),
        ]

    def _start_assistant_output(self) -> list[dict[str, Any]]:
        if self.session.active_response_id:
            return []
        response_id, item_id = self.session.begin_response()
        content_type = "audio" if self.session.output_is_audio else "text"
        item = {
            "id": item_id,
            "type": "message",
            "role": "assistant",
            "status": "in_progress",
            "content": [{"type": content_type}],
        }
        return [
            _server_event(
                "response.created",
                response={
                    "id": response_id,
                    "object": "realtime.response",
                    "status": "in_progress",
                    "output": [],
                },
            ),
            _server_event(
                "response.output_item.added",
                response_id=response_id,
                output_index=0,
                item=item,
            ),
            _server_event(
                "response.content_part.added",
                response_id=response_id,
                item_id=item_id,
                output_index=0,
                content_index=0,
                part={"type": content_type},
            ),
        ]

    def _assistant_delta(self, delta: str) -> list[dict[str, Any]]:
        events = self._start_assistant_output()
        response_id, item_id = self.session.begin_response()
        self.session.response_text += delta
        event_type = (
            "response.output_audio_transcript.delta"
            if self.session.output_is_audio
            else "response.output_text.delta"
        )
        events.append(
            _server_event(
                event_type,
                response_id=response_id,
                item_id=item_id,
                output_index=0,
                content_index=0,
                delta=delta,
            )
        )
        return events

    def _end_assistant_output(self, interrupted: bool) -> list[dict[str, Any]]:
        response_id, item_id, text = self.session.finish_response()
        if not response_id or not item_id:
            return []
        content_type = "audio" if self.session.output_is_audio else "text"
        done_type = (
            "response.output_audio_transcript.done"
            if self.session.output_is_audio
            else "response.output_text.done"
        )
        item = {
            "id": item_id,
            "type": "message",
            "role": "assistant",
            "status": "incomplete" if interrupted else "completed",
            "content": [{"type": content_type, "transcript" if self.session.output_is_audio else "text": text}],
        }
        status = "cancelled" if interrupted else "completed"
        events = [
            _server_event(done_type, response_id=response_id, item_id=item_id, output_index=0, content_index=0, transcript=text, text=text),
            _server_event("response.content_part.done", response_id=response_id, item_id=item_id, output_index=0, content_index=0, part=item["content"][0]),
            _server_event("response.output_item.done", response_id=response_id, output_index=0, item=item),
            _server_event(
                "response.done",
                response={
                    "id": response_id,
                    "object": "realtime.response",
                    "status": status,
                    "output": [item],
                },
            ),
        ]
        if self.session.output_is_audio:
            events.insert(
                1,
                _server_event(
                    "response.output_audio.done",
                    response_id=response_id,
                    item_id=item_id,
                    output_index=0,
                    content_index=0,
                ),
            )
        return events

    def _client_tool_call(self, message: dict[str, Any]) -> list[dict[str, Any]]:
        events: list[dict[str, Any]] = []
        if not self.session.active_response_id:
            response_id, _unused_item_id = self.session.begin_response()
            events.append(
                _server_event(
                    "response.created",
                    response={
                        "id": response_id,
                        "object": "realtime.response",
                        "status": "in_progress",
                        "output": [],
                    },
                )
            )
        else:
            response_id = self.session.active_response_id
        item_id = f"item_{uuid4().hex}"
        call_id = str(message.get("tool_call_id") or "")
        arguments = json.dumps(message.get("arguments") or {}, ensure_ascii=False)
        item = {
            "id": item_id,
            "type": "function_call",
            "status": "completed",
            "name": str(message.get("function_name") or ""),
            "call_id": call_id,
            "arguments": arguments,
        }
        events.extend([
            _server_event("response.output_item.added", response_id=response_id, output_index=0, item=item),
            _server_event("response.function_call_arguments.delta", response_id=response_id, item_id=item_id, output_index=0, call_id=call_id, delta=arguments),
            _server_event("response.function_call_arguments.done", response_id=response_id, item_id=item_id, output_index=0, call_id=call_id, name=item["name"], arguments=arguments),
            _server_event("response.output_item.done", response_id=response_id, output_index=0, item=item),
        ])
        self.session.active_output_item_id = None
        return events

    def _extension_event(self, message: dict[str, Any]) -> dict[str, Any] | None:
        kind = str(message.get("type") or "")
        if kind == "session-update-result" and "dynamic_variables" in self.session.capabilities:
            return _server_event(
                "x.interactive_media.session.variables.updated",
                update_id=message.get("update_id"),
                status=message.get("status"),
                message=message.get("message"),
            )
        if kind in {"node-active", "workflow-event", "workflow-variables", "workflow-error"}:
            if "workflow_events" not in self.session.capabilities:
                return None
            names = {
                "node-active": "x.interactive_media.workflow.node_active",
                "workflow-event": "x.interactive_media.workflow.event",
                "workflow-variables": "x.interactive_media.workflow.variables.updated",
                "workflow-error": "x.interactive_media.workflow.error",
            }
            return _server_event(
                names[kind],
                **{k: v for k, v in message.items() if k != "type"},
            )
        if kind in {"handoff-requested", "call-ended"}:
            if "handoff" not in self.session.capabilities:
                return None
            event_type = (
                "x.interactive_media.call.handoff_requested"
                if kind == "handoff-requested"
                else "x.interactive_media.call.ended"
            )
            return _server_event(
                event_type,
                **{k: v for k, v in message.items() if k != "type"},
            )
        return None