Module livekit.plugins.thunderphone.realtime

class RealtimeModel(_openai.RealtimeModel):
    """Run a ThunderPhone voice agent as the realtime model of an AgentSession.

    Two ways to use it:

    * **Saved agent** — pass ``agent_id``. The agent's prompt, voice, engine,
      languages, tools, greeting and silence handling come from ThunderPhone;
      the LiveKit Agent's instructions and tools are not sent.
    * **Inline** — omit ``agent_id``. Instructions and tools come from the
      LiveKit Agent, ``model`` picks the engine (``spark``, ``bolt`` or
      ``storm``) and ``voice`` the ThunderPhone voice. Function tools run in
      the LiveKit agent.

    An inline call freezes its instructions and tools once it starts, so a
    handoff to an Agent configured differently starts a new call; with a saved
    agent, handoffs keep the call. Every call ends with ThunderPhone closing the
    socket; the session does not reconnect, because a reconnect would start a
    new, separately billed call.
    """

    def __init__(
        self,
        *,
        api_key: str | None = None,
        agent_id: int | str | None = None,
        model: str | None = None,
        voice: NotGivenOr[str] = NOT_GIVEN,
        language: str | None = None,
        caller_number: str | None = None,
        callee_number: str | None = None,
        live_transcripts: bool = False,
        base_url: str = DEFAULT_BASE_URL,
        http_session: Any = None,
        conn_options: APIConnectOptions | None = None,
        **kwargs: Any,
    ) -> None:
        """Create a ThunderPhone realtime model.

        Args:
            api_key: ThunderPhone secret key (``sk_live_...``). Defaults to the
                ``THUNDERPHONE_API_KEY`` environment variable.
            agent_id: Id of a saved ThunderPhone agent. Mutually exclusive with
                ``model`` and ``voice``.
            model: Engine for inline sessions: ``"spark"``, ``"bolt"`` or ``"storm"``.
            voice: ThunderPhone voice name for inline sessions.
            language: Primary language hint for inline sessions, e.g. ``"es"``.
            caller_number: Phone number the call came from, when the session fronts a
                phone call. Shown as the call's "from" number in ThunderPhone call history.
            callee_number: Phone number that was dialed. Shown as the call's "to" number.
            live_transcripts: Stream caller transcript fragments while the caller is
                still speaking (billed extra). Off by default: one transcript per turn.
            base_url: WebSocket endpoint. Defaults to the ThunderPhone API.
            http_session: Optional aiohttp session to reuse.
            conn_options: Connection options. Retries default to 0: a ThunderPhone
                session is a call, and reconnecting would start a new one.
            **kwargs: Passed to ``livekit.plugins.openai.realtime.RealtimeModel``
                (``turn_detection``, ``modalities``, ...). ThunderPhone transcribes the
                caller with the call's own engine, so ``input_audio_transcription``
                is not sent; ``None`` still turns off the caller transcripts.
        """
        key = (api_key or os.environ.get(API_KEY_ENV, "")).strip()
        if not key.startswith(API_KEY_PREFIX):
            raise ValueError(
                f"ThunderPhone needs a secret API key starting with '{API_KEY_PREFIX}' "
                f"(pass api_key= or set {API_KEY_ENV})."
            )
        if agent_id is not None and (model or is_given(voice)):
            raise ValueError(
                "agent_id sessions take their engine and voice from the saved agent; "
                "drop model/voice or drop agent_id."
            )

        if agent_id is not None and not str(agent_id).strip():
            # The URL would drop it and the server would start an inline call,
            # while the plugin strips the agent's instructions and tools.
            raise ValueError("agent_id must not be empty")

        self._tp_agent_id = str(agent_id).strip() if agent_id is not None else None
        self._tp_engine = model
        self._tp_live_transcripts = bool(live_transcripts)

        url = _build_url(
            base_url,
            agent_id=self._tp_agent_id,
            product=model,
            language=language,
            from_number=caller_number,
            to_number=callee_number,
        )
        # The framework recycles an OpenAI socket after max_session_duration
        # (20 min by default) by closing it and reconnecting. On ThunderPhone
        # that would hang up and start a second, separately billed call.
        kwargs["max_session_duration"] = None
        super().__init__(
            model=SERVER_MODEL,
            voice=voice if is_given(voice) else "default",
            base_url=url,
            api_key=key,
            http_session=http_session,
            conn_options=conn_options or APIConnectOptions(max_retry=0),
            **kwargs,
        )
        self._provider_label = "ThunderPhone"
        # An inline call freezes its instructions and tools once it starts, so a
        # handoff to an agent configured differently gets a new session (and a
        # new call). A saved agent owns its instructions and tools and the
        # plugin never sends them, so updates are no-ops and every handoff keeps
        # the call. Turn taking is ThunderPhone's and always on; the framework
        # must not run its own VAD or turn detector against it.
        agent_mode = self._tp_agent_id is not None
        self._capabilities = dataclasses.replace(
            self._capabilities,
            mutable_instructions=agent_mode,
            mutable_tools=agent_mode,
            turn_detection=True,
            can_disable_turn_detection=False,
        )

    @property
    def model(self) -> str:
        # The engine when one was chosen; a saved agent's engine lives on ThunderPhone.
        return self._tp_engine or SERVER_MODEL

    @property
    def provider(self) -> str:
        return "ThunderPhone"

    def session(self, *, turn_detection_disabled: bool = False) -> RealtimeSession:
        if turn_detection_disabled:
            logger.warning(
                "ThunderPhone turn detection is server-side and always on; "
                "turn_detection_disabled is ignored"
            )
        sess = RealtimeSession(self)
        self._sessions.add(sess)
        return sess

class RealtimeSession(_openai.RealtimeSession):
    """One ThunderPhone call.

    Emits, in addition to the framework events:

    * ``thunderphone_call_event`` with every ``call.*`` event (transfer, keypad,
      speech ignored, ...);
    * ``thunderphone_call_ended`` with the ``call.ended`` event, after which the
      session closes itself.

    ``call_id`` holds the ThunderPhone call id once the session is live, for
    fetching the recording, transcript and grade afterwards.
    """

    def __init__(self, realtime_model: RealtimeModel) -> None:
        self._tp_model = realtime_model
        self._tp_agent_mode = realtime_model._tp_agent_id is not None
        # Inline sessions: hold session.update until the agent has supplied its
        # instructions and tools, then send one update that starts the call.
        self._tp_holding = not self._tp_agent_mode
        self._tp_pending_session: dict[str, Any] = {}
        self._tp_call_ended = False
        self._tp_warned_reply_instructions = False
        # Function call ids longer than OpenAI's 32-character cap come back
        # from the framework hashed; map the hash back to the id the server
        # expects on the function_call_output.
        self._tp_call_ids: dict[str, str] = {}
        self.call_id: int | None = None
        super().__init__(realtime_model)
        self.on("openai_server_event_received", self._tp_on_server_event)

    # ----------------------------------------------------------- outgoing

    def send_event(self, event: Any) -> None:
        payload = (
            event.model_dump(by_alias=True, exclude_unset=True, exclude_defaults=False)
            if isinstance(event, BaseModel)
            else dict(event)
        )
        if payload.get("type") == "session.update":
            payload = self._tp_shape_session_update(payload)
            if self._tp_holding:
                _merge(self._tp_pending_session, payload.get("session") or {})
                return
        elif self._tp_holding and payload.get("type") != "input_audio_buffer.append":
            # Anything but audio is meaningful only once the call exists;
            # flush the configuration first so ordering is preserved.
            self._tp_flush()
        if payload.get("type") == "response.create":
            response = payload.get("response")
            if isinstance(response, dict):
                for key in _RESPONSE_OVERRIDE_KEYS:
                    response.pop(key, None)
        if payload.get("type") == "conversation.item.create":
            item = payload.get("item")
            if isinstance(item, dict) and item.get("type") == "function_call_output":
                call_id = item.get("call_id")
                if isinstance(call_id, str):
                    item["call_id"] = self._tp_call_ids.get(call_id, call_id)
        super().send_event(payload)

    def _tp_shape_session_update(self, payload: dict[str, Any]) -> dict[str, Any]:
        session = dict(payload.get("session") or {})
        session["model"] = SERVER_MODEL
        for key in _OPENAI_ONLY_KEYS:
            session.pop(key, None)
        audio = session.get("audio")
        leg = audio.get("input") if isinstance(audio, dict) else None
        if isinstance(leg, dict):
            # ThunderPhone transcribes the caller with the call's own engine.
            leg.pop("transcription", None)
        if self._tp_agent_mode:
            # A saved agent is already running with its own prompt, tools,
            # voice, transcription and turn-taking; the server rejects any
            # change to those. Keep only what a client may still set.
            for key in _AGENT_OWNED_KEYS:
                session.pop(key, None)
            session.pop("output_modalities", None)
            audio = session.get("audio")
            if isinstance(audio, dict):
                for direction in ("input", "output"):
                    leg = audio.get(direction)
                    if isinstance(leg, dict):
                        audio[direction] = {k: v for k, v in leg.items() if k == "format"}
        else:
            config = dict(session.get("config") or {})
            # Opt into ThunderPhone's call.* events (hang-up reason, transfer, keypad).
            config.setdefault("call_events", True)
            session["config"] = config
        if self._tp_model._tp_live_transcripts:
            session["live_transcripts"] = True
        payload["session"] = session
        return payload

    def _tp_flush(self) -> None:
        if not self._tp_holding:
            return
        self._tp_holding = False
        if self._tp_pending_session:
            super().send_event(
                {
                    "event_id": utils.shortuuid("session_update_"),
                    "type": "session.update",
                    "session": self._tp_pending_session,
                }
            )
            self._tp_pending_session = {}

    def generate_reply(
        self,
        *,
        instructions: NotGivenOr[str] = NOT_GIVEN,
        tool_choice: NotGivenOr[llm.ToolChoice] = NOT_GIVEN,
        tools: NotGivenOr[list[llm.Tool]] = NOT_GIVEN,
    ) -> asyncio.Future[llm.GenerationCreatedEvent]:
        if is_given(instructions) and not self._tp_warned_reply_instructions:
            self._tp_warned_reply_instructions = True
            logger.warning(
                "ThunderPhone ignores per-reply instructions; "
                "replies use the session's instructions"
            )
        return super().generate_reply(
            instructions=instructions, tool_choice=tool_choice, tools=tools
        )

    async def _update_session(
        self,
        *,
        instructions: NotGivenOr[str] = NOT_GIVEN,
        chat_ctx: NotGivenOr[llm.ChatContext] = NOT_GIVEN,
        tools: NotGivenOr[list[llm.Tool]] = NOT_GIVEN,
    ) -> None:
        # The framework sends instructions, then chat history, then tools. On
        # ThunderPhone the first session.update starts the call and freezes
        # instructions and tools, so send both before any conversation item.
        if is_given(instructions):
            try:
                await self.update_instructions(instructions)
            except llm.RealtimeError:
                logger.exception("failed to update the instructions")
        if is_given(tools):
            try:
                await self.update_tools(tools)
            except llm.RealtimeError:
                logger.exception("failed to update the tools")
        self._tp_flush()
        if is_given(chat_ctx):
            try:
                await self.update_chat_ctx(chat_ctx)
            except llm.RealtimeError:
                logger.exception("failed to update the chat_ctx")

    # ----------------------------------------------------------- incoming

    def _emit_transcription_metrics(self, event: Any) -> None:
        # The base names the model after its OpenAI transcription config, which
        # ThunderPhone never uses: the call's engine transcribes the caller and
        # reports usage as seconds of caller audio (never tokens).
        try:
            usage = UsageTranscriptTextUsageDuration.model_validate(event.usage)
        except ValidationError:
            return
        model = self._tp_model
        self.emit(
            "metrics_collected",
            STTMetrics(
                request_id=event.event_id,
                timestamp=time.time(),
                duration=0.0,
                label=model.label,
                audio_duration=usage.seconds,
                streamed=True,
                metadata=Metadata(model_name=model.model, model_provider=model.provider),
            ),
        )

    def _tp_close_expected(self) -> bool:
        # After call.ended the server hangs up; after aclose() we do.
        return self._tp_call_ended or self._closing

    async def _create_ws_conn(self) -> aiohttp.ClientWebSocketResponse:
        ws = await super()._create_ws_conn()
        return cast(aiohttp.ClientWebSocketResponse, _CallSocket(ws, self))

    async def _run_ws(self, ws_conn: aiohttp.ClientWebSocketResponse) -> None:
        try:
            await super()._run_ws(ws_conn)
        except APIConnectionError:
            # The socket closed after the call ended or while closing, and our
            # send loop did not get there first (see _CallSocket). With the
            # outgoing channel closed the framework ends the session here.
            if not self._tp_close_expected():
                raise

    def _tp_on_server_event(self, event: dict[str, Any]) -> None:
        event_type = event.get("type")
        if event_type == "response.created":
            self._tp_claim_response(event)
            return
        if event_type == "response.output_item.done":
            item = event.get("item")
            if isinstance(item, dict) and item.get("type") == "function_call":
                call_id = item.get("call_id")
                if isinstance(call_id, str):
                    self._tp_call_ids[_shorten_call_id(call_id)] = call_id
            return
        if event_type in ("session.created", "session.updated"):
            session = event.get("session") or {}
            call_id = session.get("call_id") if isinstance(session, dict) else None
            if isinstance(call_id, int):
                self.call_id = call_id
            elif isinstance(call_id, str) and call_id.isdigit():
                self.call_id = int(call_id)
            return
        if not isinstance(event_type, str) or not event_type.startswith("call."):
            return
        self.emit(CALL_EVENT, event)
        if event_type == "call.ended" and not self._tp_call_ended:
            self._tp_call_ended = True
            logger.info("ThunderPhone call ended", extra={"lk.pii.event": event})
            self.emit(CALL_ENDED_EVENT, event)
            # The server closes the socket right after this event. Closing the
            # outgoing channel ends the send loop, which closes our side and
            # ends the session; _CallSocket holds the server's close back until
            # then so it is not read as a connection failure. Anything still
            # queued was meant for a call that no longer exists.
            self._msg_ch.close()
            while not self._msg_ch.empty():
                self._msg_ch.recv_nowait()

    def _tp_claim_response(self, event: dict[str, Any]) -> None:
        # generate_reply() resolves its future by the client_event_id the
        # server echoes in response.metadata. ThunderPhone servers that
        # predate that echo start the response but never label it, so the
        # future times out while the audio plays anyway. Credit an unlabeled
        # response to the oldest pending reply request instead; the hook runs
        # on the raw dict before the framework parses it.
        response = event.get("response")
        if not isinstance(response, dict):
            return
        metadata = response.get("metadata")
        if isinstance(metadata, dict) and metadata.get("client_event_id"):
            return
        pending = next(iter(self._response_created_futures), None)
        if pending is None:
            return
        response["metadata"] = {
            **(metadata if isinstance(metadata, dict) else {}),
            "client_event_id": pending,
        }